From c07c1f95d3aee1705f96d67c3dfc1e3d1d7fa32b Mon Sep 17 00:00:00 2001 From: Dotta Date: Sun, 23 Aug 2026 20:08:09 +0000 Subject: [PATCH] fix(runner): resume pending Codex calls exactly once Co-Authored-By: Paperclip --- .../crates/runner-core/src/durable/runner.rs | 50 ++-- .../paperclip-runner/src/cli/eval-session.ts | 2 + .../durable-prp-control-plane.ts | 234 +++++++++++++++--- .../paperclip-runner/src/live/live-session.ts | 21 ++ .../src/mock-core/durable-recovery.test.ts | 93 +++++++ .../fake-native-resume-codex-app-server.mjs | 175 +++++++++++++ 6 files changed, 521 insertions(+), 54 deletions(-) create mode 100644 packages/paperclip-runner/test/fixtures/fake-native-resume-codex-app-server.mjs diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/durable/runner.rs b/packages/paperclip-runner/runner/crates/runner-core/src/durable/runner.rs index 76a0b0d16b5..dacb4776a8f 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/durable/runner.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/durable/runner.rs @@ -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 } => { diff --git a/packages/paperclip-runner/src/cli/eval-session.ts b/packages/paperclip-runner/src/cli/eval-session.ts index 096909826b4..43bbd87d393 100644 --- a/packages/paperclip-runner/src/cli/eval-session.ts +++ b/packages/paperclip-runner/src/cli/eval-session.ts @@ -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; const snapshot = execution.snapshot as Parameters[0]; diff --git a/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts b/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts index f82ae0558e6..7d0a152d143 100644 --- a/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts +++ b/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts @@ -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> { + 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(); const discoveryEvidence: Array> = []; + let nativeResumeCall: { + callId: string; + operationId: string; + input: unknown; + } | null = null; + let releaseNativeSemanticDispatch!: () => void; + const nativeSemanticDispatchReleased = new Promise((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(); 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 | undefined + )?.payload as Record | undefined; + const semantic = payload?.semantic_tool as + | Record + | 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 + | undefined; + const semantic = payload?.semantic_tool as + | Record + | 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(); diff --git a/packages/paperclip-runner/src/live/live-session.ts b/packages/paperclip-runner/src/live/live-session.ts index ad08ff3f1e9..4cefc77343c 100644 --- a/packages/paperclip-runner/src/live/live-session.ts +++ b/packages/paperclip-runner/src/live/live-session.ts @@ -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. */ diff --git a/packages/paperclip-runner/src/mock-core/durable-recovery.test.ts b/packages/paperclip-runner/src/mock-core/durable-recovery.test.ts index d98c58f4ba5..05d1a7c5d22 100644 --- a/packages/paperclip-runner/src/mock-core/durable-recovery.test.ts +++ b/packages/paperclip-runner/src/mock-core/durable-recovery.test.ts @@ -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; + const proof = snapshot.nativeResume as Record; + const evidence = snapshot.evidence as Array>; + 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; + return entry.kind === "tool_call" && data.callId === "call-native-resume"; + })).toHaveLength(1); + expect(evidence.filter((entry) => { + const data = entry.data as Record; + 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(); diff --git a/packages/paperclip-runner/test/fixtures/fake-native-resume-codex-app-server.mjs b/packages/paperclip-runner/test/fixtures/fake-native-resume-codex-app-server.mjs new file mode 100644 index 00000000000..c7d00153c87 --- /dev/null +++ b/packages/paperclip-runner/test/fixtures/fake-native-resume-codex-app-server.mjs @@ -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)); + } +});