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 af4aeaf8ddd..2c34ce01ae1 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 @@ -904,10 +904,13 @@ describe.skipIf(!recoveryPostgresSupport.supported)("follow-up after a recovered } }, 60_000); it.each([ - { label: "never delivered to a dead runner", connectionCount: 0, claim: true, admitted: true }, - { label: "possibly delivered", connectionCount: 1, claim: true, admitted: false }, - { label: "without a dead-runner claim", connectionCount: 0, claim: false, admitted: false }, - ])("resumes from the previous turn when a warm attach rotation was $label", async ({ connectionCount, claim, admitted }) => { + { label: "never delivered to a dead runner", connectionCount: 0, claim: true, admitted: true, archiveOnly: false }, + { label: "possibly delivered", connectionCount: 1, claim: true, admitted: false, archiveOnly: false }, + { label: "without a dead-runner claim", connectionCount: 0, claim: false, admitted: false, archiveOnly: false }, + { label: "cut off before its new control plane was written", connectionCount: 0, claim: true, admitted: true, archiveOnly: true }, + { label: "cut off before its new control plane, without a dead-runner claim", connectionCount: 0, claim: false, admitted: false, + archiveOnly: true }, + ])("resumes from the previous turn when a warm attach rotation was $label", async ({ connectionCount, claim, admitted, archiveOnly }) => { const prior = await seedRecoveredPriorRun(); const followUp = await admitFollowUp(prior); const root = await restoreHeldRoot(prior, followUp); @@ -921,22 +924,28 @@ describe.skipIf(!recoveryPostgresSupport.supported)("follow-up after a recovered await mkdir(join(root, "runner"), { recursive: true, mode: 0o700 }); await writeFile(join(root, "runner", "codex-provider-state.json"), "{}"); const identity = { ...priorIdentity, runId: followUp.runId, turnId: `turn-${followUp.runId}`, itemId: `item-${followUp.runId}` }; - await mkdir(join(root, "control-plane"), { mode: 0o700 }); + if (!archiveOnly) await mkdir(join(root, "control-plane"), { mode: 0o700 }); const undelivered = JSON.stringify({ schema: "paperclip.runner.durable.control-plane-state.v2", identity, bootstrapHistory: { schema: "paperclip.runner.bootstrap-history.v1", originIdentity: identity, credentialsIssued: 0, adopted: false }, runAttachTemplate: {}, tickets: {}, leases: {}, commands: [{ commandId: "attach", controllerSeq: 1, type: "run.attach", status: "pending", payload: {} }], committedEvents: [], ackedSourceSeq: 0, connectionCount, commandDeliveryCounts: {} }); - await writeFile(join(root, "control-plane", "control-plane-state.json"), undelivered); + if (!archiveOnly) await writeFile(join(root, "control-plane", "control-plane-state.json"), undelivered); const outcome = await followUp.startLocal(claim ? { kind: "resume_dead_runner", runId: followUp.runId, leaseOwner: "successor", controllerGeneration: 2, providerAttempt: 1, restartKind: "hard", recoveryRequestId: null } as never : undefined); if (!admitted) { - expect(outcome).toBe("runner_state_identity_mismatch: authority_indeterminate"); + expect(outcome).toBe(archiveOnly ? "runner_state_identity_mismatch: durable_identity_unreadable" + : "runner_state_identity_mismatch: authority_indeterminate"); return; } expect(outcome).toBe(ADMITTED); expect(JSON.parse(await readFile(join(root, "control-plane", "control-plane-state.json"), "utf8")).identity).toEqual(priorIdentity); expect(existsSync(join(root, "authority-epochs"))).toBe(false); + if (archiveOnly) { + expect(existsSync(join(root, "continuity-breaks"))).toBe(false); + expect(await quarantined()).toEqual([]); + return; + } const [archive] = await readdir(join(root, "continuity-breaks")); expect(await readFile(join(root, "continuity-breaks", archive!, "control-plane", "control-plane-state.json"), "utf8")).toBe(undelivered); expect(await quarantined()).toEqual([]); diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 15aacf4dbd1..d3573e54916 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -5680,8 +5680,34 @@ async function retireFinishedPerTurnRunnerdOwner(input: Parameters) => boolean): string | null { + const archives = resolve(root, "authority-epochs"); + if (!isSafeNativeStateDirectory(archives)) return null; + const candidates = readdirSync(archives).map(name => resolve(archives, name)).filter(epoch => { + const state = resolve(epoch, "control-plane", "control-plane-state.json"); + return isSafeNativeStateDirectory(epoch) && existsSync(state) && existsSync(resolve(epoch, "runner-state.json")); + }).sort((left, right) => lstatSync(resolve(right, "control-plane", "control-plane-state.json")).mtimeMs + - lstatSync(resolve(left, "control-plane", "control-plane-state.json")).mtimeMs); + const latest = candidates[0]; + if (!latest) return null; + const prior = record(readRunnerdControlPlaneState(latest).identity); + const runner = record(JSON.parse(readBoundedNativeFile(resolve(latest, "runner-state.json"), + NATIVE_RUNNER_STATE_MAX_BYTES, "runner_state_too_large").toString("utf8"))); + const digest = createHash("sha256").update(JSON.stringify(prior)).digest("hex").slice(0, 24); + if (basename(latest) !== `epoch-${digest}` || !matches(prior) || + runner.schema !== RUNNERD_STATE_SCHEMA || runner.lifecycle !== "suspended" || + ["runnerInstanceId", "environmentLeaseId", "runId", "normalizedSessionId"].some(key => runner[key] !== prior[key])) return null; + return latest; +} + +function interruptedRotationSource(root: string, execution: NativeExecutionInput): string | null { try { + if (existsSync(resolve(root, "runner", "runner-state.json"))) return null; + if (!existsSync(resolve(root, "control-plane", "control-plane-state.json"))) { + if (existsSync(resolve(root, "control-plane")) && readdirSync(resolve(root, "control-plane")).length > 0) return null; + return latestSuspendedArchivedEpoch(root, prior => typeof prior.runId === "string" && + prior.runId !== execution.binding.runId && prior.normalizedSessionId === nativeSessionKey(execution)); + } const control = readRunnerdControlPlaneState(root); const identity = record(control.identity); const commands = Array.isArray(control.commands) ? control.commands.map(record) : []; @@ -5693,26 +5719,9 @@ function undeliveredAttachRotationSource(root: string, execution: NativeExecutio Object.keys(record(control.commandDeliveryCounts)).length !== 0 || control.warmTransition != null || control.completedWarmTransition != null || control.stagedRunAttachment != null || history.adopted === true || history.credentialsIssued !== 0 || - commands.length !== 1 || commands[0]!.type !== "run.attach" || commands[0]!.status !== "pending" || - existsSync(resolve(root, "runner", "runner-state.json"))) return null; - const archives = resolve(root, "authority-epochs"); - if (!isSafeNativeStateDirectory(archives)) return null; - const candidates = readdirSync(archives).map(name => resolve(archives, name)).filter(epoch => { - const state = resolve(epoch, "control-plane", "control-plane-state.json"); - return isSafeNativeStateDirectory(epoch) && existsSync(state) && existsSync(resolve(epoch, "runner-state.json")); - }).sort((left, right) => lstatSync(resolve(right, "control-plane", "control-plane-state.json")).mtimeMs - - lstatSync(resolve(left, "control-plane", "control-plane-state.json")).mtimeMs); - const latest = candidates[0]; - if (!latest) return null; - const prior = record(readRunnerdControlPlaneState(latest).identity); - const runner = record(JSON.parse(readBoundedNativeFile(resolve(latest, "runner-state.json"), - NATIVE_RUNNER_STATE_MAX_BYTES, "runner_state_too_large").toString("utf8"))); - const digest = createHash("sha256").update(JSON.stringify(prior)).digest("hex").slice(0, 24); - if (basename(latest) !== `epoch-${digest}` || prior.runId === identity.runId || - ["runnerInstanceId", "environmentLeaseId", "normalizedSessionId"].some(key => prior[key] !== identity[key]) || - runner.schema !== RUNNERD_STATE_SCHEMA || runner.lifecycle !== "suspended" || - ["runnerInstanceId", "environmentLeaseId", "runId", "normalizedSessionId"].some(key => runner[key] !== prior[key])) return null; - return latest; + commands.length !== 1 || commands[0]!.type !== "run.attach" || commands[0]!.status !== "pending") return null; + return latestSuspendedArchivedEpoch(root, prior => prior.runId !== identity.runId && + ["runnerInstanceId", "environmentLeaseId", "normalizedSessionId"].every(key => prior[key] === identity[key])); } catch { return null; } @@ -5721,13 +5730,15 @@ function undeliveredAttachRotationSource(root: string, execution: NativeExecutio async function rollBackUndeliveredAttachRotation(root: string, execution: NativeExecutionInput, onLog?: (stream: "stdout" | "stderr", chunk: string) => Promise): Promise { if (!existsSync(root) || !isSafeNativeStateDirectory(root)) return; - const epoch = undeliveredAttachRotationSource(root, execution); + const epoch = interruptedRotationSource(root, execution); if (!epoch) return; const archivedAt = Date.now(); - const breakRoot = resolve(root, "continuity-breaks", `${archivedAt}-${randomUUID()}`); - mkdirSync(breakRoot, { recursive: true, mode: 0o700 }); noteNativeRunnerEvidence(archivedAt); - renameSync(resolve(root, "control-plane"), resolve(breakRoot, "control-plane")); + if (existsSync(resolve(root, "control-plane"))) { + const breakRoot = resolve(root, "continuity-breaks", `${archivedAt}-${randomUUID()}`); + mkdirSync(breakRoot, { recursive: true, mode: 0o700 }); + renameSync(resolve(root, "control-plane"), resolve(breakRoot, "control-plane")); + } mkdirSync(resolve(root, "runner"), { recursive: true, mode: 0o700 }); renameSync(resolve(epoch, "runner-state.json"), resolve(root, "runner", "runner-state.json")); renameSync(resolve(epoch, "control-plane"), resolve(root, "control-plane"));