diff --git a/server/src/services/native-runtime/native-bootstrap-proof.ts b/server/src/services/native-runtime/native-bootstrap-proof.ts index 85338dbf122..0c6fde2767f 100644 --- a/server/src/services/native-runtime/native-bootstrap-proof.ts +++ b/server/src/services/native-runtime/native-bootstrap-proof.ts @@ -56,8 +56,9 @@ export function runnerdStateHasNoProviderTurn(state: Record): b 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); - if (!commands.every(command => RUNNERD_BOOTSTRAP_COMMAND_TYPES.has(String(command.type)) && - (command.status === "pending" || command.status === "completed"))) return false; + if (!commands.every(command => RUNNER_MAINTENANCE_COMMAND_TYPES.has(String(command.type)) || + (RUNNERD_BOOTSTRAP_COMMAND_TYPES.has(String(command.type)) && + (command.status === "pending" || command.status === "completed")))) return false; const events = state.committedEvents.map(record); // Every event since the first must still be here, in order, so none is hidden. if (typeof state.ackedSourceSeq !== "number" || !Number.isSafeInteger(state.ackedSourceSeq) || diff --git a/server/src/services/native-runtime/native-runner-checkpoint.ts b/server/src/services/native-runtime/native-runner-checkpoint.ts index c39886e6a6a..f4b47b94e9c 100644 --- a/server/src/services/native-runtime/native-runner-checkpoint.ts +++ b/server/src/services/native-runtime/native-runner-checkpoint.ts @@ -260,7 +260,8 @@ export async function loadNativeUnclaimedBootstrapCheckpoint( } catch { return null; } } -const BOOTSTRAP_CHECKPOINT_PARTS = new Set(["header", "cursor", "command:1", "command:2"]); +const BOOTSTRAP_CHECKPOINT_PARTS = new Set(["header", "cursor"]); +const MAX_BOOTSTRAP_CHECKPOINT_COMMANDS = 8; /** Session-open events are few; a longer journal is never a bootstrap. */ const MAX_BOOTSTRAP_CHECKPOINT_EVENTS = 16; @@ -282,13 +283,15 @@ export async function nativeRunnerCheckpointRemoteBootstrapProof( db: Pick, binding: Binding, expected: { runnerInstanceId: string; normalizedSessionId: string }, ): Promise { try { - const limit = BOOTSTRAP_CHECKPOINT_PARTS.size + MAX_BOOTSTRAP_CHECKPOINT_EVENTS; + const limit = BOOTSTRAP_CHECKPOINT_PARTS.size + MAX_BOOTSTRAP_CHECKPOINT_COMMANDS + MAX_BOOTSTRAP_CHECKPOINT_EVENTS; const parts = await db.select({ part: nativeRunnerCheckpoints.part }).from(nativeRunnerCheckpoints).where(and( eq(nativeRunnerCheckpoints.companyId, binding.companyId), eq(nativeRunnerCheckpoints.issueId, binding.issueId), eq(nativeRunnerCheckpoints.runId, binding.runId))).limit(limit + 1); const eventParts = parts.map(row => row.part).filter(part => /^event:\d+$/.test(part)); - if (!parts.length || parts.length > limit - || parts.some(row => !BOOTSTRAP_CHECKPOINT_PARTS.has(row.part) && !/^event:\d+$/.test(row.part))) return null; + const commandParts = parts.map(row => row.part).filter(part => /^command:\d+$/.test(part)); + if (!parts.length || parts.length > limit || eventParts.length > MAX_BOOTSTRAP_CHECKPOINT_EVENTS + || commandParts.length > MAX_BOOTSTRAP_CHECKPOINT_COMMANDS + || parts.some(row => !BOOTSTRAP_CHECKPOINT_PARTS.has(row.part) && !/^(?:event|command):\d+$/.test(row.part))) return null; const checkpoint = await loadNativeRunnerCheckpoint(db, binding); if (!checkpoint || checkpoint.predecessor !== null || checkpoint.recoveryEvidence.runnerInstanceId !== expected.runnerInstanceId @@ -301,11 +304,13 @@ export async function nativeRunnerCheckpointRemoteBootstrapProof( }; const header = read("header"); const cursor = read("cursor"); - if (header?.schema !== "paperclip.runner.controller-checkpoint.v2" || !cursor) return null; + if (header?.schema !== "paperclip.runner.controller-checkpoint.v2" || !cursor + || cursor.commandCount !== commandParts.length) return null; const state: Record = { ...(header.state as Record), ackedSourceSeq: cursor.ackedSourceSeq, - commands: ["command:1", "command:2"].map(read).filter(command => command !== undefined), + commands: commandParts.map(part => Number(part.slice("command:".length))).sort((a, b) => a - b) + .map(seq => read(`command:${seq}`)), committedEvents: eventParts.map(part => Number(part.slice("event:".length))).sort((a, b) => a - b) .map(seq => read(`event:${seq}`)), }; diff --git a/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts b/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts index d0443de99bb..31ef5944755 100644 --- a/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts +++ b/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts @@ -2589,6 +2589,33 @@ process.on("SIGTERM", () => child.kill("SIGTERM")); expect(run).toMatchObject({ status: "running", errorCode: null }); }); + it("restarts a drained remote session whose controller only read its goal after opening it", async () => { + const readGoal = (core: DurablePrpControlPlane) => { + openSession(core, ["session.started", "session.capabilities.updated", "session.goal.snapshot", "session.goal.snapshot"]); + core.queueCommand("session.goal.get", {}); + const goal = core.store.state.commands.at(-1)!; + goal.status = "completed"; + goal.result = { goal: null } as never; + }; + const fixture = await seedClaimedRemoteBootstrap("SESSION-GOAL-READ", { mutate: readGoal }); + expect(await claimNativeRestartRecoveries({ db: fixture.db, controller: successor, + restartKind: "hard", runIds: [fixture.runId] })).toEqual([expect.objectContaining({ + kind: "bootstrap_incomplete", runId: fixture.runId, controllerGeneration: 2 })]); + const [coordinator] = await fixture.db.select().from(nativeRunFinalizations) + .where(eq(nativeRunFinalizations.runId, fixture.runId)); + expect(coordinator.recoveryHistory.at(-1)).toMatchObject({ disposition: "bootstrap_incomplete", + reason: "remote_session_opened_without_turn" }); + const executor = await seedClaimedRemoteBootstrap("SESSION-GOAL-READ-EXECUTOR", { mutate: readGoal }); + const root = resolve(runtimeRoot, `goal-read-${executor.runId}`); + try { + expect(await restoreNativeRunnerCheckpoint(executor.db, { companyId, issueId: executor.issueId, runId: executor.runId }, + root)).not.toBeNull(); + expect(runnerdStateProvesNoRemoteProviderTurn(root)).toBe(true); + } finally { + await rm(root, { recursive: true, force: true }); + } + }); + it("lets the executor prove a drained remote session had no turn from its restored checkpoint", async () => { const fixture = await seedClaimedRemoteBootstrap("SESSION-OPENED-EXECUTOR", { mutate: core => openSession(core) }); const root = resolve(runtimeRoot, `opened-${fixture.runId}`); @@ -2607,6 +2634,8 @@ process.on("SIGTERM", () => child.kill("SIGTERM")); it.each<[string, Parameters[1], { eventType: string } | null]>([ ["a turn was issued", { mutate: core => { openSession(core); core.queueCommand("turn.start", { prompt: "Prior work must stay held." }); } }, null], + ["a goal was set after a read", { mutate: core => { openSession(core); core.queueCommand("session.goal.get", {}); + core.queueCommand("session.goal.set", { objective: "Prior work must stay held." }); } }, null], ["a turn event was committed", { mutate: core => openSession(core, ["session.started", "turn.started"]) }, null], ["an acknowledged event is missing", { mutate: core => { openSession(core); core.store.state.ackedSourceSeq = 4; } }, null], ["a provider turn event was projected", { mutate: core => openSession(core) }, { eventType: "turn.started" }],