fix: checkpoint every usage update with incremental protocol parsers

Use provider terminal evidence for completeness; an OpenCode successful EOF closes its step-based stream. Preserve immutable snapshots and propagate partial completeness through final results.

Co-Authored-By: Paperclip <noreply@paperclip.ing>

packages/adapter-utils/src/usage-checkpoint.test.ts

@@ -3,20 +3,24 @@ import { createUsageCheckpointLog } from "./usage-checkpoint.js";
 import { runChildProcess } from "./server-utils.js";
 import type { AdapterUsageCheckpoint } from "./types.js";
 
-const parse = (stdout: string): AdapterUsageCheckpoint => {
-  const events = stdout.trim().split("\n").map(line => JSON.parse(line));
-  return { costUsd: events.reduce((sum, event) => sum + event.usage.costUsd, 0), complete: false };
+const createParser = () => {
+  let costUsd = 0;
+  return (stdout: string): AdapterUsageCheckpoint => {
+    const events = stdout.trim().split("\n").map(line => JSON.parse(line));
+    costUsd += events.reduce((sum, event) => sum + event.usage.costUsd, 0);
+    return { costUsd, complete: events.some(event => event.type === "result") };
+  };
 };
 const event = JSON.stringify({ type: "usage", usage: { inputTokens: 2, costUsd: 0.25 }, text: "private content" });
 
 describe("usage checkpoint stream", () => {
   it("flushes an unterminated Pi price and closes each attempt independently", async () => {
-    const saved = vi.fn(); const parser = vi.fn(parse);
+    const saved = vi.fn(); const parser = vi.fn(createParser());
     const first = createUsageCheckpointLog(vi.fn(), saved, parser);
     await first("stdout", event);
     expect(saved).not.toHaveBeenCalled();
     await first.flush({ complete: true });
-    const second = createUsageCheckpointLog(vi.fn(), saved, parse);
+    const second = createUsageCheckpointLog(vi.fn(), saved, createParser());
     await second("stdout", event + "\n");
     await second.flush({ complete: true });
     expect(saved.mock.calls[0][0]).toMatchObject({ costUsd: 0.25, complete: true });
@@ -28,7 +32,7 @@ describe("usage checkpoint stream", () => {
   it("surfaces a persistence failure outside the local process's best-effort log path", async () => {
     const failure = new Error("Receipt storage unavailable");
     const output = vi.fn(); const saved = vi.fn().mockRejectedValue(failure);
-    const log = createUsageCheckpointLog(output, saved, parse);
+    const log = createUsageCheckpointLog(output, saved, createParser());
     const result = await runChildProcess("checkpoint-test", process.execPath, ["-e", `process.stdout.write(${JSON.stringify(event + "\n")})`], {
       cwd: process.cwd(), env: {}, timeoutSec: 5, graceSec: 1, onLog: log,
     });
@@ -38,19 +42,32 @@ describe("usage checkpoint stream", () => {
   });
 
   it("does not promote partial usage to complete merely because the stream was flushed", async () => {
-    const saved = vi.fn(); const log = createUsageCheckpointLog(vi.fn(), saved, parse);
+    const saved = vi.fn(); const log = createUsageCheckpointLog(vi.fn(), saved, createParser());
     await log("stdout", event); await log.flush();
     expect(saved.mock.calls.at(-1)![0]).toMatchObject({ costUsd: 0.25, complete: false });
   });
 
-  it("bounds replay work by stream size while retaining the exact final total", async () => {
-    const parser = vi.fn(parse); const saved = vi.fn();
+  it("persists each changed total and the final receipt before publishing its chunk", async () => {
+    const order: string[] = [];
+    const saved = vi.fn(async (receipt: AdapterUsageCheckpoint) => { order.push(`saved:${receipt.costUsd}:${receipt.complete}`); });
+    const log = createUsageCheckpointLog(async () => { order.push("log"); }, saved, createParser());
+    await log("stdout", event + "\n");
+    await log("stdout", event + "\n");
+    await log("stdout", JSON.stringify({ type: "result", usage: { costUsd: 0 } }) + "\n");
+    expect(order).toEqual(["saved:0.25:false", "log", "saved:0.5:false", "log", "saved:0.5:true", "log"]);
+  });
+
+  it("saves every new receipt before logging with linear parsing work", async () => {
+    const parser = vi.fn(createParser()); const saved = vi.fn();
     const log = createUsageCheckpointLog(vi.fn(), saved, parser);
-    for (let i = 0; i < 1024; i++) await log("stdout", event + "\n");
+    for (let i = 0; i < 1024; i++) {
+      await log("stdout", event + "\n");
+      expect(saved.mock.calls.at(-1)![0].costUsd).toBe((i + 1) * 0.25);
+    }
     await log.flush({ complete: true });
     expect(saved.mock.calls.at(-1)![0]).toMatchObject({ costUsd: 256, complete: true });
-    expect(parser.mock.calls.length).toBeLessThanOrEqual(12);
-    const finalSize = parser.mock.calls.at(-1)![0].length;
-    expect(parser.mock.calls.reduce((sum, [input]) => sum + input.length, 0)).toBeLessThan(3 * finalSize);
+    expect(saved).toHaveBeenCalledTimes(1025); // all updates plus completion
+    expect(parser).toHaveBeenCalledTimes(1024);
+    expect(parser.mock.calls.reduce((sum, [input]) => sum + input.length, 0)).toBeLessThanOrEqual(1024 * (event.length + 1));
   });
 });

packages/adapter-utils/src/usage-checkpoint.ts

@@ -24,17 +24,18 @@ function project(value: unknown, depth = 0, field = ""): unknown {
       : project(entry, depth + 1, key)]));
 }
 
-/** A fresh instance belongs to one CLI attempt. Call flush() after its process
+/** The consumer incrementally parses only new records (never the full history).
+ * A fresh instance belongs to one CLI attempt. Call flush() after its process
  * exits, before retrying or final result handling. Checkpoint failures are
  * retained outside runChildProcess's best-effort onLog error handler. */
 export function createUsageCheckpointLog(
   onLog: (stream: "stdout" | "stderr", chunk: string) => Promise<void>,
   onUsage: ((receipt: AdapterUsageCheckpoint) => Promise<void>) | undefined,
-  parse: (stdout: string) => AdapterUsageCheckpoint | null,
+  consume: (records: string) => AdapterUsageCheckpoint | null,
 ) {
   const attemptId = randomUUID();
   let remainder = "", accounting = "", previous = "";
-  let nextReplaySize = 0;
+  let snapshot: AdapterUsageCheckpoint | null = null;
   let failure: unknown;
   let failed = false;
   function retain(line: string) {
@@ -43,14 +44,15 @@ export function createUsageCheckpointLog(
     const compact = JSON.stringify(project(raw));
     if (!compact || !/usage|tokens|cost|"result"|"turn.completed"|"agent_end"|"model"|"modelID"/.test(compact)) return;
     accounting += compact + "\n";
-    if (accounting.length > 8 * 1024 * 1024) throw new Error("Accounting checkpoint history exceeds 8 MiB");
+    if (accounting.length > 8 * 1024 * 1024) throw new Error("Accounting checkpoint chunk exceeds 8 MiB");
   }
-  async function publish(final = false, complete = false) {
-    if (!onUsage || !accounting || (!final && accounting.length < nextReplaySize)) return;
-    const parsed = parse(accounting);
-    // Geometric replay bounds total parsing work by stream size. The first
-    // observation is saved immediately; flush always saves the final snapshot.
-    nextReplaySize = accounting.length * 2;
+  async function publish(complete = false) {
+    if (!onUsage) return;
+    if (accounting) {
+      snapshot = consume(accounting);
+      accounting = "";
+    }
+    const parsed = snapshot;
     if (!parsed) return;
     if (!parsed.complete && parsed.costUsd == null && parsed.costUsdExact == null &&
       !Object.values(parsed.usage ?? {}).some(value => typeof value === "number" && value > 0)) return;
@@ -79,7 +81,7 @@ export function createUsageCheckpointLog(
       if (failed) throw failure;
       if (!onUsage) return;
       if (remainder) { retain(remainder); remainder = ""; }
-      await publish(true, options.complete ?? false);
+      await publish(options.complete ?? false);
     },
   });
 }

packages/adapters/claude-local/src/server/execute.ts

@@ -65,7 +65,7 @@ import {
 import {
   claudeModelUsageTotals,
   claudeModelReceipts,
-  parseClaudeStreamJson,
+  parseClaudeStreamJson, createClaudeStreamParser,
   describeClaudeFailure,
   detectClaudeLoginRequired,
   extractClaudeRetryNotBefore,
@@ -964,8 +964,9 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
       });
     }
 
+    const consumeAccounting = createClaudeStreamParser();
     const accountingLog = createUsageCheckpointLog(onLog, ctx.onUsage, stdout => {
-      const parsed = parseClaudeStreamJson(stdout);
+      const parsed = consumeAccounting(stdout);
       return { usage: parsed.usage ?? undefined, usageBasis: "per_run", costUsd: parsed.costUsd,
         provider: "anthropic", biller: isBedrockAuth(effectiveEnv) ? "aws_bedrock" : "anthropic",
         billingType, model: Object.keys(parseObject(parsed.resultJson?.modelUsage)).length > 1 ? "mixed" : parsed.model || model,
@@ -989,7 +990,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
       },
       localProcessSandbox,
     });
-    await accountingLog.flush({ complete: !proc.timedOut && !proc.signal });
+    await accountingLog.flush();
 
     const parsedStream = parseClaudeStreamJson(proc.stdout);
     const parsed = parsedStream.resultJson ?? parseJson(proc.stdout);
@@ -1092,6 +1093,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
         exitCode: proc.exitCode,
         signal: proc.signal,
         timedOut: false,
+        usageComplete: parsedStream.resultJson !== null,
         usage: parsedStream.usage ?? undefined,
         usageBasis: "per_run",
         provider: "anthropic",
@@ -1262,6 +1264,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
       exitCode: proc.exitCode,
       signal: proc.signal,
       timedOut: false,
+      usageComplete: parsedStream.resultJson !== null || parsed.type === "result",
       errorMessage,
       errorCode: resolvedErrorCode,
       errorFamily,

packages/adapters/claude-local/src/server/parse.ts

@@ -67,93 +67,102 @@ export function claudeModelReceipts(modelUsage: unknown) {
 }
 
 export function parseClaudeStreamJson(stdout: string) {
+  return createClaudeStreamParser()(stdout);
+}
+
+/** Consume complete JSONL records once, retaining protocol accounting state. */
+export function createClaudeStreamParser() {
   let sessionId: string | null = null;
   let model = "";
   let finalResult: Record<string, unknown> | null = null;
   const assistantTexts: string[] = [];
   const messageUsage = new Map<string, UsageSummary>();
   let anonymousMessage = 0;
+  const observedTotals = { inputTokens: 0, outputTokens: 0, cachedInputTokens: 0 };
 
-  for (const rawLine of stdout.split(/\r?\n/)) {
-    const line = rawLine.trim();
-    if (!line) continue;
-    const event = parseJson(line);
-    if (!event) continue;
+  return (stdout: string) => {
+    for (const rawLine of stdout.split(/\r?\n/)) {
+      const line = rawLine.trim();
+      if (!line) continue;
+      const event = parseJson(line);
+      if (!event) continue;
 
-    const type = asString(event.type, "");
-    if (type === "system" && asString(event.subtype, "") === "init") {
-      sessionId = asString(event.session_id, sessionId ?? "") || sessionId;
-      model = asString(event.model, model);
-      continue;
-    }
-
-    if (type === "assistant") {
-      sessionId = asString(event.session_id, sessionId ?? "") || sessionId;
-      const message = parseObject(event.message);
-      const observed = parseObject(message.usage);
-      if (Object.keys(observed).length > 0) {
-        const key = asString(message.id, "") || `anonymous:${anonymousMessage++}`;
-        const previous = messageUsage.get(key);
-        messageUsage.set(key, {
-          inputTokens: Math.max(previous?.inputTokens ?? 0, asNumber(observed.input_tokens, 0) + asNumber(observed.cache_creation_input_tokens, 0)),
-          cachedInputTokens: Math.max(previous?.cachedInputTokens ?? 0, asNumber(observed.cache_read_input_tokens, 0)),
-          outputTokens: Math.max(previous?.outputTokens ?? 0, asNumber(observed.output_tokens, 0)),
-        });
+      const type = asString(event.type, "");
+      if (type === "system" && asString(event.subtype, "") === "init") {
+        sessionId = asString(event.session_id, sessionId ?? "") || sessionId;
+        model = asString(event.model, model);
+        continue;
       }
-      const content = Array.isArray(message.content) ? message.content : [];
-      for (const entry of content) {
-        if (typeof entry !== "object" || entry === null || Array.isArray(entry)) continue;
-        const block = entry as Record<string, unknown>;
-        if (asString(block.type, "") === "text") {
-          const text = asString(block.text, "");
-          if (text) assistantTexts.push(text);
+
+      if (type === "assistant") {
+        sessionId = asString(event.session_id, sessionId ?? "") || sessionId;
+        const message = parseObject(event.message);
+        const observed = parseObject(message.usage);
+        if (Object.keys(observed).length > 0) {
+          const key = asString(message.id, "") || `anonymous:${anonymousMessage++}`;
+          const previous = messageUsage.get(key);
+          const next = {
+            inputTokens: Math.max(previous?.inputTokens ?? 0, asNumber(observed.input_tokens, 0) + asNumber(observed.cache_creation_input_tokens, 0)),
+            cachedInputTokens: Math.max(previous?.cachedInputTokens ?? 0, asNumber(observed.cache_read_input_tokens, 0)),
+            outputTokens: Math.max(previous?.outputTokens ?? 0, asNumber(observed.output_tokens, 0)),
+          };
+          observedTotals.inputTokens += next.inputTokens - (previous?.inputTokens ?? 0);
+          observedTotals.outputTokens += next.outputTokens - (previous?.outputTokens ?? 0);
+          observedTotals.cachedInputTokens += next.cachedInputTokens - (previous?.cachedInputTokens ?? 0);
+          messageUsage.set(key, next);
         }
+        const content = Array.isArray(message.content) ? message.content : [];
+        for (const entry of content) {
+          if (typeof entry !== "object" || entry === null || Array.isArray(entry)) continue;
+          const block = entry as Record<string, unknown>;
+          if (asString(block.type, "") === "text") {
+            const text = asString(block.text, "");
+            if (text) assistantTexts.push(text);
+          }
+        }
+        continue;
+      }
+
+      if (type === "result") {
+        finalResult = event;
+        sessionId = asString(event.session_id, sessionId ?? "") || sessionId;
       }
-      continue;
     }
 
-    if (type === "result") {
-      finalResult = event;
-      sessionId = asString(event.session_id, sessionId ?? "") || sessionId;
+    if (!finalResult) {
+      return {
+        sessionId,
+        model,
+        costUsd: null as number | null,
+        usage: messageUsage.size ? { ...observedTotals } : null,
+        usageBasis: "per_run" as const,
+        summary: assistantTexts.join("\n\n").trim(),
+        resultJson: null as Record<string, unknown> | null,
+      };
     }
-  }
 
-  if (!finalResult) {
+    const modelUsageTotals = claudeModelUsageTotals(finalResult.modelUsage);
+    const usageObj = parseObject(finalResult.usage);
+    const usage: UsageSummary = modelUsageTotals ?? {
+      inputTokens: asNumber(usageObj.input_tokens, 0) + asNumber(usageObj.cache_creation_input_tokens, 0),
+      cachedInputTokens: asNumber(usageObj.cache_read_input_tokens, 0),
+      outputTokens: asNumber(usageObj.output_tokens, 0),
+    };
+    const costRaw = finalResult.total_cost_usd;
+    const costUsd = typeof costRaw === "number" && Number.isFinite(costRaw) ? costRaw : null;
+    const summary = asString(finalResult.result, assistantTexts.join("\n\n")).trim();
+
     return {
       sessionId,
       model,
-      costUsd: null as number | null,
-      usage: messageUsage.size ? [...messageUsage.values()].reduce<UsageSummary>((total, part) => ({
-        inputTokens: total.inputTokens + part.inputTokens, outputTokens: total.outputTokens + part.outputTokens,
-        cachedInputTokens: (total.cachedInputTokens ?? 0) + (part.cachedInputTokens ?? 0),
-      }), { inputTokens: 0, outputTokens: 0, cachedInputTokens: 0 }) : null,
+      costUsd,
+      usage,
+      // modelUsage covers exactly this CLI invocation, so mark it per-run to
+      // keep the server from applying its session-cumulative delta heuristic.
       usageBasis: "per_run" as const,
-      summary: assistantTexts.join("\n\n").trim(),
-      resultJson: null as Record<string, unknown> | null,
+      summary,
+      resultJson: finalResult,
     };
-  }
-
-  const modelUsageTotals = claudeModelUsageTotals(finalResult.modelUsage);
-  const usageObj = parseObject(finalResult.usage);
-  const usage: UsageSummary = modelUsageTotals ?? {
-    inputTokens: asNumber(usageObj.input_tokens, 0) + asNumber(usageObj.cache_creation_input_tokens, 0),
-    cachedInputTokens: asNumber(usageObj.cache_read_input_tokens, 0),
-    outputTokens: asNumber(usageObj.output_tokens, 0),
-  };
-  const costRaw = finalResult.total_cost_usd;
-  const costUsd = typeof costRaw === "number" && Number.isFinite(costRaw) ? costRaw : null;
-  const summary = asString(finalResult.result, assistantTexts.join("\n\n")).trim();
-
-  return {
-    sessionId,
-    model,
-    costUsd,
-    usage,
-    // modelUsage covers exactly this CLI invocation, so mark it per-run to
-    // keep the server from applying its session-cumulative delta heuristic.
-    usageBasis: "per_run" as const,
-    summary,
-    resultJson: finalResult,
   };
 }
 

packages/adapters/codex-local/src/server/execute.ts

@@ -62,7 +62,7 @@ import {
   type LocalProcessSandboxOptions,
 } from "@paperclipai/adapter-utils/local-process-sandbox";
 import {
-  parseCodexJsonl,
+  parseCodexJsonl, createCodexJsonlParser,
   classifyCodexAuthRefreshFailure,
   extractCodexRetryNotBefore,
   isCodexHarnessCrash,
@@ -1319,8 +1319,9 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
         }
       };
 
+      const consumeAccounting = createCodexJsonlParser();
       const accountingLog = createUsageCheckpointLog(onLog, ctx.onUsage, stdout => {
-        const parsed = parseCodexJsonl(stdout);
+        const parsed = consumeAccounting(stdout);
         return { usage: parsed.usage, usageBasis: "per_run", provider: "openai", biller: resolveCodexBiller(effectiveEnv, billingType), billingType, model, costUsd: null, complete: parsed.sawProtocolTerminalEvent };
       });
       try {
@@ -1347,7 +1348,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
           settleRunDisposition: paperclipBridge?.settleRunDisposition,
           localProcessSandbox,
         });
-        await accountingLog.flush({ complete: !proc.timedOut && !proc.signal });
+        await accountingLog.flush();
         const cleanedStderr = stripCodexRolloutNoise(proc.stderr);
         return {
           proc: {
@@ -1397,6 +1398,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
           exitCode: null,
           signal: attempt.monitor.terminationSignal ?? attempt.proc.signal,
           timedOut: false,
+          usageComplete: attempt.parsed.sawProtocolTerminalEvent,
           errorMessage,
           errorCode: "codex_output_inactivity_monitor",
           errorFamily: null,
@@ -1429,7 +1431,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
           exitCode: attempt.proc.exitCode,
           signal: attempt.proc.signal,
           timedOut: true,
-        usageComplete: attempt.parsed.sawProtocolTerminalEvent,
+          usageComplete: attempt.parsed.sawProtocolTerminalEvent,
           usage: attempt.parsed.usage,
           usageBasis: "per_run",
           provider: "openai",
@@ -1516,6 +1518,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
         exitCode: attempt.proc.exitCode,
         signal: attempt.proc.signal,
         timedOut: false,
+        usageComplete: attempt.parsed.sawProtocolTerminalEvent,
         errorMessage:
           (attempt.proc.exitCode ?? 0) === 0
             ? null

packages/adapters/codex-local/src/server/parse.ts

@@ -28,6 +28,11 @@ export type CodexAuthRefreshFailureClass =
   | "refresh_token_invalidated";
 
 export function parseCodexJsonl(stdout: string) {
+  return createCodexJsonlParser()(stdout);
+}
+
+/** Consume complete JSONL records once, retaining protocol accounting state. */
+export function createCodexJsonlParser() {
   let sessionId: string | null = null;
   let finalMessage: string | null = null;
   let errorMessage: string | null = null;
@@ -39,62 +44,64 @@ export function parseCodexJsonl(stdout: string) {
     outputTokens: 0,
   };
 
-  for (const rawLine of stdout.split(/\r?\n/)) {
-    const line = rawLine.trim();
-    if (!line) continue;
+  return (stdout: string) => {
+    for (const rawLine of stdout.split(/\r?\n/)) {
+      const line = rawLine.trim();
+      if (!line) continue;
 
-    const event = parseJson(line);
-    if (!event) continue;
+      const event = parseJson(line);
+      if (!event) continue;
 
-    const type = asString(event.type, "");
-    if (type) sawProtocolEvent = true;
-    if (type === "error" || type === "turn.completed" || type === "turn.failed") {
-      sawProtocolTerminalEvent = true;
-    }
-    if (type === "thread.started") {
-      sessionId = asString(event.thread_id, sessionId ?? "") || sessionId;
-      continue;
-    }
+      const type = asString(event.type, "");
+      if (type) sawProtocolEvent = true;
+      if (type === "error" || type === "turn.completed" || type === "turn.failed") {
+        sawProtocolTerminalEvent = true;
+      }
+      if (type === "thread.started") {
+        sessionId = asString(event.thread_id, sessionId ?? "") || sessionId;
+        continue;
+      }
 
-    if (type === "error") {
-      const msg = asString(event.message, "").trim();
-      if (msg) errorMessage = msg;
-      continue;
-    }
+      if (type === "error") {
+        const msg = asString(event.message, "").trim();
+        if (msg) errorMessage = msg;
+        continue;
+      }
 
-    if (type === "item.completed") {
-      const item = parseObject(event.item);
-      if (asString(item.type, "") === "agent_message") {
-        const text = asString(item.text, "");
-        if (text) finalMessage = text;
+      if (type === "item.completed") {
+        const item = parseObject(event.item);
+        if (asString(item.type, "") === "agent_message") {
+          const text = asString(item.text, "");
+          if (text) finalMessage = text;
+        }
+        continue;
+      }
+
+      if (type === "turn.completed") {
+        const usageObj = parseObject(event.usage);
+        usage.inputTokens = asNumber(usageObj.input_tokens, usage.inputTokens);
+        usage.cachedInputTokens = asNumber(usageObj.cached_input_tokens, usage.cachedInputTokens);
+        usage.outputTokens = asNumber(usageObj.output_tokens, usage.outputTokens);
+        continue;
+      }
+
+      if (type === "turn.failed") {
+        const err = parseObject(event.error);
+        const msg = asString(err.message, "").trim();
+        if (msg) errorMessage = msg;
       }
-      continue;
     }
 
-    if (type === "turn.completed") {
-      const usageObj = parseObject(event.usage);
-      usage.inputTokens = asNumber(usageObj.input_tokens, usage.inputTokens);
-      usage.cachedInputTokens = asNumber(usageObj.cached_input_tokens, usage.cachedInputTokens);
-      usage.outputTokens = asNumber(usageObj.output_tokens, usage.outputTokens);
-      continue;
-    }
-
-    if (type === "turn.failed") {
-      const err = parseObject(event.error);
-      const msg = asString(err.message, "").trim();
-      if (msg) errorMessage = msg;
-    }
-  }
-
-  return {
-    sessionId,
-    summary: finalMessage?.trim() ?? "",
-    // Codex includes cache hits in input_tokens; Paperclip stores them separately.
-    usage: { ...usage, inputTokens: Math.max(0, usage.inputTokens - usage.cachedInputTokens) },
-    usageBasis: "per_run" as const,
-    errorMessage,
-    sawProtocolEvent,
-    sawProtocolTerminalEvent,
+    return {
+      sessionId,
+      summary: finalMessage?.trim() ?? "",
+      // Codex includes cache hits in input_tokens; Paperclip stores them separately.
+      usage: { ...usage, inputTokens: Math.max(0, usage.inputTokens - usage.cachedInputTokens) },
+      usageBasis: "per_run" as const,
+      errorMessage,
+      sawProtocolEvent,
+      sawProtocolTerminalEvent,
+    };
   };
 }
 

packages/adapters/cursor-local/src/server/execute.ts

@@ -54,7 +54,7 @@ import {
   joinPromptSections,
 } from "@paperclipai/adapter-utils/server-utils";
 import { DEFAULT_CURSOR_LOCAL_MODEL, SANDBOX_INSTALL_COMMAND } from "../index.js";
-import { firstCursorDiagnosticLine, parseCursorJsonl, isCursorUnknownSessionError } from "./parse.js";
+import { firstCursorDiagnosticLine, parseCursorJsonl, createCursorJsonlParser, isCursorUnknownSessionError } from "./parse.js";
 import { prepareCursorSandboxCommand } from "./remote-command.js";
 import { normalizeCursorStreamLine } from "../shared/stream.js";
 import { hasCursorTrustBypassArg } from "../shared/trust.js";
@@ -627,8 +627,9 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
       });
     }
 
+    const consumeAccounting = createCursorJsonlParser();
     const accountingLog = createUsageCheckpointLog(onLog, ctx.onUsage, stdout => {
-      const parsed = parseCursorJsonl(stdout);
+      const parsed = consumeAccounting(stdout);
       const provider = resolveProviderFromModel(model);
       return { usage: parsed.usage, costUsd: parsed.costUsd, usageBasis: "per_run", provider, biller: resolveCursorBiller(effectiveEnv, billingType, provider), billingType, model, complete: parsed.sawResult };
     });
@@ -676,7 +677,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
       settleRunDisposition: paperclipBridge?.settleRunDisposition,
     });
     await flushStdoutChunk("", true);
-    await accountingLog.flush({ complete: !proc.timedOut && !proc.signal });
+    await accountingLog.flush();
 
     return {
       proc,
@@ -745,6 +746,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
       exitCode: attempt.proc.exitCode,
       signal: attempt.proc.signal,
       timedOut: false,
+      usageComplete: attempt.parsed.sawResult,
       usageBasis: "per_run",
       errorMessage:
         (attempt.proc.exitCode ?? 0) === 0

packages/adapters/cursor-local/src/server/parse.ts

@@ -61,6 +61,11 @@ function readSessionId(event: Record<string, unknown>): string | null {
 }
 
 export function parseCursorJsonl(stdout: string) {
+  return createCursorJsonlParser()(stdout);
+}
+
+/** Consume complete JSONL records once, retaining protocol accounting state. */
+export function createCursorJsonlParser() {
   let sessionId: string | null = null;
   let sawResult = false;
   const messages: string[] = [];
@@ -80,94 +85,96 @@ export function parseCursorJsonl(stdout: string) {
     outputTokens: 0,
   };
 
-  for (const rawLine of stdout.split(/\r?\n/)) {
-    const line = normalizeCursorStreamLine(rawLine).line;
-    if (!line) continue;
+  return (stdout: string) => {
+    for (const rawLine of stdout.split(/\r?\n/)) {
+      const line = normalizeCursorStreamLine(rawLine).line;
+      if (!line) continue;
 
-    const event = parseJson(line);
-    if (!event) continue;
+      const event = parseJson(line);
+      if (!event) continue;
 
-    const foundSession = readSessionId(event);
-    if (foundSession) sessionId = foundSession;
+      const foundSession = readSessionId(event);
+      if (foundSession) sessionId = foundSession;
 
-    const type = asString(event.type, "").trim();
+      const type = asString(event.type, "").trim();
 
-    if (type === "assistant") {
-      messages.push(...collectAssistantText(event.message));
-      continue;
-    }
-
-    if (type === "result") {
-      sawResult = true;
-      const usageObj = parseObject(event.usage);
-      usage.inputTokens += asNumber(
-        usageObj.input_tokens,
-        asNumber(usageObj.inputTokens, 0),
-      );
-      usage.cachedInputTokens += asNumber(
-        usageObj.cached_input_tokens,
-        asNumber(usageObj.cachedInputTokens, asNumber(usageObj.cache_read_input_tokens, 0)),
-      );
-      usage.outputTokens += asNumber(
-        usageObj.output_tokens,
-        asNumber(usageObj.outputTokens, 0),
-      );
-      addCost(event.total_cost_usd ?? event.cost_usd ?? event.cost);
-
-      const isError = event.is_error === true || asString(event.subtype, "").toLowerCase() === "error";
-      const resultText = asString(event.result, "").trim();
-      if (resultText && messages.length === 0) {
-        messages.push(resultText);
+      if (type === "assistant") {
+        messages.push(...collectAssistantText(event.message));
+        continue;
       }
-      if (isError) {
-        const resultError = asErrorText(event.error ?? event.message ?? event.result).trim();
-        if (resultError) errorMessage = resultError;
+
+      if (type === "result") {
+        sawResult = true;
+        const usageObj = parseObject(event.usage);
+        usage.inputTokens += asNumber(
+          usageObj.input_tokens,
+          asNumber(usageObj.inputTokens, 0),
+        );
+        usage.cachedInputTokens += asNumber(
+          usageObj.cached_input_tokens,
+          asNumber(usageObj.cachedInputTokens, asNumber(usageObj.cache_read_input_tokens, 0)),
+        );
+        usage.outputTokens += asNumber(
+          usageObj.output_tokens,
+          asNumber(usageObj.outputTokens, 0),
+        );
+        addCost(event.total_cost_usd ?? event.cost_usd ?? event.cost);
+
+        const isError = event.is_error === true || asString(event.subtype, "").toLowerCase() === "error";
+        const resultText = asString(event.result, "").trim();
+        if (resultText && messages.length === 0) {
+          messages.push(resultText);
+        }
+        if (isError) {
+          const resultError = asErrorText(event.error ?? event.message ?? event.result).trim();
+          if (resultError) errorMessage = resultError;
+        }
+        continue;
       }
-      continue;
-    }
 
-    if (type === "error") {
-      const message = asErrorText(event.message ?? event.error ?? event.detail).trim();
-      if (message) errorMessage = message;
-      continue;
-    }
-
-    if (type === "system") {
-      const subtype = asString(event.subtype, "").trim().toLowerCase();
-      if (subtype === "error") {
+      if (type === "error") {
         const message = asErrorText(event.message ?? event.error ?? event.detail).trim();
         if (message) errorMessage = message;
+        continue;
+      }
+
+      if (type === "system") {
+        const subtype = asString(event.subtype, "").trim().toLowerCase();
+        if (subtype === "error") {
+          const message = asErrorText(event.message ?? event.error ?? event.detail).trim();
+          if (message) errorMessage = message;
+        }
+        continue;
+      }
+
+      // Compatibility with older stream-json shapes.
+      if (type === "text") {
+        const part = parseObject(event.part);
+        const text = asString(part.text, "").trim();
+        if (text) messages.push(text);
+        continue;
+      }
+
+      if (type === "step_finish") {
+        const part = parseObject(event.part);
+        const tokens = parseObject(part.tokens);
+        const cache = parseObject(tokens.cache);
+        usage.inputTokens += asNumber(tokens.input, 0) + asNumber(cache.write, 0);
+        usage.cachedInputTokens += asNumber(cache.read, 0);
+        usage.outputTokens += asNumber(tokens.output, 0);
+        addCost(part.cost);
+        continue;
       }
-      continue;
     }
 
-    // Compatibility with older stream-json shapes.
-    if (type === "text") {
-      const part = parseObject(event.part);
-      const text = asString(part.text, "").trim();
-      if (text) messages.push(text);
-      continue;
-    }
-
-    if (type === "step_finish") {
-      const part = parseObject(event.part);
-      const tokens = parseObject(part.tokens);
-      const cache = parseObject(tokens.cache);
-      usage.inputTokens += asNumber(tokens.input, 0) + asNumber(cache.write, 0);
-      usage.cachedInputTokens += asNumber(cache.read, 0);
-      usage.outputTokens += asNumber(tokens.output, 0);
-      addCost(part.cost);
-      continue;
-    }
-  }
-
-  return {
-    sawResult,
-    sessionId,
-    summary: messages.join("\n\n").trim(),
-    usage,
-    costUsd: reportedCost && !missingCost ? totalCostUsd : null,
-    errorMessage,
+    return {
+      sawResult,
+      sessionId,
+      summary: messages.join("\n\n").trim(),
+      usage: { ...usage },
+      costUsd: reportedCost && !missingCost ? totalCostUsd : null,
+      errorMessage,
+    };
   };
 }
 

packages/adapters/gemini-local/src/server/execute.ts

@@ -63,7 +63,7 @@ import {
   isGeminiTransientNetworkError,
   isGeminiTurnLimitResult,
   isGeminiSessionUnrecoverableError,
-  parseGeminiJsonl,
+  parseGeminiJsonl, createGeminiJsonlParser,
 } from "./parse.js";
 import { firstNonEmptyLine } from "./utils.js";
 import {
@@ -643,8 +643,9 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
       });
     }
 
+    const consumeAccounting = createGeminiJsonlParser();
     const accountingLog = createUsageCheckpointLog(onLog, ctx.onUsage, stdout => {
-      const parsed = parseGeminiJsonl(stdout);
+      const parsed = consumeAccounting(stdout);
       return { usage: parsed.usage, costUsd: parsed.costUsd, usageBasis: "per_run", provider: "google", biller: "google", billingType, model, complete: parsed.resultEvent !== null };
     });
     const proc = await runAdapterExecutionTargetProcess(runId, runtimeExecutionTarget, command, args, {
@@ -659,7 +660,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
       runLogTail: paperclipBridge?.runLogTail,
       settleRunDisposition: paperclipBridge?.settleRunDisposition,
     });
-    await accountingLog.flush({ complete: !proc.timedOut && !proc.signal });
+    await accountingLog.flush();
     return {
       proc,
       parsed: parseGeminiJsonl(proc.stdout),
@@ -757,6 +758,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
       exitCode: attempt.proc.exitCode,
       signal: attempt.proc.signal,
       timedOut: false,
+      usageComplete: attempt.parsed.resultEvent !== null,
       usageBasis: "per_run",
       errorMessage: failed ? fallbackErrorMessage : null,
       // Forward the transport-level error code from the run-disposition seam

packages/adapters/gemini-local/src/server/parse.ts

@@ -74,6 +74,11 @@ function accumulateUsage(
 }
 
 export function parseGeminiJsonl(stdout: string) {
+  return createGeminiJsonlParser()(stdout);
+}
+
+/** Consume complete JSONL records once, retaining protocol accounting state. */
+export function createGeminiJsonlParser() {
   let sessionId: string | null = null;
   const messages: string[] = [];
   let errorMessage: string | null = null;
@@ -86,116 +91,118 @@ export function parseGeminiJsonl(stdout: string) {
     outputTokens: 0,
   };
 
-  for (const rawLine of stdout.split(/\r?\n/)) {
-    const line = rawLine.trim();
-    if (!line) continue;
+  return (stdout: string) => {
+    for (const rawLine of stdout.split(/\r?\n/)) {
+      const line = rawLine.trim();
+      if (!line) continue;
 
-    const event = parseJson(line);
-    if (!event) continue;
+      const event = parseJson(line);
+      if (!event) continue;
 
-    const foundSessionId = readSessionId(event);
-    if (foundSessionId) sessionId = foundSessionId;
+      const foundSessionId = readSessionId(event);
+      if (foundSessionId) sessionId = foundSessionId;
 
-    const type = asString(event.type, "").trim();
+      const type = asString(event.type, "").trim();
 
-    if (type === "assistant") {
-      messages.push(...collectMessageText(event.message));
-      const messageObj = parseObject(event.message);
-      const content = Array.isArray(messageObj.content) ? messageObj.content : [];
-      for (const partRaw of content) {
-        const part = parseObject(partRaw);
-        if (asString(part.type, "").trim() === "question") {
-          question = {
-            prompt: asString(part.prompt, "").trim(),
-            choices: (Array.isArray(part.choices) ? part.choices : []).map((choiceRaw) => {
-              const choice = parseObject(choiceRaw);
-              return {
-                key: asString(choice.key, "").trim(),
-                label: asString(choice.label, "").trim(),
-                description: asString(choice.description, "").trim() || undefined,
-              };
-            }),
-          };
-          break; // only one question per message
+      if (type === "assistant") {
+        messages.push(...collectMessageText(event.message));
+        const messageObj = parseObject(event.message);
+        const content = Array.isArray(messageObj.content) ? messageObj.content : [];
+        for (const partRaw of content) {
+          const part = parseObject(partRaw);
+          if (asString(part.type, "").trim() === "question") {
+            question = {
+              prompt: asString(part.prompt, "").trim(),
+              choices: (Array.isArray(part.choices) ? part.choices : []).map((choiceRaw) => {
+                const choice = parseObject(choiceRaw);
+                return {
+                  key: asString(choice.key, "").trim(),
+                  label: asString(choice.label, "").trim(),
+                  description: asString(choice.description, "").trim() || undefined,
+                };
+              }),
+            };
+            break; // only one question per message
+          }
         }
+        continue;
       }
-      continue;
-    }
 
-    // Gemini CLI v0.38+ stream-json schema emits assistant turns as:
-    // {"type":"message","role":"assistant","content":"...","delta":true}
-    // These are discrete final messages (one per assistant turn), not
-    // cumulative streaming tokens, so collecting all of them produces the
-    // expected concatenated turn-by-turn summary rather than duplicated text.
-    if (type === "message") {
-      const role = asString(event.role, "").trim().toLowerCase();
-      if (role === "assistant") {
-        messages.push(...collectMessageText(event.content));
+      // Gemini CLI v0.38+ stream-json schema emits assistant turns as:
+      // {"type":"message","role":"assistant","content":"...","delta":true}
+      // These are discrete final messages (one per assistant turn), not
+      // cumulative streaming tokens, so collecting all of them produces the
+      // expected concatenated turn-by-turn summary rather than duplicated text.
+      if (type === "message") {
+        const role = asString(event.role, "").trim().toLowerCase();
+        if (role === "assistant") {
+          messages.push(...collectMessageText(event.content));
+        }
+        continue;
       }
-      continue;
-    }
 
-    if (type === "result") {
-      resultEvent = event;
-      accumulateUsage(usage, event.usage ?? event.usageMetadata ?? event.stats);
-      const resultText =
-        asString(event.result, "").trim() ||
-        asString(event.text, "").trim() ||
-        asString(event.response, "").trim();
-      if (resultText && messages.length === 0) messages.push(resultText);
-      const reportedCost = [event.total_cost_usd, event.cost_usd, event.cost].find((value) => typeof value === "number" && Number.isFinite(value) && value >= 0);
-      if (typeof reportedCost === "number") costUsd = reportedCost;
-      const status = asString(event.status, "").toLowerCase();
-      const isError =
-        event.is_error === true ||
-        asString(event.subtype, "").toLowerCase() === "error" ||
-        status === "error" ||
-        status === "failed";
-      if (isError) {
-        const text = asErrorText(event.error ?? event.message ?? event.result).trim();
-        if (text) errorMessage = text;
+      if (type === "result") {
+        resultEvent = event;
+        accumulateUsage(usage, event.usage ?? event.usageMetadata ?? event.stats);
+        const resultText =
+          asString(event.result, "").trim() ||
+          asString(event.text, "").trim() ||
+          asString(event.response, "").trim();
+        if (resultText && messages.length === 0) messages.push(resultText);
+        const reportedCost = [event.total_cost_usd, event.cost_usd, event.cost].find((value) => typeof value === "number" && Number.isFinite(value) && value >= 0);
+        if (typeof reportedCost === "number") costUsd = reportedCost;
+        const status = asString(event.status, "").toLowerCase();
+        const isError =
+          event.is_error === true ||
+          asString(event.subtype, "").toLowerCase() === "error" ||
+          status === "error" ||
+          status === "failed";
+        if (isError) {
+          const text = asErrorText(event.error ?? event.message ?? event.result).trim();
+          if (text) errorMessage = text;
+        }
+        continue;
       }
-      continue;
-    }
 
-    if (type === "error") {
-      const text = asErrorText(event.error ?? event.message ?? event.detail).trim();
-      if (text) errorMessage = text;
-      continue;
-    }
-
-    if (type === "system") {
-      const subtype = asString(event.subtype, "").trim().toLowerCase();
-      if (subtype === "error") {
+      if (type === "error") {
         const text = asErrorText(event.error ?? event.message ?? event.detail).trim();
         if (text) errorMessage = text;
+        continue;
+      }
+
+      if (type === "system") {
+        const subtype = asString(event.subtype, "").trim().toLowerCase();
+        if (subtype === "error") {
+          const text = asErrorText(event.error ?? event.message ?? event.detail).trim();
+          if (text) errorMessage = text;
+        }
+        continue;
+      }
+
+      if (type === "text") {
+        const part = parseObject(event.part);
+        const text = asString(part.text, "").trim();
+        if (text) messages.push(text);
+        continue;
+      }
+
+      if (type === "step_finish" || event.usage || event.usageMetadata) {
+        accumulateUsage(usage, event.usage ?? event.usageMetadata);
+        const reportedCost = [event.total_cost_usd, event.cost_usd, event.cost].find((value) => typeof value === "number" && Number.isFinite(value) && value >= 0);
+        if (typeof reportedCost === "number") costUsd = reportedCost;
+        continue;
       }
-      continue;
     }
 
-    if (type === "text") {
-      const part = parseObject(event.part);
-      const text = asString(part.text, "").trim();
-      if (text) messages.push(text);
-      continue;
-    }
-
-    if (type === "step_finish" || event.usage || event.usageMetadata) {
-      accumulateUsage(usage, event.usage ?? event.usageMetadata);
-      const reportedCost = [event.total_cost_usd, event.cost_usd, event.cost].find((value) => typeof value === "number" && Number.isFinite(value) && value >= 0);
-      if (typeof reportedCost === "number") costUsd = reportedCost;
-      continue;
-    }
-  }
-
-  return {
-    sessionId,
-    summary: messages.join("\n\n").trim(),
-    usage,
-    costUsd,
-    errorMessage,
-    resultEvent,
-    question,
+    return {
+      sessionId,
+      summary: messages.join("\n\n").trim(),
+      usage: { ...usage },
+      costUsd,
+      errorMessage,
+      resultEvent,
+      question,
+    };
   };
 }
 

packages/adapters/opencode-local/src/server/execute.ts

@@ -53,7 +53,7 @@ import {
   readPaperclipIssueWorkModeFromContext,
   resolveLegacyPaperclipDesiredSkillNames,
 } from "@paperclipai/adapter-utils/server-utils";
-import { isOpenCodeUnknownSessionError, parseOpenCodeJsonl } from "./parse.js";
+import { isOpenCodeUnknownSessionError, parseOpenCodeJsonl, createOpenCodeJsonlParser } from "./parse.js";
 import {
   ensureOpenCodeModelConfiguredAndAvailable,
   isTruthyEnvFlag,
@@ -646,8 +646,9 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
         });
       }
 
+      const consumeAccounting = createOpenCodeJsonlParser();
       const accountingLog = createUsageCheckpointLog(onLog, ctx.onUsage, stdout => {
-        const parsed = parseOpenCodeJsonl(stdout);
+        const parsed = consumeAccounting(stdout);
         const provider = parseModelProvider(model || null);
         return { usage: parsed.usage, costUsd: parsed.costUsd, usageBasis: "per_run", provider, biller: resolveOpenCodeBiller(runtimeEnv, provider), billingType: "unknown", model, complete: false };
       });
@@ -664,7 +665,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
         runLogTail: paperclipBridge?.runLogTail,
         settleRunDisposition: paperclipBridge?.settleRunDisposition,
       });
-      await accountingLog.flush({ complete: !proc.timedOut && !proc.signal });
+      await accountingLog.flush({ complete: proc.exitCode === 0 && !proc.timedOut && !proc.signal });
       return {
         proc,
         rawStderr: proc.stderr,
@@ -685,7 +686,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
           exitCode: attempt.proc.exitCode,
           signal: attempt.proc.signal,
           timedOut: true,
-        usageComplete: false,
+          usageComplete: false,
           usage: attempt.parsed.usage,
           usageBasis: "per_run",
           provider: parseModelProvider(model || null),
@@ -730,6 +731,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
         exitCode: synthesizedExitCode,
         signal: attempt.proc.signal,
         timedOut: false,
+        usageComplete: attempt.proc.exitCode === 0 && !attempt.proc.signal,
       usageBasis: "per_run",
         errorMessage: (synthesizedExitCode ?? 0) === 0 ? null : fallbackErrorMessage,
         // Forward the transport-level error code from the run-disposition seam.

packages/adapters/opencode-local/src/server/parse.ts

@@ -20,6 +20,11 @@ function errorText(value: unknown): string {
 }
 
 export function parseOpenCodeJsonl(stdout: string) {
+  return createOpenCodeJsonlParser()(stdout);
+}
+
+/** Consume complete JSONL records once, retaining protocol accounting state. */
+export function createOpenCodeJsonlParser() {
   let sessionId: string | null = null;
   const messages: string[] = [];
   const errors: string[] = [];
@@ -32,61 +37,63 @@ export function parseOpenCodeJsonl(stdout: string) {
   let costUsd: number | null = null;
   let missingCost = false;
 
-  for (const rawLine of stdout.split(/\r?\n/)) {
-    const line = rawLine.trim();
-    if (!line) continue;
+  return (stdout: string) => {
+    for (const rawLine of stdout.split(/\r?\n/)) {
+      const line = rawLine.trim();
+      if (!line) continue;
 
-    const event = parseJson(line);
-    if (!event) continue;
+      const event = parseJson(line);
+      if (!event) continue;
 
-    const currentSessionId = asString(event.sessionID, "").trim();
-    if (currentSessionId) sessionId = currentSessionId;
+      const currentSessionId = asString(event.sessionID, "").trim();
+      if (currentSessionId) sessionId = currentSessionId;
 
-    const type = asString(event.type, "");
+      const type = asString(event.type, "");
 
-    if (type === "text") {
-      const part = parseObject(event.part);
-      const text = asString(part.text, "").trim();
-      if (text) messages.push(text);
-      continue;
-    }
+      if (type === "text") {
+        const part = parseObject(event.part);
+        const text = asString(part.text, "").trim();
+        if (text) messages.push(text);
+        continue;
+      }
 
-    if (type === "step_finish") {
-      const part = parseObject(event.part);
-      const tokens = parseObject(part.tokens);
-      const cache = parseObject(tokens.cache);
-      usage.inputTokens += asNumber(tokens.input, 0) + asNumber(cache.write, 0);
-      usage.cachedInputTokens += asNumber(cache.read, 0);
-      usage.outputTokens += asNumber(tokens.output, 0) + asNumber(tokens.reasoning, 0);
-      if (typeof part.cost === "number" && Number.isFinite(part.cost) && part.cost >= 0) costUsd = (costUsd ?? 0) + part.cost;
-      else missingCost = true;
-      continue;
-    }
+      if (type === "step_finish") {
+        const part = parseObject(event.part);
+        const tokens = parseObject(part.tokens);
+        const cache = parseObject(tokens.cache);
+        usage.inputTokens += asNumber(tokens.input, 0) + asNumber(cache.write, 0);
+        usage.cachedInputTokens += asNumber(cache.read, 0);
+        usage.outputTokens += asNumber(tokens.output, 0) + asNumber(tokens.reasoning, 0);
+        if (typeof part.cost === "number" && Number.isFinite(part.cost) && part.cost >= 0) costUsd = (costUsd ?? 0) + part.cost;
+        else missingCost = true;
+        continue;
+      }
 
-    if (type === "tool_use") {
-      const part = parseObject(event.part);
-      const state = parseObject(part.state);
-      if (asString(state.status, "") === "error") {
-        const text = asString(state.error, "").trim();
-        if (text) toolErrors.push(text);
+      if (type === "tool_use") {
+        const part = parseObject(event.part);
+        const state = parseObject(part.state);
+        if (asString(state.status, "") === "error") {
+          const text = asString(state.error, "").trim();
+          if (text) toolErrors.push(text);
+        }
+        continue;
+      }
+
+      if (type === "error") {
+        const text = errorText(event.error ?? event.message).trim();
+        if (text) errors.push(text);
+        continue;
       }
-      continue;
     }
 
-    if (type === "error") {
-      const text = errorText(event.error ?? event.message).trim();
-      if (text) errors.push(text);
-      continue;
-    }
-  }
-
-  return {
-    sessionId,
-    summary: messages.join("\n\n").trim(),
-    usage,
-    costUsd: missingCost ? null : costUsd,
-    errorMessage: errors.length > 0 ? errors.join("\n") : null,
-    toolErrors,
+    return {
+      sessionId,
+      summary: messages.join("\n\n").trim(),
+      usage: { ...usage },
+      costUsd: missingCost ? null : costUsd,
+      errorMessage: errors.length > 0 ? errors.join("\n") : null,
+      toolErrors,
+    };
   };
 }
 

packages/adapters/pi-local/src/server/execute.ts

@@ -55,7 +55,7 @@ import {
   runChildProcess,
 } from "@paperclipai/adapter-utils/server-utils";
 import { shellQuote } from "@paperclipai/adapter-utils/ssh";
-import { isPiUnknownSessionError, parsePiJsonl } from "./parse.js";
+import { isPiUnknownSessionError, parsePiJsonl, createPiJsonlParser } from "./parse.js";
 import { ensurePiModelConfiguredAndAvailable } from "./models.js";
 import { preparePiRuntimeConfig } from "./runtime-config.js";
 import { SANDBOX_INSTALL_COMMAND } from "../index.js";
@@ -737,8 +737,9 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
         }
       };
 
+      const consumeAccounting = createPiJsonlParser();
       const accountingLog = createUsageCheckpointLog(bufferedOnLog, ctx.onUsage, stdout => {
-        const parsed = parsePiJsonl(stdout);
+        const parsed = consumeAccounting(stdout);
         return { usage: parsed.usage, costUsd: parsed.usage.costUsd, usageBasis: "per_run", provider, biller: resolvePiBiller(runtimeEnv, provider), billingType: "unknown", model, complete: parsed.sawAgentEnd };
       });
       const proc = await runAdapterExecutionTargetProcess(runId, runtimeExecutionTarget, command, args, {
@@ -753,7 +754,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
         runLogTail: paperclipBridge?.runLogTail,
         settleRunDisposition: paperclipBridge?.settleRunDisposition,
       });
-      await accountingLog.flush({ complete: !proc.timedOut && !proc.signal });
+      await accountingLog.flush();
 
       // Flush any remaining buffer content
       if (stdoutBuffer) {
@@ -780,7 +781,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
           exitCode: attempt.proc.exitCode,
           signal: attempt.proc.signal,
           timedOut: true,
-        usageComplete: attempt.parsed.sawAgentEnd,
+          usageComplete: attempt.parsed.sawAgentEnd,
         usage: attempt.parsed.usage, usageBasis: "per_run", provider, biller: resolvePiBiller(runtimeEnv, provider), model, billingType: "unknown", costUsd: attempt.parsed.usage.costUsd,
           errorMessage: `Timed out after ${timeoutSec}s`,
           clearSession: clearSessionOnMissingSession,
@@ -813,6 +814,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
         exitCode: effectiveExitCode,
         signal: attempt.proc.signal,
         timedOut: false,
+        usageComplete: attempt.parsed.sawAgentEnd,
       usageBasis: "per_run",
         errorMessage: (effectiveExitCode ?? 0) === 0 ? null : fallbackErrorMessage,
         // Forward the transport-level error code from the run-disposition seam.

packages/adapters/pi-local/src/server/parse.ts

@@ -29,7 +29,12 @@ function extractTextContent(content: string | Array<{ type: string; text?: strin
     .join("");
 }
 
-export function parsePiJsonl(stdout: string): ParsedPiOutput {
+export function parsePiJsonl(stdout: string) {
+  return createPiJsonlParser()(stdout);
+}
+
+/** Consume complete JSONL records once, retaining protocol accounting state. */
+export function createPiJsonlParser() {
   const result: ParsedPiOutput = {
     sawAgentEnd: false,
     sessionId: null,
@@ -52,186 +57,188 @@ export function parsePiJsonl(stdout: string): ParsedPiOutput {
   }
   let currentToolCall: { toolCallId: string; toolName: string; args: unknown } | null = null;
 
-  for (const rawLine of stdout.split(/\r?\n/)) {
-    const line = rawLine.trim();
-    if (!line) continue;
+  return (stdout: string) => {
+    for (const rawLine of stdout.split(/\r?\n/)) {
+      const line = rawLine.trim();
+      if (!line) continue;
 
-    const event = parseJson(line);
-    if (!event) continue;
+      const event = parseJson(line);
+      if (!event) continue;
 
-    const eventType = asString(event.type, "");
+      const eventType = asString(event.type, "");
 
-    // Pi can exit successfully after a provider failure. The terminal assistant
-    // message carries that failure in both message_end and turn_end envelopes.
-    const terminalMessages = eventType === "agent_end"
-      ? (Array.isArray(event.messages) ? event.messages : [])
-      : eventType === "message_end" || eventType === "turn_end"
-        ? [event.message]
-        : [];
-    for (const rawMessage of terminalMessages) {
-      const message = asRecord(rawMessage);
-      if (message?.role !== "assistant" || message.stopReason !== "error") continue;
-      const error = asString(message.errorMessage, "").trim() || "Pi provider request failed.";
-      if (!result.errors.includes(error)) result.errors.push(error);
-    }
-
-    // RPC protocol messages - skip these (internal implementation detail)
-    if (eventType === "response" || eventType === "extension_ui_request" || eventType === "extension_ui_response" || eventType === "extension_error") {
-      continue;
-    }
-
-    // Agent lifecycle
-    if (eventType === "agent_start") {
-      continue;
-    }
-
-    if (eventType === "agent_end") {
-      result.sawAgentEnd = true;
-      const messages = event.messages as Array<Record<string, unknown>> | undefined;
-      if (messages && messages.length > 0) {
-        const lastMessage = messages[messages.length - 1];
-        if (lastMessage?.role === "assistant") {
-          const content = lastMessage.content as string | Array<{ type: string; text?: string }>;
-          result.finalMessage = extractTextContent(content);
-        }
+      // Pi can exit successfully after a provider failure. The terminal assistant
+      // message carries that failure in both message_end and turn_end envelopes.
+      const terminalMessages = eventType === "agent_end"
+        ? (Array.isArray(event.messages) ? event.messages : [])
+        : eventType === "message_end" || eventType === "turn_end"
+          ? [event.message]
+          : [];
+      for (const rawMessage of terminalMessages) {
+        const message = asRecord(rawMessage);
+        if (message?.role !== "assistant" || message.stopReason !== "error") continue;
+        const error = asString(message.errorMessage, "").trim() || "Pi provider request failed.";
+        if (!result.errors.includes(error)) result.errors.push(error);
       }
-      continue;
-    }
 
-    if (eventType === "auto_retry_end") {
-      const succeeded = event.success === true;
-      if (!succeeded) {
-        const finalError = asString(event.finalError, "").trim();
-        result.errors.push(finalError || "Pi exhausted automatic retries without producing a response.");
+      // RPC protocol messages - skip these (internal implementation detail)
+      if (eventType === "response" || eventType === "extension_ui_request" || eventType === "extension_ui_response" || eventType === "extension_error") {
+        continue;
       }
-      continue;
-    }
 
-    // Turn lifecycle
-    if (eventType === "turn_start") {
-      continue;
-    }
-
-    if (eventType === "turn_end") {
-      const message = asRecord(event.message);
-      if (message) {
-        const content = message.content as string | Array<{ type: string; text?: string }>;
-        const text = extractTextContent(content);
-        if (text) {
-          result.finalMessage = text;
-          result.messages.push(text);
-        }
-        
-        // Extract usage and cost from assistant message
-        const usage = asRecord(message.usage);
-        if (usage) {
-          result.usage.inputTokens += asNumber(usage.input, 0) + asNumber(usage.cacheWrite, 0);
-          result.usage.outputTokens += asNumber(usage.output, 0);
-          result.usage.cachedInputTokens += asNumber(usage.cacheRead, 0);
-          
-          // Pi stores cost in usage.cost.total (and broken down in usage.cost.input, etc.)
-          const cost = asRecord(usage.cost);
-          addCost(cost?.total);
-        }
+      // Agent lifecycle
+      if (eventType === "agent_start") {
+        continue;
       }
-      
-      // Tool results are in toolResults array
-      const toolResults = event.toolResults as Array<Record<string, unknown>> | undefined;
-      if (toolResults) {
-        for (const tr of toolResults) {
-          const toolCallId = asString(tr.toolCallId, "");
-          const content = tr.content;
-          const isError = tr.isError === true;
-          
-          // Find matching tool call by toolCallId
-          const existingCall = result.toolCalls.find((tc) => tc.toolCallId === toolCallId);
-          if (existingCall) {
-            existingCall.result = typeof content === "string" ? content : JSON.stringify(content);
-            existingCall.isError = isError;
+
+      if (eventType === "agent_end") {
+        result.sawAgentEnd = true;
+        const messages = event.messages as Array<Record<string, unknown>> | undefined;
+        if (messages && messages.length > 0) {
+          const lastMessage = messages[messages.length - 1];
+          if (lastMessage?.role === "assistant") {
+            const content = lastMessage.content as string | Array<{ type: string; text?: string }>;
+            result.finalMessage = extractTextContent(content);
           }
         }
+        continue;
       }
-      continue;
-    }
 
-    // Message updates (streaming)
-    if (eventType === "message_update") {
-      const assistantEvent = asRecord(event.assistantMessageEvent);
-      if (assistantEvent) {
-        const msgType = asString(assistantEvent.type, "");
-        if (msgType === "text_delta") {
-          const delta = asString(assistantEvent.delta, "");
-          if (delta) {
-            // Append to last message or create new
-            if (result.messages.length === 0) {
-              result.messages.push(delta);
-            } else {
-              result.messages[result.messages.length - 1] += delta;
+      if (eventType === "auto_retry_end") {
+        const succeeded = event.success === true;
+        if (!succeeded) {
+          const finalError = asString(event.finalError, "").trim();
+          result.errors.push(finalError || "Pi exhausted automatic retries without producing a response.");
+        }
+        continue;
+      }
+
+      // Turn lifecycle
+      if (eventType === "turn_start") {
+        continue;
+      }
+
+      if (eventType === "turn_end") {
+        const message = asRecord(event.message);
+        if (message) {
+          const content = message.content as string | Array<{ type: string; text?: string }>;
+          const text = extractTextContent(content);
+          if (text) {
+            result.finalMessage = text;
+            result.messages.push(text);
+          }
+
+          // Extract usage and cost from assistant message
+          const usage = asRecord(message.usage);
+          if (usage) {
+            result.usage.inputTokens += asNumber(usage.input, 0) + asNumber(usage.cacheWrite, 0);
+            result.usage.outputTokens += asNumber(usage.output, 0);
+            result.usage.cachedInputTokens += asNumber(usage.cacheRead, 0);
+
+            // Pi stores cost in usage.cost.total (and broken down in usage.cost.input, etc.)
+            const cost = asRecord(usage.cost);
+            addCost(cost?.total);
+          }
+        }
+
+        // Tool results are in toolResults array
+        const toolResults = event.toolResults as Array<Record<string, unknown>> | undefined;
+        if (toolResults) {
+          for (const tr of toolResults) {
+            const toolCallId = asString(tr.toolCallId, "");
+            const content = tr.content;
+            const isError = tr.isError === true;
+
+            // Find matching tool call by toolCallId
+            const existingCall = result.toolCalls.find((tc) => tc.toolCallId === toolCallId);
+            if (existingCall) {
+              existingCall.result = typeof content === "string" ? content : JSON.stringify(content);
+              existingCall.isError = isError;
             }
           }
         }
+        continue;
       }
-      continue;
-    }
 
-    if (eventType === "error") {
-      const message = asString(event.message, "").trim();
-      if (message) {
-        result.errors.push(message);
+      // Message updates (streaming)
+      if (eventType === "message_update") {
+        const assistantEvent = asRecord(event.assistantMessageEvent);
+        if (assistantEvent) {
+          const msgType = asString(assistantEvent.type, "");
+          if (msgType === "text_delta") {
+            const delta = asString(assistantEvent.delta, "");
+            if (delta) {
+              // Append to last message or create new
+              if (result.messages.length === 0) {
+                result.messages.push(delta);
+              } else {
+                result.messages[result.messages.length - 1] += delta;
+              }
+            }
+          }
+        }
+        continue;
       }
-      continue;
-    }
 
-    // Tool execution
-    if (eventType === "tool_execution_start") {
-      const toolCallId = asString(event.toolCallId, "");
-      const toolName = asString(event.toolName, "");
-      const args = event.args;
-      currentToolCall = { toolCallId, toolName, args };
-      result.toolCalls.push({
-        toolCallId,
-        toolName,
-        args,
-        result: null,
-        isError: false,
-      });
-      continue;
-    }
-
-    if (eventType === "tool_execution_end") {
-      const toolCallId = asString(event.toolCallId, "");
-      const toolName = asString(event.toolName, "");
-      const toolResult = event.result;
-      const isError = event.isError === true;
-      
-      // Find the tool call by toolCallId (not toolName, to handle multiple calls to same tool)
-      const existingCall = result.toolCalls.find((tc) => tc.toolCallId === toolCallId);
-      if (existingCall) {
-        existingCall.result = typeof toolResult === "string" ? toolResult : JSON.stringify(toolResult);
-        existingCall.isError = isError;
+      if (eventType === "error") {
+        const message = asString(event.message, "").trim();
+        if (message) {
+          result.errors.push(message);
+        }
+        continue;
       }
-      currentToolCall = null;
-      continue;
-    }
 
-    // Usage tracking if available in the event (fallback for standalone usage events)
-    if (eventType === "usage" || event.usage) {
-      const usage = asRecord(event.usage);
-      if (usage) {
-        // Support both Pi format (input/output/cacheRead) and generic format (inputTokens/outputTokens/cachedInputTokens)
-        result.usage.inputTokens += asNumber(usage.inputTokens ?? usage.input, 0) + asNumber(usage.cacheWrite, 0);
-        result.usage.outputTokens += asNumber(usage.outputTokens ?? usage.output, 0);
-        result.usage.cachedInputTokens += asNumber(usage.cachedInputTokens ?? usage.cacheRead, 0);
-        
-        // Cost may be in usage.costUsd (direct) or usage.cost.total (Pi format)
-        const cost = asRecord(usage.cost);
-        addCost(cost?.total ?? usage.costUsd);
+      // Tool execution
+      if (eventType === "tool_execution_start") {
+        const toolCallId = asString(event.toolCallId, "");
+        const toolName = asString(event.toolName, "");
+        const args = event.args;
+        currentToolCall = { toolCallId, toolName, args };
+        result.toolCalls.push({
+          toolCallId,
+          toolName,
+          args,
+          result: null,
+          isError: false,
+        });
+        continue;
+      }
+
+      if (eventType === "tool_execution_end") {
+        const toolCallId = asString(event.toolCallId, "");
+        const toolName = asString(event.toolName, "");
+        const toolResult = event.result;
+        const isError = event.isError === true;
+
+        // Find the tool call by toolCallId (not toolName, to handle multiple calls to same tool)
+        const existingCall = result.toolCalls.find((tc) => tc.toolCallId === toolCallId);
+        if (existingCall) {
+          existingCall.result = typeof toolResult === "string" ? toolResult : JSON.stringify(toolResult);
+          existingCall.isError = isError;
+        }
+        currentToolCall = null;
+        continue;
+      }
+
+      // Usage tracking if available in the event (fallback for standalone usage events)
+      if (eventType === "usage" || event.usage) {
+        const usage = asRecord(event.usage);
+        if (usage) {
+          // Support both Pi format (input/output/cacheRead) and generic format (inputTokens/outputTokens/cachedInputTokens)
+          result.usage.inputTokens += asNumber(usage.inputTokens ?? usage.input, 0) + asNumber(usage.cacheWrite, 0);
+          result.usage.outputTokens += asNumber(usage.outputTokens ?? usage.output, 0);
+          result.usage.cachedInputTokens += asNumber(usage.cachedInputTokens ?? usage.cacheRead, 0);
+
+          // Cost may be in usage.costUsd (direct) or usage.cost.total (Pi format)
+          const cost = asRecord(usage.cost);
+          addCost(cost?.total ?? usage.costUsd);
+        }
       }
     }
-  }
 
-  if (missingCost) result.usage.costUsd = null;
-  return result;
+    if (missingCost) result.usage.costUsd = null;
+    return { ...result, usage: { ...result.usage }, messages: [...result.messages], errors: [...result.errors], toolCalls: [...result.toolCalls] };
+  };
 }
 
 export function isPiUnknownSessionError(stdout: string, stderr: string): boolean {

server/src/__tests__/adapter-cost-failure-paths.test.ts

@@ -50,8 +50,18 @@ describe("CLI adapter accounting on timeout", () => {
       context: {}, onLog: async () => {}, onUsage,
     };
     const execute = await fixture.execute();
-    await execute(context);
-    expect(onUsage.mock.calls.at(-1)![0]).toMatchObject({ usage: fixture.tokens, costUsd: fixture.price, complete: true });
+    const initial = await execute(context);
+    const complete = !["claude", "pi"].includes(fixture.name);
+    expect(initial.usageComplete).toBe(complete);
+    expect(onUsage.mock.calls.at(-1)![0]).toMatchObject({ usage: fixture.tokens, costUsd: fixture.price, complete });
+    if (["claude", "pi", "opencode"].includes(fixture.name)) {
+      processResult.mockImplementationOnce(async (_run, _target, _command, _args, options) => {
+        await options.onLog("stdout", stdout).catch(() => {});
+        return { exitCode: 1, signal: null, timedOut: false, stdout, stderr: "process failed", pid: 123, startedAt: new Date().toISOString() };
+      });
+      expect((await execute(context)).usageComplete).toBe(false);
+      expect(onUsage.mock.calls.at(-1)![0].complete).toBe(false);
+    }
     if (fixture.name === "claude") {
       const mixed = [
         { type: "system", subtype: "init", model: "actual-model" },
@@ -97,3 +107,36 @@ describe("CLI adapter accounting on timeout", () => {
     expect(result.costUsd).toBe(fixture.price);
   });
 });
+
+
+describe("incremental protocol accounting", () => {
+  it("keeps Claude message updates idempotent and preserves immutable snapshots", async () => {
+    const { createClaudeStreamParser } = await import("../../../packages/adapters/claude-local/src/server/parse.js");
+    const consume = createClaudeStreamParser();
+    consume(JSON.stringify({ type: "system", subtype: "init", model: "actual-model" }));
+    const message = (id: string, output: number) => JSON.stringify({ type: "assistant", message: { id, usage: { input_tokens: 2, output_tokens: output } } });
+    const first = consume(message("one", 1));
+    expect(first.usage).toMatchObject({ inputTokens: 2, outputTokens: 1 });
+    expect(consume(message("one", 3)).usage).toMatchObject({ inputTokens: 2, outputTokens: 3 });
+    expect(consume(message("two", 2)).usage).toMatchObject({ inputTokens: 4, outputTokens: 5 });
+    expect(first.usage).toMatchObject({ inputTokens: 2, outputTokens: 1 });
+    const final = consume(JSON.stringify({ type: "result", total_cost_usd: 1, usage: { input_tokens: 9, output_tokens: 8 } }));
+    expect(final).toMatchObject({ model: "actual-model", costUsd: 1, usage: { inputTokens: 9, outputTokens: 8 } });
+  });
+  it.each(cases.filter(item => !["claude", "kimi"].includes(item.name)))("updates $name totals once per newly received record", async fixture => {
+    const factories = {
+      codex: async () => (await import("../../../packages/adapters/codex-local/src/server/parse.js")).createCodexJsonlParser(),
+      cursor: async () => (await import("../../../packages/adapters/cursor-local/src/server/parse.js")).createCursorJsonlParser(),
+      gemini: async () => (await import("../../../packages/adapters/gemini-local/src/server/parse.js")).createGeminiJsonlParser(),
+      pi: async () => (await import("../../../packages/adapters/pi-local/src/server/parse.js")).createPiJsonlParser(),
+      opencode: async () => (await import("../../../packages/adapters/opencode-local/src/server/parse.js")).createOpenCodeJsonlParser(),
+    };
+    const consume = await factories[fixture.name as keyof typeof factories]();
+    const line = JSON.stringify(fixture.event);
+    const first = consume(line);
+    const second = consume(line);
+    expect(first.usage).toMatchObject(fixture.tokens!);
+    const multiplier = fixture.name === "codex" ? 1 : 2;
+    expect(second.usage).toMatchObject(Object.fromEntries(Object.entries(fixture.tokens!).map(([key, count]) => [key, count * multiplier])));
+  });
+});