diff --git a/layers/registry/server/utils/repository-purpose-effect.ts b/layers/registry/server/utils/repository-purpose-effect.ts index b3b3e992..9ae0c344 100644 --- a/layers/registry/server/utils/repository-purpose-effect.ts +++ b/layers/registry/server/utils/repository-purpose-effect.ts @@ -9,9 +9,12 @@ import { REPOSITORY_PURPOSE_PROMPT_VERSION, REPOSITORY_PURPOSE_REFRESH_SECONDS, } from './repository-purpose' +import { markRepoMissing } from './sync-repo' export type RepositoryPurposeJudge = (state: RepositoryPurposeEvidence) => Promise export interface RepositoryIdentity { owner: string, repo: string } +export interface RepositoryPurposeSourceMissing { _tag: 'source_missing', status: 404 | 410 } +export type RepositoryPurposeRefresh = RepositoryPurposeFinding | RepositoryPurposeSourceMissing export async function listRepositoryPurposeCandidates(db: D1Database, now: number): Promise { // Completed jobs are retained in `jobs` forever, so a correlated read over @@ -45,8 +48,10 @@ export async function listRepositoryPurposeCandidates(db: D1Database, now: numbe export async function readRepositoryPurposeEvidence( input: RepositoryIdentity, bindings: GithubBindings, -): Promise { +): Promise { const summary = await getRepoSummary(input.owner, input.repo, bindings) + if (summary.status === 404 || summary.status === 410) + return { _tag: 'source_missing', status: summary.status } if (!summary.data?.headCommitSha || !summary.data.headTreeSha || summary.data.meta.private) throw new Error(`Repository purpose source unavailable: ${summary.status}`) const { meta, headCommitSha, headTreeSha } = summary.data @@ -95,10 +100,14 @@ export async function persistRepositoryPurpose(db: D1Database, finding: Reposito export async function refreshRepositoryPurpose(deps: { db: D1Database - readEvidence: (input: RepositoryIdentity) => Promise + readEvidence: (input: RepositoryIdentity) => Promise judge: RepositoryPurposeJudge -}, input: RepositoryIdentity, now: number): Promise { +}, input: RepositoryIdentity, now: number): Promise { const evidence = await deps.readEvidence(input) + if ('_tag' in evidence) { + await markRepoMissing(deps.db, input.owner, input.repo, now) + return evidence + } const cached = await deps.db.prepare(` SELECT purpose,probability,reason,model,source_commit,prompt_version,evidence,answer FROM repository_purpose WHERE owner=? AND repo=? AND source_commit=? @@ -117,7 +126,7 @@ export async function refreshRepositoryPurpose(deps: { export async function checkRepositoryPurposeAdmission( db: D1Database, input: RepositoryIdentity & { ownerVerified: boolean }, - classify: () => Promise, + classify: () => Promise, ) { const row = await db.prepare(` SELECT EXISTS(SELECT 1 FROM skills WHERE owner=?1 AND repo=?2) AS stored, @@ -129,5 +138,7 @@ export async function checkRepositoryPurposeAdmission( if (row?.stored || row?.eligible || row?.reviewed || input.ownerVerified) return { _tag: 'continue' as const } const finding = await classify() + if (finding._tag === 'source_missing') + return { _tag: 'held' as const, reason: `repo fetch ${finding.status}` } return decideRepositoryPurposeAdmission({ purpose: finding.purpose, hasStoredSkills: false, humanEligible: false, ownerVerified: false }) } diff --git a/layers/registry/server/utils/sync-repo.ts b/layers/registry/server/utils/sync-repo.ts index b22db39e..95eb5d50 100644 --- a/layers/registry/server/utils/sync-repo.ts +++ b/layers/registry/server/utils/sync-repo.ts @@ -654,7 +654,7 @@ async function markRepoTooLarge(db: D1Database, owner: string, repo: string, now * gone while the repo was healthy, and overwriting them here would let the * unchanged-path recovery below resurrect skills the tree itself dropped. */ -async function markRepoMissing(db: D1Database, owner: string, repo: string, now: number): Promise { +export async function markRepoMissing(db: D1Database, owner: string, repo: string, now: number): Promise { await Promise.all([ db .prepare( diff --git a/server/jobs/registry/repository-purpose.ts b/server/jobs/registry/repository-purpose.ts index c2823913..74c1e376 100644 --- a/server/jobs/registry/repository-purpose.ts +++ b/server/jobs/registry/repository-purpose.ts @@ -21,6 +21,10 @@ export default defineJob({ readEvidence: identity => readRepositoryPurposeEvidence(identity, bindings), judge: state => ctx.env.AI.run(REPOSITORY_PURPOSE_MODEL, { state, questions: repositoryPurposeQuestions }), }, input, Math.floor(Date.now() / 1000)) + if (result._tag === 'source_missing') { + await ctx.fail(`repo fetch ${result.status}`) + return + } emitOperationalEvent(createWideEvent({ operation: 'repository-purpose', outcome: result.purpose, repo: `${input.owner}/${input.repo}` })) ctx.reportStats?.({ rowsFetched: 1, rowsInserted: 1 }) }, diff --git a/test/unit/github-repo-identity.test.ts b/test/unit/github-repo-identity.test.ts index 336a73f3..080ecef8 100644 --- a/test/unit/github-repo-identity.test.ts +++ b/test/unit/github-repo-identity.test.ts @@ -25,8 +25,8 @@ describe('github repository identity', () => { })) vi.stubGlobal('fetch', fetch) await expect(readRepositoryPurposeEvidence({ owner: 'acme', repo: 'private' }, {})) - .rejects - .toThrow('Repository purpose source unavailable: 404') + .resolves + .toEqual({ _tag: 'source_missing', status: 404 }) expect(fetch).toHaveBeenCalledTimes(1) }) diff --git a/test/unit/repository-purpose-effect.test.ts b/test/unit/repository-purpose-effect.test.ts index 49eee2ad..8726869d 100644 --- a/test/unit/repository-purpose-effect.test.ts +++ b/test/unit/repository-purpose-effect.test.ts @@ -16,6 +16,38 @@ function judge() { } describe('purpose records', () => { + it.each([404, 410] as const)('quarantines a missing source (%s) and stops scheduling its purpose', async (status) => { + const sqlite = createSqliteD1(allMigrations()) + const missing = { _tag: 'source_missing' as const, status } + const judgeMock = vi.fn(judge) + try { + sqlite.raw.exec(` + INSERT INTO repos(owner,repo,repo_skill_count) VALUES ('org','directory',3); + INSERT INTO skills(owner,repo,name,slug,display_name) VALUES + ('org','directory','review','org/review','Review'), + ('org','directory','gone','org/gone','Gone'); + UPDATE skills SET sync_status='path_missing' WHERE name='gone'; + `) + expect(await refreshRepositoryPurpose({ db: sqlite.db, readEvidence: async () => missing, judge: judgeMock }, evidence, 1000)).toEqual(missing) + expect(judgeMock).not.toHaveBeenCalled() + expect(sqlite.raw.prepare('SELECT broken_since FROM repos').get()).toEqual({ broken_since: 1000 }) + expect(sqlite.raw.prepare('SELECT sync_status,source_resolved,seo_indexable FROM skills WHERE name=\'review\'').get()) + .toEqual({ sync_status: 'repo_missing', source_resolved: 0, seo_indexable: 0 }) + expect(sqlite.raw.prepare('SELECT sync_status FROM skills WHERE name=\'gone\'').get()).toEqual({ sync_status: 'path_missing' }) + expect(await listRepositoryPurposeCandidates(sqlite.db, 200000)).toEqual([]) + } + finally { sqlite.close() } + }) + + it('returns a terminal admission rejection for a missing new source', async () => { + const sqlite = createSqliteD1(allMigrations()) + try { + expect(await checkRepositoryPurposeAdmission(sqlite.db, { ...evidence, ownerVerified: false }, async () => ({ _tag: 'source_missing', status: 404 }))) + .toEqual({ _tag: 'held', reason: 'repo fetch 404' }) + } + finally { sqlite.close() } + }) + it.each(['official', 'trusted-author', 'trusted-curator'])('retains a named %s decision for an organization', async (tier) => { const sqlite = createSqliteD1(allMigrations()) const classify = vi.fn() diff --git a/test/unit/repository-purpose-job.test.ts b/test/unit/repository-purpose-job.test.ts new file mode 100644 index 00000000..53eb86f0 --- /dev/null +++ b/test/unit/repository-purpose-job.test.ts @@ -0,0 +1,43 @@ +import { afterEach, describe, expect, it, vi } from 'vitest' +import { listRepositoryPurposeCandidates } from '../../layers/registry/server/utils/repository-purpose-effect' +import job from '../../server/jobs/registry/repository-purpose' +import { allMigrations, createSqliteD1 } from './helpers/d1-sqlite' + +vi.hoisted(() => vi.stubGlobal('defineJob', (definition: T) => definition)) + +describe('repository purpose job', () => { + afterEach(() => vi.unstubAllGlobals()) + + it('ends a missing source on its first attempt without calling AI', async () => { + const sqlite = createSqliteD1(allMigrations()) + const fail = vi.fn() + const run = vi.fn() + vi.stubGlobal('fetch', vi.fn(async () => Response.json({ data: { repository: null } }))) + try { + sqlite.raw.exec('INSERT INTO repos(owner,repo,repo_skill_count) VALUES (\'org\',\'gone\',3)') + const ctx = { db: sqlite.db, env: { AI: { run } }, fail } as unknown as Parameters[1] + await job.handle({ owner: 'org', repo: 'gone' }, ctx) + expect(fail).toHaveBeenCalledExactlyOnceWith('repo fetch 404') + expect(run).not.toHaveBeenCalled() + expect(await listRepositoryPurposeCandidates(sqlite.db, Math.floor(Date.now() / 1000) + 2 * 86400)).toEqual([]) + } + finally { sqlite.close() } + }) + + it('leaves temporary source failures retryable without quarantining the source', async () => { + const sqlite = createSqliteD1(allMigrations()) + const fail = vi.fn() + const run = vi.fn() + vi.stubGlobal('fetch', vi.fn(async () => new Response('Unavailable', { status: 503 }))) + try { + sqlite.raw.exec('INSERT INTO repos(owner,repo,repo_skill_count) VALUES (\'org\',\'busy\',3)') + const ctx = { db: sqlite.db, env: { AI: { run } }, fail } as unknown as Parameters[1] + await expect(job.handle({ owner: 'org', repo: 'busy' }, ctx)).rejects.toThrow('Repository purpose source unavailable: 503') + expect(fail).not.toHaveBeenCalled() + expect(run).not.toHaveBeenCalled() + expect(sqlite.raw.prepare('SELECT broken_since FROM repos').get()).toEqual({ broken_since: null }) + expect(await listRepositoryPurposeCandidates(sqlite.db, 200000)).toEqual([{ owner: 'org', repo: 'busy' }]) + } + finally { sqlite.close() } + }) +})