fix(facts): make dream extract_facts idempotent so fence rows don't duplicate each cycle (#2932)

Port of #1837 (mvanhorn) onto current master. The extract_facts cycle phase unconditionally wipe-and-reinserted every page's fence-owned DB rows, so re-running a cycle on unchanged content churned rows and — on the Postgres engine reported in #1781 — accumulated duplicates each run. The phase now de-dupes extracted facts by the canonical (claim, source) content key and reconciles the page-scoped DB index: no-op when already in sync, insert-only for new keys, wipe/reinsert only when stale rows need cleanup. Adjustments over the original PR to fit current master: - preserve #1972's abortSignal threading into the batch embed call - preserve #1928's excludeSourcePrefixes: ['cli:'] on every wipe, and exclude cli:-origin conversation facts from the existing-row set so they neither count as stale (which would force a wipe every cycle) nor get compared against the fence Fixes #1781 Co-authored-by: Sinabina <sinabina@Sinabinas-MacBook-Pro-4.local> Co-authored-by: Matt Van Horn <455140+mvanhorn@users.noreply.github.com> Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
2026-07-27 22:15:33 +00:00 · 2026-07-17 14:29:07 -07:00
parent e1cefd0654
commit 54a8070640
2 changed files with 229 additions and 30 deletions
@@ -11,15 +11,17 @@
 *   1. Reads the markdown body (DB-side fetch via engine.getPage).
 *   2. Parses the `## Facts` fence with parseFactsFence.
 *   3. Maps ParsedFact → FenceExtractedFact via extractFactsFromFenceText.
- *   4. Wipes the page's DB index via deleteFactsForPage.
- *   5. Re-inserts via engine.insertFacts batch.
+ *   4. De-dupes rows by canonical (claim, source) content key.
+ *   5. Reconciles the page-scoped DB index: no-op when already in sync,
+ *      insert only missing keys when possible, or wipe/reinsert when stale
+ *      DB rows need cleanup (#1781 — the unconditional wipe-and-reinsert
+ *      made every cycle non-idempotent, re-appending duplicate rows).
 *
