fix(state): transcript step records duplicate the prior conversation — record per-step deltas (#175)

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);
+});

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<TranscriptRecorder['step']>[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,

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 () => {

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') {