From 3fa0a5acb593f02c3cdc6b9bcb83119825dc1415 Mon Sep 17 00:00:00 2001 From: Garry Tan Date: Wed, 12 Aug 2026 11:18:32 -0700 Subject: [PATCH] fix(dream): do not permanently cache truncated or degenerate significance verdicts (#3918) Wave-assembled from PR #3918 by @Masashi-Ono0611. Co-Authored-By: masashiono0611 --- src/core/cycle/synthesize.ts | 115 ++++++++++++++++-- test/cycle-synthesize.test.ts | 105 ++++++++++++++++ test/cycle/synthesize-gateway-adapter.test.ts | 50 ++++++++ test/e2e/dream-synthesize-pglite.test.ts | 114 ++++++++++++++++- 4 files changed, 374 insertions(+), 10 deletions(-) diff --git a/src/core/cycle/synthesize.ts b/src/core/cycle/synthesize.ts index 262f87bcb..975485665 100644 --- a/src/core/cycle/synthesize.ts +++ b/src/core/cycle/synthesize.ts @@ -502,7 +502,22 @@ export async function runPhaseSynthesize( } try { const verdict = await judgeSignificance(judge, t, config.verdictModel); - await engine.putDreamVerdict(t.filePath, t.contentHash, verdict); + if (verdict.unreliable) { + // Degenerate judgement (truncated, refused/content-filtered, or + // unparseable LLM output). Do + // NOT write it to dream_verdicts: a cached `worth_processing: + // false` is permanent for this content hash, and a brain whose + // verdict model reliably truncates (e.g. a reasoning model under + // a tight budget) would silently reject every transcript forever. + // Log + skip so the next cycle re-judges. + process.stderr.write( + `[dream] verdict for ${t.basename} was ${verdict.unreliable} ` + + `(${verdict.reasons.join('; ')}); not caching in dream_verdicts — ` + + `next cycle will re-judge ${t.filePath}\n`, + ); + } else { + await engine.putDreamVerdict(t.filePath, t.contentHash, verdict); + } verdicts.push({ filePath: t.filePath, worth: verdict.worth_processing, reasons: verdict.reasons, cached: false }); if (verdict.worth_processing) worthProcessing.push(t); } catch (e) { @@ -1030,15 +1045,36 @@ export function makeJudgeClient(verdictModel: string): JudgeClient | null { }); // Map gateway.ChatResult → Anthropic.Message shape. judgeSignificance - // reads `.content[0].type === 'text'` and `.content[0].text`; other - // fields are best-effort for downstream telemetry parity. + // reads `.content[0].type === 'text'`, `.content[0].text`, and + // `.stop_reason`; other fields are best-effort for downstream + // telemetry parity. The stopReason mapping is load-bearing: + // 'length' → 'max_tokens' lets judgeSignificance detect a truncated + // verdict (reasoning models can burn the whole max_tokens budget on + // reasoning tokens, leaving empty/partial text) and + // 'refusal'/'content_filter' → 'refusal' surfaces a blocked response, + // instead of silently treating either as a clean end-of-turn. + // 'other' (gateway's mapStopReason catch-all — unknown provider + // finish reasons, which includes both non-standard SUCCESSFUL stops + // and the AI SDK's 'error'/'unknown' labels) maps to 'end_turn' + // deliberately. Rationale: (a) treating 'other' as abnormal would + // permanently disable verdict caching on providers whose successful + // stops the gateway doesn't recognize; (b) the residual risk is + // narrow — an errored response only gets cached if it still contains + // a complete, parseable JSON object with a boolean worth_processing + // (unparseable output is never cached), and such a complete verdict + // is trustworthy regardless of the finish label. Distinguishing + // error/unknown from benign-unknown belongs in gateway.ts's + // mapStopReason (out of scope here — see PR notes). return { id: '', type: 'message', role: 'assistant', model: modelStr, content: [{ type: 'text', text: result.text }], - stop_reason: 'end_turn', + stop_reason: result.stopReason === 'length' ? 'max_tokens' + : result.stopReason === 'tool_calls' ? 'tool_use' + : (result.stopReason === 'refusal' || result.stopReason === 'content_filter') ? 'refusal' + : 'end_turn', stop_sequence: null, usage: { input_tokens: result.usage.input_tokens, @@ -1052,6 +1088,21 @@ export function makeJudgeClient(verdictModel: string): JudgeClient | null { interface VerdictResult { worth_processing: boolean; reasons: string[]; + /** + * Set when the judgement is degenerate and must NOT be cached in + * dream_verdicts: + * 'truncated' — the response hit max_tokens (reasoning models can spend + * the whole budget on reasoning tokens before emitting the + * verdict JSON), so the verdict is unreliable; + * 'refusal' — the model refused or a provider content filter blocked + * the response (stop_reason=refusal); + * 'unparseable' — no JSON object with a boolean `worth_processing` could + * be parsed out of the response. + * The synthesize verdict loop skips putDreamVerdict for these so the next + * cycle re-judges the transcript instead of permanently trusting a + * degenerate `worth_processing: false`. + */ + unreliable?: 'truncated' | 'refusal' | 'unparseable'; } export async function judgeSignificance( @@ -1102,11 +1153,32 @@ Two reasons max, one phrase each.`; const msg = await client.create({ model: verdictModel, - max_tokens: 200, + // 1024, not the 200 this call shipped with for a year: the verdict JSON + // itself needs <100 tokens, but reasoning models spend output budget on + // reasoning tokens BEFORE the visible text, so a 200-token cap gets + // eaten whole and the verdict comes back empty/truncated. Matches the + // judge-call ballpark elsewhere in the repo (grade-takes.ts: 600, + // eval-contradictions/judge.ts: 1024). + max_tokens: 1024, system: sys, messages: [{ role: 'user', content: `Transcript ${t.basename}:\n\n${trimmed}` }], }); + // stop_reason === 'max_tokens' means the response was cut off; 'refusal' + // means the model refused or a content filter blocked it. Even if a + // parseable JSON object survives either condition, don't trust it as a + // durable verdict — mark it unreliable so the caller skips the + // dream_verdicts write and the next cycle re-judges. (Legacy SDK-shape + // clients and mocks without a stop_reason field land on undefined here, + // which is treated as a clean stop — preserves the pre-existing contract.) + // Widen: the pinned Anthropic SDK's stop_reason union predates 'refusal', + // but the gateway adapter (and newer SDKs) can emit it. + const stopReasonRaw = (msg as { stop_reason?: string | null }).stop_reason; + const truncated = stopReasonRaw === 'max_tokens'; + const refused = stopReasonRaw === 'refusal'; + const abnormalStop: VerdictResult['unreliable'] | undefined = + truncated ? 'truncated' : refused ? 'refusal' : undefined; + for (const block of msg.content) { if (block.type === 'text') { const text = block.text.trim(); @@ -1114,16 +1186,41 @@ Two reasons max, one phrase each.`; if (!m) continue; try { const parsed = JSON.parse(m[0]) as { worth_processing?: unknown; reasons?: unknown }; - const worth = parsed.worth_processing === true; + // A JSON object without a boolean worth_processing (`{}`, + // `{"worth_processing": "true"}`) is NOT a verdict — fall through to + // the unparseable branch rather than coercing to a cacheable false. + if (typeof parsed.worth_processing !== 'boolean') continue; + const worth = parsed.worth_processing; const reasons = Array.isArray(parsed.reasons) ? parsed.reasons.filter((r): r is string => typeof r === 'string').slice(0, 4) : []; - return { worth_processing: worth, reasons }; + return abnormalStop + ? { worth_processing: worth, reasons, unreliable: abnormalStop } + : { worth_processing: worth, reasons }; } catch { /* fall through */ } } } - // Couldn't parse — default to NOT processing (cheap fallback). - return { worth_processing: false, reasons: ['judge response unparseable'] }; + // Couldn't parse — default to NOT processing this cycle, but flag the + // result unreliable so it is never cached as a permanent verdict. + if (truncated) { + return { + worth_processing: false, + reasons: ['judge response truncated (stop_reason=max_tokens)'], + unreliable: 'truncated', + }; + } + if (refused) { + return { + worth_processing: false, + reasons: ['judge response refused or content-filtered (stop_reason=refusal)'], + unreliable: 'refusal', + }; + } + return { + worth_processing: false, + reasons: ['judge response unparseable'], + unreliable: 'unparseable', + }; } // ── Subagent prompt ────────────────────────────────────────────────── diff --git a/test/cycle-synthesize.test.ts b/test/cycle-synthesize.test.ts index 2e60bbf2f..9d6668ab4 100644 --- a/test/cycle-synthesize.test.ts +++ b/test/cycle-synthesize.test.ts @@ -334,6 +334,111 @@ describe('judgeSignificance', () => { expect(r.worth_processing).toBe(false); expect(r.reasons[0]).toContain('unparseable'); }); + + test('marks unparseable output unreliable so the caller never caches it', async () => { + const client: JudgeClient = { + create: async () => ({ + content: [{ type: 'text', text: 'no json here' }], + stop_reason: 'end_turn', + } as any), + }; + const r = await judgeSignificance(client, makeTranscript()); + expect(r.worth_processing).toBe(false); + expect(r.unreliable).toBe('unparseable'); + }); + + test('marks truncated response (stop_reason=max_tokens) unreliable — reasoning models can burn the budget', async () => { + // Simulates a reasoning model that spent the whole max_tokens budget on + // reasoning tokens: visible text is a partial JSON fragment and + // stop_reason is 'max_tokens'. + const client: JudgeClient = { + create: async () => ({ + content: [{ type: 'text', text: '{"worth_process' }], + stop_reason: 'max_tokens', + } as any), + }; + const r = await judgeSignificance(client, makeTranscript()); + expect(r.worth_processing).toBe(false); + expect(r.unreliable).toBe('truncated'); + expect(r.reasons[0]).toContain('truncated'); + }); + + test('marks truncated response unreliable even when a parseable JSON object survives the cut', async () => { + const client: JudgeClient = { + create: async () => ({ + content: [{ type: 'text', text: '{"worth_processing": true, "reasons": ["r1"]}' }], + stop_reason: 'max_tokens', + } as any), + }; + const r = await judgeSignificance(client, makeTranscript()); + // The parsed values still drive THIS cycle... + expect(r.worth_processing).toBe(true); + expect(r.reasons).toEqual(['r1']); + // ...but the verdict must not be banked as permanent. + expect(r.unreliable).toBe('truncated'); + }); + + test('clean parse with stop_reason=end_turn stays cacheable (no unreliable marker)', async () => { + const client: JudgeClient = { + create: async () => ({ + content: [{ type: 'text', text: '{"worth_processing": true, "reasons": ["r1"]}' }], + stop_reason: 'end_turn', + } as any), + }; + const r = await judgeSignificance(client, makeTranscript()); + expect(r.worth_processing).toBe(true); + expect(r.unreliable).toBeUndefined(); + }); + + test('marks refused/content-filtered response (stop_reason=refusal) unreliable', async () => { + const client: JudgeClient = { + create: async () => ({ + content: [{ type: 'text', text: '{"worth_processing": false, "reasons": ["blocked"]}' }], + stop_reason: 'refusal', + } as any), + }; + const r = await judgeSignificance(client, makeTranscript()); + expect(r.unreliable).toBe('refusal'); + }); + + test('JSON object without worth_processing key is unparseable, not a cacheable false', async () => { + const client: JudgeClient = { + create: async () => ({ + content: [{ type: 'text', text: '{}' }], + stop_reason: 'end_turn', + } as any), + }; + const r = await judgeSignificance(client, makeTranscript()); + expect(r.worth_processing).toBe(false); + expect(r.unreliable).toBe('unparseable'); + }); + + test('non-boolean worth_processing ("true") is unparseable, not a cacheable false', async () => { + const client: JudgeClient = { + create: async () => ({ + content: [{ type: 'text', text: '{"worth_processing": "true", "reasons": ["r1"]}' }], + stop_reason: 'end_turn', + } as any), + }; + const r = await judgeSignificance(client, makeTranscript()); + expect(r.worth_processing).toBe(false); + expect(r.unreliable).toBe('unparseable'); + }); + + test('judge max_tokens budget is exactly 1024 (reasoning-model headroom, cost-capped)', async () => { + let captured: number | undefined; + const client: JudgeClient = { + create: async (p: any) => { + captured = p.max_tokens; + return { + content: [{ type: 'text', text: '{"worth_processing": false, "reasons": []}' }], + stop_reason: 'end_turn', + } as any; + }, + }; + await judgeSignificance(client, makeTranscript()); + expect(captured).toBe(1024); + }); }); // ─── v0.41.13: UTF-16 safety in judgeSignificance ───────────────────── diff --git a/test/cycle/synthesize-gateway-adapter.test.ts b/test/cycle/synthesize-gateway-adapter.test.ts index d22bba85b..ccb4dcaf7 100644 --- a/test/cycle/synthesize-gateway-adapter.test.ts +++ b/test/cycle/synthesize-gateway-adapter.test.ts @@ -235,6 +235,56 @@ describe('JudgeClient.create — gateway routing + shape adapter', () => { expect(caught).toBeInstanceOf(AIConfigError); }); }); + + test('A10: ChatResult.stopReason propagates — length → max_tokens, end → end_turn', async () => { + await withEnv({ ANTHROPIC_API_KEY: 'sk-test-A10' }, async () => { + const judge = makeJudgeClient('claude-haiku-4-5-20251001'); + + let nextStopReason: ChatResult['stopReason'] = 'length'; + __setChatTransportForTests(async (): Promise => ({ + text: '{"worth_processing"', + blocks: [], + stopReason: nextStopReason, + usage: { input_tokens: 5, output_tokens: 200, cache_read_tokens: 0, cache_creation_tokens: 0 }, + model: 'test:stub', + providerId: 'test', + })); + + const params = { + model: 'claude-haiku-4-5-20251001', + max_tokens: 100, + system: 's', + messages: [{ role: 'user' as const, content: 'u' }], + }; + + // Pre-fix the adapter pinned stop_reason to 'end_turn', hiding + // truncation from judgeSignificance. 'length' must surface as the + // Anthropic-shape 'max_tokens'. + const truncatedMsg = await judge!.create(params); + expect(truncatedMsg.stop_reason).toBe('max_tokens'); + + nextStopReason = 'end'; + const cleanMsg = await judge!.create(params); + expect(cleanMsg.stop_reason).toBe('end_turn'); + + // String() widening: the pinned Anthropic SDK's stop_reason union + // predates 'refusal', but the adapter emits it for blocked responses. + nextStopReason = 'refusal'; + const refusedMsg = await judge!.create(params); + expect(String(refusedMsg.stop_reason)).toBe('refusal'); + + nextStopReason = 'content_filter'; + const filteredMsg = await judge!.create(params); + expect(String(filteredMsg.stop_reason)).toBe('refusal'); + + // 'other' is the gateway's catch-all for unknown provider finish + // reasons — some non-standard providers report successful stops that + // way, so it must stay a cacheable clean stop. + nextStopReason = 'other'; + const otherMsg = await judge!.create(params); + expect(otherMsg.stop_reason).toBe('end_turn'); + }); + }); }); describe('R3 — parsed-verdict semantic parity (IRON RULE regression)', () => { diff --git a/test/e2e/dream-synthesize-pglite.test.ts b/test/e2e/dream-synthesize-pglite.test.ts index a859f5af2..a0f4f3e4d 100644 --- a/test/e2e/dream-synthesize-pglite.test.ts +++ b/test/e2e/dream-synthesize-pglite.test.ts @@ -12,7 +12,8 @@ * Run: bun test test/e2e/dream-synthesize-pglite.test.ts */ -import { describe, test, expect } from 'bun:test'; +import { describe, test, expect, afterEach } from 'bun:test'; +import { __setChatTransportForTests, resetGateway } from '../../src/core/ai/gateway.ts'; import { mkdtempSync, rmSync, writeFileSync } from 'fs'; import { tmpdir } from 'os'; import { join } from 'path'; @@ -517,6 +518,117 @@ describe('E2E synthesize — verdict cache (Q-2)', () => { }, 30_000); }); +describe('E2E synthesize — degenerate verdicts are NOT cached in dream_verdicts', () => { + // A truncated (stop_reason=length) or unparseable judge response used to be + // banked as a permanent `worth_processing: false` row — a brain whose + // verdict model reliably truncates (e.g. a reasoning model whose reasoning + // tokens ate the old 200-token budget) silently rejected every transcript + // forever. These tests pin the fix: degenerate verdicts skip the cache + // write (with a stderr warning) so the next cycle re-judges. + + afterEach(() => { + __setChatTransportForTests(null); + resetGateway(); + }); + + async function captureStderr(body: () => Promise): Promise<{ result: T; stderr: string }> { + const chunks: string[] = []; + const original = process.stderr.write.bind(process.stderr); + // eslint-disable-next-line @typescript-eslint/no-explicit-any + (process.stderr as any).write = (chunk: any, ..._args: any[]): boolean => { + const s = typeof chunk === 'string' ? chunk : chunk.toString(); + chunks.push(s); + return true; + }; + try { + const result = await body(); + return { result, stderr: chunks.join('') }; + } finally { + // eslint-disable-next-line @typescript-eslint/no-explicit-any + (process.stderr as any).write = original; + } + } + + /** Run body with a fake ANTHROPIC_API_KEY so makeJudgeClient constructs; the + * transport stub means no network call ever happens. */ + async function withFakeAnthropicKey(body: () => Promise): Promise { + const saved = process.env.ANTHROPIC_API_KEY; + process.env.ANTHROPIC_API_KEY = 'sk-test-degenerate-verdict'; + try { + return await body(); + } finally { + if (saved === undefined) delete process.env.ANTHROPIC_API_KEY; + else process.env.ANTHROPIC_API_KEY = saved; + } + } + + async function runWithStubbedJudge(opts: { + text: string; + stopReason: 'end' | 'length'; + }): Promise<{ verdictRow: unknown; stderr: string }> { + 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 filePath = join(rig.corpusDir, '2026-05-01-session.txt'); + const body = 'a meaningful conversation\n'.repeat(200); + writeFileSync(filePath, body); + + __setChatTransportForTests(async () => ({ + text: opts.text, + blocks: [], + stopReason: opts.stopReason, + usage: { input_tokens: 10, output_tokens: 200, cache_read_tokens: 0, cache_creation_tokens: 0 }, + model: 'test:stub', + providerId: 'test', + })); + + const { stderr } = await withFakeAnthropicKey(() => + captureStderr(() => + runPhaseSynthesize(rig.engine, { brainDir: rig.brainDir, dryRun: true }), + ), + ); + + const { createHash } = await import('node:crypto'); + const hash = createHash('sha256').update(body, 'utf8').digest('hex'); + const verdictRow = await rig.engine.getDreamVerdict(filePath, hash); + return { verdictRow, stderr }; + } finally { + await rig.cleanup(); + } + } + + test('truncated judge response (stop_reason=length) → no dream_verdicts row + warning', async () => { + const { verdictRow, stderr } = await runWithStubbedJudge({ + text: '{"worth_process', // reasoning ate the budget; partial JSON + stopReason: 'length', + }); + expect(verdictRow).toBeNull(); + expect(stderr).toMatch(/\[dream\] verdict for 2026-05-01-session was truncated/); + expect(stderr).toMatch(/not caching in dream_verdicts/); + }, 30_000); + + test('unparseable judge response → no dream_verdicts row + warning', async () => { + const { verdictRow, stderr } = await runWithStubbedJudge({ + text: 'not json at all', + stopReason: 'end', + }); + expect(verdictRow).toBeNull(); + expect(stderr).toMatch(/\[dream\] verdict for 2026-05-01-session was unparseable/); + expect(stderr).toMatch(/not caching in dream_verdicts/); + }, 30_000); + + test('control: clean parseable verdict is still cached', async () => { + const { verdictRow, stderr } = await runWithStubbedJudge({ + text: '{"worth_processing": false, "reasons": ["routine ops"]}', + stopReason: 'end', + }); + expect(verdictRow).not.toBeNull(); + expect((verdictRow as { worth_processing: boolean }).worth_processing).toBe(false); + expect(stderr).not.toMatch(/not caching in dream_verdicts/); + }, 30_000); +}); + describe('E2E synthesize — PGLite inline subagent drain (takeover of #2699)', () => { test('drains private subagent queue inline so the parent can observe completion', async () => { const rig = await setupRig();