fix(cycle): tombstone zero-yield pages so extract_atoms stops rediscovering them (#2144) (#2145)

Idempotency was keyed on atom rows alone — a page the LLM judges
un-atomizable leaves no row, so it re-entered the discovery window every
run. Two production consequences: --drain false-stopped with
no_progress once the window head was mostly zero-yield pages (remaining
frozen while batches report +0), and every nightly re-spent extraction
budget on the same pages.

Fix:
- After a SUCCESSFUL chat call that parses to zero atoms, stamp the
  source page with frontmatter.atoms_scan_hash = contentHash16. LLM
  failures take the catch path and stay retryable.
- discoverExtractablePages + countExtractAtomsBacklog (both variants)
  exclude pages whose stamp matches the CURRENT content hash prefix —
  content edits re-eligibilize, mirroring atom-row staleness semantics.
- Drain no_progress now recounts the backlog on a zero-atom batch and
  only stops when it genuinely didn't shrink — tombstoning IS progress.

Tests: +2 pure-loop drain cases (shrinking backlog continues / flat
backlog stops) and +3 PGLite integration cases (stamp + exclusion /
content-change re-eligibility / failed chat does not stamp).
29 pass / 0 fail across the two files; tsc clean.

Co-authored-by: 陈源泉 <chenyuanquan@chenyuanquandeMac-mini.local>
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

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

src/core/cycle/extract-atoms.ts

@@ -239,6 +239,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
@@ -312,6 +313,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
@@ -326,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, '')) >= $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
@@ -571,6 +574,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;

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

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<ChatResult> => { 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');
+  });
+});