From e6b538efd24e81fa0b5ffa2321a4533dd93115f3 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Wed, 11 Nov 2026 10:53:50 -0600 Subject: [PATCH] fix(native): retire a superseded warm owner instead of failing the new run (#20844) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - Native runs keep a warm Runner between turns of the same task, so the next run can start fast > - When the next run needs a new provider session (for example after an agent setting or skill changes), it first retires the previous run's warm Runner > - That retirement uses the previous owner's close. When the close cannot prove checkpoint suspension, the new run failed before any provider work, with the previous owner's error `provider_transport_failed: runner did not durably suspend before checkpoint` > - A governed-wait park (a saved question or review request) always leaves such an owner: its question callback stays unsettled by design. A failed Launcher stop RPC leaves one too > - Nothing retried the run, and the task rested as "recovery needed" with no provider work done > - This pull request lets the new run retire the old owner when that owner is contained, and forces the containment when the close left it unproven > - The benefit is that a fresh run never fails on its predecessor's close, and the chaos harness now covers this class ## Linked Issues or Issue Description **What happened?** A new run on a task whose previous run left a warm Runner retained failed within two minutes, before any provider turn. The error was `provider_transport_failed: runner did not durably suspend before checkpoint`. The run row recorded `executionRecovery: { kind: "bootstrap", providerWorkStarted: false }`. Two runs of the same agent on two tasks failed this way 47 seconds apart, after a configuration change made both start new provider sessions. One predecessor had parked on a `request_human_input` question; its callback never settled. **Expected behavior** A run that has done no provider work does not fail because its predecessor's Runner could not prove suspension. The new session never resumes the old checkpoint. Core must only prove that the old owner is contained, and then start. **Steps to reproduce** 1. Run a warm native task that parks on a saved question (chaos workload Q1). 2. After the asking run settles, change the agent's adapter configuration, so the next run cannot resume the saved session. 3. Answer the question. The follow-up run retires the parked owner and fails with the error above. Related: #18373 added the stopped-owner reset path that this change extends. ## What Changed - `closeWarmNativeSession` (server native executor): during retirement for a different normalized session, accept an unrecoverable close when the old owner is contained. A plugin task owner must show its `executionStopped: true` receipt and a completed handoff. Without a task owner, the Runner's own durable suspension (`settlement.runnerSuspended`) is the proof. Unsettled Core tool callbacks of the discarded conversation no longer block the reset. - With a plugin task owner, any close error qualifies once the Launcher receipt exists. This includes a failed `stop` that `releaseRunnerProcessOwnership` rethrows, and "Controller journal drain timed out": the discarded session's journal lives in its own state root. Owners without a task keep the narrower rule. `--matrix` now rejects session faults. - `createPluginRunnerTaskOwner`: add `retireSupersededTask()`. When close left the receipt unproven, the new run sends one confirmed `stop` (not resent after a confirmed stop), takes the credential snapshot, and calls `complete`. It never reads the old state. - Same-session continuation, a handoff that still fails, other close errors and failed instruction collection still block replacement. - Chaos harness: add session fault SR1 (`tests/restart-chaos/session-faults.ts`). With Q1 and `--lifecycle warm` (or `--daytona-warm`), the board changes `adapterConfig.idleTimeoutMs` after the asking run settles and before it answers. `--tasks 2 --concurrency 2` is the two-concurrent-runs variant. Q1's oracle applies, plus `session_rotated_before_answer` and `follow_up_new_session`. - Docs: `doc/plugins/PLUGIN_RUNNER_EXECUTION.md` and the chaos README describe the new boundary and cell. ## Verification - RED on master: `native-session-executor.test.ts -t "permanently rejected plugin close during reset"` fails the `pending-semantic` and `stop-unconfirmed-at-close` cases with `NativeSessionCloseUnrecoverableError: provider_transport_failed: runner did not durably suspend before checkpoint`. GREEN with this change: all identity-transition cases pass (49). - Chaos SR1 local, master server code (`--env local --lifecycle warm --workloads Q1 --faults SR1`, seeds 301 with 2 tasks and 303 with 1 task): all tasks pass, and the follow-up started a new provider session. The local parked owner closes cleanly, so the local cell does not reproduce the failure. It needs a Daytona task owner. - Chaos SR1 local with this change (seed 302, 2 concurrent tasks on one agent): 2/2 pass, `follow_up_new_session` true for both. - Not run: the Daytona SR1 variant. No runner image matches this checkout's runner content ID, and this change does not touch runner inputs. - `pnpm exec vitest run --config tests/restart-chaos/vitest.config.ts question-handoff scheduler config` passes. Server and harness `tsc --noEmit` pass. ## Risks - The reset no longer waits for unsettled Core tool callbacks of the discarded session. Their run is already terminal and their provider is stopped, so no result can reach a provider, and the new session never replays them. Core-side tool executions keep their own receipts under the old run. - A local owner without a task needs the Runner's durable suspension. A local owner that did not suspend still blocks replacement, as before. - Previous-release story: this changes no persisted state. Warm owners live only in controller memory, so a deploy drops them as before. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, agentic coding with tool use (shell, file edits, test runs) in Claude Code. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 --- doc/plugins/PLUGIN_RUNNER_EXECUTION.md | 20 +++++--- .../native-session-executor.test.ts | 34 ++++++++++--- .../native-runtime/native-session-executor.ts | 25 +++++++--- .../native-runtime/plugin-runner-task.ts | 22 ++++++++- tests/restart-chaos/README.md | 31 ++++++++++++ tests/restart-chaos/config.ts | 19 +++++-- tests/restart-chaos/journey-matrix.ts | 2 + tests/restart-chaos/observer.ts | 4 +- tests/restart-chaos/question-handoff.test.ts | 48 ++++++++++++++++++ tests/restart-chaos/question-handoff.ts | 39 +++++++++++++-- tests/restart-chaos/scheduler.ts | 9 ++-- tests/restart-chaos/session-faults.ts | 49 +++++++++++++++++++ tests/restart-chaos/types.ts | 5 +- 13 files changed, 272 insertions(+), 35 deletions(-) create mode 100644 tests/restart-chaos/session-faults.ts 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>;