fix(runner): answer held Paperclip tool calls after a local restart before the next turn (#20227)

## 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 <noreply@anthropic.com>
This commit is contained in:
DottaandClaude Opus 5.5 authored and GitHub committed 2026-11-09 05:26:12 -06:00
1 parent c5da0b88ad
commit 3f2a940d5b
5 files changed
+97 -6

No files matched your search

@@ -2262,7 +2262,20 @@ export class DurablePrpControlPlane {
this.#semanticInputSourceRunId = null;
}
async recoverPendingSemanticInputs(): Promise<void> {
#semanticRecoveryPass: Promise<void> | null = null;
recoverPendingSemanticInputs(): Promise<void> {
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<void> {
if (this.#retainsSourceInputs() || !this.#recoverSemanticToolInput) return;
const identity = this.#identity;
for (const saved of [...this.#store.state.committedEvents]) {
@@ -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<void>((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) => {
@@ -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<void>((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) : [];
@@ -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";
@@ -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))