diff --git a/server/src/services/native-runtime/native-bootstrap-proof.ts b/server/src/services/native-runtime/native-bootstrap-proof.ts index 786db4d3034..1e0476d4e27 100644 --- a/server/src/services/native-runtime/native-bootstrap-proof.ts +++ b/server/src/services/native-runtime/native-bootstrap-proof.ts @@ -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 { + 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, + runnerJournal: Record | 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)); } 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 74520fb3c55..664e41b7ffb 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -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) => void]>([ + ["session.open completed", f => { + (f.controlPlane.commands as Array>)[1]!.status = "completed"; + (f.runner.processedCommands as Record>).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).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 = {}) => ({ + sourceSeq, eventType, priority: 0, + envelope: { kind: "event", payload: { eventType, payload } }, + }); + const drainedAfterSessionOpen = (pid: number) => { + const fixture = connectedBeforeSessionOpen(); + const commands = fixture.controlPlane.commands as Array>; + 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>).command_prp_00000002 = { + commandType: "session.open", status: "completed", result: { processId: pid, status: "started" } }; + return fixture; + }; + async function provesLocally(fixture: ReturnType, 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).command_prp_00000003 = { + commandType: "runner.suspend", status: "completed" }; + expect(await provesLocally(suspended)).toBe(true); + }); + + it.each<[string, (fixture: ReturnType) => 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).command_prp_00000003 = { commandType: "turn.start", status: "pending" }; }], + ["the session resumed an earlier conversation", f => { + const open = (f.controlPlane.commands as Array>)[1]!; + (open.result as { result: Record }).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>)[1]!; + delete (open.result as { result: Record }).result.processId; + }], + ["the runner journal is missing", f => { f.runner = null as unknown as Record; }], + ])("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", () => { diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 4efe6dfe7ed..35798b3fa77 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -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; }