fix(native): re-bootstrap a runner that a graceful drain suspended during bootstrap (#19665)

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work
> - Native runs start a local `paperclip-runnerd` process. The control
plane keeps a durable state root for each session.
> - After a restart, recovery gives a run with no provider evidence the
`bootstrap_incomplete` disposition. The executor quarantines the old
root and starts the bootstrap again, but only when the root proves that
no provider work started.
> - #19524 made that proof accept a root killed after `run.prepare` and
before `session.open` (a hard kill).
> - A graceful restart (SIGTERM) leaves different shapes. The drain
sends `runner.drain` and `runner.suspend`, and the kill can land just
after `session.open` failed or just after it started a fresh session.
The proof refused both shapes. The run failed with
`runner_state_identity_mismatch: bootstrap_not_proven_incomplete`,
although no turn was ever sent.
> - This pull request extends the proof to both drain shapes and keeps
provider work fail-closed.
> - The benefit is that a graceful restart during native bootstrap
continues automatically.

## Linked Issues or Issue Description

Refs #19524

**What happened?**
Restart chaos waves (ACPX Claude, K2 graceful restart about 10 s after
the run started) failed with `adapter_failed` and
`runner_state_identity_mismatch: bootstrap_not_proven_incomplete`.
Startup recovery chose `bootstrap_incomplete`, and the executor proof
refused the state root. Two real roots were kept:

1. `session.open` failed (the ACPX sidecar timed out), then
`runner.drain` completed and `runner.suspend` completed. There was 1
connection, no committed or acknowledged event, an empty runner outbox,
and `nextSourceSeq: 1`.
2. `session.open` completed with a freshly started session (`status:
"started"`). The committed events were only `session.started`,
`session.capabilities.updated` and `session.goal.snapshot`, and the
runner outbox held one `harness.diagnostic` (the sidecar process). No
turn command existed, and the sidecar process had exited. The database
had not yet recorded the session events, so the classifier saw no
provider evidence.

**Steps to reproduce**
1. Start an ACPX Claude task on a local runner.
2. Send SIGTERM to the control plane about 10 s after the run starts,
before the first turn.
3. Let the control plane restart.

**Expected behavior**
Recovery starts the bootstrap again, and the task continues without a
user action.

**Actual behavior**
The run fails and the task stops.

## What Changed

- `native-bootstrap-proof.ts`, pre-session proof: a `session.open` that
is pending or `failed` is now accepted, as are the runner lifecycle
commands `runner.drain`, `runner.suspend` and `runner.shutdown` in any
state. The controller journal and the runner journal must both agree.
Any committed or acknowledged event, any turn command, a completed
`session.open`, and any queued runner event still fail closed.
- New `runnerdStateIsOpenedSessionBeforeTurn`. It accepts a completed
`session.open` only when all of these are true:
  - the session was freshly started (`status: "started"`, not resumed);
- the commands are only `run.prepare`, `session.open` and runner
maintenance commands;
- every committed event and every event in the runner outbox is a
session setup or runner lifecycle event;
  - `ackedSourceSeq` equals the count of committed events, with no gap;
  - the runner journal is present;
  - every recorded provider process ID has exited on this host.
- `native-session-executor.ts`: the executor applies the opened-session
proof only for a local execution target. Remote runners stay fail-closed
for that shape, because their process IDs belong to another host. The
remote checkpoint classifier still reads only commands 1 and 2.

## Verification

- Red before the fix: "allows a fresh bootstrap after a drain suspended
a runner whose session.open failed" and "allows a fresh bootstrap after
a drain that landed between session.open and the first turn" both failed
with `expected false to be true`. Green after the fix: the `native
incomplete-bootstrap evidence` block passes 32/32.
- Fail-closed counter-tests:
  - completed `session.open`;
  - committed event, acknowledged event, or queued runner event;
  - turn command, `turn.stop`, or a turn processed by the runner;
  - `turn.started` or `provider.event`;
  - resumed session;
  - compacted events;
  - no recorded provider process;
  - missing runner journal;
  - live provider process;
  - remote runner.
- I ran the new proof read-only against the three real kept roots: the
K1 root from #19524, the K2 `session.open` failed root and the K2
opened-session root. All three return `true` for a local runner. The
opened-session root returns `false` when the processes are on another
host.
- `native-session-executor.test.ts` and
`native-restart-recovery.test.ts` filtered to bootstrap, restart, remote
and drain: 333 passed.
`native-runner-restart-recovery.integration.test.ts` and
`native-bootstrap-retry.integration.test.ts`: 106 passed. `tsc --noEmit
-p server/tsconfig.json` passes.
- Live restart chaos replay: see the PR comment.

## Risks

- After a `session.open` failure or timeout, the provider sidecar might
have created an empty session. No turn command was sent, so no provider
work can repeat.
- An opened session before any turn is discarded, and a new one is
started. The proof requires a fresh `started` session, so no earlier
conversation is lost. Resumed sessions stay fail-closed.
- The opened-session proof is local only. It requires every recorded
provider process to be gone, so a live provider process blocks it.
- Out of scope: the Retry offer when the proof fails legitimately
(RC-05/06).

> For core feature work, check [`ROADMAP.md`](ROADMAP.md) first and
discuss it in `#dev` before opening the PR. Feature PRs that overlap
with planned core work may need to be redirected — check the roadmap
first. See `CONTRIBUTING.md`.

## Model Used

- Anthropic Claude Opus 5.5 (`claude-opus-5-5`), used through Claude
Code with tool use (shell, file edits, test execution). Default context
window. Extended reasoning was available.

## 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
- [x] All Paperclip CI gates are green
- [x] 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-06 17:55:47 -06:00
1 parent 9197abf673
commit e89be6ee83
3 files changed
+263 -9

No files matched your search

@@ -53,9 +53,7 @@ export function runnerdStateIsIncompleteBootstrap(
if (record(state.bootstrapHistory).adopted === true) return false;
if (!Array.isArray(state.commands)) return false;
const commands = state.commands.map(record);
if (!commands.every(command => RUNNERD_BOOTSTRAP_COMMAND_TYPES.has(String(command.type)) &&
(command.status === "pending" ||
(command.type === "run.prepare" && command.status === "completed")))) return false;
if (!commands.every(command => preOpenCommandIsSafe(command.type, command.status))) return false;
// The controller cannot see runner events that were emitted but never
// committed. When the runner journal exists it must agree that none were.
if (runnerJournal === undefined) return true;
@@ -63,6 +61,111 @@ export function runnerdStateIsIncompleteBootstrap(
const processed = Object.values(record(runnerJournal.processedCommands)).map(record);
return Array.isArray(runnerJournal.outbox) && runnerJournal.outbox.length === 0 &&
runnerJournal.nextSourceSeq === 1 && runnerJournal.ackedSourceSeq === 0 &&
processed.every(command => RUNNERD_BOOTSTRAP_COMMAND_TYPES.has(String(command.commandType)) &&
(command.commandType === "run.prepare" || command.status !== "completed"));
processed.every(command => preOpenCommandIsSafe(command.commandType, command.status));
}
/** Runner process lifecycle commands a graceful drain sends; none reaches the provider. */
const RUNNER_LIFECYCLE_COMMAND_TYPES = new Set(["runner.drain", "runner.suspend", "runner.shutdown"]);
/**
* Before any provider session exists: run.prepare may have completed,
* session.open may be pending or have failed (no session was handed back),
* and a drain may have sent runner lifecycle commands in any state.
*/
function preOpenCommandIsSafe(type: unknown, status: unknown): boolean {
if (type === "run.prepare") return status === "pending" || status === "completed";
if (type === "session.open") return status === "pending" || status === "failed";
return RUNNER_LIFECYCLE_COMMAND_TYPES.has(String(type));
}
/** Commands that manage the runner process or read session state; none sends provider input. */
const RUNNER_MAINTENANCE_COMMAND_TYPES = new Set([
"runner.drain", "runner.suspend", "runner.shutdown", "session.snapshot", "session.goal.get",
]);
/** Events a runner emits while opening a fresh session or managing its own process, before any turn. */
const PRE_TURN_EVENT_TYPES = new Set([
"session.starting", "session.started", "session.capabilities.updated", "session.goal.snapshot",
"harness.starting", "harness.ready", "harness.diagnostic",
"runner.connected", "runner.disconnected", "runner.reconnected", "runner.diagnostic", "runner.backpressure",
"runner.drain", "runner.draining", "runner.suspending", "runner.suspended", "runner.shutdown", "runner.stopped",
]);
/** Collects every recorded process and process-group id so each can be proven gone. */
function collectRecordedProcessIds(value: unknown, into: Array<{ id: unknown; group: boolean }>, depth = 0): boolean {
if (depth > 10) return false;
if (Array.isArray(value)) return value.every(entry => collectRecordedProcessIds(entry, into, depth + 1));
if (!value || typeof value !== "object") return true;
for (const [key, entry] of Object.entries(value)) {
if (key === "processId" || key === "pid") into.push({ id: entry, group: false });
else if (key === "processGroupId") { if (entry !== null) into.push({ id: entry, group: true }); }
else if (!collectRecordedProcessIds(entry, into, depth + 1)) return false;
}
return true;
}
function prpEventType(event: Record<string, unknown>): string {
return String(event.eventType ?? record(record(event.envelope).payload).eventType);
}
export type OpenedSessionProofOptions = {
/** True only when the runner and its provider processes ran on this host. */
localProcesses: boolean;
/** Whether a recorded process (or process-group) id has definitely exited. */
processGone: (id: unknown, group: boolean) => boolean;
};
/**
* A drain or restart can land after session.open started a fresh provider
* session but before the controller issued any turn. That root holds an
* empty provider session and no provider work, so recovery may quarantine it
* and bootstrap again when all of these hold:
*
* - controller commands are run.prepare and session.open (both completed, the
* session freshly started rather than a resumed conversation) plus runner
* maintenance only;
* - every committed event and every event left in the runner outbox is
* session setup or runner lifecycle, and compaction left no gap;
* - the runner journal is present and processed no other command; and
* - the session's provider process and every other recorded process id has
* exited on this host.
*
* Remote runners stay fail-closed here: their process ids belong to another host.
*/
export function runnerdStateIsOpenedSessionBeforeTurn(
state: Record<string, unknown>,
runnerJournal: Record<string, unknown> | null | undefined,
options: OpenedSessionProofOptions,
): boolean {
if (!options.localProcesses || !runnerJournal) return false;
if (!RUNNERD_CONTROL_PLANE_STATE_SCHEMAS.has(String(state.schema))) return false;
if (state.warmTransition != null || state.completedWarmTransition != null ||
state.stagedRunAttachment != null) return false;
if (record(state.bootstrapHistory).adopted === true) return false;
if (!Array.isArray(state.commands) || !Array.isArray(state.committedEvents)) return false;
const commands = state.commands.map(record);
const prepares = commands.filter(command => command.type === "run.prepare");
const opens = commands.filter(command => command.type === "session.open");
if (prepares.length !== 1 || prepares[0]!.status !== "completed") return false;
if (opens.length !== 1 || opens[0]!.status !== "completed") return false;
if (!commands.every(command => RUNNERD_BOOTSTRAP_COMMAND_TYPES.has(String(command.type)) ||
RUNNER_MAINTENANCE_COMMAND_TYPES.has(String(command.type)))) return false;
const opened = record(record(opens[0]!.result).result);
if (opened.status !== "started" || typeof opened.processId !== "number") return false;
const committed = state.committedEvents.map(record);
if (state.ackedSourceSeq !== committed.length ||
!committed.every((event, index) => event.sourceSeq === index + 1)) return false;
if (!Array.isArray(runnerJournal.outbox)) return false;
const outbox = runnerJournal.outbox.map(record);
if (![...committed, ...outbox].every(event => PRE_TURN_EVENT_TYPES.has(prpEventType(event)))) return false;
const processed = Object.values(record(runnerJournal.processedCommands)).map(record);
if (!processed.every(command => RUNNERD_BOOTSTRAP_COMMAND_TYPES.has(String(command.commandType)) ||
RUNNER_MAINTENANCE_COMMAND_TYPES.has(String(command.commandType)))) return false;
const ids: Array<{ id: unknown; group: boolean }> = [{ id: opened.processId, group: false }];
if (!collectRecordedProcessIds([...committed, ...outbox, ...processed.map(command => command.result)], ids)) {
return false;
}
return ids.every(({ id, group }) => options.processGone(id, group));
}
@@ -1718,6 +1718,145 @@ describe("native incomplete-bootstrap evidence", () => {
it("stays fail-closed when the runner journal is unreadable", async () => {
expect(await provesWith(connectedBeforeSessionOpen(), "torn")).toBe(false);
});
// A graceful drain (SIGTERM) during ACPX bootstrap: session.open failed when
// the sidecar timed out, then the drain sent runner.drain and runner.suspend.
// No event was emitted, so nothing reached the provider.
const drainedBeforeSessionOpened = (sessionOpenStatus = "failed") => {
const fixture = connectedBeforeSessionOpen();
const lifecycle = [
{ type: "runner.drain", status: "completed", result: { result: { retainedEventsDrained: true, status: "completed" } } },
{ type: "runner.suspend", status: "completed", result: { result: { status: "completed" } } },
];
fixture.controlPlane.commands = [
{ type: "run.prepare", status: "completed", result: { result: { status: "prepared" } } },
{ type: "session.open", status: sessionOpenStatus, result: { result: { code: "command_execution_failed" } } },
...lifecycle,
];
fixture.controlPlane.duplicateCommandResults = 2;
fixture.runner.lifecycle = "suspended";
fixture.runner.processedCommands = {
command_prp_00000001: { commandType: "run.prepare", status: "completed" },
command_prp_00000002: { commandType: "session.open", status: sessionOpenStatus },
command_close_drain: { commandType: "runner.drain", status: "completed" },
command_close_suspend: { commandType: "runner.suspend", status: "completed" },
};
return fixture;
};
it("allows a fresh bootstrap after a drain suspended a runner whose session.open failed", async () => {
expect(await provesWith(drainedBeforeSessionOpened())).toBe(true);
expect(await provesWith(drainedBeforeSessionOpened("pending"))).toBe(true);
const shutdown = drainedBeforeSessionOpened();
(shutdown.controlPlane.commands as unknown[]).push({ type: "runner.shutdown", status: "pending" });
expect(await provesWith(shutdown)).toBe(true);
});
it.each<[string, (fixture: ReturnType<typeof connectedBeforeSessionOpen>) => void]>([
["session.open completed", f => {
(f.controlPlane.commands as Array<Record<string, unknown>>)[1]!.status = "completed";
(f.runner.processedCommands as Record<string, Record<string, unknown>>).command_prp_00000002!.status = "completed";
}],
["an event was committed", f => { f.controlPlane.committedEvents = [{ sourceSeq: 1, eventType: "runner.suspended" }]; }],
["an event was acknowledged", f => { f.controlPlane.ackedSourceSeq = 1; }],
["a turn was issued", f => { (f.controlPlane.commands as unknown[]).push({ type: "turn.start", status: "pending" }); }],
["the drain stopped a turn", f => { (f.controlPlane.commands as unknown[]).push({ type: "turn.stop", status: "completed" }); }],
["the runner queued an event", f => { f.runner.nextSourceSeq = 2; f.runner.outbox = [{ sourceSeq: 1, eventType: "runner.suspended" }]; }],
["the runner processed a turn", f => { (f.runner.processedCommands as Record<string, unknown>).command_prp_00000005 = { commandType: "turn.start", status: "pending" }; }],
])("stays fail-closed after a drain when %s", async (_label, mutate) => {
const fixture = drainedBeforeSessionOpened();
mutate(fixture);
expect(await provesWith(fixture)).toBe(false);
});
// A graceful drain (SIGTERM) can land after session.open started a fresh
// provider session but before the controller issued any turn. The committed
// events are session setup only, and the session's provider process is gone.
const deadPid = async () => {
const child = spawn(process.execPath, ["-e", ""], { stdio: "ignore" });
await new Promise(resolve => child.once("exit", resolve));
return child.pid!;
};
const sessionEvent = (sourceSeq: number, eventType: string, payload: Record<string, unknown> = {}) => ({
sourceSeq, eventType, priority: 0,
envelope: { kind: "event", payload: { eventType, payload } },
});
const drainedAfterSessionOpen = (pid: number) => {
const fixture = connectedBeforeSessionOpen();
const commands = fixture.controlPlane.commands as Array<Record<string, unknown>>;
commands[1] = { type: "session.open", status: "completed", result: {
commandType: "session.open", status: "completed",
result: { driver: "acpx_runtime", processId: pid, provider: "acpx", providerSessionId: "provider-session",
sessionId: "account-session", status: "started" },
} };
fixture.controlPlane.committedEvents = [
sessionEvent(1, "session.started", { driver: "acpx_runtime", processId: pid, provider: "acpx" }),
sessionEvent(2, "session.capabilities.updated"),
sessionEvent(3, "session.goal.snapshot", { goal: null }),
];
fixture.controlPlane.ackedSourceSeq = 3;
fixture.runner.nextSourceSeq = 5;
fixture.runner.ackedSourceSeq = 3;
fixture.runner.outbox = [{ sourceSeq: 4, priority: 1, eventType: "harness.diagnostic", envelope: { kind: "event",
payload: { eventType: "harness.diagnostic", payload: { code: "acpx_process",
details: { pid, processGroupId: null, role: "sidecar" } } } } }];
(fixture.runner.processedCommands as Record<string, Record<string, unknown>>).command_prp_00000002 = {
commandType: "session.open", status: "completed", result: { processId: pid, status: "started" } };
return fixture;
};
async function provesLocally(fixture: ReturnType<typeof connectedBeforeSessionOpen>, localProcesses = true) {
const root = await mkdtemp(join(tmpdir(), "native-bootstrap-drained-"));
try {
await mkdir(join(root, "control-plane"), { recursive: true });
await mkdir(join(root, "runner"), { recursive: true });
await writeFile(join(root, "control-plane", "control-plane-state.json"), JSON.stringify(fixture.controlPlane));
await writeFile(join(root, "runner", "runner-state.json"), JSON.stringify(fixture.runner));
return runnerdStateProvesIncompleteBootstrap(root, { localProcesses });
} finally {
await rm(root, { recursive: true, force: true });
}
}
it("allows a fresh bootstrap after a drain that landed between session.open and the first turn", async () => {
const pid = await deadPid();
expect(await provesLocally(drainedAfterSessionOpen(pid))).toBe(true);
const suspended = drainedAfterSessionOpen(pid);
(suspended.controlPlane.commands as unknown[]).push({ type: "runner.suspend", status: "completed" });
(suspended.controlPlane.committedEvents as unknown[]).push(sessionEvent(4, "runner.suspended"));
suspended.controlPlane.ackedSourceSeq = 4;
(suspended.runner.processedCommands as Record<string, unknown>).command_prp_00000003 = {
commandType: "runner.suspend", status: "completed" };
expect(await provesLocally(suspended)).toBe(true);
});
it.each<[string, (fixture: ReturnType<typeof connectedBeforeSessionOpen>) => void]>([
["a turn was issued", f => { (f.controlPlane.commands as unknown[]).push({ type: "turn.start", status: "pending" }); }],
["a turn started", f => { (f.controlPlane.committedEvents as unknown[]).push(sessionEvent(4, "turn.started")); f.controlPlane.ackedSourceSeq = 4; }],
["the runner queued provider output", f => { (f.runner.outbox as unknown[]).push(sessionEvent(5, "provider.event")); }],
["the runner processed a turn", f => { (f.runner.processedCommands as Record<string, unknown>).command_prp_00000003 = { commandType: "turn.start", status: "pending" }; }],
["the session resumed an earlier conversation", f => {
const open = (f.controlPlane.commands as Array<Record<string, unknown>>)[1]!;
(open.result as { result: Record<string, unknown> }).result.status = "resumed";
}],
["committed events were compacted", f => { f.controlPlane.ackedSourceSeq = 7; }],
["session.open recorded no provider process", f => {
const open = (f.controlPlane.commands as Array<Record<string, unknown>>)[1]!;
delete (open.result as { result: Record<string, unknown> }).result.processId;
}],
["the runner journal is missing", f => { f.runner = null as unknown as Record<string, unknown>; }],
])("stays fail-closed after session.open when %s", async (_label, mutate) => {
const fixture = drainedAfterSessionOpen(await deadPid());
mutate(fixture);
expect(await provesLocally(fixture)).toBe(false);
});
it("stays fail-closed after session.open while the provider process is alive", async () => {
expect(await provesLocally(drainedAfterSessionOpen(process.pid))).toBe(false);
});
it("stays fail-closed after session.open for a runner whose processes are on another host", async () => {
expect(await provesLocally(drainedAfterSessionOpen(await deadPid()), false)).toBe(false);
});
});
describe("native provider usage normalization", () => {
@@ -90,7 +90,7 @@ import { createNativeProviderFailureObservation } from "./native-provider-failur
import { nativeStartupRetryExhaustedAction, nativeStartupRetryWait, planNativeStartupRetry } from "./native-startup-retry.js";
import { admitUnclaimedNativeBootstrapRetry } from "./native-bootstrap-retry.js";
import { RUNNERD_CONTROL_PLANE_STATE_SCHEMAS, runnerdStateIsIncompleteBootstrap,
runnerdStateIsUnconnectedBootstrap } from "./native-bootstrap-proof.js";
runnerdStateIsOpenedSessionBeforeTurn, runnerdStateIsUnconnectedBootstrap } from "./native-bootstrap-proof.js";
import { createNativeDeliverableTextObservation } from "./native-deliverable-text.js";
import { createNativeGitHubAccess, type NativeGitHubAccess } from "./native-github-access.js";
import { resolveGitHubOperationCredentials } from "../github-operation-credentials.js";
@@ -5589,7 +5589,10 @@ async function migrateRunnerdStateRootForExecution(input: {
savePluginRunnerCheckpoint(scoped, result.state, identity);
}
if (input.restartRecovery?.kind === "bootstrap_incomplete") {
if (runnerdStateProvesIncompleteBootstrap(scoped)) {
// A local runner's process ids belong to this host, so a fresh provider
// session it opened before any turn can be proven gone here.
const localProcesses = !input.runnerExecutionTarget || input.runnerExecutionTarget.kind === "local";
if (runnerdStateProvesIncompleteBootstrap(scoped, { localProcesses })) {
quarantineRunnerdStateRoot(scoped, "identity_indeterminate");
return;
}
@@ -6072,9 +6075,18 @@ export async function restoreRemoteBootstrapCheckpointForRecovery(input: {
await restoreNativeRunnerCheckpoint(input.db, input.execution.binding, input.root, nativeSessionKey(input.execution));
}
export function runnerdStateProvesIncompleteBootstrap(root: string): boolean {
export function runnerdStateProvesIncompleteBootstrap(
root: string,
options: { localProcesses: boolean } = { localProcesses: false },
): boolean {
try {
return runnerdStateIsIncompleteBootstrap(readRunnerdControlPlaneState(root), readRunnerdBootstrapJournal(root));
const state = readRunnerdControlPlaneState(root);
const journal = readRunnerdBootstrapJournal(root);
return runnerdStateIsIncompleteBootstrap(state, journal) ||
runnerdStateIsOpenedSessionBeforeTurn(state, journal, {
localProcesses: options.localProcesses,
processGone: (id, group) => localProcessDefinitelyGone(id, group),
});
} catch {
return false;
}