diff --git a/src/core/cycle/extract-atoms-drain.ts b/src/core/cycle/extract-atoms-drain.ts index c6f474d2a..e981333a4 100644 --- a/src/core/cycle/extract-atoms-drain.ts +++ b/src/core/cycle/extract-atoms-drain.ts @@ -118,7 +118,14 @@ export async function runExtractAtomsDrain( // Stop if a batch made zero forward progress — extraction is failing or // everything left is ineligible (e.g. all skipped). Prevents a hot loop // that spends budget without draining. - if (r.extracted === 0 && r.skipped === 0) { stopped = 'no_progress'; break; } + // + // #2144: a zero-ATOM batch can still be progress — tombstoned + // zero-yield pages shrink the backlog without producing atoms. Only + // stop when the backlog count genuinely didn't move. + if (r.extracted === 0 && r.skipped === 0) { + const after = await deps.countRemaining(); + if (after === null || before === null || after >= before) { stopped = 'no_progress'; break; } + } } const remaining = await deps.countRemaining(); diff --git a/src/core/cycle/extract-atoms.ts b/src/core/cycle/extract-atoms.ts index 49dc9b749..dd94e2107 100644 --- a/src/core/cycle/extract-atoms.ts +++ b/src/core/cycle/extract-atoms.ts @@ -254,6 +254,7 @@ export async function discoverExtractablePages( AND COALESCE(p.frontmatter->>'dream_generated', '') <> 'true' ${RAW_SOURCE_HOLDER_EXCLUSION_SQL} AND length(COALESCE(p.compiled_truth, '')) >= $3 + AND COALESCE(p.frontmatter->>'atoms_scan_hash', '') <> substring(p.content_hash from 1 for 16) ${hasFilter ? "AND p.slug = ANY($5::text[])" : ''} AND NOT EXISTS ( SELECT 1 @@ -327,6 +328,7 @@ export async function countExtractAtomsBacklog( AND COALESCE(p.frontmatter->>'dream_generated', '') <> 'true' ${RAW_SOURCE_HOLDER_EXCLUSION_SQL} AND length(COALESCE(p.compiled_truth, '')) >= $3 + AND COALESCE(p.frontmatter->>'atoms_scan_hash', '') <> substring(p.content_hash from 1 for 16) AND NOT EXISTS ( SELECT 1 FROM pages atom WHERE atom.type = 'atom' AND atom.source_id = $1 @@ -341,6 +343,7 @@ export async function countExtractAtomsBacklog( AND COALESCE(p.frontmatter->>'dream_generated', '') <> 'true' ${RAW_SOURCE_HOLDER_EXCLUSION_SQL} AND length(COALESCE(p.compiled_truth, '')) >= $2 + AND COALESCE(p.frontmatter->>'atoms_scan_hash', '') <> substring(p.content_hash from 1 for 16) AND NOT EXISTS ( SELECT 1 FROM pages atom WHERE atom.type = 'atom' AND atom.source_id = p.source_id @@ -586,6 +589,25 @@ export async function runPhaseExtractAtoms( const atoms = parseAtomsResponse(result.text); if (atoms.length === 0) { + // #2144: tombstone zero-yield pages so they stop being rediscovered. + // Idempotency is keyed on atom rows — a page that yields no atoms + // leaves no row, so pre-fix it re-entered the discovery window every + // run (wedging --drain with a false no_progress and re-spending + // nightly budget on the same pages). Stamp the content hash we + // scanned; discovery skips the page only while its content is + // unchanged (edits re-eligibilize, mirroring atom-row staleness). + // Only stamped after a SUCCESSFUL chat call — LLM failures take the + // catch path below and stay retryable. + if (!opts.dryRun && item.kind === 'page') { + try { + await engine.executeRaw( + `UPDATE pages + SET frontmatter = frontmatter || jsonb_build_object('atoms_scan_hash', $1::text) + WHERE source_id = $2 AND slug = $3 AND deleted_at IS NULL`, + [item.contentHash.slice(0, 16), sourceId, item.slug], + ); + } catch { /* fail-soft: page stays rediscoverable */ } + } if (item.kind === 'transcript') transcriptsProcessed++; else pagesProcessed++; continue; diff --git a/test/extract-atoms-drain.test.ts b/test/extract-atoms-drain.test.ts index fa8aaa19d..accc15459 100644 --- a/test/extract-atoms-drain.test.ts +++ b/test/extract-atoms-drain.test.ts @@ -228,3 +228,42 @@ describe('extract-atoms-drain Minion handler retries on provider_failure (issue ); }); }); + +describe('#2144: zero-yield tombstone progress semantics', () => { + it('continues when a zero-atom batch still shrinks the backlog (tombstoned pages)', async () => { + let batches = 0; + const result = await runExtractAtomsDrain( + { + withLock: passThroughLock, + // consumed: before#1=4, after#1=2 (<4 → progress), before#2=2, + // after#2=0 (<2 → progress), before#3=0 → drained; final repeats 0. + countRemaining: seq([4, 2, 2, 0, 0]), + runBatch: async () => { batches++; return { extracted: 0, skipped: 0 }; }, + now: () => 0, + }, + { windowMs: 1_000_000 }, + ); + expect(result.stopped).toBe('drained'); + expect(result.batches).toBe(2); + expect(result.extracted).toBe(0); + expect(result.remaining).toBe(0); + expect(batches).toBe(2); + }); + + it('stops no_progress when a zero-atom batch leaves the backlog flat', async () => { + let batches = 0; + const result = await runExtractAtomsDrain( + { + withLock: passThroughLock, + countRemaining: seq([5, 5]), + runBatch: async () => { batches++; return { extracted: 0, skipped: 0 }; }, + now: () => 0, + }, + { windowMs: 1_000_000 }, + ); + expect(result.stopped).toBe('no_progress'); + expect(result.batches).toBe(1); + expect(result.remaining).toBe(5); + expect(batches).toBe(1); + }); +}); diff --git a/test/extract-atoms-page-discovery.test.ts b/test/extract-atoms-page-discovery.test.ts index d1b4fcefb..c36494c9e 100644 --- a/test/extract-atoms-page-discovery.test.ts +++ b/test/extract-atoms-page-discovery.test.ts @@ -444,3 +444,50 @@ describe('v0.41.2.1: runPhaseExtractAtoms — dual-source merge + idempotency', expect(discovered.details?.atoms_extracted).toBe(1); }); }); + +describe('#2144: zero-yield tombstone', () => { + test('zero-yield page is stamped and excluded from rediscovery', async () => { + await seedPage({ slug: 'article/zero-yield', type: 'article' }); + // Successful LLM call that yields no atoms. + const result = await runPhaseExtractAtoms(engine, { _transcripts: [], _chat: stubChat('[]') }); + expect(result.details?.pages_processed).toBe(1); + expect(result.details?.atoms_extracted).toBe(0); + + // Stamp landed: atoms_scan_hash = first 16 chars of the page's content_hash. + const rows = await engine.executeRaw<{ scan: string; ch: string }>( + `SELECT frontmatter->>'atoms_scan_hash' AS scan, content_hash AS ch + FROM pages WHERE slug = 'article/zero-yield'`, + ); + expect(rows[0].scan).toBe(rows[0].ch.slice(0, 16)); + + // No longer rediscovered. + const discovered = await discoverExtractablePages(engine, 'default'); + expect(discovered.find((d) => d.slug === 'article/zero-yield')).toBeUndefined(); + }); + + test('content change re-eligibilizes a tombstoned page', async () => { + await seedPage({ slug: 'article/evolves', type: 'article' }); + await runPhaseExtractAtoms(engine, { _transcripts: [], _chat: stubChat('[]') }); + expect((await discoverExtractablePages(engine, 'default')).length).toBe(0); + + // Simulate an edit: content_hash moves while the stale stamp stays. + await engine.executeRaw( + `UPDATE pages SET content_hash = 'fresh-hash-after-edit' WHERE slug = $1 AND source_id = 'default'`, + ['article/evolves'], + ); + const rediscovered = await discoverExtractablePages(engine, 'default'); + expect(rediscovered.map((d) => d.slug)).toContain('article/evolves'); + }); + + test('failed chat does NOT stamp — page stays retryable', async () => { + await seedPage({ slug: 'article/transient-failure', type: 'article' }); + const failingChat = async (_o: ChatOpts): Promise => { throw new Error('rate limit'); }; + await runPhaseExtractAtoms(engine, { _transcripts: [], _chat: failingChat as never }); + const rows = await engine.executeRaw<{ scan: string | null }>( + `SELECT frontmatter->>'atoms_scan_hash' AS scan FROM pages WHERE slug = 'article/transient-failure'`, + ); + expect(rows[0].scan).toBeNull(); + const discovered = await discoverExtractablePages(engine, 'default'); + expect(discovered.map((d) => d.slug)).toContain('article/transient-failure'); + }); +});