fix(runner): resume pending Codex calls exactly once

Co-Authored-By: Paperclip <noreply@paperclip.ing>
2026-10-26 10:26:26 +01:00 · 2026-08-23 20:08:09 +00:00
1 parent a70b0f70b6
commit c07c1f95d3
@@ -1096,26 +1096,48 @@ fn poll_provider(
                input,
            } => {
                state.last_activity_at_unix_ms = current_unix_ms()?;
+                let pending_before_resume = state
+                    .provider_tool_bridge
+                    .pending_calls()
+                    .any(|pending| pending.call_id == call_id);
                let call = state
                    .provider_tool_bridge
                    .begin_call(call_id, operation_id, input)
                    .map_err(|error| {
                        DurableRunnerError::invalid(format!("provider tool call rejected: {error}"))
                    })?;
-                enqueue_event(
-                    state,
-                    config,
-                    "semantic_tool.input",
-                    0,
-                    json!({
-                        "semantic_tool": {
-                            "schema": "paperclip.prp.semantic_tool.v1", "schemaVersion": 1,
-                            "phase": "input", "operationId": call.operation_id,
-                            "callId": call.call_id, "input": call.input,
-                        }
-                    }),
-                    Some(&state.item_id.clone()),
-                )?;
+                if pending_before_resume {
+                    enqueue_event(
+                        state,
+                        config,
+                        "semantic_tool.reconciled",
+                        0,
+                        json!({
+                            "semantic_tool": {
+                                "schema": "paperclip.prp.semantic_tool.v1", "schemaVersion": 1,
+                                "phase": "reconciled", "operationId": call.operation_id,
+                                "callId": call.call_id,
+                            },
+                            "reason": "provider_replayed_pending_call",
+                        }),
+                        Some(&state.item_id.clone()),
+                    )?;
+                } else {
+                    enqueue_event(
+                        state,
+                        config,
+                        "semantic_tool.input",
+                        0,
+                        json!({
+                            "semantic_tool": {
+                                "schema": "paperclip.prp.semantic_tool.v1", "schemaVersion": 1,
+                                "phase": "input", "operationId": call.operation_id,
+                                "callId": call.call_id, "input": call.input,
+                            }
+                        }),
+                        Some(&state.item_id.clone()),
+                    )?;
+                }
                store.save(state)?;
            }
            crate::codex_provider::ProviderEvent::Notification { method, params } => {
@@ -59,6 +59,7 @@ interface EvalSessionRequest {
  runnerd: { path: string; sha256: string };
  limits: { turnTimeoutMs: number; maxAgentTurns: number; maxEstimatedCostNanodollars: number };
  session: CreateCapabilityLiveSessionInput;
+  nativeResume?: { operationId: string };
  /** Codex collaboration/tool-preamble instructions are enabled unless explicitly false. */
  includeCollaborationModeInstructions?: boolean;
 }
@@ -123,6 +124,7 @@ try {
    toolExposure: request.session.toolExposure,
    includeCollaborationModeInstructions:
      request.includeCollaborationModeInstructions,
+    nativeResume: request.nativeResume,
  });
  const turn = execution.turn as Record<string, unknown>;
  const snapshot = execution.snapshot as Parameters<typeof reconcileCapabilityLiveUsage>[0];
