From 186215dc04baa063207cb0e04505e9866743917e Mon Sep 17 00:00:00 2001 From: marshall Date: Tue, 14 Jul 2026 01:55:59 +0000 Subject: [PATCH] =?UTF-8?q?fix(state):=20record=20per-step=20transcript=20?= =?UTF-8?q?deltas=20=E2=80=94=20cumulative=20response.messages=20duplicate?= =?UTF-8?q?d=20resume=20context=20(#175)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ai@6's onStepFinish exposes event.response.messages as the CUMULATIVE response list for the whole run so far, not a per-step delta (live evidence: per-record counts [2,4,6], record 0 a strict prefix of record 1). All four transcript recording sites wrote it verbatim, so files grew O(N^2) and reconstructTranscript concatenated overlapping records — the #108 resume path (findResumable -> priorHandle) handed the Worker/CI-fix session a duplicated conversation. - recordStepDeltas(recorder): an onStepFinish handler that tracks the count already recorded and slices the new tail, so each record is a true delta. Wired at all four sites (planner/worker/reviewer/ci-fix); the recorder stays a dumb append. - reconstructTranscript de-overlaps records whose leading run equals the accumulated messages, so transcripts written by prior releases (cumulative) reconstruct duplicate-free too; per-step-delta records don't overlap and append whole. - Tests: recordStepDeltas emits [2,2,2] deltas from cumulative [2,4,6] events (+ a no-new-messages step records nothing); reconstruct de-overlaps a pre-fix cumulative transcript to 4 messages, not 6. Part of #196. Shares its root cause with #176 (compaction sizing). --- .../aitm/src/loop/run-loop-adapter.test.ts | 47 +++++++++++++++++++ packages/aitm/src/loop/run-loop-adapter.ts | 35 +++++++++----- .../aitm/src/state/transcript-store.test.ts | 17 +++++++ packages/aitm/src/state/transcript-store.ts | 21 ++++++++- 4 files changed, 107 insertions(+), 13 deletions(-) diff --git a/packages/aitm/src/loop/run-loop-adapter.test.ts b/packages/aitm/src/loop/run-loop-adapter.test.ts index 7dee381..2e04fb5 100644 --- a/packages/aitm/src/loop/run-loop-adapter.test.ts +++ b/packages/aitm/src/loop/run-loop-adapter.test.ts @@ -28,6 +28,7 @@ import { TOOL_SEARCH_TOOL_NAME } from '../mcp/tool-search.ts'; import type { Plan } from '../plan/schema.ts'; import type { PrGroup, RunState } from '../state/schema.ts'; import { StateStore } from '../state/state-store.ts'; +import type { TranscriptRecorder } from '../state/transcript-store.ts'; import type { ReviewerResult } from '../subagents/reviewer.ts'; import type { WorkerDelivery, WorkerResult } from '../subagents/worker.ts'; import { @@ -48,6 +49,7 @@ import { persistRollingContext, planToPrGroups, type RunLoopAdapterSeams, + recordStepDeltas, reminderAgentSystemPrompt, resolvePlannerTools, resolveWorkerTools, @@ -1084,3 +1086,48 @@ test('deferred loading end-to-end: an over-threshold MCP server surfaces name-on ); await mcp.close(); }); + +// ---- recordStepDeltas: per-step transcript deltas from cumulative SDK events (issue #175) ---- + +test('recordStepDeltas: records only the per-step delta from cumulative onStepFinish events (issue #175)', () => { + const recorded: Array<{ count: number; usage: unknown }> = []; + const recorder: TranscriptRecorder = { + step: async (messages, usage) => { + recorded.push({ count: messages.length, usage }); + }, + compaction: async () => {}, + end: async () => {}, + }; + const onStep = recordStepDeltas(recorder); + const m = (i: number) => ({ role: 'assistant' as const, content: `m${i}` }); + // ai@6 hands the callback the CUMULATIVE response list each step: [m0,m1], [m0..m3], [m0..m5]. + onStep({ response: { messages: [m(0), m(1)] }, usage: { totalTokens: 7 } }); + onStep({ response: { messages: [m(0), m(1), m(2), m(3)] } }); + onStep({ response: { messages: [m(0), m(1), m(2), m(3), m(4), m(5)] } }); + assert.deepEqual( + recorded.map((r) => r.count), + [2, 2, 2], + 'per-step deltas, not the cumulative [2, 4, 6]', + ); + assert.equal( + (recorded[0]?.usage as { totalTokens?: number } | undefined)?.totalTokens, + 7, + 'usage forwarded with the first delta', + ); +}); + +test('recordStepDeltas: a step with no new messages records nothing (issue #175)', () => { + let calls = 0; + const recorder: TranscriptRecorder = { + step: async () => { + calls += 1; + }, + compaction: async () => {}, + end: async () => {}, + }; + const onStep = recordStepDeltas(recorder); + const m = (i: number) => ({ role: 'assistant' as const, content: `m${i}` }); + onStep({ response: { messages: [m(0)] } }); + onStep({ response: { messages: [m(0)] } }); // unchanged cumulative → empty delta → no write + assert.equal(calls, 1); +}); diff --git a/packages/aitm/src/loop/run-loop-adapter.ts b/packages/aitm/src/loop/run-loop-adapter.ts index 0f7d836..78041a8 100644 --- a/packages/aitm/src/loop/run-loop-adapter.ts +++ b/packages/aitm/src/loop/run-loop-adapter.ts @@ -207,6 +207,25 @@ async function beginTranscript( } } +// An `onStepFinish` handler that records only the per-step message delta (issue #175). `ai@6` hands +// the callback the CUMULATIVE response-message list each step (per-step lengths [2, 4, 6] on a live +// run), not a delta — recording it verbatim grew transcript files O(N²) and made resume replay a +// duplicated conversation. Tracking the count already recorded and slicing keeps each record a true +// delta. One handler per recorder (fresh closure state); exported for tests. +export function recordStepDeltas( + recorder: TranscriptRecorder, +): (event: { + response: { messages: readonly ModelMessage[] }; + usage?: Parameters[1]; +}) => void { + let recorded = 0; + return (event) => { + const delta = event.response.messages.slice(recorded); + recorded = event.response.messages.length; + if (delta.length > 0) void recorder.step(delta, event.usage); + }; +} + // Apply operator-configured PreToolUse/PostToolUse hooks over a resolved tool record (issue #121), // after the MCP/local partial-fill so both MCP-supplied and local tools are covered. No hooks // configured → the record is returned untouched. Exported for tests. @@ -616,9 +635,7 @@ async function defaultPlanGroups( }), timeout: { stepMs: input.resolved.llmStepTimeoutMs }, ...(plannerUsage ? { onUsage: plannerUsage } : {}), - ...(plannerRecorder - ? { onStepFinish: (event) => plannerRecorder.step(event.response.messages, event.usage) } - : {}), + ...(plannerRecorder ? { onStepFinish: recordStepDeltas(plannerRecorder) } : {}), }); const result = await runPlanner(agent, { goal: input.goal, @@ -767,9 +784,7 @@ export function defaultMakeOrchestrator(ctx: OrchestratorBridgeCtx): WorkLoopOrc ), timeout: stepTimeout, ...(reviewerUsage ? { onUsage: reviewerUsage } : {}), - ...(recorder - ? { onStepFinish: (event) => recorder.step(event.response.messages, event.usage) } - : {}), + ...(recorder ? { onStepFinish: recordStepDeltas(recorder) } : {}), }); const result = await runReviewerSubagent(agent, { pr, @@ -844,9 +859,7 @@ export function defaultMakeOrchestrator(ctx: OrchestratorBridgeCtx): WorkLoopOrc timeout: stepTimeout, ...(providerOptions !== undefined ? { providerOptions } : {}), ...(workerUsage ? { onUsage: workerUsage } : {}), - ...(recorder - ? { onStepFinish: (event) => recorder.step(event.response.messages, event.usage) } - : {}), + ...(recorder ? { onStepFinish: recordStepDeltas(recorder) } : {}), }); const result = await runWorkerSubagent(agent, { group, @@ -932,9 +945,7 @@ export function defaultMakeOrchestrator(ctx: OrchestratorBridgeCtx): WorkLoopOrc ...(ciFixUsage ? { onUsage: ciFixUsage } : {}), ...(input.resolved.formatCommand ? { formatCommand: input.resolved.formatCommand } : {}), ...(input.resolved.verifyCommand ? { verifyCommand: input.resolved.verifyCommand } : {}), - ...(ciRecorder - ? { onStepFinish: (event) => ciRecorder.step(event.response.messages, event.usage) } - : {}), + ...(ciRecorder ? { onStepFinish: recordStepDeltas(ciRecorder) } : {}), ...(ciResume ? { resumeMessages: ciResume } : {}), }, group, diff --git a/packages/aitm/src/state/transcript-store.test.ts b/packages/aitm/src/state/transcript-store.test.ts index 880b28b..5cc55aa 100644 --- a/packages/aitm/src/state/transcript-store.test.ts +++ b/packages/aitm/src/state/transcript-store.test.ts @@ -61,6 +61,23 @@ test('reconstructTranscript skips unknown kinds (forward compatible)', () => { assert.deepEqual(reconstructTranscript(raw).messages, [msg('assistant', 'a')]); }); +test('reconstructTranscript: de-overlaps a pre-#175 cumulative transcript (no duplicated context)', () => { + // Files written before #175 stored step.messages as the CUMULATIVE response list: each record + // re-includes everything so far ([a,b], then [a,b,c,d]). Naive concatenation duplicated the prefix. + const a = msg('user', 'goal'); + const b = msg('assistant', 'a1'); + const c = msg('tool', 'r1'); + const d = msg('assistant', 'a2'); + const raw = [ + JSON.stringify({ kind: 'step', ts: 't1', messages: [a, b] }), + JSON.stringify({ kind: 'step', ts: 't2', messages: [a, b, c, d] }), + JSON.stringify({ kind: 'run-end', ts: 't3', outcome: 'submitted' }), + ].join('\n'); + const { messages, complete } = reconstructTranscript(raw); + assert.deepEqual(messages, [a, b, c, d], 'the overlapping prefix is dropped — 4 messages, not 6'); + assert.equal(complete, true); +}); + // --- TranscriptStore (fs) --- test('append/replay round-trip: a recorded run reconstructs to the message array the agent last held', async () => { diff --git a/packages/aitm/src/state/transcript-store.ts b/packages/aitm/src/state/transcript-store.ts index ea29097..873e836 100644 --- a/packages/aitm/src/state/transcript-store.ts +++ b/packages/aitm/src/state/transcript-store.ts @@ -53,10 +53,26 @@ function ordinalOf(file: string, prefix: string): number | null { return m ? Number(m[1]) : null; } +// True when `prefix` is a leading run of `arr` (deep-equal, element by element with an early exit). +// Used to drop the cumulative overlap in pre-#175 transcripts during reconstruction. +function isLeadingRun(arr: readonly ModelMessage[], prefix: readonly ModelMessage[]): boolean { + if (prefix.length === 0) return true; + if (arr.length < prefix.length) return false; + for (let i = 0; i < prefix.length; i++) { + if (JSON.stringify(arr[i]) !== JSON.stringify(prefix[i])) return false; + } + return true; +} + // Reconstruct the message array the agent would see next from a transcript's raw JSONL, plus whether // the run completed (a `run-end` record present). `step` messages concatenate; a `compaction` record // replaces the accumulated array (its summary subsumes earlier steps), and steps after it append. A // corrupt/truncated line is skipped with a warning; unknown `kind`s are ignored (forward-compatible). +// +// Transcripts written before issue #175 stored `step.messages` as the CUMULATIVE response list per +// step (each record re-includes the whole conversation), so a record whose leading run equals the +// accumulated messages is de-overlapped to just its new suffix. Post-fix (per-step-delta) records +// don't overlap and append whole — the same rule handles both, keeping resume context duplicate-free. export function reconstructTranscript( raw: string, onWarn: (message: string) => void = () => {}, @@ -74,7 +90,10 @@ export function reconstructTranscript( } if (!isRecord(record)) continue; if (record.kind === 'step' && Array.isArray(record.messages)) { - messages.push(...(record.messages as ModelMessage[])); + const recorded = record.messages as ModelMessage[]; + messages.push( + ...(isLeadingRun(recorded, messages) ? recorded.slice(messages.length) : recorded), + ); } else if (record.kind === 'compaction' && Array.isArray(record.messages)) { messages = [...(record.messages as ModelMessage[])]; } else if (record.kind === 'run-end') {