- * After the phase, the DB index for every affected page byte-matches
- * the fence (modulo embeddings + runtime-derived fields). Pages with
- * no fence go through delete-then-empty-insert — DB rows for that
- * page coordinate are wiped; legacy NULL-source_markdown_slug rows
- * survive because deleteFactsForPage targets source_markdown_slug =
- * slug only.
+ * After the phase, the DB index for every affected page matches the
+ * fence's canonical (claim, source) row set (modulo embeddings +
+ * runtime-derived fields). Pages with no fence wipe DB rows for that
+ * page coordinate only; legacy NULL-source_markdown_slug rows survive
+ * because deleteFactsForPage targets source_markdown_slug = slug only.
 *
 * Empty-fence guard (Codex R2-#7): the phase refuses to do its
 * destructive reconciliation pass when legacy rows (row_num IS NULL,
@@ -35,7 +37,11 @@ import type { BrainEngine } from '../engine.ts';
 import { writeReceipt } from '../extract/receipt-writer.ts';
 import { upsertExtractRollup } from '../extract/rollup-writer.ts';
 import { parseFactsFence } from '../facts-fence.ts';
-import { extractFactsFromFenceText } from '../facts/extract-from-fence.ts';
+import {
+  extractFactsFromFenceText,
+  FENCE_SOURCE_DEFAULT,
+  type FenceExtractedFact,
+} from '../facts/extract-from-fence.ts';
 import {
  runPhantomRedirectPass,
  emptyPhantomPassResult,
@@ -44,6 +50,51 @@ import {
 import { embed, isAvailable } from '../ai/gateway.ts';
 import { isAborted } from '../abort-check.ts';

+interface ExistingPageFact {
+  fact: string;
+  source: string | null;
+  row_num: number | string | null;
+}
+
+function factContentKey(fact: string, source: string | null | undefined): string {
+  return `${fact}\u0000${source ?? FENCE_SOURCE_DEFAULT}`;
+}
+
+function dedupeFactsByContentKey(facts: FenceExtractedFact[]): FenceExtractedFact[] {
+  const seen = new Set<string>();
+  const deduped: FenceExtractedFact[] = [];
+  for (const fact of facts) {
+    const key = factContentKey(fact.fact, fact.source);
+    if (seen.has(key)) continue;
+    seen.add(key);
+    deduped.push(fact);
+  }
+  return deduped;
+}
+
+/**
+ * Fence-owned DB rows for one page coordinate. Excludes `cli:`-origin
+ * conversation facts (#1928) — they are not fence-owned, so they must
+ * neither count as "stale" (which would force a wipe every cycle) nor
+ * be compared against the fence's row set. Mirrors the
+ * excludeSourcePrefixes filter deleteFactsForPage applies on the wipe.
+ */
+async function listExistingFactsForPage(
+  engine: BrainEngine,
+  slug: string,
+  sourceId: string,
+): Promise<ExistingPageFact[]> {
+  return engine.executeRaw<ExistingPageFact>(
+    `SELECT fact, source, row_num
+       FROM facts
+      WHERE source_id = $1
+        AND source_markdown_slug = $2
+        AND COALESCE(source, '') NOT LIKE 'cli:%'
+      ORDER BY row_num ASC, id ASC`,
+    [sourceId, slug],
+  );
+}
+
 export interface ExtractFactsOpts {
  /** Subset of slugs to reconcile. undefined = walk every page in the brain. */
  slugs?: string[];
@@ -220,28 +271,70 @@ export async function runExtractFacts(

    if (parsed.facts.length > 0) result.pagesWithFacts += 1;

-    if (opts.dryRun) continue;
-
-    // Wipe-and-reinsert per page. The delete targets source_markdown_slug =
-    // slug only, so NULL-source_markdown_slug legacy rows survive (the
-    // partial-UNIQUE-index keyspace). #1928: `cli:`-origin facts (conversation
-    // facts from extract-conversation-facts) are NOT fence-owned — the page
-    // carries no `## Facts` fence to recreate them — so they MUST survive this
-    // reconcile. Exclude them from the wipe.
-    const deleted = await engine.deleteFactsForPage(slug, sourceId, {
-      excludeSourcePrefixes: ['cli:'],
-    });
-    result.factsDeleted += deleted.deleted;
-
-    if (parsed.facts.length === 0) continue;
-
    // v0.35.4 (D-ENG-1) — thread page.effective_date as the fallback
    // valid_from. Without this, fence rows without explicit `validFrom:`
    // land with `valid_from = now()` (import timestamp) and every
    // trajectory query against the page returns import dates instead of
    // claim dates.
    const pageEffectiveDate = page.effective_date ? new Date(page.effective_date) : null;
-    const extracted = extractFactsFromFenceText(parsed.facts, slug, sourceId, { pageEffectiveDate });
+    const extracted = dedupeFactsByContentKey(
+      extractFactsFromFenceText(parsed.facts, slug, sourceId, { pageEffectiveDate }),
+    );
+
+    if (opts.dryRun) continue;
+
+    // #1781 — reconcile instead of unconditional wipe-and-reinsert. Compare
+    // the fence's canonical (claim, source) row set against the page's
+    // fence-owned DB rows: no-op when already in sync, insert only missing
+    // keys when possible, wipe/reinsert only when stale rows need cleanup.
+    const existing = await listExistingFactsForPage(engine, slug, sourceId);
+    const existingKeys = new Set(existing.map(f => factContentKey(f.fact, f.source)));
+    const desiredByKey = new Map(extracted.map(f => [factContentKey(f.fact, f.source), f]));
+
+    if (extracted.length === 0) {
+      if (existing.length > 0) {
+        // The delete targets source_markdown_slug = slug only, so
+        // NULL-source_markdown_slug legacy rows survive (the
+        // partial-UNIQUE-index keyspace). #1928: `cli:`-origin facts
+        // (conversation facts from extract-conversation-facts) are NOT
+        // fence-owned — the page carries no `## Facts` fence to recreate
+        // them — so they MUST survive this reconcile.
+        const deleted = await engine.deleteFactsForPage(slug, sourceId, {
+          excludeSourcePrefixes: ['cli:'],
+        });
+        result.factsDeleted += deleted.deleted;
+      }
+      continue;
+    }
+
+    const hasStaleExisting = existing.some(f => !desiredByKey.has(factContentKey(f.fact, f.source)));
+    const hasDuplicateExisting = existing.length !== existingKeys.size;
+    const hasRowNumDrift = existing.some(f => {
+      const desired = desiredByKey.get(factContentKey(f.fact, f.source));
+      return desired !== undefined && Number(f.row_num) !== desired.row_num;
+    });
+
+    if (
+      existing.length === extracted.length &&
+      !hasStaleExisting &&
+      !hasDuplicateExisting &&
+      !hasRowNumDrift
+    ) {
+      continue;
+    }
+
+    let toInsert = extracted.filter(f => !existingKeys.has(factContentKey(f.fact, f.source)));
+    if (hasStaleExisting || hasDuplicateExisting || hasRowNumDrift) {
+      // Fall back to the legacy page-level reconcile when old DB rows must
+      // be removed. Same delete scoping as above: legacy
+      // NULL-source_markdown_slug rows and `cli:`-origin conversation
+      // facts (#1928) survive.
+      const deleted = await engine.deleteFactsForPage(slug, sourceId, {
+        excludeSourcePrefixes: ['cli:'],
+      });
+      result.factsDeleted += deleted.deleted;
+      toInsert = extracted;
+    }

    // v0.35.4 (D-CDX-3) — batch-embed before insert. Without this,
    // cycle-inserted facts land with `embedding = NULL`, which breaks
@@ -250,17 +343,17 @@ export async function runExtractFacts(
    // unavailable (no API key configured), facts still insert with
    // NULL embeddings — drift_score gracefully returns null and
    // clustering falls back to recency.
-    if (isAvailable('embedding') && extracted.length > 0) {
+    if (isAvailable('embedding') && toInsert.length > 0) {
      try {
-        const texts = extracted.map(e => e.fact);
+        const texts = toInsert.map(e => e.fact);
        // #1972: forward the abort signal so a cancelled cycle's in-flight
        // batch embed (a network call) is itself abortable, not just the loop.
        const embeddings = await embed(texts, { abortSignal: opts.signal });
        // Defensive: embed should return one vector per input; if the
        // gateway returns a partial array (provider partial-batch retry
        // returning fewer than requested), only fill what we have.
-        for (let i = 0; i < extracted.length && i < embeddings.length; i++) {
-          extracted[i].embedding = embeddings[i];
+        for (let i = 0; i < toInsert.length && i < embeddings.length; i++) {
+          toInsert[i].embedding = embeddings[i];
        }
      } catch (err) {
        // Embedding failure is non-fatal — facts still get inserted, just
@@ -271,7 +364,9 @@ export async function runExtractFacts(
      }
    }

-    const inserted = await engine.insertFacts(extracted, { source_id: sourceId }); // gbrain-allow-direct-insert: extract_facts cycle phase reconciles fence → DB
+    if (toInsert.length === 0) continue;
+
+    const inserted = await engine.insertFacts(toInsert, { source_id: sourceId }); // gbrain-allow-direct-insert: extract_facts cycle phase reconciles fence → DB
    result.factsInserted += inserted.inserted;
  }

@@ -12,6 +12,7 @@
 import { describe, test, expect, beforeAll, afterAll, beforeEach } from 'bun:test';
 import { PGLiteEngine } from '../src/core/pglite-engine.ts';
 import { runExtractFacts } from '../src/core/cycle/extract-facts.ts';
+import { parseFactsFence } from '../src/core/facts-fence.ts';

 let engine: PGLiteEngine;

@@ -99,11 +100,114 @@ describe('runExtractFacts — happy path', () => {
    );

    expect(r2.guardTriggered).toBe(false);
+    expect(r2.factsInserted).toBe(0);
+    expect(r2.factsDeleted).toBe(0);
    expect(after2.rows.map((r: { fact: string }) => r.fact))
      .toEqual(after1.rows.map((r: { fact: string }) => r.fact));
    expect(after2.rows).toHaveLength(2);
  });

+  test('dedupes duplicate fence rows by claim and source without rewriting the fence', async () => {
+    const body = FACT_FENCE(
+      `| 1 | A | fact | 1.0 | world | medium | 2026-01-01 |  | s |  |
+| 2 | A | fact | 1.0 | world | medium | 2026-01-01 |  | s |  |`,
+    );
+    await putPage('people/alice', body);
+
+    const r1 = await runExtractFacts(engine, { slugs: ['people/alice'] });
+    const r2 = await runExtractFacts(engine, { slugs: ['people/alice'] });
+
+    expect(r1.factsInserted).toBe(1);
+    expect(r2.factsInserted).toBe(0);
+    expect(r2.factsDeleted).toBe(0);
+
+    // The cycle dedups the derived DB index; it does not destructively
+    // rewrite user-authored markdown fence rows.
+    const page = await engine.getPage('people/alice', { sourceId: 'default' });
+    expect(parseFactsFence(page?.compiled_truth ?? '').facts).toHaveLength(2);
+
+    // eslint-disable-next-line @typescript-eslint/no-explicit-any
+    const rows = await (engine as any).db.query(
+      `SELECT fact, source FROM facts WHERE source_markdown_slug = 'people/alice'`,
+    );
+    expect(rows.rows).toHaveLength(1);
+    expect(rows.rows[0]).toMatchObject({ fact: 'A', source: 's' });
+  });
+
+  test('same claim with a different source is not treated as duplicate', async () => {
+    await putPage('people/alice', FACT_FENCE(
+      `| 1 | Same claim | fact | 1.0 | world | medium | 2026-01-01 |  | source-a |  |`,
+    ));
+    await runExtractFacts(engine, { slugs: ['people/alice'] });
+
+    await putPage('people/alice', FACT_FENCE(
+      `| 1 | Same claim | fact | 1.0 | world | medium | 2026-01-01 |  | source-a |  |
+| 2 | Same claim | fact | 1.0 | world | medium | 2026-01-01 |  | source-b |  |`,
+    ));
+    const r = await runExtractFacts(engine, { slugs: ['people/alice'] });
+
+    expect(r.factsInserted).toBe(1);
+    expect(r.factsDeleted).toBe(0);
+
+    // eslint-disable-next-line @typescript-eslint/no-explicit-any
+    const rows = await (engine as any).db.query(
+      `SELECT fact, source FROM facts WHERE source_markdown_slug = 'people/alice' ORDER BY row_num`,
+    );
+    expect(rows.rows).toEqual([
+      expect.objectContaining({ fact: 'Same claim', source: 'source-a' }),
+      expect.objectContaining({ fact: 'Same claim', source: 'source-b' }),
+    ]);
+  });
+
+  test('new fact added to the fence is inserted once without re-appending existing facts', async () => {
+    await putPage('people/alice', FACT_FENCE(
+      `| 1 | Existing | fact | 1.0 | world | medium | 2026-01-01 |  | s |  |`,
+    ));
+    await runExtractFacts(engine, { slugs: ['people/alice'] });
+
+    await putPage('people/alice', FACT_FENCE(
+      `| 1 | Existing | fact | 1.0 | world | medium | 2026-01-01 |  | s |  |
+| 2 | New | fact | 1.0 | world | medium | 2026-01-01 |  | s |  |`,
+    ));
+
+    const r = await runExtractFacts(engine, { slugs: ['people/alice'] });
+    expect(r.factsInserted).toBe(1);
+    expect(r.factsDeleted).toBe(0);
+
+    // eslint-disable-next-line @typescript-eslint/no-explicit-any
+    const rows = await (engine as any).db.query(
+      `SELECT fact FROM facts WHERE source_markdown_slug = 'people/alice' ORDER BY row_num`,
+    );
+    expect(rows.rows.map((row: { fact: string }) => row.fact)).toEqual(['Existing', 'New']);
+  });
+
+  test('cli:-origin conversation facts (#1928) neither break idempotency nor get wiped', async () => {
+    await putPage('people/alice', FACT_FENCE(
+      `| 1 | Fence fact | fact | 1.0 | world | medium | 2026-01-01 |  | s |  |`,
+    ));
+    // A conversation fact on the same page coordinate — NOT fence-owned.
+    await engine.insertFacts(
+      [{ fact: 'conversation fact', kind: 'fact', source: 'cli:extract-conversation-facts', row_num: 99, source_markdown_slug: 'people/alice' }],
+      { source_id: 'default' },
+    );
+
+    const r1 = await runExtractFacts(engine, { slugs: ['people/alice'] });
+    const r2 = await runExtractFacts(engine, { slugs: ['people/alice'] });
+
+    // The cli: row must not count as "stale" — a wipe/reinsert every cycle
+    // would defeat idempotency (and churn factsDeleted/factsInserted).
+    expect(r1.factsInserted).toBe(1);
+    expect(r2.factsInserted).toBe(0);
+    expect(r2.factsDeleted).toBe(0);
+
+    // eslint-disable-next-line @typescript-eslint/no-explicit-any
+    const rows = await (engine as any).db.query(
+      `SELECT fact FROM facts WHERE source_markdown_slug = 'people/alice' ORDER BY row_num`,
+    );
+    expect(rows.rows.map((row: { fact: string }) => row.fact))
+      .toEqual(['Fence fact', 'conversation fact']);
+  });
+
  test('removed-from-fence row is deleted from DB (wipe-and-reinsert pattern)', async () => {
    // Seed: 2 facts.
    await putPage('people/alice', FACT_FENCE(