fix(synthesize): dedupe successful transcripts after corpus moves (#3424)

* fix(synthesize): dedupe across corpus moves * fix(synthesize): dedupe legacy CHUNKED completions; keep plain-completed suppression Repairs three gaps in the corpus-move dedupe (v2 content-hash keys): 1. Legacy chunked completions now suppress v2 resubmission. The scan previously matched only keys ending ':<hash16>' (legacy single-chunk), so every transcript synthesized under the pre-v2 chunked family 'dream:synth:<path>:<hash16>:c<i>of<n>' re-ran as a full paid v2 synthesis after upgrade. findLegacyCompletion now also matches the chunked family, counting a transcript as done only when the FULL chunk set c0..c(n-1) completed; partial sets fall through to a fresh v2 run (reason: already_synthesized_legacy_chunked for full sets). 2+3. Legacy suppression reverts to plain status='completed', dropping the result->>'stop_reason' = 'end_turn' filter. This restores the pre-v2 cost-safe semantics (queue-level idempotency blocks re-submission of completed jobs regardless of stop_reason, pinned in test/minions.test.ts) and sidesteps the double-encoded-jsonb result rows the naive ->> read missed. The tightening was not documented as intended in the PR. Tests: legacy chunked full-set suppression + partial-set resubmission; double-encoded jsonb result row still recognized. Co-authored-by: zsimovanforgeops <justin@caddolandworks.com> Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> --------- Co-authored-by: Forge (Ron) <forge@zsimovan.dev> Co-authored-by: Garry Tan <garrytan@gmail.com> Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
2026-07-30 03:12:32 +00:00 · 2026-07-27 14:17:06 -07:00
parent 10079efe40
commit d2ac2aef49
2 changed files with 245 additions and 59 deletions
@@ -33,7 +33,7 @@ import { chat as gatewayChat, validateModelId, type ChatResult } from '../ai/gat
 import { AIConfigError } from '../ai/errors.ts';
 import { normalizeModelId } from '../model-id.ts';
 import { hasAnthropicKey } from '../ai/anthropic-key.ts';
-import { join, dirname, isAbsolute, resolve } from 'node:path';
+import { basename, join, dirname, isAbsolute, resolve } from 'node:path';
 import type { BrainEngine } from '../engine.ts';
 import type { PhaseResult, PhaseError } from '../cycle.ts';
 import { MinionQueue } from '../minions/queue.ts';
@@ -560,20 +560,31 @@ export async function runPhaseSynthesize(
    const skipReports: Array<{ filePath: string; reason: string }> = [];

    const maxCharsPerChunk = computeChunkCharBudget(config.model, config.maxPromptTokens);
+    const successfulLegacyKeys = await loadSuccessfulLegacySynthesisKeys(
+      engine,
+      opts.sourceId ?? 'default',
+    );

    for (const t of worthProcessing) {
      const hash16 = t.contentHash.slice(0, 16);
      const hash6 = t.contentHash.slice(0, 6);

-      // D8: single→multi-chunk migration safety. If a completed legacy
-      // single-chunk job exists for this content_hash, treat as already-
-      // synthesized and skip. Prevents duplicate writes when a transcript
-      // that was previously single-chunk now multi-chunks (because budget
-      // shrank or model changed).
-      if (await hasLegacySingleChunkCompletion(engine, t.filePath, hash16)) {
+      // D8: legacy-key migration safety. If this content hash already
+      // completed under the pre-v2 path-based key family — single-chunk OR
+      // a full chunked set — treat as already-synthesized and skip.
+      // Prevents a full paid re-synthesis when the corpus root moves or
+      // the chunking outcome changes across versions.
+      const legacyCompletion = findLegacyCompletion(
+        successfulLegacyKeys,
+        t.filePath,
+        hash16,
+      );
+      if (legacyCompletion) {
        skipReports.push({
          filePath: t.filePath,
-          reason: 'already_synthesized_legacy_single_chunk',
+          reason: legacyCompletion === 'chunked'
+            ? 'already_synthesized_legacy_chunked'
+            : 'already_synthesized_legacy_single_chunk',
        });
        continue;
      }
@@ -617,15 +628,15 @@ export async function runPhaseSynthesize(
          // so put_page writes land there instead of the hardcoded 'default'.
          ...(opts.sourceId ? { source_id: opts.sourceId } : {}),
        };
-        // Idempotency key parity:
-        //   - single-chunk → legacy `dream:synth:<filePath>:<hash16>` (byte-
-        //     equivalent across versions; preserves dedup for unchanged
-        //     transcripts on upgrade).
-        //   - multi-chunk → `<legacy>:c<i>of<n>` per chunk; durable across
-        //     runs because D9 splitTranscriptByBudget is hash-deterministic.
+        // Keep producer identity stable when the corpus root moves. Source and
+        // complete filename remain explicit so equal bytes in different source
+        // or filename namespaces do not collide.
+        const synthesisKey =
+          `dream:synth-v2:${encodeURIComponent(opts.sourceId ?? 'default')}` +
+          `:filename:${encodeURIComponent(basename(t.filePath))}:${hash16}`;
        const idempotency_key = isChunked
-          ? `dream:synth:${t.filePath}:${hash16}:c${i}of${chunks.length}`
-          : `dream:synth:${t.filePath}:${hash16}`;
+          ? `${synthesisKey}:c${i}of${chunks.length}`
+          : synthesisKey;
        const submitOpts: Partial<MinionJobInput> = {
          max_stalled: 3,
          on_child_fail: 'continue',
@@ -1281,29 +1292,73 @@ async function collectChildPutPageSlugs(
 }

 /**
- * D8: query for any `completed` legacy single-chunk job at the canonical
- * idempotency key shape `dream:synth:<filePath>:<hash16>`. Used at fan-out
- * time to detect transcripts that were synthesized under the pre-chunking
- * code path; those should NOT be re-submitted under chunked keys.
+ * D8: load every `completed` legacy job key in the pre-v2 path-based
+ * family `dream:synth:<filePath>:<hash16>[:c<i>of<n>]`. Used at fan-out
+ * time to detect transcripts already synthesized under an old key shape;
+ * those should NOT be re-submitted under v2 keys. (v2 keys start with
+ * `dream:synth-v2:` and don't match the LIKE prefix — the queue's own
+ * idempotency dedupe already covers them.)
 *
- * Reuses the existing `minion_jobs.idempotency_key` index — no schema
- * additions. One indexed lookup per worth-processing transcript.
+ * Plain `status = 'completed'` deliberately mirrors the queue-level
+ * idempotency semantics the legacy keys relied on: a completed job blocks
+ * re-submission regardless of `result.stop_reason` (pinned in
+ * test/minions.test.ts). Filtering on stop_reason here would re-pay for
+ * transcripts the old code path never re-ran, and reading `result` at all
+ * would need the `(result #>> '{}')` double-encoded-jsonb defense.
+ *
+ * Loads source-scoped completions once per phase; no schema additions
+ * and no repeated history scan for each transcript.
 */
-async function hasLegacySingleChunkCompletion(
+async function loadSuccessfulLegacySynthesisKeys(
  engine: BrainEngine,
+  sourceId: string,
+): Promise<string[]> {
+  const rows = await engine.executeRaw<{ idempotency_key: string }>(
+    `SELECT idempotency_key
+       FROM minion_jobs
+      WHERE name = 'subagent'
+        AND status = 'completed'
+        AND COALESCE(NULLIF(data->>'source_id', ''), 'default') = $1
+        AND idempotency_key LIKE 'dream:synth:%'`,
+    [sourceId],
+  );
+  return rows.map(row => row.idempotency_key);
+}
+
+/**
+ * Match a transcript (by filename + content hash) against completed legacy
+ * keys. `'single'` when a `dream:synth:<path>:<hash16>` completion exists;
+ * `'chunked'` when a FULL chunk set `:c0of<n>`..`:c<n-1>of<n>` completed
+ * (chunk indices are 0-based). Partial chunk sets return null so the
+ * transcript gets a fresh v2 synthesis instead of shipping with holes.
+ */
+function findLegacyCompletion(
+  successfulKeys: string[],
  filePath: string,
  hash16: string,
-): Promise<boolean> {
-  const legacyKey = `dream:synth:${filePath}:${hash16}`;
-  const rows = await engine.executeRaw<{ status: string }>(
-    `SELECT status
-       FROM minion_jobs
-      WHERE idempotency_key = $1
-        AND status = 'completed'
-      LIMIT 1`,
-    [legacyKey],
-  );
-  return rows.length > 0;
+): 'single' | 'chunked' | null {
+  const filename = basename(filePath);
+  const hashSuffix = `:${hash16}`;
+  /** total chunk count n → completed 0-based chunk indices */
+  const chunkSets = new Map<number, Set<number>>();
+  for (const key of successfulKeys) {
+    const chunk = /:c(\d+)of(\d+)$/.exec(key);
+    const base = chunk ? key.slice(0, -chunk[0].length) : key;
+    if (!base.endsWith(hashSuffix)) continue;
+    const historicalPath = base.slice('dream:synth:'.length, -hashSuffix.length);
+    if (basename(historicalPath) !== filename) continue;
+    if (!chunk) return 'single';
+    const i = Number(chunk[1]);
+    const n = Number(chunk[2]);
+    if (n < 1 || i < 0 || i >= n) continue;
+    let seen = chunkSets.get(n);
+    if (!seen) chunkSets.set(n, seen = new Set());
+    seen.add(i);
+  }
+  for (const [n, seen] of chunkSets) {
+    if (seen.size === n) return 'chunked';
+  }
+  return null;
 }

 // ── Dream-provenance DB stamp (#2569) ────────────────────────────────
@@ -8,10 +8,12 @@
 * Coverage:
 *   - D5 cap-hit: chunks > maxChunks → log + skip with no minion_jobs row
 *     and no dream_verdicts cache write (closes the poison-pill class).
- *   - D8 legacy single-chunk migration: pre-seed a `completed` legacy job
- *     for the same content hash → next synthesize skips submission.
+ *   - D8 legacy-key migration: a completed old-root job (single-chunk or a
+ *     full chunked set) for the same filename + content hash suppresses
+ *     duplicate synthesis; partial chunk sets and double-encoded result
+ *     rows are covered.
 *   - Chunked path: fat transcript spawns N children with chunk-suffixed
- *     idempotency keys; single-chunk path keeps the legacy key shape.
+ *     path-independent idempotency keys; single-chunk omits the suffix.
 *
 * Run: bun test test/e2e/dream-synthesize-chunking.test.ts
 */
@@ -168,9 +170,10 @@ describe('E2E synthesize chunking — D5 cap hit', () => {
  }, 30_000);
 });

-describe('E2E synthesize chunking — D8 legacy single-chunk migration', () => {
-  test('completed legacy idempotency key → skip submission entirely', async () => {
+describe('E2E synthesize chunking — D8 legacy-key migration', () => {
+  test('successful legacy synthesis survives a corpus-root move', async () => {
    const rig = await setupRig();
+    const oldCorpusDir = mkdtempSync(join(tmpdir(), 'gbrain-chunk-old-corpus-'));
    try {
      await rig.engine.setConfig('dream.synthesize.enabled', 'true');
      await rig.engine.setConfig('dream.synthesize.session_corpus_dir', rig.corpusDir);
@@ -181,29 +184,155 @@ describe('E2E synthesize chunking — D8 legacy single-chunk migration', () => {
      writeFileSync(filePath, content);
      const contentHash = await seedVerdict(rig.engine, filePath, content);

-      // Pre-seed a completed `subagent` job at the legacy idempotency key.
-      const legacyKey = `dream:synth:${filePath}:${contentHash.slice(0, 16)}`;
+      // The successful historical job used a different corpus root.
+      const oldFilePath = corpusPath(oldCorpusDir, basename);
+      const legacyKey = `dream:synth:${oldFilePath}:${contentHash.slice(0, 16)}`;
      await rig.engine.executeRaw(
-        `INSERT INTO minion_jobs (name, queue, status, idempotency_key, finished_at)
-         VALUES ('subagent', 'default', 'completed', $1, now())`,
+        `INSERT INTO minion_jobs
+           (name, queue, status, data, result, idempotency_key, finished_at)
+         VALUES
+           ('subagent', 'default', 'completed', '{}'::jsonb,
+            '{"stop_reason":"end_turn"}'::jsonb, $1, now())`,
        [legacyKey],
      );

      await withoutAnthropicKey(async () => {
-        const result = await runPhaseSynthesize(rig.engine, {
-          brainDir: rig.brainDir,
-          dryRun: false,
+        await withSubagentAutoCancel(rig.engine, async () => {
+          const result = await runPhaseSynthesize(rig.engine, {
+            brainDir: rig.brainDir,
+            dryRun: false,
+          });
+          const details = result.details as {
+            children_submitted: number;
+            skips: Array<{ reason: string }>;
+          };
+          expect(details.children_submitted).toBe(0);
+          expect(details.skips).toHaveLength(1);
+          expect(details.skips[0].reason).toBe('already_synthesized_legacy_single_chunk');
+        });
+      });
+
+      // No new subagent job: still exactly one historical success.
+      const jobs = await rig.engine.executeRaw<{ cnt: string | number }>(
+        `SELECT count(*) AS cnt FROM minion_jobs WHERE name = 'subagent'`,
+      );
+      expect(Number(jobs[0].cnt)).toBe(1);
+    } finally {
+      rmSync(oldCorpusDir, { recursive: true, force: true });
+      await rig.cleanup();
+    }
+  }, 30_000);
+
+  test('legacy CHUNKED completion suppresses v2 resubmission; partial chunk set does not', async () => {
+    const rig = await setupRig();
+    const oldCorpusDir = mkdtempSync(join(tmpdir(), 'gbrain-chunk-old-corpus-'));
+    try {
+      await rig.engine.setConfig('dream.synthesize.enabled', 'true');
+      await rig.engine.setConfig('dream.synthesize.session_corpus_dir', rig.corpusDir);
+
+      // Transcript A: previously synthesized as a FULL 2-chunk legacy run.
+      const fullName = '2026-04-26-chunked-complete.txt';
+      const fullPath = corpusPath(rig.corpusDir, fullName);
+      const fullContent = 'fully chunk-synthesized lines\n'.repeat(200);
+      writeFileSync(fullPath, fullContent);
+      const fullHash16 = (await seedVerdict(rig.engine, fullPath, fullContent)).slice(0, 16);
+
+      // Transcript B: legacy run completed only chunk 0 of 3 (partial).
+      const partialName = '2026-04-27-chunked-partial.txt';
+      const partialPath = corpusPath(rig.corpusDir, partialName);
+      const partialContent = 'partially chunk-synthesized lines\n'.repeat(200);
+      writeFileSync(partialPath, partialContent);
+      const partialHash16 = (await seedVerdict(rig.engine, partialPath, partialContent)).slice(0, 16);
+
+      // All legacy rows lived under a different (moved-away) corpus root.
+      const legacyKeys = [
+        `dream:synth:${corpusPath(oldCorpusDir, fullName)}:${fullHash16}:c0of2`,
+        `dream:synth:${corpusPath(oldCorpusDir, fullName)}:${fullHash16}:c1of2`,
+        `dream:synth:${corpusPath(oldCorpusDir, partialName)}:${partialHash16}:c0of3`,
+      ];
+      for (const key of legacyKeys) {
+        await rig.engine.executeRaw(
+          `INSERT INTO minion_jobs
+             (name, queue, status, data, result, idempotency_key, finished_at)
+           VALUES
+             ('subagent', 'default', 'completed', '{}'::jsonb,
+              '{"stop_reason":"end_turn"}'::jsonb, $1, now())`,
+          [key],
+        );
+      }
+
+      await withoutAnthropicKey(async () => {
+        await withSubagentAutoCancel(rig.engine, async () => {
+          const result = await runPhaseSynthesize(rig.engine, {
+            brainDir: rig.brainDir,
+            dryRun: false,
+          });
+          const details = result.details as {
+            children_submitted: number;
+            skips: Array<{ filePath: string; reason: string }>;
+          };
+          // A skipped (full legacy chunk set); B resubmitted (partial set).
+          expect(details.children_submitted).toBe(1);
+          expect(details.skips).toHaveLength(1);
+          expect(details.skips[0].filePath).toBe(fullPath);
+          expect(details.skips[0].reason).toBe('already_synthesized_legacy_chunked');
+        });
+      });
+
+      // 3 seeded legacy rows + exactly 1 new v2 job for the partial transcript.
+      const rows = await rig.engine.executeRaw<{ idempotency_key: string }>(
+        `SELECT idempotency_key FROM minion_jobs
+          WHERE name = 'subagent' AND idempotency_key LIKE 'dream:synth-v2:%'`,
+      );
+      expect(rows).toHaveLength(1);
+      expect(rows[0].idempotency_key).toContain(encodeURIComponent(partialName));
+    } finally {
+      rmSync(oldCorpusDir, { recursive: true, force: true });
+      await rig.cleanup();
+    }
+  }, 30_000);
+
+  test('legacy completed row with double-encoded jsonb result still suppresses', async () => {
+    const rig = await setupRig();
+    try {
+      await rig.engine.setConfig('dream.synthesize.enabled', 'true');
+      await rig.engine.setConfig('dream.synthesize.session_corpus_dir', rig.corpusDir);
+
+      const basename = '2026-04-28-double-encoded.txt';
+      const filePath = corpusPath(rig.corpusDir, basename);
+      const content = 'double-encoded result lines\n'.repeat(200);
+      writeFileSync(filePath, content);
+      const contentHash = await seedVerdict(rig.engine, filePath, content);
+
+      // Historical row whose `result` was double-encoded (jsonb string
+      // scalar — the #2339 class). `result->>'stop_reason'` yields NULL on
+      // this row; a completed legacy job must suppress regardless.
+      const legacyKey = `dream:synth:${filePath}:${contentHash.slice(0, 16)}`;
+      await rig.engine.executeRaw(
+        `INSERT INTO minion_jobs
+           (name, queue, status, data, result, idempotency_key, finished_at)
+         VALUES
+           ('subagent', 'default', 'completed', '{}'::jsonb,
+            to_jsonb('{"stop_reason":"end_turn"}'::text), $1, now())`,
+        [legacyKey],
+      );
+
+      await withoutAnthropicKey(async () => {
+        await withSubagentAutoCancel(rig.engine, async () => {
+          const result = await runPhaseSynthesize(rig.engine, {
+            brainDir: rig.brainDir,
+            dryRun: false,
+          });
+          const details = result.details as {
+            children_submitted: number;
+            skips: Array<{ reason: string }>;
+          };
+          expect(details.children_submitted).toBe(0);
+          expect(details.skips).toHaveLength(1);
+          expect(details.skips[0].reason).toBe('already_synthesized_legacy_single_chunk');
        });
-        const details = result.details as {
-          children_submitted: number;
-          skips: Array<{ reason: string }>;
-        };
-        expect(details.children_submitted).toBe(0);
-        expect(details.skips).toHaveLength(1);
-        expect(details.skips[0].reason).toBe('already_synthesized_legacy_single_chunk');
      });

-      // No NEW subagent job: still exactly one (the seeded completed row).
      const jobs = await rig.engine.executeRaw<{ cnt: string | number }>(
        `SELECT count(*) AS cnt FROM minion_jobs WHERE name = 'subagent'`,
      );
@@ -215,7 +344,7 @@ describe('E2E synthesize chunking — D8 legacy single-chunk migration', () => {
 });

 describe('E2E synthesize chunking — fan-out shape', () => {
-  test('single-chunk transcript uses legacy idempotency key (parity on upgrade)', async () => {
+  test('single-chunk transcript key excludes the corpus root', async () => {
    const rig = await setupRig();
    try {
      await rig.engine.setConfig('dream.synthesize.enabled', 'true');
@@ -239,13 +368,14 @@ describe('E2E synthesize chunking — fan-out shape', () => {
        });
      });

-      const expectedKey = `dream:synth:${filePath}:${contentHash.slice(0, 16)}`;
+      const expectedKey =
+        `dream:synth-v2:default:filename:${encodeURIComponent(basename)}:${contentHash.slice(0, 16)}`;
      const rows = await rig.engine.executeRaw<{ idempotency_key: string }>(
        `SELECT idempotency_key FROM minion_jobs WHERE name = 'subagent' ORDER BY id`,
      );
      expect(rows).toHaveLength(1);
      expect(rows[0].idempotency_key).toBe(expectedKey);
-      // Specifically: legacy key shape has NO ":c<idx>of<n>" suffix.
+      // Single-chunk keys have no ":c<idx>of<n>" suffix.
      expect(rows[0].idempotency_key).not.toMatch(/:c\d+of\d+$/);
    } finally {
      await rig.cleanup();
@@ -283,10 +413,11 @@ describe('E2E synthesize chunking — fan-out shape', () => {
        `SELECT idempotency_key FROM minion_jobs WHERE name = 'subagent' ORDER BY id`,
      );
      expect(rows.length).toBeGreaterThan(1);
-      // Every key matches the chunked shape `dream:synth:<path>:<hash16>:c<i>of<N>`.
+      const baseKey =
+        `dream:synth-v2:default:filename:${encodeURIComponent(basename)}:${hash16}`;
      for (const r of rows) {
        expect(r.idempotency_key).toMatch(
-          new RegExp(`^dream:synth:${escapeRe(filePath)}:${hash16}:c\\d+of\\d+$`),
+          new RegExp(`^${escapeRe(baseKey)}:c\\d+of\\d+$`),
        );
      }
      // Chunk indices are unique 0..N-1.