diff --git a/doc/plugins/PLUGIN_RUNNER_EXECUTION.md b/doc/plugins/PLUGIN_RUNNER_EXECUTION.md index 1a871bf041f..a4bc0be0b53 100644 --- a/doc/plugins/PLUGIN_RUNNER_EXECUTION.md +++ b/doc/plugins/PLUGIN_RUNNER_EXECUTION.md @@ -336,12 +336,20 @@ not authorize new work or file access through that lease. Temporary credential homes are not needed after the originating run ends. An explicit reset to a different normalized session can retire a warm owner whose close failed only because it could not prove checkpoint suspension. Core -requires the exact lease-bound task's `executionStopped: true` receipt, a -successful `complete` handoff, and settled Core tool results first. It then -collects stopped instruction files before it releases the in-memory owner. The old checkpoint remains evidence; -it is not reused by the new session. A failed handoff, failed file collection, -unsettled Core tool result, other close error, or continuation of the same -session still blocks replacement. +requires the exact lease-bound task's `executionStopped: true` receipt and a +successful `complete` handoff first. When the close left them unproven, the new +run requests them itself: one confirmed `stop` (not resent once confirmed), +the credential snapshot, then `complete`. A warm owner without a task needs its +Runner's own durable suspension instead. Unsettled Core tool results do not +block this reset: a governed-wait park leaves one by design, its run is already +terminal, its provider is stopped, and the new session never replays it. Core +then collects stopped instruction files before it releases the in-memory owner. +With that task receipt, any close error qualifies, including a controller +journal drain that outlived the close grace: the old journal lives in the +discarded session's own state root. The old checkpoint remains evidence; it is +not reused by the new session. A handoff that still fails, failed file +collection, another close error from an owner without a task, or continuation +of the same session still blocks replacement. A controller restart is not required for this stopped-owner reset path. Core ends a dispatched task lease with `stop`, `status`, and `complete`. When 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 03e71ad6519..b723fd1d0b5 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -11882,15 +11882,19 @@ describe("native warm session supervision", () => { await closeIdleWarmNativeSessionsForRestart(); }); - it.each(["retired", "unconfirmed", "other-error", "collection-failed", "same-session", "same-id-provider", "unbound", "pending-semantic"] as const)( + it.each(["retired", "unconfirmed", "other-error", "collection-failed", "same-session", "same-id-provider", "unbound", "pending-semantic", "stop-unconfirmed-at-close", "suspended-unbound", "journal-drain-timeout"] as const)( "handles a permanently rejected plugin close during reset (%s)", async mode => { const { base, next, run } = fixture(`identity-poisoned-close-${mode}`); const identity = { runnerInstanceId: "runner", environmentLeaseId: "lease", runId: base.binding.runId, normalizedSessionId: base.session.normalizedSessionId!, turnId: "turn", itemId: "item" }; let stopped = false; + let stopFailures = mode === "stop-unconfirmed-at-close" ? 1 : 0; const task = vi.fn(async (operation: any) => { if (operation.kind === "describe") return { kind: "description", taskId: "old-task", launchPreparationVersion: 1, projectAuthCheckVersion: 1, runner: { version: "fixture", digest: `sha256:${"a".repeat(64)}` } }; + // The Launcher RPC can fail while the old owner closes; the successor's + // forced retirement must send its own confirmed stop. + if (operation.kind === "stop" && stopFailures-- > 0) throw new Error("launcher stop unavailable"); if (operation.kind === "stop") stopped = true; if (operation.kind === "status") return { kind: "status", taskId: "old-task", phase: stopped ? "cancelled" : "running", executionStopped: stopped }; if (operation.kind === "complete" && mode === "unconfirmed") throw new Error("handoff unconfirmed"); @@ -11910,15 +11914,29 @@ describe("native warm session supervision", () => { runnerInstanceId: "runner", runnerIngressAuthorized: true, runnerExecutionTarget: { kind: "remote", transport: "sandbox", remoteCwd: ".", leaseId: "lease", runnerTask: task } }); const registration = await owner.registration({} as never, base.binding.runId); - const failure = mode === "other-error" ? new Error("controller retirement unconfirmed") : new NativeSessionCloseUnrecoverableError(undefined, mode !== "pending-semantic"); + // A drain that outlives the close grace fails before the checkpoint's stop + // and completion, as releaseRunnerProcessOwnership does in the transport. + const drainTimedOut = mode === "journal-drain-timeout"; + const failure = mode === "other-error" ? new Error("controller retirement unconfirmed") + : drainTimedOut ? new Error("Controller journal drain timed out; preserve the session before replacing its writer.") + // releaseRunnerProcessOwnership rethrows the checkpoint's failed stop + // before the transport can raise NativeSessionCloseUnrecoverableError. + : mode === "stop-unconfirmed-at-close" ? new Error("launcher stop unavailable") + // A local Runner that durably suspended is contained without a task owner. + : mode === "suspended-unbound" ? new NativeSessionCloseUnrecoverableError({ runnerSuspended: true, providerDrained: true }, false) + : new NativeSessionCloseUnrecoverableError(undefined, mode !== "pending-semantic"); let closePromise: Promise | undefined; const close = vi.fn(() => closePromise ??= (async () => { await registration.release(); - await registration.checkpoint("unsettled").catch(() => undefined); + if (!drainTimedOut) await registration.checkpoint("unsettled").catch(() => undefined); throw failure; })()); const session = { close }; - if (mode !== "unbound") backend.bindManagedSession(session as never); + // Without a task owner, a close error other than proven suspension still blocks. + if (mode !== "unbound" && mode !== "suspended-unbound" && mode !== "other-error") backend.bindManagedSession(session as never); + // A governed-wait park leaves its Core tool callback unsettled, and a + // failed Launcher stop leaves no receipt: neither belongs to the new session. + const succeeds = mode === "retired" || mode === "pending-semantic" || mode === "stop-unconfirmed-at-close" || mode === "suspended-unbound" || drainTimedOut; const collectStopped = vi.fn(async () => { expect(owner.isRetired()).toBe(true); if (mode === "collection-failed") throw new Error("file collection failed"); @@ -11949,7 +11967,7 @@ describe("native warm session supervision", () => { const attempt = () => mode === "same-session" ? executePaperclipNativeSession({ db: leaseDb(replacement), execution: replacement, runnerInstanceId: "runner", refreshTools: true }) : run(replacement); - if (mode === "retired") { + if (succeeds) { await attempt(); expect(state.execute).toHaveBeenCalledTimes(2); expect(retirementFailed).not.toHaveBeenCalled(); @@ -11967,7 +11985,9 @@ describe("native warm session supervision", () => { expect(state.execute).toHaveBeenCalledTimes(2); } } - expect(task.mock.calls.filter(([operation]) => operation.kind === "stop")).toHaveLength(1); + // One confirmed stop per owner; a failed Launcher RPC is the only resend. + expect(task.mock.calls.filter(([operation]) => operation.kind === "stop")) + .toHaveLength(mode === "stop-unconfirmed-at-close" ? 2 : 1); expect(task.mock.calls.some(([operation]) => operation.kind === "read_state")).toBe(false); } finally { collectStopped.mockResolvedValue(undefined); @@ -17142,7 +17162,7 @@ describe("runnerd provider runtime wiring", () => { state.toolAuthorityDefinitions.mockImplementation(async binding => binding.readRemoteTaskFile ? [{ name: "register_deliverable" }] as never : []); const readWorkspaceFile = vi.fn(async () => Buffer.from("proof")); const release = vi.fn(async () => {}), resume = vi.fn(async () => "resumed" as const); - const owner = vi.spyOn(pluginTaskRuntime, "createPluginRunnerTaskOwner").mockReturnValue({ launcher, registration, driver, readWorkspaceFile, release, resume, isRetired: () => false, retiredBeforeClaim: () => false, credentialsUpdated: vi.fn(), rebind:vi.fn() }); + const owner = vi.spyOn(pluginTaskRuntime, "createPluginRunnerTaskOwner").mockReturnValue({ launcher, registration, driver, readWorkspaceFile, release, resume, isRetired: () => false, retireSupersededTask: async () => false, retiredBeforeClaim: () => false, credentialsUpdated: vi.fn(), rebind:vi.fn() }); const task = vi.fn(async (operation: unknown) => { environmentTaskOperationSchema.parse(operation); return { kind: "description", launchPreparationVersion: 1, projectAuthCheckVersion: 1, releaseResumeVersion: 1, ...(kind === "codex" ? { workspaceFileReadVersion: 1 } : {}), taskId: "task", runner: { version: "fixture", digest: `sha256:${"a".repeat(64)}` } }; diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 8ce1b6faf79..f59b41980fa 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -731,14 +731,23 @@ async function closeWarmNativeSession(entry: WarmNativeSession, reason: string, try { try { await entry.session.close({ reason }); } catch (error) { - // A new conversation does not consume the old checkpoint. Its failed - // suspension must not poison every later reset after Launcher has proved - // that the exact task stopped and completed its credential handoff. - // Other close errors can mean controller authority is still live. - const retiredWithoutCheckpoint = options?.discardUnsettledSession === true - && error instanceof NativeSessionCloseUnrecoverableError - && error.semanticToolsSettled - && pluginTaskSessionOwners.get(entry.session)?.isRetired() === true; + // A new conversation does not consume the old checkpoint, so the old + // owner's unprovable close must not fail it once that owner is contained: + // Launcher proves the exact task stopped and completed its credential + // handoff (the successor forces this when close left it unproven), or, + // without a task owner, the Runner itself durably suspended. Unsettled + // Core tool callbacks (a governed-wait park leaves one) belong to the + // discarded conversation: its run is terminal, its provider is stopped, + // and the new conversation never replays them. With a Launcher receipt + // any close error qualifies, including a controller journal drain that + // outlived the close grace: the old journal lives in the discarded + // session's own state root, and its Runner is stopped. Without a task + // owner, other close errors can mean controller authority is still live. + const discardable = options?.discardUnsettledSession === true; + const owner = discardable ? pluginTaskSessionOwners.get(entry.session) : undefined; + const retiredWithoutCheckpoint = discardable && (owner + ? owner.isRetired() || await owner.retireSupersededTask().catch(() => false) + : error instanceof NativeSessionCloseUnrecoverableError && error.settlement?.runnerSuspended === true); if (!retiredWithoutCheckpoint) { try { await entry.instructionWorkingCopy?.retirementFailed?.(); } catch (receiptError) { throw new AggregateError([error, receiptError], "Warm instruction retirement and receipt both failed"); } diff --git a/server/src/services/native-runtime/plugin-runner-task.ts b/server/src/services/native-runtime/plugin-runner-task.ts index 72893bff48c..f294e3d5f45 100644 --- a/server/src/services/native-runtime/plugin-runner-task.ts +++ b/server/src/services/native-runtime/plugin-runner-task.ts @@ -429,7 +429,27 @@ export function createPluginRunnerTaskOwner(input: { } return result.outcome; }; - return { launcher, registration, driver, readWorkspaceFile, release, resume, + // A successor that discards this owner's session must not inherit its + // failed close. Repeat the unsettled checkpoint's task obligations: confirmed + // stop (sent once), credential handoff, then completion. Never reads state. + const retireSupersededTask = async () => { + if (!submitted || detached) return false; + if (!completed) { + stopRequested = true; + if (stoppedState?.executionStopped !== true) await stopAndConfirm(); + if (input.accountIdentity && input.onCredentials) { + const snapshot = await call({ kind: "credentials", accountIdentity: input.accountIdentity }); + if (snapshot.kind !== "credentials") throw new Error("Plugin Runner credential snapshot unavailable"); + if (snapshot.credential !== null) await input.onCredentials(snapshot.credential); + } + await call({ kind: "complete" }); + completed = true; + const identity = preparedIdentity ?? launchIdentity ?? input.priorIdentity; + if (identity) await input.onRetired?.(identity); + } + return stoppedState?.executionStopped === true && completed && !detached; + }; + return { launcher, registration, driver, readWorkspaceFile, release, resume, retireSupersededTask, // A warm transport outlives its first lease and invocation's private home. credentialsUpdated: async (event: PrpEvent) => { await input.onCredentialsUpdated?.(event); }, // This is a lease-bound Launcher receipt, not a terminal phase or a sent diff --git a/tests/restart-chaos/README.md b/tests/restart-chaos/README.md index c9660eaea57..6d08b5bd47f 100644 --- a/tests/restart-chaos/README.md +++ b/tests/restart-chaos/README.md @@ -882,6 +882,37 @@ The Q4 oracle checks that two questions were asked and both were answered, that ended `succeeded`, that no run of the task failed or was cancelled, and that the brief carries both answers. Q4 is not a default workload. +## Session replaced under a retained owner (SR1) + +```sh +# One task: the parked question's warm owner is superseded before the answer. +node cli/node_modules/tsx/dist/cli.mjs tests/restart-chaos/main.ts --env local --lifecycle warm \ + --adapters codex --workloads Q1 --faults SR1 --concurrency 1 --tasks 1 --seed 301 +# Two tasks on the same agent: one change supersedes both owners at once. +node cli/node_modules/tsx/dist/cli.mjs tests/restart-chaos/main.ts --env local --lifecycle warm \ + --adapters codex --workloads Q1 --faults SR1 --concurrency 2 --tasks 2 --seed 302 +``` + +A session fault restarts nothing. It covers a staging failure class: a new run on a task whose +previous run left a warm Runner retained needed a new provider session, retired that owner first, +and failed with `provider_transport_failed: runner did not durably suspend before checkpoint` +before any provider work. A Q1 park leaves the owner with an unsettled question callback, the +condition that made its close unprovable. + +The failure needs a Daytona task owner: locally a parked owner closes cleanly, so the local cell +passes before and after the fix and guards only the ordinary reset path. Run the Daytona variant +(`--env daytona --daytona-warm`) with an image that matches the checkout to exercise the +superseded-owner retirement itself. + +SR1 runs Q1 with `--lifecycle warm` (or `--daytona-warm`). After the asking run settles and before +the board answers, the harness changes the agent's `adapterConfig.idleTimeoutMs` through +`PATCH /api/agents/:id`, as a user editing the agent would. That setting is part of the session's +configuration fingerprint, so the answer's follow-up cannot resume the saved session. The Q1 oracle +applies unchanged (`no_failed_attempt` and `issue_done` cover the follow-up), plus +`session_rotated_before_answer` and `follow_up_new_session` (a follow-up that resumed the saved +session would not exercise the retirement). Because tasks share their adapter's agent, `--tasks 2 --concurrency 2` +is the two-concurrent-runs variant. SR1 does not combine with K* or PV* faults. + ## Database contention (L1, `--db-pressure`) ```sh diff --git a/tests/restart-chaos/config.ts b/tests/restart-chaos/config.ts index fb4540aba0c..c8a5b30596c 100644 --- a/tests/restart-chaos/config.ts +++ b/tests/restart-chaos/config.ts @@ -12,6 +12,7 @@ import { parseNativeTrigger, parseNativeTriggerOutage } from "./native-trigger.j import { parseRepeatedRestartOptions } from "./repeated-restarts.js"; import { daytonaRetainsSandboxes } from "./ttft-baseline-stats.js"; import { PROVIDER_FAULT_KINDS, PROVIDER_FAULT_WORKLOADS, TOOL_COMPLETED_WORKLOADS, UNINTERRUPTED_PROVIDER_MESSAGES, isProviderFault, parseProviderDelays, parseProviderMessages, type ProviderFaultKind } from "./provider-faults.js"; +import { SESSION_FAULT_KINDS, SESSION_FAULT_WORKLOADS, isSessionFault, type SessionFaultKind } from "./session-faults.js"; import { PROVIDER_FAULT_ADAPTERS, parseProviderFaultRestart } from "./acpx-provider-faults.js"; import { DISK_FULL_FAULT_KINDS, isDiskFullFault, type DiskFullFaultKind } from "./disk-full-faults.js"; import { @@ -184,12 +185,13 @@ export function parseCommand(argv: readonly string[], now = Date.now()): ParsedC } if (lifecycle !== undefined && lifecycle !== "per_turn" && lifecycle !== "warm" && lifecycle !== "default") throw new Error("--lifecycle must be per_turn, warm or default"); const adapters = list(values.adapters, ADAPTER_IDS, "adapters", ["codex", "claude"]); - // `--faults` takes restart faults (K*), provider message faults (PV*, provider-faults.ts) and the - // sandbox disk-full fault (DF1, disk-full-faults.ts). - const allFaults = list(values.faults, [...FAULT_KINDS, ...PROVIDER_FAULT_KINDS, ...DISK_FULL_FAULT_KINDS], "faults", + // `--faults` takes restart faults (K*), provider message faults (PV*, provider-faults.ts), the + // sandbox disk-full fault (DF1, disk-full-faults.ts) and session faults (SR*, session-faults.ts). + const allFaults = list(values.faults, [...FAULT_KINDS, ...PROVIDER_FAULT_KINDS, ...DISK_FULL_FAULT_KINDS, ...SESSION_FAULT_KINDS], "faults", stormMinutes ? STORM_DEFAULT_FAULTS : crashPoint ? ["K1"] : ["K1", "K2", "K4"]); const providerFaults = allFaults.filter(isProviderFault); const diskFullFaults = allFaults.filter(isDiskFullFault); + const sessionFaults = allFaults.filter(isSessionFault); const options: WaveOptions = { env, adapters, @@ -197,9 +199,10 @@ export function parseCommand(argv: readonly string[], now = Date.now()): ParsedC workloads: list(values.workloads, workloadChoices, "workloads", workloadChoices.filter(id => id !== "G1" && id !== "H1" && id !== "R1" && id !== "S1" && id !== "L1" && id !== "F1" && id !== "T1" && id !== "Q4" && !takesJourneyTriggers(id) && (id !== "M1" || adapters.length >= 2))), // A crash-point wave schedules no fault; K1 is the restart its self-kill gets. - faults: allFaults.filter((fault): fault is FaultKind => !isProviderFault(fault) && !isDiskFullFault(fault)), + faults: allFaults.filter((fault): fault is FaultKind => !isProviderFault(fault) && !isDiskFullFault(fault) && !isSessionFault(fault)), ...(providerFaults.length ? { providerFaults } : {}), ...(diskFullFaults.length ? { diskFullFaults } : {}), + ...(sessionFaults.length ? { sessionFaults } : {}), triggers: list(values.triggers, TRIGGER_IDS, "triggers", ["time"]), concurrency: integer(values.concurrency, "concurrency", 4, 1, 20), tasks: integer(values.tasks, "tasks", stormMinutes ? 200 : 8, 1, 200), @@ -272,6 +275,14 @@ export function parseCommand(argv: readonly string[], now = Date.now()): ParsedC || values["provider-fault-restart"] !== undefined || values["provider-fault-restart-after-seconds"] !== undefined) { throw new Error("--provider-fault-message, --provider-fault-delays and --provider-fault-restart need a provider fault (PV1, PV2 or PV3) in --faults"); } + if (sessionFaults.length) { + if (stormMinutes) throw new Error("Session faults (SR*) are planned per task; do not combine them with --storm-minutes"); + if (options.faults.length || providerFaults.length) throw new Error("Session faults (SR*) restart nothing; run them in a wave without K* or PV* faults"); + const other = options.workloads.filter(id => !SESSION_FAULT_WORKLOADS.includes(id)); + if (other.length) throw new Error(`Session faults (SR*) need ${SESSION_FAULT_WORKLOADS.join(", ")}; remove ${other.join(", ")} from --workloads`); + // Only a retained warm owner can be superseded; a per-turn Runner exits after each turn. + if (options.lifecycle !== "warm" && !options.daytonaWarm) throw new Error("Session faults (SR*) need a retained owner: add --lifecycle warm (or --daytona-warm)"); + } validateMultiServerOptions(options); // K15 emulates one deploy identity per generation; a second server beside it would run in the real host's process space. if (options.faults.includes("K15") && options.faults.some(isMultiServerFault)) { diff --git a/tests/restart-chaos/journey-matrix.ts b/tests/restart-chaos/journey-matrix.ts index b8e7f6f6b71..19582904745 100644 --- a/tests/restart-chaos/journey-matrix.ts +++ b/tests/restart-chaos/journey-matrix.ts @@ -77,6 +77,8 @@ export function applyJourneyOptions(options: WaveOptions, input: { matrix: boole if (input.tasksGiven) throw new Error("--matrix runs one task per cell; do not pass --tasks"); if (input.triggersGiven) throw new Error("--matrix uses each workload's own triggers; do not pass --triggers"); if (options.faults.includes("K14")) throw new Error("--matrix does not cross K14 (an upgrade happens once per wave)"); + // The matrix crosses restart and provider faults only; SR1 would silently run as its K0 baseline. + if (options.sessionFaults?.length) throw new Error("--matrix does not cross session faults (SR*); run SR1 as its own wave"); options.matrix = true; options.tasks = matrixCells(options).length; } diff --git a/tests/restart-chaos/observer.ts b/tests/restart-chaos/observer.ts index e39f9ac5900..7d029633b09 100644 --- a/tests/restart-chaos/observer.ts +++ b/tests/restart-chaos/observer.ts @@ -16,6 +16,7 @@ import { restartProcessStartMatches } from "../runner-e2e/process-tree-owner.js" import type { UiRetryClick } from "./board-ui.js"; import { runOracleCommand } from "./oracle-sandbox.js"; import { awaitingQuestionAnswer, collectQuestionArtifacts, questionStep } from "./question-handoff.js"; +import { isSessionFault, rotateAgentSession } from "./session-faults.js"; import { captureRemoteWorkspaceFiles, readRemoteWorkspaceFile } from "./remote-workspace-files.js"; import { taskFaultSchedule } from "./scheduler.js"; import { g1ChildTitle, g1Decision, workloadDefinitions, type WorkloadAgents } from "./workloads.js"; @@ -857,7 +858,8 @@ export class TaskObserver { await questionStep({ api: this.api, issueId: this.record.issueId, nonce: this.plan.nonce, record: this.record, sample, now, // An `answered` fault waits for the answer, so the board may answer before it fires. boardMayAct: this.boardMayAct(sample, "answered"), sequence: this.plan.workload === "Q4", - note: (kind, data) => this.timeline.add(kind, { taskId: this.plan.taskId, ...data }) }); + note: (kind, data) => this.timeline.add(kind, { taskId: this.plan.taskId, ...data }), + ...(isSessionFault(this.plan.fault) ? { beforeAnswer: () => rotateAgentSession(this.api, this.lab.agents[this.plan.adapter].id, this.plan.index, now) } : {}) }); } } diff --git a/tests/restart-chaos/question-handoff.test.ts b/tests/restart-chaos/question-handoff.test.ts index 7b5ae3a013c..d1082ad0507 100644 --- a/tests/restart-chaos/question-handoff.test.ts +++ b/tests/restart-chaos/question-handoff.test.ts @@ -5,6 +5,8 @@ import { awaitingQuestionAnswer, gradeQuestionHandoff, gradeQuestionSequence, q1AnswerMarker, q1Code, q4SecondQuestion, q4Tone, q4ToneMarker, questionStep, type QuestionArtifacts, } from "./question-handoff.js"; +import { planTasks, taskFaultSchedule } from "./scheduler.js"; +import { rotateAgentSession } from "./session-faults.js"; import type { OracleArtifacts, TaskRecord, TaskSample } from "./types.js"; import { workloadDefinitions } from "./workloads.js"; @@ -164,4 +166,50 @@ describe("Q1 saved question", () => { await questionStep({ api, issueId: "i1", nonce: NONCE, record: task, sample: sample("running"), now: 1_000, boardMayAct: true, note: () => undefined }); expect(task.question).toMatchObject({ seenAt: 1_000, savedAt: 400 }); }); + + describe("SR1: session replaced while the asking owner is retained", () => { + it("is a warm Q1-only fault that schedules no restart", () => { + const command = parseCommand(["--adapters", "codex", "--workloads", "Q1", "--faults", "SR1", "--lifecycle", "warm", "--tasks", "2", "--concurrency", "2"]); + if (command.mode !== "wave") throw new Error("expected a wave"); + expect(command.options).toMatchObject({ faults: [], sessionFaults: ["SR1"] }); + const tasks = planTasks(command.options); + expect(tasks.map(task => task.fault)).toEqual(["SR1", "SR1"]); + expect(tasks.flatMap(taskFaultSchedule)).toEqual([]); + expect(() => parseCommand(["--workloads", "Q1", "--faults", "SR1"])).toThrow("--lifecycle warm"); + expect(() => parseCommand(["--workloads", "A1", "--faults", "SR1", "--lifecycle", "warm"])).toThrow("remove A1"); + expect(() => parseCommand(["--workloads", "Q1", "--faults", "SR1,K1", "--lifecycle", "warm"])).toThrow("without K* or PV*"); + // The matrix would run only its K0 baseline and pass without replacing a session. + expect(() => parseCommand(["--matrix", "--workloads", "Q1", "--faults", "SR1", "--lifecycle", "warm"])).toThrow("session faults"); + }); + + it("changes the agent's session configuration after the asking run settles, then answers on a later poll", async () => { + const task = record(); + const { api, posts } = fakeApi([pendingQuestion]); + const patches: Array<{ route: string; body: unknown }> = []; + const patchApi = { async patch(route: string, body?: unknown) { patches.push({ route, body }); return {} as T; } }; + const notes: string[] = []; + const step = (status: string, now: number) => questionStep({ api, issueId: "i1", nonce: NONCE, record: task, sample: sample(status), now, + boardMayAct: true, note: kind => notes.push(kind), beforeAnswer: () => rotateAgentSession(patchApi, "agent-1", 1, now) }); + await step("running", 1_000); + expect(patches).toHaveLength(0); + await step("succeeded", 2_000); + expect(patches).toEqual([{ route: "/api/agents/agent-1", body: { adapterConfig: { idleTimeoutMs: 300_002 } } }]); + expect(posts).toHaveLength(0); + await step("succeeded", 3_000); + expect(patches).toHaveLength(1); + expect(posts).toHaveLength(1); + expect(notes).toEqual(["question_pending", "session_rotated", "question_answered"]); + const graded = gradeQuestionHandoff(NONCE, artifacts({ rotation: task.question!.rotation })); + expect(graded.checks.find(check => check.name === "session_rotated_before_answer")).toMatchObject({ ok: true }); + // A resumed session would not exercise the superseded-owner retirement. + expect(graded.checks.find(check => check.name === "follow_up_new_session")).toMatchObject({ ok: false }); + const sessions = (first: string, second: string) => question({ rotation: task.question!.rotation }).runs + .map((run, index) => ({ ...run, nativeSessionId: index === 0 ? first : second })); + const replaced = gradeQuestionHandoff(NONCE, artifacts({ rotation: task.question!.rotation, runs: sessions("s1", "s2") })); + expect(replaced).toMatchObject({ verdict: "pass" }); + expect(gradeQuestionHandoff(NONCE, artifacts({ rotation: task.question!.rotation, runs: sessions("s1", "s1") })).verdict).toBe("fail"); + const refused = gradeQuestionHandoff(NONCE, artifacts({ rotation: { at: 2_000, idleTimeoutMs: 300_002, httpStatus: 422, error: "invalid" } })); + expect(refused.verdict).toBe("fail"); + }); + }); }); diff --git a/tests/restart-chaos/question-handoff.ts b/tests/restart-chaos/question-handoff.ts index a548a8a7ac6..cc9d81efa59 100644 --- a/tests/restart-chaos/question-handoff.ts +++ b/tests/restart-chaos/question-handoff.ts @@ -12,6 +12,7 @@ import type { ChaosApi } from "./api.js"; import type { OracleArtifacts, OracleResult, TaskRecord, TaskSample } from "./types.js"; import type { WorkloadDefinition } from "./workloads.js"; +import type { SessionRotation } from "./session-faults.js"; type Json = Record; @@ -30,6 +31,8 @@ export interface QuestionRecord { answer?: QuestionAnswer; /** Q4: every answer the board gave, in order (`answer` is the first). */ answers?: QuestionAnswer[]; + /** SR1: the agent configuration change made after the asking run settled, before the answer. */ + rotation?: SessionRotation; } export interface QuestionAnswer { @@ -39,7 +42,9 @@ export interface QuestionAnswer { } export interface QuestionArtifacts extends QuestionRecord { - runs: Array<{ id: string; status: string; errorCode: string | null; createdAt: string | null; startedAt?: string | null }>; + runs: Array<{ id: string; status: string; errorCode: string | null; createdAt: string | null; startedAt?: string | null; + /** SR1: the run's normalized provider session, read only when the board replaced the session. */ + nativeSessionId?: string | null }>; statuses: Record; } @@ -103,7 +108,9 @@ function questionInteractions(rows: Json[] | null | undefined): Json[] { type QuestionStepInput = { api: Pick; issueId: string; nonce: string; record: TaskRecord; sample: TaskSample; now: number; boardMayAct: boolean; note: (kind: string, data: Record) => void; /** Q4: answer each saved question once, in order. */ - sequence?: boolean }; + sequence?: boolean; + /** SR1: replace the saved session while the asking run's owner is still retained. */ + beforeAnswer?: () => Promise }; /** Q1 and Q4 board step, called on every poll. */ export async function questionStep(input: QuestionStepInput): Promise { @@ -126,6 +133,12 @@ export async function questionStep(input: QuestionStepInput): Promise { } // The saved question must settle on its own first: answer only once nothing of the task is running. if (!input.boardMayAct || (input.sample.runs ?? []).some(run => ACTIVE.has(run.status))) return; + if (input.beforeAnswer && !question.rotation) { + question.rotation = await input.beforeAnswer(); + input.note("session_rotated", { ...question.rotation }); + // Answer on a later poll, so the change is committed before the wake. + return; + } const code = q1Code(input.nonce); const questions = (pending.payload?.questions ?? []) as Json[]; const target = questions[0]; @@ -199,15 +212,32 @@ export async function collectQuestionArtifacts(api: Pick, issue const runs = await api.get(`/api/issues/${issueId}/runs`, 10_000).catch(() => []); const interactions = questionInteractions(await api.get(`/api/issues/${issueId}/interactions`, 10_000).catch(() => [])); const question = record.question ?? { interactionIds: [], seenAt: null, askedByRunId: null, pendingNow: false }; + // SR1 proves its follow-up started a new provider session, so a pass is not a resumed session. + const sessions = new Map(); + if (question.rotation) for (const run of runs ?? []) { + const id = String(run.runId ?? run.id); + const detail = await api.get(`/api/heartbeat-runs/${id}`, 10_000).catch(() => null); + sessions.set(id, typeof detail?.nativeSessionId === "string" ? detail.nativeSessionId : null); + } return { ...question, interactionIds: [...new Set([...question.interactionIds, ...interactions.map(interaction => String(interaction.id))])], statuses: Object.fromEntries(interactions.map(interaction => [String(interaction.id), String(interaction.status)])), runs: (runs ?? []).map(run => ({ id: String(run.runId ?? run.id), status: String(run.status), errorCode: run.errorCode ?? null, - createdAt: run.createdAt ?? null, startedAt: run.startedAt ?? null })), + createdAt: run.createdAt ?? null, startedAt: run.startedAt ?? null, + ...(sessions.has(String(run.runId ?? run.id)) ? { nativeSessionId: sessions.get(String(run.runId ?? run.id)) } : {}) })), }; } +/** SR1: a run created after the answer used a different provider session than the asking run. */ +function followUpSessionCheck(runs: QuestionArtifacts["runs"], asked: QuestionArtifacts["runs"][number] | undefined, answeredAt: number | null) { + const followUps = runs.filter(run => run.id !== asked?.id && answeredAt !== null && run.createdAt && Date.parse(run.createdAt) > answeredAt); + const before = asked?.nativeSessionId ?? null; + const replaced = before !== null && followUps.some(run => typeof run.nativeSessionId === "string" && run.nativeSessionId !== before); + return { name: "follow_up_new_session", ok: replaced, + detail: `asked=${before ?? "unknown"} followUps=${followUps.map(run => run.nativeSessionId ?? "unknown").join(",") || "none"}` }; +} + export function gradeQuestionHandoff(nonce: string, artifacts: OracleArtifacts): OracleResult { const question = artifacts.question; const code = q1Code(nonce); @@ -231,6 +261,9 @@ export function gradeQuestionHandoff(nonce: string, artifacts: OracleArtifacts): detail: asked ? `${asked.id} ${asked.status}${asked.errorCode ? ` ${asked.errorCode}` : ""}` : `run=${question?.askedByRunId ?? "unknown"}` }, { name: "no_failed_attempt", ok: failed.length === 0, detail: failed.map(run => `${run.id}:${run.status}:${run.errorCode ?? ""}`).join(",") || "none" }, { name: "nothing_ran_before_answer", ok: between.length === 0, detail: between.map(run => run.id).join(",") || "none" }, + ...(question?.rotation ? [followUpSessionCheck(runs, asked, answeredAt), { name: "session_rotated_before_answer", ok: question.rotation.httpStatus === 200 + && (answeredAt === null || question.rotation.at < answeredAt), + detail: `idleTimeoutMs=${question.rotation.idleTimeoutMs} http=${question.rotation.httpStatus}${question.rotation.error ? ` ${question.rotation.error}` : ""}` }] : []), { name: "brief_has_answer", ok: brief.includes(q1AnswerMarker(nonce, code)), detail: q1AnswerMarker(nonce, code) }, { name: "brief_has_three_bullets", ok: bullets >= 3, detail: `bullets=${bullets}` }, ]; diff --git a/tests/restart-chaos/scheduler.ts b/tests/restart-chaos/scheduler.ts index fdde7f8865a..ef829e802b4 100644 --- a/tests/restart-chaos/scheduler.ts +++ b/tests/restart-chaos/scheduler.ts @@ -1,5 +1,6 @@ import { isProviderFault, withProviderFaultRules } from "./provider-faults.js"; import { isDiskFullFault } from "./disk-full-faults.js"; +import { isSessionFault } from "./session-faults.js"; import { takesJourneyTriggers } from "./journeys.js"; import { workloadDefinitions } from "./workloads.js"; import { isTimeFault, timeFaultHoldMs } from "./time-faults-config.js"; @@ -80,8 +81,8 @@ const STORM_RESULT_FAULTS: readonly FaultKind[] = ["K1", "K2"]; * A `result` task in a storm also fires its own K1 or K2 just after its run * accepts a result, the window of RC-30, RC-33 and RC-41. */ -function stormCellFaults(options: Pick, trigger: TriggerId): readonly TaskFaultKind[] { - if (!options.stormMinutes) return [...options.faults, ...(options.providerFaults ?? []), ...(options.diskFullFaults ?? [])]; +function stormCellFaults(options: Pick, trigger: TriggerId): readonly TaskFaultKind[] { + if (!options.stormMinutes) return [...options.faults, ...(options.providerFaults ?? []), ...(options.diskFullFaults ?? []), ...(options.sessionFaults ?? [])]; if (trigger === "time") return ["K13"]; if (trigger !== "result") throw new Error(`A K13 storm fires on its own clock; --triggers ${trigger} does not combine with it (use time or result)`); const faults = options.faults.filter(fault => STORM_RESULT_FAULTS.includes(fault)); @@ -97,7 +98,7 @@ function pinnedFireAfterMs(faultAfterMs: number | undefined, drawn: number): num return faultAfterMs ?? drawn; } -export function planTasks(options: Pick): TaskPlan[] { +export function planTasks(options: Pick): TaskPlan[] { const rng = createRng(options.seed); const triggers = options.triggers?.length ? options.triggers : ["time" as const]; for (const trigger of triggers) { @@ -298,7 +299,7 @@ export function taskFaultSchedule(task: TaskPlan): ScheduledFault[] { // `--provider-fault-restart` asked for a K1 once the provider fault applied (the `provider` trigger). const restart = task.providerFault?.restart; if (restart) return [{ kind: restart.kind, holdMs: restart.holdMs ?? 0, atMs: 0, trigger: "provider", delayMs: restart.afterAppliedMs }]; - if (task.fault === "K13" || task.fault === "K0" || isProviderFault(task.fault) || isDiskFullFault(task.fault)) return []; + if (task.fault === "K13" || task.fault === "K0" || isProviderFault(task.fault) || isDiskFullFault(task.fault) || isSessionFault(task.fault)) return []; const planned: ScheduledFault = { kind: task.fault, holdMs: task.holdMs, atMs: task.fireAfterMs, ...(task.restartSteps ? { restartSteps: task.restartSteps } : {}) }; const trigger = task.trigger ?? "time"; diff --git a/tests/restart-chaos/session-faults.ts b/tests/restart-chaos/session-faults.ts new file mode 100644 index 00000000000..83822c803b9 --- /dev/null +++ b/tests/restart-chaos/session-faults.ts @@ -0,0 +1,49 @@ +/** + * Session faults (SR1): the agent's saved provider session is replaced while a + * warm owner of the same task is still retained. Nothing restarts. + * + * SR1 runs with the Q1 workload. After the asking run parks on its saved + * question and settles, the board changes the agent's adapter configuration + * through the agent route a user would use. The answer then wakes a follow-up + * that cannot resume the saved session, so it must retire the parked owner + * (whose question callback is still unsettled) before it starts a new one. + * The retirement must not fail the follow-up: Q1's `no_failed_attempt` and + * `issue_done` checks are the oracle. Two Q1 tasks on the same agent are the + * two-concurrent-runs variant: one change rotates both tasks' sessions. + */ +import type { ChaosApi } from "./api.js"; +import type { WorkloadId } from "./types.js"; + +export const SESSION_FAULT_KINDS = ["SR1"] as const; +export type SessionFaultKind = (typeof SESSION_FAULT_KINDS)[number]; +export const SESSION_FAULT_WORKLOADS: readonly WorkloadId[] = ["Q1"]; + +export function isSessionFault(value: string): value is SessionFaultKind { + return (SESSION_FAULT_KINDS as readonly string[]).includes(value); +} + +/** What the board changed, recorded on the task's question. */ +export interface SessionRotation { + at: number; + idleTimeoutMs: number; + httpStatus: number | null; + error: string | null; +} + +/** + * Change one adapter setting that is part of the session's configuration + * fingerprint and is harmless to the run: the warm idle window, by a few + * milliseconds unique to this task. + */ +export async function rotateAgentSession(api: Pick, agentId: string, taskIndex: number, now: number): Promise { + const idleTimeoutMs = 300_000 + 1 + taskIndex; + const rotation: SessionRotation = { at: now, idleTimeoutMs, httpStatus: null, error: null }; + try { + await api.patch(`/api/agents/${agentId}`, { adapterConfig: { idleTimeoutMs } }); + rotation.httpStatus = 200; + } catch (error) { + rotation.httpStatus = typeof (error as { status?: unknown }).status === "number" ? (error as { status: number }).status : null; + rotation.error = error instanceof Error ? error.message : String(error); + } + return rotation; +} diff --git a/tests/restart-chaos/types.ts b/tests/restart-chaos/types.ts index 734a70493a0..7c28ff11f6c 100644 --- a/tests/restart-chaos/types.ts +++ b/tests/restart-chaos/types.ts @@ -21,9 +21,10 @@ export const K4_MIN_HOLD_MS = 120_000; * A task's planned fault. "K13" (restart storm) tasks have none of their own; the wave's storm crosses them. * PV1–PV3 (provider message faults, provider-faults.ts) restart nothing: the task's rule is armed in the Codex shim. * "K0" is a `--matrix` baseline cell: the same journey with no fault. + * SR1 (session faults, session-faults.ts) restarts nothing: the board replaces the saved session before answering. * DF1 (disk-full-faults.ts) restarts nothing either: the task's final workspace saves find the sandbox disk full. */ -export type TaskFaultKind = FaultKind | "K13" | "K0" | ProviderFaultKind | DiskFullFaultKind; +export type TaskFaultKind = FaultKind | "K13" | "K0" | ProviderFaultKind | DiskFullFaultKind | import("./session-faults.js").SessionFaultKind; /** * G1: an approval-gated action across a restart. H1: a budget hard stop crossed mid-run with a restart (its own wave). @@ -126,6 +127,8 @@ export interface WaveOptions { providerFaults?: ProviderFaultKind[]; /** Sandbox disk-full faults from `--faults` (DF1); see disk-full-faults.ts. */ diskFullFaults?: DiskFullFaultKind[]; + /** Session faults from `--faults` (SR1); see session-faults.ts. */ + sessionFaults?: Array; providerFaultMessage?: ProviderMessage; /** `--provider-fault-message S1=interrupt_ack,...`: a workload's own message, over `providerFaultMessage`. */ providerFaultMessageByWorkload?: Partial>;