diff --git a/src/core/cycle/synthesize.ts b/src/core/cycle/synthesize.ts index 02a2a268f..4f1166cec 100644 --- a/src/core/cycle/synthesize.ts +++ b/src/core/cycle/synthesize.ts @@ -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::` (byte- - // equivalent across versions; preserves dedup for unchanged - // transcripts on upgrade). - // - multi-chunk → `:cof` 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 = { 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::`. 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::[:cof]`. 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 { + 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::` completion exists; + * `'chunked'` when a FULL chunk set `:c0of`..`:cof` 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 { - 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>(); + 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) ──────────────────────────────── diff --git a/test/e2e/dream-synthesize-chunking.test.ts b/test/e2e/dream-synthesize-chunking.test.ts index fedb06aea..71e6ce616 100644 --- a/test/e2e/dream-synthesize-chunking.test.ts +++ b/test/e2e/dream-synthesize-chunking.test.ts @@ -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 ":cof" suffix. + // Single-chunk keys have no ":cof" 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:::cof`. + 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.