fix(registry): stop purpose retries for missing sources

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<unknown>
 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<RepositoryIdentity[]> {
   // 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<RepositoryPurposeEvidence> {
+): Promise<RepositoryPurposeEvidence | RepositoryPurposeSourceMissing> {
   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<RepositoryPurposeEvidence>
+  readEvidence: (input: RepositoryIdentity) => Promise<RepositoryPurposeEvidence | RepositoryPurposeSourceMissing>
   judge: RepositoryPurposeJudge
-}, input: RepositoryIdentity, now: number): Promise<RepositoryPurposeFinding> {
+}, input: RepositoryIdentity, now: number): Promise<RepositoryPurposeRefresh> {
   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<RepositoryPurposeFinding>,
+  classify: () => Promise<RepositoryPurposeRefresh>,
 ) {
   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 })
 }

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<void> {
+export async function markRepoMissing(db: D1Database, owner: string, repo: string, now: number): Promise<void> {
   await Promise.all([
     db
       .prepare(

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 })
   },

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)
   })
 

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()

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', <T>(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<typeof job.handle>[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<typeof job.handle>[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() }
+  })
+})