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 977a426dccb..47c09f6bde5 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -1638,6 +1638,82 @@ describe("native incomplete-bootstrap evidence", () => { await rm(root, { recursive: true, force: true }); } }); + + // A restart can land after the runner connected and acknowledged + // run.prepare but before session.open finished. Both journals then prove no + // provider session or turn exists, so recovery may bootstrap again. + const connectedBeforeSessionOpen = () => ({ + controlPlane: { + schema: "paperclip.runner.durable.control-plane-state.v2", + identity: { runId: "run-1", normalizedSessionId: "session-1" }, + bootstrapHistory: { schema: "paperclip.runner.bootstrap-history.v1", credentialsIssued: 1, adopted: false }, + commands: [ + { type: "run.prepare", status: "completed", result: { status: "completed" } }, + { type: "session.open", status: "pending", result: null }, + ] as Array>, + committedEvents: [] as unknown[], + ackedSourceSeq: 0, + connectionCount: 1, + freshBootstraps: 1, + } as Record, + runner: { + schema: "paperclip.runner.durable.state.v1", + lifecycle: "ready", + nextSourceSeq: 1, + ackedSourceSeq: 0, + outbox: [] as unknown[], + processedCommands: { + command_prp_00000001: { commandType: "run.prepare", status: "completed" }, + command_prp_00000002: { commandType: "session.open", status: "pending", + result: { code: "execution_indeterminate" } }, + } as Record>, + } as Record, + }); + + async function provesWith(fixture: ReturnType, runner: "present" | "absent" | "torn" = "present") { + const root = await mkdtemp(join(tmpdir(), "native-bootstrap-connected-")); + try { + await mkdir(join(root, "control-plane"), { recursive: true }); + await writeFile(join(root, "control-plane", "control-plane-state.json"), JSON.stringify(fixture.controlPlane)); + if (runner !== "absent") { + await mkdir(join(root, "runner"), { recursive: true }); + await writeFile(join(root, "runner", "runner-state.json"), + runner === "torn" ? "{\"schema\":" : JSON.stringify(fixture.runner)); + } + return runnerdStateProvesIncompleteBootstrap(root); + } finally { + await rm(root, { recursive: true, force: true }); + } + } + + it("allows a fresh bootstrap after connect and run.prepare when session.open never completed", async () => { + expect(await provesWith(connectedBeforeSessionOpen())).toBe(true); + expect(await provesWith(connectedBeforeSessionOpen(), "absent")).toBe(true); + const preparing = connectedBeforeSessionOpen(); + (preparing.controlPlane.commands as Array>)[0]!.status = "pending"; + expect(await provesWith(preparing)).toBe(true); + }); + + it.each<[string, (fixture: ReturnType) => void]>([ + ["session.open completed", f => { (f.controlPlane.commands as Array>)[1]!.status = "completed"; }], + ["an event was committed", f => { f.controlPlane.committedEvents = [{ eventType: "session.started" }]; }], + ["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" }); }], + ["run.prepare failed", f => { (f.controlPlane.commands as Array>)[0]!.status = "failed"; }], + ["the state was adopted", f => { f.controlPlane.bootstrapHistory = { adopted: true }; }], + ["a warm transition is retained", f => { f.controlPlane.warmTransition = { receipt: {} }; }], + ["the runner emitted an uncommitted event", f => { f.runner.nextSourceSeq = 2; f.runner.outbox = [{ sourceSeq: 1 }]; }], + ["the runner completed session.open", f => { (f.runner.processedCommands as Record>).command_prp_00000002!.status = "completed"; }], + ["the runner processed a turn", f => { (f.runner.processedCommands as Record).command_prp_00000003 = { commandType: "turn.start", status: "pending" }; }], + ])("stays fail-closed when %s", async (_label, mutate) => { + const fixture = connectedBeforeSessionOpen(); + mutate(fixture); + expect(await provesWith(fixture)).toBe(false); + }); + + it("stays fail-closed when the runner journal is unreadable", async () => { + expect(await provesWith(connectedBeforeSessionOpen(), "torn")).toBe(false); + }); }); describe("native provider usage normalization", () => { diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index ac41ebd8d80..9a9382c8363 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -5952,32 +5952,94 @@ function localProcessDefinitelyGone(value: unknown, group = false): boolean { } } -function runnerdStateIsIncompleteBootstrap(state: Record): boolean { +const RUNNERD_BOOTSTRAP_COMMAND_TYPES = new Set(["run.prepare", "session.open"]); + +/** The runner never connected, so every bootstrap command is still undelivered. */ +function runnerdStateIsUnconnectedBootstrap(state: Record): boolean { const commands = Array.isArray(state.commands) ? state.commands.map(record) : []; const committedEvents = Array.isArray(state.committedEvents) ? state.committedEvents : []; return RUNNERD_CONTROL_PLANE_STATE_SCHEMAS.has(String(state.schema)) && state.connectionCount === 0 && committedEvents.length === 0 && commands.every(command => command.status === "pending" && - (command.type === "run.prepare" || command.type === "session.open")); + RUNNERD_BOOTSTRAP_COMMAND_TYPES.has(String(command.type))); +} + +/** + * A provider turn can start only after session.open completes and the + * controller issues a turn command. A root whose controller journal stops + * before that point (run.prepare may be acknowledged, session.open never is, + * no event was committed or acknowledged) and whose runner journal emitted no + * event either holds no provider work, so restart recovery may quarantine it + * and bootstrap again. Any other shape stays fail-closed. + */ +function runnerdStateIsIncompleteBootstrap( + state: Record, + runnerJournal?: Record | null, +): boolean { + if (runnerdStateIsUnconnectedBootstrap(state)) return true; + if (!RUNNERD_CONTROL_PLANE_STATE_SCHEMAS.has(String(state.schema))) return false; + if (typeof state.connectionCount !== "number" || !Number.isSafeInteger(state.connectionCount) || + state.connectionCount < 0) return false; + if (!Array.isArray(state.committedEvents) || state.committedEvents.length !== 0) return false; + if (state.ackedSourceSeq !== 0) 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)) 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; + // 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; + if (runnerJournal === null) return false; + 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")); +} + +/** `undefined` when the runner never journaled, `null` when unreadable. */ +function readRunnerdBootstrapJournal(root: string): Record | null | undefined { + const runnerRoot = resolve(root, "runner"); + if (!existsSync(runnerRoot)) return undefined; + if (!isSafeNativeStateDirectory(runnerRoot)) return null; + const statePath = resolve(runnerRoot, "runner-state.json"); + if (!existsSync(statePath)) return undefined; + try { + const state = record(JSON.parse(readBoundedNativeFile(statePath, NATIVE_RUNNER_STATE_MAX_BYTES, + "runner_state_too_large").toString("utf8"))); + return state.schema === "paperclip.runner.durable.state.v1" ? state : null; + } catch { + return null; + } +} + +function readRunnerdControlPlaneState(root: string): Record { + return record( + JSON.parse( + readBoundedNativeFile( + resolve(root, "control-plane", "control-plane-state.json"), + NATIVE_CONTROL_PLANE_STATE_MAX_BYTES, + "runner_durable_identity_too_large", + ).toString("utf8"), + ), + ); } export function runnerdStateProvesIncompleteBootstrap(root: string): boolean { try { - const statePath = resolve( - root, - "control-plane", - "control-plane-state.json", - ); - const state = record( - JSON.parse( - readBoundedNativeFile( - statePath, - NATIVE_CONTROL_PLANE_STATE_MAX_BYTES, - "runner_durable_identity_too_large", - ).toString("utf8"), - ), - ); - return runnerdStateIsIncompleteBootstrap(state); + return runnerdStateIsIncompleteBootstrap(readRunnerdControlPlaneState(root), readRunnerdBootstrapJournal(root)); + } catch { + return false; + } +} + +function runnerdStateProvesUnconnectedBootstrap(root: string): boolean { + try { + return runnerdStateIsUnconnectedBootstrap(readRunnerdControlPlaneState(root)); } catch { return false; } @@ -14159,7 +14221,7 @@ async function createRunnerdBackendWithinSessionClaim( if (!durableIdentityMatchesExecution(identity, input.execution) || identity?.runnerInstanceId !== effectiveRunnerInstanceId || identity?.environmentLeaseId !== effectiveEnvironmentLeaseId || - !(controllerState ? runnerdStateIsIncompleteBootstrap(record(controllerState)) : runnerdStateProvesIncompleteBootstrap(root))) return false; + !(controllerState ? runnerdStateIsUnconnectedBootstrap(record(controllerState)) : runnerdStateProvesUnconnectedBootstrap(root))) return false; // Workspace reuse or replacement can acquire a lease for a new session. // Claim this exact new session atomically under readable real directories. // Missing files inside an existing session never authorize a fresh start.