@@ -1987,6 +1987,11 @@ export interface DurableEvalSessionInput {
  explicitClaims: string[];
  turnTimeoutMs: number;
  toolExposure?: "eager" | "lazy";
+  /** Crash Runner D after this semantic input commits, then resume it natively. */
+  nativeResume?: { operationId: string };
+  /** Test-only provider process override; production eval callers omit these. */
+  providerCommand?: string;
+  providerArgs?: string[];
  /** Defaults on for Codex; set false only for an eval that needs the baseline prompt. */
  includeCollaborationModeInstructions?: boolean;
  provider?: "codex" | "opencode" | "claude_managed" | "aws_agentcore" | "acpx";
@@ -2222,6 +2227,12 @@ function sanitizedAwsEvalEnvironment(
 export async function runDurableEvalSession(
  input: DurableEvalSessionInput,
 ): Promise<Record<string, unknown>> {
+  if (
+    input.nativeResume !== undefined &&
+    (input.provider ?? "codex") !== "codex"
+  ) {
+    throw new Error("native resume eval requires the Codex provider");
+  }
  const identity = {
    runnerInstanceId: `runner_${createHash("sha256").update(input.attemptId).digest("hex").slice(0, 16)}`,
    environmentLeaseId: `lease_${createHash("sha256").update(`${input.attemptId}:lease`).digest("hex").slice(0, 16)}`,
@@ -2258,6 +2269,16 @@ export async function runDurableEvalSession(
  const authorized = dispatcher.listTools(identity.runId);
  const loadedOperations = new Set<string>();
  const discoveryEvidence: Array<Record<string, unknown>> = [];
+  let nativeResumeCall: {
+    callId: string;
+    operationId: string;
+    input: unknown;
+  } | null = null;
+  let releaseNativeSemanticDispatch!: () => void;
+  const nativeSemanticDispatchReleased = new Promise<void>((resolveDispatch) => {
+    releaseNativeSemanticDispatch = resolveDispatch;
+  });
+  let nativeSemanticHandlerCallCount = 0;
  const operations: Array<{
    operationId: string;
    version: number;
@@ -2343,6 +2364,11 @@ export async function runDurableEvalSession(
          contentItems: [{ type: "inputText", text: JSON.stringify(invoked) }],
        };
      }
+      if (call.operationId === input.nativeResume?.operationId) {
+        nativeResumeCall ??= structuredClone(call);
+        await nativeSemanticDispatchReleased;
+        nativeSemanticHandlerCallCount += 1;
+      }
      return dispatcher.dispatch({ runId: identity.runId, ...call });
    },
    // The connection lease is an ownership/security boundary, not a provider-turn
@@ -2433,13 +2459,15 @@ export async function runDurableEvalSession(
            : {
                kind: providerKind,
                command:
-                  providerKind === "opencode"
+                  input.providerCommand ??
+                  (providerKind === "opencode"
                    ? process.execPath
-                    : (process.env.PAPERCLIP_CODEX_COMMAND ?? "codex"),
+                    : (process.env.PAPERCLIP_CODEX_COMMAND ?? "codex")),
                args:
-                  providerKind === "opencode"
+                  input.providerArgs ??
+                  (providerKind === "opencode"
                    ? [opencodeProxyPath]
-                    : providerArgs,
+                    : providerArgs),
                cwd: tmpdir(),
                model: input.model,
                instructions:
@@ -2453,33 +2481,43 @@ export async function runDurableEvalSession(
  core.queueCommand("session.open", { reuse: "same_session" });
  core.queueCommand("turn.start", { text: input.prompt });
  await core.start();
-  const handle = spawnRunner({
-    connectUrl: core.connectUrl,
-    stateDirectory: runnerStateDirectory,
-    identity,
-    ticket: core.issueBootstrapTicket(),
-    maxOutboxBytes: 256 * 1024,
-    p0ReserveBytes: 64 * 1024,
-    maxRuntimeMs: input.turnTimeoutMs + 30_000,
-    runnerBinaryPath: input.runnerBinaryPath,
-    environment:
-      providerKind === "opencode"
-        ? {
-            ...process.env,
-            PAPERCLIP_OPENCODE_COMMAND: input.opencodeCommand ?? "opencode",
-            PAPERCLIP_OPENCODE_RUNTIME_DIR: resolve(root, "opencode"),
-            PAPERCLIP_RUNNER_INSTANCE_ID: identity.runnerInstanceId,
-            PAPERCLIP_RUN_ID: identity.runId,
-            PAPERCLIP_NORMALIZED_SESSION_ID: identity.normalizedSessionId,
-          }
-        : providerKind === "claude_managed"
-          ? createSanitizedClaudeManagedEnvironment(process.env)
-          : providerKind === "aws_agentcore"
-            ? sanitizedAwsEvalEnvironment(process.env)
-            : providerKind === "acpx"
-              ? createSanitizedAcpxEnvironment(process.env, acpxProfile!.agent)
-              : process.env,
-  });
+  const launchRunner = (): RunnerProcessHandle =>
+    spawnRunner({
+        connectUrl: core.connectUrl,
+        stateDirectory: runnerStateDirectory,
+        identity,
+        ticket: core.issueBootstrapTicket(),
+        maxOutboxBytes: 256 * 1024,
+        p0ReserveBytes: 64 * 1024,
+        maxRuntimeMs: input.turnTimeoutMs + 30_000,
+        runnerBinaryPath: input.runnerBinaryPath,
+        environment:
+          providerKind === "opencode"
+            ? {
+                ...process.env,
+                PAPERCLIP_OPENCODE_COMMAND: input.opencodeCommand ?? "opencode",
+                PAPERCLIP_OPENCODE_RUNTIME_DIR: resolve(root, "opencode"),
+                PAPERCLIP_RUNNER_INSTANCE_ID: identity.runnerInstanceId,
+                PAPERCLIP_RUN_ID: identity.runId,
+                PAPERCLIP_NORMALIZED_SESSION_ID: identity.normalizedSessionId,
+              }
+            : providerKind === "claude_managed"
+              ? createSanitizedClaudeManagedEnvironment(process.env)
+              : providerKind === "aws_agentcore"
+                ? sanitizedAwsEvalEnvironment(process.env)
+                : providerKind === "acpx"
+                  ? createSanitizedAcpxEnvironment(
+                      process.env,
+                      acpxProfile!.agent,
+                    )
+                  : process.env,
+    });
+  let handle = launchRunner();
+  const runnerProcessPids = [handle.child.pid].filter(
+    (pid): pid is number => typeof pid === "number",
+  );
+  let runnerRestarts = 0;
+  let crashedRunner: RunnerProcessResult | null = null;
  try {
    const resolvedEvalRuntimeRequests = new Set<string>();
    let providerTerminalFailure: {
@@ -2488,6 +2526,48 @@ export async function runDurableEvalSession(
      retryable: boolean;
    } | null = null;
    const deadline = Date.now() + input.turnTimeoutMs;
+    if (input.nativeResume !== undefined) {
+      while (nativeResumeCall === null && Date.now() < deadline) {
+        await new Promise((resolveWait) => setTimeout(resolveWait, 20));
+      }
+      if (nativeResumeCall === null) {
+        throw new Error(
+          `native resume operation ${input.nativeResume.operationId} did not commit before timeout`,
+        );
+      }
+      handle.child.kill("SIGKILL");
+      crashedRunner = await waitForProcess(handle, 10_000);
+      if (crashedRunner.signal !== "SIGKILL") {
+        throw new Error(`native resume runner exited with ${String(crashedRunner.signal)}`);
+      }
+      runnerRestarts = 1;
+      handle = launchRunner();
+      if (typeof handle.child.pid === "number")
+        runnerProcessPids.push(handle.child.pid);
+      const pendingCall = nativeResumeCall;
+      const pendingCallWasReconciled = (): boolean =>
+        core.store.state.committedEvents.some((event) => {
+          if (event.eventType !== "semantic_tool.reconciled") return false;
+          const payload = (
+            event.envelope.payload as Record<string, unknown> | undefined
+          )?.payload as Record<string, unknown> | undefined;
+          const semantic = payload?.semantic_tool as
+            | Record<string, unknown>
+            | undefined;
+          return (
+            semantic?.callId === pendingCall.callId &&
+            semantic?.operationId === pendingCall.operationId
+          );
+        });
+      while (Date.now() < deadline) {
+        if (pendingCallWasReconciled()) break;
+        await new Promise((resolveWait) => setTimeout(resolveWait, 20));
+      }
+      if (!pendingCallWasReconciled()) {
+        throw new Error("replacement Runner D did not reconcile the pending Codex tool call");
+      }
+      releaseNativeSemanticDispatch();
+    }
    while (Date.now() < deadline) {
      if (providerKind === "acpx") {
        for (const committed of core.store.state.committedEvents) {
@@ -2621,6 +2701,8 @@ export async function runDurableEvalSession(
    };
    let providerSessionId: string | null = null;
    let providerPid: number | null = null;
+    const providerProcessPids: number[] = [];
+    const providerThreadIds: string[] = [];
    let sidecarPid: number | null = null;
    let agentPid: number | null = null;
    let providerDriver: string | null = null;
@@ -2641,6 +2723,7 @@ export async function runDurableEvalSession(
          providerSessionId = descriptor.providerSessionId;
        if (typeof descriptor?.processId === "number") {
          providerPid = descriptor.processId;
+          providerProcessPids.push(descriptor.processId);
          if (descriptor.driver === "acpx_runtime")
            sidecarPid = descriptor.processId;
        }
@@ -2662,6 +2745,8 @@ export async function runDurableEvalSession(
          providerExecutionKind = descriptor.executionKind;
        if (typeof descriptor?.service === "string")
          providerService = descriptor.service;
+        if (typeof payload?.threadId === "string")
+          providerThreadIds.push(payload.threadId);
      }
      if (record.eventType === "harness.diagnostic") {
        if (
@@ -2780,17 +2865,32 @@ export async function runDurableEvalSession(
        if (runDelta) {
          usage = {
            ...usage,
-            providerRequests:
-              runDelta.requests ??
-              runDelta.providerRequests ??
-              usage.providerRequests,
-            inputTokens: runDelta.inputTokens ?? 0,
-            outputTokens: runDelta.outputTokens ?? 0,
-            cachedInputTokens: runDelta.cacheReadTokens ?? 0,
-            reasoningTokens: runDelta.reasoningTokens ?? 0,
+            providerRequests: Math.max(
+              usage.providerRequests ?? 0,
+              runDelta.requests ?? runDelta.providerRequests ?? 0,
+            ),
+            inputTokens: Math.max(
+              usage.inputTokens ?? 0,
+              runDelta.inputTokens ?? 0,
+            ),
+            outputTokens: Math.max(
+              usage.outputTokens ?? 0,
+              runDelta.outputTokens ?? 0,
+            ),
+            cachedInputTokens: Math.max(
+              usage.cachedInputTokens ?? 0,
+              runDelta.cacheReadTokens ?? 0,
+            ),
+            reasoningTokens: Math.max(
+              usage.reasoningTokens ?? 0,
+              runDelta.reasoningTokens ?? 0,
+            ),
            providerCostNanodollars:
              typeof runDelta.providerCostUsd === "number"
-                ? Math.round(runDelta.providerCostUsd * 1_000_000_000)
+                ? Math.max(
+                    usage.providerCostNanodollars ?? 0,
+                    Math.round(runDelta.providerCostUsd * 1_000_000_000),
+                  )
                : usage.providerCostNanodollars,
          };
        }
@@ -2846,6 +2946,58 @@ export async function runDurableEvalSession(
    if (!assistantText)
      assistantText = "Provider completed the requested Paperclip operation.";
    const finalState = JSON.stringify(authority.snapshot());
+    const nativeResumeProof =
+      input.nativeResume === undefined || nativeResumeCall === null
+        ? undefined
+        : (() => {
+            const semanticEvents = records.flatMap((record) => {
+              const payload = record.payload as
+                | Record<string, unknown>
+                | undefined;
+              const semantic = payload?.semantic_tool as
+                | Record<string, unknown>
+                | undefined;
+              return semantic?.callId === nativeResumeCall.callId &&
+                semantic.operationId === nativeResumeCall.operationId
+                ? [{ eventType: record.eventType, semantic }]
+                : [];
+            });
+            const initialComments = JSON.parse(initialState) as {
+              comments?: unknown[];
+            };
+            const finalComments = authority.snapshot().comments;
+            return {
+              schema: "paperclip.runner.native-resume-proof/v1",
+              triggered: true,
+              runnerRestarts,
+              runnerProcessPids,
+              providerProcessPids,
+              sameProviderThread:
+                providerThreadIds.length >= 2 &&
+                new Set(providerThreadIds).size === 1,
+              providerThreadDigest: providerThreadIds[0]
+                ? `sha256:${createHash("sha256").update(providerThreadIds[0]).digest("hex")}`
+                : null,
+              callId: nativeResumeCall.callId,
+              operationId: nativeResumeCall.operationId,
+              inputEventCount: semanticEvents.filter(
+                (event) => event.eventType === "semantic_tool.input",
+              ).length,
+              resultEventCount: semanticEvents.filter(
+                (event) => event.eventType === "semantic_tool.result",
+              ).length,
+              semanticHandlerCallCount: nativeSemanticHandlerCallCount,
+              controlPlaneEffectCount:
+                finalComments.length - (initialComments.comments?.length ?? 0),
+              runnerReconciledEventCount: semanticEvents.filter(
+                (event) => event.eventType === "semantic_tool.reconciled",
+              ).length,
+              crashedRunnerSignal: crashedRunner?.signal ?? null,
+              providerCalls: usage.providerCalls ?? 1,
+              providerRequests: usage.providerRequests ?? 0,
+              costNanodollars: usage.providerCostNanodollars ?? 0,
+            };
+          })();
    const now = new Date().toISOString();
    return {
      turn: { turnId: identity.turnId, status: "completed", assistantText },
@@ -2883,6 +3035,7 @@ export async function runDurableEvalSession(
        status: "idle",
        activeTurnId: null,
        semanticResult,
+        ...(nativeResumeProof === undefined ? {} : { nativeResume: nativeResumeProof }),
        createdAt: now,
        updatedAt: now,
        authority: {
@@ -3008,6 +3161,7 @@ export async function runDurableEvalSession(
      },
    };
  } finally {
+    releaseNativeSemanticDispatch();
    handle.child.kill("SIGKILL");
    await core.stop();
    await authority.stop();
@@ -258,6 +258,27 @@ export interface CapabilityLiveSessionSnapshot {
  terminalTurns?: CapabilityLiveTurnTerminalFact[];
  /** Exact-once provider usage/cost receipts retained across every attempt. */
  usageLedger?: CapabilityLiveUsageReceipt[];
+  /** Process-level evidence for an interrupted Codex semantic-call resume. */
+  nativeResume?: {
+    schema: "paperclip.runner.native-resume-proof/v1";
+    triggered: true;
+    runnerRestarts: number;
+    runnerProcessPids: number[];
+    providerProcessPids: number[];
+    sameProviderThread: boolean;
+    providerThreadDigest: string | null;
+    callId: string;
+    operationId: string;
+    inputEventCount: number;
+    resultEventCount: number;
+    semanticHandlerCallCount: number;
+    controlPlaneEffectCount: number;
+    runnerReconciledEventCount: number;
+    crashedRunnerSignal: string | null;
+    providerCalls: number;
+    providerRequests: number;
+    costNanodollars: number;
+  };
  /** Bounded mock-control-plane snapshots for DevTools time travel and diffs. */
  stateHistory?: CapabilityLiveStateRevision[];
  /** Complete, bounded per-turn workspace changes verified by the runner. */
@@ -14,12 +14,14 @@ import { connect, type Socket } from "node:net";
 import { tmpdir } from "node:os";
 import { resolve } from "node:path";
 import { setTimeout as delay } from "node:timers/promises";
+import { fileURLToPath } from "node:url";

 import { describe, expect, it } from "vitest";

 import {
  DurableRecoveryMockCore,
  durableRecoveryInternals,
+  runDurableEvalSession,
  runDurableRecoveryRecovery,
  runDurableToolBridgeConformance,
 } from "./durable-recovery.js";
@@ -366,6 +368,97 @@ async function runRunnerProcess(options: {
 }

 describe.sequential("Durable transport and recovery", () => {
+  it("resumes the original pending Codex tool call once after Runner D is killed", async () => {
+    const scratch =
+      process.env.PAPERCLIP_RUN_SCRATCH_DIR ??
+      process.env.PAPERCLIP_SCRATCH_DIR ??
+      tmpdir();
+    const root = mkdtempSync(resolve(scratch, "native-codex-resume-test-"));
+    const providerFixture = fileURLToPath(new URL(
+      "../../test/fixtures/fake-native-resume-codex-app-server.mjs",
+      import.meta.url,
+    ));
+    try {
+      const execution = await runDurableEvalSession({
+        attemptId: "native-resume-test",
+        prompt: "Record one progress update.",
+        model: "gpt-native-resume-fixture",
+        runnerBinaryPath: durableRecoveryInternals.runnerBinary,
+        providerCommand: process.execPath,
+        providerArgs: [providerFixture, resolve(root, "provider-state.json")],
+        seed: {
+          actors: [{
+            id: "actor-1",
+            companyId: "company-1",
+            name: "Resume tester",
+            role: "engineer",
+            status: "active",
+            budgetId: "budget-1",
+            capabilityGrants: [],
+          }],
+          tasks: [{
+            id: "task-1",
+            companyId: "company-1",
+            identifier: "TST-1",
+            title: "Resume one pending operation",
+            description: null,
+            status: "todo",
+            priority: "high",
+            workMode: "standard",
+            parentId: null,
+            assigneeActorId: "actor-1",
+            checkoutRunId: null,
+            executionRunId: null,
+            startedAt: null,
+            completedAt: null,
+          }],
+          comments: [],
+        },
+        actorId: "actor-1",
+        taskId: "task-1",
+        capabilities: [],
+        explicitClaims: [],
+        turnTimeoutMs: 10_000,
+        nativeResume: { operationId: "report_progress" },
+      });
+      const snapshot = execution.snapshot as Record<string, unknown>;
+      const proof = snapshot.nativeResume as Record<string, unknown>;
+      const evidence = snapshot.evidence as Array<Record<string, unknown>>;
+      const finalState = JSON.parse(String(snapshot.mockState)) as { comments: unknown[] };
+
+      expect(proof).toMatchObject({
+        schema: "paperclip.runner.native-resume-proof/v1",
+        triggered: true,
+        runnerRestarts: 1,
+        sameProviderThread: true,
+        callId: "call-native-resume",
+        operationId: "report_progress",
+        inputEventCount: 1,
+        resultEventCount: 1,
+        semanticHandlerCallCount: 1,
+        controlPlaneEffectCount: 1,
+        runnerReconciledEventCount: 1,
+        crashedRunnerSignal: "SIGKILL",
+        providerCalls: 1,
+        providerRequests: 1,
+        costNanodollars: 0,
+      });
+      expect(new Set(proof.runnerProcessPids as number[]).size).toBe(2);
+      expect(new Set(proof.providerProcessPids as number[]).size).toBe(2);
+      expect(evidence.filter((entry) => {
+        const data = entry.data as Record<string, unknown>;
+        return entry.kind === "tool_call" && data.callId === "call-native-resume";
+      })).toHaveLength(1);
+      expect(evidence.filter((entry) => {
+        const data = entry.data as Record<string, unknown>;
+        return entry.kind === "tool_result" && data.callId === "call-native-resume";
+      })).toHaveLength(1);
+      expect(finalState.comments).toHaveLength(1);
+    } finally {
+      rmSync(root, { recursive: true, force: true });
+    }
+  }, 30_000);
+
  it("repeatedly completes the provider bridge and explicit runner shutdown", async () => {
    for (let attempt = 0; attempt < 10; attempt += 1) {
      const result = await runDurableToolBridgeConformance();
@@ -0,0 +1,175 @@
+#!/usr/bin/env node
+import { readFileSync, writeFileSync } from "node:fs";
+
+const statePath = process.argv[2];
+if (!statePath) throw new Error("state path is required");
+
+function load() {
+  try {
+    return JSON.parse(readFileSync(statePath, "utf8"));
+  } catch (error) {
+    if (error?.code !== "ENOENT") throw error;
+    return {
+      threadId: "thread-native-resume",
+      sessionId: "session-native-resume",
+      turnId: "turn-native-resume",
+      pendingTool: null,
+      completed: false,
+    };
+  }
+}
+
+let state = load();
+let buffer = "";
+
+function save() {
+  writeFileSync(statePath, `${JSON.stringify(state)}\n`, { mode: 0o600 });
+}
+
+function send(value) {
+  process.stdout.write(`${JSON.stringify(value)}\n`);
+}
+
+function sendPendingTool() {
+  if (!state.pendingTool) return;
+  send({
+    id: state.pendingTool.requestId,
+    method: "item/tool/call",
+    params: {
+      threadId: state.threadId,
+      turnId: state.turnId,
+      callId: state.pendingTool.callId,
+      tool: "report_progress",
+      arguments: {
+        idempotencyKey: "native-resume-progress",
+        body: "One native resume effect.",
+      },
+    },
+  });
+}
+
+function handleRequest(message) {
+  const { id, method } = message;
+  if (method === "initialize") {
+    send({ id, result: { user: { sessionId: state.sessionId } } });
+    return;
+  }
+  if (method === "thread/start" || method === "thread/resume") {
+    send({
+      id,
+      result: {
+        model: "gpt-native-resume-fixture",
+        modelProvider: "openai-fixture",
+        thread: { id: state.threadId, sessionId: state.sessionId },
+      },
+    });
+    if (method === "thread/resume") sendPendingTool();
+    return;
+  }
+  if (method === "thread/read") {
+    send({
+      id,
+      result: {
+        thread: {
+          id: state.threadId,
+          sessionId: state.sessionId,
+          turns: [{ id: state.turnId, status: state.completed ? "completed" : "inProgress" }],
+          tokenUsage: {
+            total: {
+              inputTokens: 10,
+              cachedInputTokens: 0,
+              outputTokens: 2,
+              reasoningOutputTokens: 0,
+            },
+          },
+        },
+      },
+    });
+    return;
+  }
+  if (method === "turn/start") {
+    state.pendingTool = {
+      requestId: "rpc-native-resume",
+      callId: "call-native-resume",
+    };
+    save();
+    send({ id, result: { turn: { id: state.turnId, status: "inProgress" } } });
+    send({
+      method: "turn/started",
+      params: {
+        threadId: state.threadId,
+        turn: { id: state.turnId, status: "inProgress" },
+      },
+    });
+    sendPendingTool();
+    return;
+  }
+  send({ id, error: { code: -32601, message: `unsupported method ${method}` } });
+}
+
+function handle(message) {
+  if (message.method) {
+    if (message.id !== undefined) handleRequest(message);
+    return;
+  }
+  if (!state.pendingTool || String(message.id) !== state.pendingTool.requestId) return;
+  const toolResult = message.result;
+  state.pendingTool = null;
+  state.completed = true;
+  save();
+  send({
+    method: "item/completed",
+    params: {
+      threadId: state.threadId,
+      turnId: state.turnId,
+      item: {
+        id: "message-native-resume",
+        type: "agentMessage",
+        text: `Tool result: ${JSON.stringify(toolResult)}`,
+      },
+    },
+  });
+  send({
+    method: "thread/tokenUsage/updated",
+    params: {
+      threadId: state.threadId,
+      turnId: state.turnId,
+      tokenUsage: {
+        total: {
+          inputTokens: 10,
+          cachedInputTokens: 0,
+          outputTokens: 2,
+          reasoningTokens: 0,
+        },
+        runDelta: {
+          requests: 1,
+          inputTokens: 10,
+          cacheReadTokens: 0,
+          outputTokens: 2,
+          reasoningTokens: 0,
+          providerCostUsd: 0,
+        },
+      },
+    },
+  });
+  send({
+    method: "turn/completed",
+    params: {
+      threadId: state.threadId,
+      turn: { id: state.turnId, status: "completed" },
+    },
+  });
+}
+
+process.stdin.setEncoding("utf8");
+process.stdin.resume();
+process.stdin.on("data", (chunk) => {
+  buffer += chunk;
+  for (;;) {
+    const newline = buffer.indexOf("\n");
+    if (newline < 0) break;
+    const line = buffer.slice(0, newline).trim();
+    buffer = buffer.slice(newline + 1);
+    if (line) handle(JSON.parse(line));
+  }
+});