From 3f2a940d5bb6247f1c3b385ac3293f569bd0dc80 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Mon, 9 Nov 2026 05:26:12 -0600 Subject: [PATCH] fix(runner): answer held Paperclip tool calls after a local restart before the next turn (#20227) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - When a restart interrupts a turn, the Runner keeps each Paperclip tool call the model made as "held" until Paperclip answers it, and it refuses to start the next turn while any call is held > - Paperclip answers held calls after a reattach through its tool recovery path. That path was wired only for remote runners > - In restart chaos wave `w-20261109T084023-s9403`, t101 (local codex, K2 then K1) held a `read_document` call that nothing answered, so the continuation's `turn.start` failed and the task needed a Retry > - This pull request wires held-call recovery for local runners too, and makes `turn.start` answer or wait for held calls first > - The benefit is that a turn lost to a restart no longer strands the run behind one unanswered call ## Linked Issues or Issue Description Refs #19831. Restart chaos wave `w-20261109T084023-s9403` (master `fc5c79a327`), task t101 (codex C1, K2, local). **What happened:** - Run `4b5810d5` called `read_document` (`exec-ac8fa9b4`) around the K2 drain at 09:08:18. A K1 then killed the next server start. - At 09:08:36 the third controller reattached. The Runner restored the conversation and reported the old turn lost (`provider_turn_lost_on_restore`, recoverable). The held call stayed pending, by design: a held call accepts a late result. - The controller had the call: the quarantine detail lists `exec-ac8fa9b4` under `semanticTools.pending`. But `recoverPendingSemanticInputs()` returned at once: for a local run `controllerPersistence` is unset, so `recoverSemanticToolInput` was `undefined`. - At 09:08:46 the continuation's `turn.start` failed with "cannot prepare the next provider turn while tool calls are active (pending: 1 [read_document:exec-ac8fa9b4…])". The run ended `native_session_interrupted`. ## What Changed - `server/src/services/native-runtime/native-session-executor.ts`: the runnerd transport now always gets `recoverSemanticToolInput`, not only when `controllerPersistence` (remote) is set. It uses the same epoch check and `SessionAuthorityEpoch.recover`. Recovery re-authorizes and re-runs reads, and replays receipts for mutations; it never repeats an uncertain mutation. - `packages/paperclip-runner/src/live/runnerd-codex-transport.ts` (`#startTurnWork`): when the control plane has unsettled semantic inputs, it runs `recoverPendingSemanticInputs()` before queueing `turn.start`, and waits for that pass, bounded by the turn-start deadline. Recovered results are queued as commands ahead of `turn.start`, and the Runner applies commands in order. If recovery has not finished by the deadline, the turn start fails with `native_turn_start_held_tool_calls_unresolved` instead of sending `turn.start` behind it. When everything is settled there is no wait, so a normal turn start is unchanged. Live calls of a turn being replaced are not waited on; the replaced start still rejects them. - `packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts`: `recoverPendingSemanticInputs()` passes are serialized. A call made while the startup pass is still answering a held call waits for that pass, then runs its own, so the turn-start gate cannot finish early by skipping a call that is already being recovered. ## Verification Red/green: - `native-session-executor.test.ts` "answers a local run's held Paperclip tool calls after a restart (restart chaos t101)". On master `recoverSemanticToolInput` is `undefined` for a local run; with this PR it is wired. - `runnerd-codex-transport.test.ts` "answers a held call before starting the next turn (restart chaos t101)", against real runnerd and fake Codex, with a slow startup recovery pass already in flight when `turn/start` arrives. On master `turn/start` never ran recovery, and without the serialization `turn.start` is queued before the held call's result. With this PR the held call is recovered once and its result is queued before `turn.start`. Regression and static checks: - `runnerd-codex-transport.test.ts`, `durable-prp-control-plane.test.ts`, `runnerd-codex-transport-adoption.test.ts`: 510 passed. - `native-session-executor.test.ts` and `native-session-resume.test.ts`: 1368 passed, 1 failed. The failing case, "archives a damaged prior epoch … with real runnerd", fails identically on master on this Mac (`ENOENT … authority-epochs`), so CI is its reference. - `tsc --noEmit` passes for `packages/paperclip-runner` and `server`. ## Risks - Local runs now recover held calls after a reattach, as remote runs already did. Reads re-run under current authority. A mutation without a saved receipt still returns no result and stays held, as before. - `turn.start` may now wait up to its existing deadline while a recovery pass runs, then fails clearly if it has not finished. It never waits when nothing is unsettled, so normal turns get no added latency before `turn.start`. - No schema change and no change to persisted run state formats. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, via Claude Code with tool use (shell, file edits). ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 --- .../durable-prp-control-plane.ts | 15 +++++- .../src/live/runnerd-codex-transport.test.ts | 54 +++++++++++++++++++ .../src/live/runnerd-codex-transport.ts | 18 ++++++- .../native-session-executor.test.ts | 8 +++ .../native-runtime/native-session-executor.ts | 8 +-- 5 files changed, 97 insertions(+), 6 deletions(-) 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 eaa91d2006e..e2e61d45b72 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 @@ -2262,7 +2262,20 @@ export class DurablePrpControlPlane { this.#semanticInputSourceRunId = null; } - async recoverPendingSemanticInputs(): Promise { + #semanticRecoveryPass: Promise | null = null; + + recoverPendingSemanticInputs(): Promise { + const pass = (this.#semanticRecoveryPass ?? Promise.resolve()) + .catch(() => undefined) + .then(() => this.#recoverPendingSemanticInputsOnce()); + this.#semanticRecoveryPass = pass; + void pass.finally(() => { + if (this.#semanticRecoveryPass === pass) this.#semanticRecoveryPass = null; + }).catch(() => undefined); + return pass; + } + + async #recoverPendingSemanticInputsOnce(): Promise { if (this.#retainsSourceInputs() || !this.#recoverSemanticToolInput) return; const identity = this.#identity; for (const saved of [...this.#store.state.committedEvents]) { diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts index c772fb40098..d9627e81fa6 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts @@ -1398,6 +1398,60 @@ it("refuses a terminal call that arrives before a held call's result (restart ch } }, 20_000); +it("answers a held call before starting the next turn (restart chaos t101)", async () => { + const stateDirectory = await mkdtemp(join(tmpdir(), "runnerd-held-before-turn-")); + const OriginalCore = durableControlPlane.DurablePrpControlPlane; + let core!: DurablePrpControlPlane; + const coreSpy = vi.spyOn(durableControlPlane, "DurablePrpControlPlane").mockImplementation(function (options) { + return core = new OriginalCore(options); + }); + const order: string[] = []; + const recovered = vi.fn(async (call: { callId: string }) => { + order.push(`recover:${call.callId}`); + await new Promise((resolveRead) => setTimeout(resolveRead, 300)); + return { result: { success: true, contentItems: [{ type: "inputText", text: "brief body" }] } }; + }); + const bundle = createCapabilityRunnerdCodexTransport({ + runnerBinary: defaultCapabilityRunnerdBinary(), codexCommand: fakeCodex, + codexArgs: fakeCodexArgs(stateDirectory), stateDirectory, recoverSemanticToolInput: recovered, + }); + bundle.transport.setServerRequestHandler(vi.fn(async () => ({ success: true, contentItems: [] }))); + try { + await bundle.transport.request("thread/start", { cwd: tmpdir(), dynamicTools: codexSemanticToolSpecs() }); + const identity = core.store.state.identity; + const correlation = { runId: identity.runId, normalizedSessionId: identity.normalizedSessionId, + turnId: identity.turnId, itemId: identity.itemId }; + core.store.state.committedEvents.unshift({ sourceSeq: 0, sourceEventId: "held-read-input", eventType: "semantic_tool.input", + priority: 1, deliveryCount: 1, logicalEffectCount: 1, envelope: { payload: { runId: identity.runId, + normalizedSessionId: identity.normalizedSessionId, sourceEventId: "held-read-input", eventType: "semantic_tool.input", + itemId: identity.itemId, payload: { semantic_tool: { callId: "exec-held-read", operationId: "read_document", + input: { key: "brief" }, correlation } } } } }); + expect(core.semanticToolResultsSettled()).toBe(false); + const queueCommand = core.queueCommand.bind(core); + vi.spyOn(core, "queueCommand").mockImplementation((type, ...rest) => { + order.push(type); + return queueCommand(type, ...rest); + }); + const startupPass = core.recoverPendingSemanticInputs(); + await vi.waitFor(() => expect(recovered).toHaveBeenCalledOnce()); + const turnStarted = expectTurnStarted(bundle.transport.notifications()[Symbol.asyncIterator]()); + void turnStarted.catch(() => undefined); + await bundle.transport.request("turn/start", { input: [{ type: "text", text: "Continue after the restart." }] }); + await turnStarted; + await startupPass; + expect(recovered).toHaveBeenCalledOnce(); + expect(order.slice(0, 2)).toEqual(["recover:exec-held-read", "semantic_tool.result"]); + expect(order.indexOf("semantic_tool.result")).toBeLessThan(order.indexOf("turn.start")); + const result = core.store.state.commands.find(command => command.type === "semantic_tool.result"); + expect(result).toMatchObject({ payload: { callId: "exec-held-read", operationId: "read_document" } }); + expect(result!.status).not.toBe("pending"); + } finally { + coreSpy.mockRestore(); + await bundle.transport.close().catch(() => undefined); + await rm(stateDirectory, { recursive: true, force: true }); + } +}, 30_000); + it.each(["after_budget", "within_budget", "interrupted_within_budget", "persistence_failure"] as const)( "fences reusable suspension against late semantic completion (%s)", async (mode) => { diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts index 62b46db025e..700127cbc2e 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts @@ -7262,7 +7262,23 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { if (!Number.isSafeInteger(turnStartTimeoutMs) || turnStartTimeoutMs <= 0) { throw new Error("turnStartTimeoutMs must be a positive safe integer"); } - const commandDeadline = Date.now() + turnStartTimeoutMs; + let commandDeadline = Date.now() + turnStartTimeoutMs; + const heldCalls = this.#core; + if (typeof heldCalls?.semanticToolResultsSettled === "function" && !heldCalls.semanticToolResultsSettled()) { + let recovering = true; + void Promise.resolve(heldCalls.recoverPendingSemanticInputs()) + .catch(error => this.#failTransport(error instanceof Error ? error : new Error(String(error)))) + .finally(() => { recovering = false; }); + while (recovering && Date.now() < commandDeadline) { + this.#assertStartupOpen(); + this.#throwIfFailed(); + await new Promise((resolveWait) => setTimeout(resolveWait, 10)); + } + this.#assertStartupOpen(); + this.#throwIfFailed(); + if (recovering) throw new Error("native_turn_start_held_tool_calls_unresolved: recovering an earlier turn's held Paperclip tool calls did not finish before the turn-start deadline"); + commandDeadline = Date.now() + turnStartTimeoutMs; + } const compactionTurn = params[COMPACTION_TURN] === true; const compactionPayload = compactionTurn ? runnerdCompactionTurnPayload(this.options.provider ?? "codex") : null; const input = Array.isArray(params.input) ? params.input.map(record) : []; diff --git a/server/src/services/native-runtime/native-session-executor.test.ts b/server/src/services/native-runtime/native-session-executor.test.ts index d96451efb43..5a577b4ab17 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -764,6 +764,14 @@ describe("remote controller restart adoption", () => { await rm(stateBase, { recursive: true, force: true }); } }); + it("answers a local run's held Paperclip tool calls after a restart (restart chaos t101)", async () => { + state.createBackend.mockClear(); + state.createTransport.mockClear(); + await createRunnerdBackend({ db: leaseDb(execution), execution, runnerInstanceId: "runner" }); + state.createBackend.mock.calls.at(-1)![1].codexTransportFactory!(); + expect((state.createTransport.mock.calls.at(-1)![0] as { recoverSemanticToolInput?: unknown }).recoverSemanticToolInput) + .toEqual(expect.any(Function)); + }); it("rejects a replaced sandbox before reading or mutating it", async () => { const { execute, target } = fixture(); target.sandboxLeaseAcquisition.providerLeaseId = "replacement"; diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 74ae7795aca..9dfa84c231a 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -14904,8 +14904,8 @@ async function createRunnerdBackendWithinSessionClaim( stateDirectory: root, runnerBinary: remoteDot?.runnerBinary ?? resolvePaperclipRunnerBinary(), runnerArtifact: input.controllerPersistence?.runnerArtifact, runnerReconnectGraceMs: remoteDot ? REMOTE_RUNNER_RECONNECT_GRACE_MS : undefined, - recoverSemanticToolInput: input.controllerPersistence ? (call: { operationId: string; callId: string; input: unknown }) => - authorityEpoch.recover({ tool: call.operationId, callId: call.callId, arguments: call.input }) : undefined, + recoverSemanticToolInput: (call: { operationId: string; callId: string; input: unknown }) => + authorityEpoch.recover({ tool: call.operationId, callId: call.callId, arguments: call.input }), ...(remoteDot ?? {}), identity: { runnerInstanceId: effectiveRunnerInstanceId, environmentLeaseId: effectiveEnvironmentLeaseId, runId: input.execution.binding.runId, normalizedSessionId: nativeSessionKey(input.execution), @@ -16499,11 +16499,11 @@ async function createRunnerdBackendWithinSessionClaim( stateDirectory: root, runnerBinary: controllerRunnerBinary, runnerArtifact: input.controllerPersistence?.runnerArtifact ?? providerPackAuthority?.runnerArtifact, - recoverSemanticToolInput: input.controllerPersistence ? call => { + recoverSemanticToolInput: call => { const current = sessionToolAuthorityEpochs.get(sessionScopeId); if (!current || current.runId !== call.correlation.runId) throw new Error("native_tool_authority_epoch_revoked"); return current.recover({ tool: call.operationId, callId: call.callId, arguments: call.input }); - } : undefined, + }, codexCommand: remoteCodexBinary ?? undefined, sourceCodexHome: remoteTarget ? resolveSourceCodexHome(resolveNativeProviderEnvironment(input.execution.provider, runnerEnvironment))