diff --git a/server/src/services/native-runtime/native-recovered-owner-followup.test.ts b/server/src/services/native-runtime/native-recovered-owner-followup.test.ts index 178faddc960..ae9be61febe 100644 --- a/server/src/services/native-runtime/native-recovered-owner-followup.test.ts +++ b/server/src/services/native-runtime/native-recovered-owner-followup.test.ts @@ -90,7 +90,9 @@ describe.skipIf(!recoveryPostgresSupport.supported)("follow-up after a recovered * later committed its result and released the controller lease, but no * suspended Runner receipt or failover backup ever reached the controller. * Only the database controller checkpoint and recovery pointer survive. */ - async function seedRecoveredPriorRun(control: "settled" | "pending-command" | "unanswered-tool" = "settled", + async function seedRecoveredPriorRun(control: "settled" | "pending-command" | "unanswered-tool" + | "failed-turn-start" | "rejected-turn-start" | "indeterminate-turn-start" | "unknown-outcome-turn-start" + | "unsuperseded-unknown-outcome-turn-start" | "failed-suspend" = "settled", options: { controller?: { bootId: string; pid: number; processStartedAt: Date }; provider?: "codex" | "acpx" | "opencode" } = {}) { const issueId = randomUUID(), workspaceId = randomUUID(), runId = randomUUID(); await db.insert(issues).values({ id: issueId, companyId, title: "Continue after recovery", assigneeAgentId: agentId, status: "in_progress" }); @@ -127,10 +129,25 @@ describe.skipIf(!recoveryPostgresSupport.supported)("follow-up after a recovered try { // Shape the file-backed journal first; the database checkpoint captures // it when persistence moves to the controller's external writer. - const commands = ["run.prepare", "session.open", "turn.start", "runner.suspend"].map(type => controller.queueCommand(type)); + const unknownOutcome = { status: "failed", result: { commandType: "turn.start", status: "failed", + result: { code: "execution_indeterminate", message: "runner recovered after journaling this command; it will not execute twice" } } }; + const answers = { + "failed-turn-start": { status: "failed", result: { status: "failed" } }, + "rejected-turn-start": { status: "rejected", result: { status: "rejected" } }, + "indeterminate-turn-start": { status: "indeterminate", result: { status: "indeterminate" } }, + "unknown-outcome-turn-start": unknownOutcome, + } as const; + const answered = answers as Record; + const commands = ["run.prepare", "session.open", ...(control in answered ? ["turn.start"] : []), "turn.start", "runner.suspend"] + .map(type => controller.queueCommand(type)); for (const command of control === "pending-command" ? commands.slice(0, -1) : commands) { Object.assign(command, { status: "completed", result: { status: "completed" } }); } + if (control in answered) Object.assign(commands[2]!, answers[control as keyof typeof answers]); + if (control === "unsuperseded-unknown-outcome-turn-start") { + Object.assign(commands.find(command => command.type === "turn.start")!, unknownOutcome); + } + if (control === "failed-suspend") Object.assign(commands.at(-1)!, { status: "failed", result: { status: "failed" } }); const event = (sourceSeq: number, sourceEventId: string, eventType: string) => ({ sourceSeq, sourceEventId, eventType, priority: 0 as const, envelope: { ...identity }, deliveryCount: 1, logicalEffectCount: 1 }); controller.store.state.committedEvents.push( @@ -628,6 +645,19 @@ describe.skipIf(!recoveryPostgresSupport.supported)("follow-up after a recovered expect(await retry.start()).toBe("native_controller_recovery_unsupported:metadata_format"); }, 60_000); + it.each([ + { control: "failed-turn-start", outcome: ADMITTED }, + { control: "rejected-turn-start", outcome: ADMITTED }, + { control: "indeterminate-turn-start", outcome: ADMITTED }, + { control: "unknown-outcome-turn-start", outcome: ADMITTED }, + { control: "unsuperseded-unknown-outcome-turn-start", outcome: stranded("journal_commands_unsettled(turn.start.failed)") }, + { control: "failed-suspend", outcome: stranded("journal_commands_unsettled(runner.suspend.failed)") }, + ] as const)("retires a finished owner whose journal holds an answered $control command", async ({ control, outcome }) => { + const prior = await seedRecoveredPriorRun(control); + const followUp = await admitFollowUp(prior); + expect(await followUp.start()).toBe(outcome); + }, 60_000); + it("admits again when a retry rebuilds the same projection from the database", async () => { const prior = await seedRecoveredPriorRun(); const followUp = await admitFollowUp(prior); diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index c74977841dd..4782abec0a4 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -267,7 +267,7 @@ import { remoteRecoveryProbeReason, } from "./remote-runner-recovery-diagnostics.js"; import { readRemoteRunnerRecoveryProbe, REMOTE_RUNNER_RECOVERY_PROBE_BUDGET_MS } from "./remote-runner-recovery-probe.js"; -import { describeOwnFailedAttachment, inspectRemoteWarmSessionRecovery, isSettledRemoteWarmControlState, summarizePostTerminalEvents, type RemoteWarmSessionRecovery } from "./remote-warm-session-recovery.js"; +import { describeOwnFailedAttachment, inspectRemoteWarmSessionRecovery, isSettledRemoteWarmControlState, summarizePostTerminalEvents, type RemoteWarmSessionRecovery, unsettledControllerCommands } from "./remote-warm-session-recovery.js"; import { assertNativeWarmIngressRetirement, nativeWarmIngressAllocation, nativeWarmIngressIdentity, type NativeWarmIngressClaim, type NativeWarmIngressOwner } from "./native-warm-ingress-ownership.js"; import { resolveNativeStagedAttachmentRecovery, type NativeStagedAttachmentRecovery } from "./native-staged-attachment-recovery.js"; import { emitStartupReuse, environmentStartupReuse } from "@paperclipai/adapter-utils/startup-reuse"; @@ -5458,7 +5458,7 @@ function readSettledRunnerdProjection(root: string, identity: Record command.status !== "completed") + const unsettled = unsettledControllerCommands(settledCommands) .map(command => `${retirementReasonName("commandType", command.type)}.${retirementReasonName("commandStatus", command.status)}`); return refuse(unsettled.length > 0 ? `journal_commands_unsettled(${[...new Set(unsettled)].slice(0, 4).join(",")})` : "journal_terminal_unsettled"); diff --git a/server/src/services/native-runtime/remote-warm-session-recovery.ts b/server/src/services/native-runtime/remote-warm-session-recovery.ts index 40324702407..d1759466ed7 100644 --- a/server/src/services/native-runtime/remote-warm-session-recovery.ts +++ b/server/src/services/native-runtime/remote-warm-session-recovery.ts @@ -31,10 +31,32 @@ function postTerminalEvents(control: Record): unknown[] { /** A session is settled when its commands finished and nothing is live or waiting. * Events after the terminal describe the past, so refusing cannot undo them: they * are reported instead. Every event must still authenticate as this session's. */ +const RUNNER_LIFECYCLE_COMMAND_TYPES = new Set(["runner.suspend", "runner.drain", "runner.shutdown"]); + +function outcomeUnknown(command: Record): boolean { + return command.status === "indeterminate" || + (command.status === "failed" && (record(command.result).code === "execution_indeterminate" || + record(record(command.result).result).code === "execution_indeterminate")); +} + +export function unsettledControllerCommands(commands: readonly Record[]): Record[] { + const lastCompletedTurnStart = Math.max(-1, ...commands + .filter(command => command.type === "turn.start" && command.status === "completed") + .map(command => Number(command.controllerSeq))); + return commands.filter(command => { + if (command.status === "completed") return false; + if (outcomeUnknown(command)) { + return !(command.type === "turn.start" && Number(command.controllerSeq) < lastCompletedTurnStart); + } + return !((command.status === "failed" || command.status === "rejected") && + !RUNNER_LIFECYCLE_COMMAND_TYPES.has(String(command.type))); + }); +} + export function isSettledRemoteWarmControlState(control: Record, identity: Record): boolean { const commands = control.commands; const events = control.committedEvents; - if (!Array.isArray(commands) || commands.some(command => record(command).status !== "completed") || + if (!Array.isArray(commands) || unsettledControllerCommands(commands.map(record)).length > 0 || !Array.isArray(events) || Object.entries(identity).some(([key, value]) => record(control.identity)[key] !== value)) return false; const terminalIndex = events.findLastIndex(event => record(event).eventType === "run.terminal"); if (terminalIndex < 0) return false;