From 4158799d1b5d7604def6adf903993578b8a93bfc Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Mon, 9 Nov 2026 20:31:18 -0600 Subject: [PATCH] test(chaos): add provider message faults (PV1-PV3) and a saved-question scenario (#20432) 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. > - The restart chaos harness (`tests/restart-chaos`) tests that agent work survives faults. > - The harness has two fault families: control-plane restarts (K*) and storage faults (ST*). > - On 2026-11-09, two production failures came from a third family. The provider was alive, but a protocol message was late or missing (#20330, #20336). > - The harness could not produce that family, so it did not catch these failures. > - This pull request adds provider message faults (PV1 delay, PV2 drop, PV3 duplicate) on the real path between the runner and the Codex app-server. It also adds a saved-question scenario (Q1) that uses the faults. > - The benefit: the harness can now find this class of failure, and it fails against both pre-fix commits. ## Linked Issues or Issue Description Refs #20330 Refs #20336 **What happened?** A Codex interruption terminal that arrived after 18 s failed a governed wait (#20330). A dropped terminal after a saved handoff failed shutdown settlement with "Couldn't finish this attempt" (#20336). The chaos harness could not inject either fault. **Expected behavior** The harness can delay, drop or duplicate a chosen provider protocol message for one task. A scenario fails on the pre-fix product code and passes on master. **Steps to reproduce** Run the wave commands in Verification. Run them on master and in a worktree at the pre-fix commit with `tests/restart-chaos` copied from this branch. ## What Changed - `provider-faults.ts` and `provider-shim/`: the PV1–PV3 fault family. - A resolve hook goes into the wave's server through `NODE_OPTIONS`. It answers only the `@openai/codex/package.json` lookup that product code uses to find the local Codex. It returns a stand-in package in the instance temp root, and then removes itself from `NODE_OPTIONS`. - The stand-in `codex` bin is a stdio shim. It starts the real Codex with the same arguments and relays JSON lines unchanged. It changes only the message that the task's rule names: the `turn/completed` after a `turn/interrupt` (`terminal`) or the response to `turn/interrupt` (`interrupt_ack`). Each rule applies once per task. The task nonce binds the rule to the task. - Every decision goes to `/provider-faults.jsonl`. A PV task fails `provider_fault_applied` when its fault did not happen. - There is no product change. Thus the same harness runs against older product commits. - `question-handoff.ts`: Q1 workload. The agent saves one question with `request_human_input`. After the asking run settles, the board answers through the respond route. The oracle checks that the question was asked once, the asking run succeeded, no run failed, nothing ran before the answer, and the brief has the answer. - PV faults run only with `--env local --adapters codex`, only with workloads that interrupt a turn (Q1, S1, G1), and not in the same wave as restart faults. A PV1 delay passes only when the held message was released. - Registry edits: `--faults` accepts PV1–PV3. New options are `--provider-fault-message` and `--provider-fault-delays` (a sweep, cycled across tasks). The edits are in `types.ts`, `config.ts`, `scheduler.ts`, `workloads.ts`, `observer.ts` (Q1 board step) and `main.ts` (install the seam and add the check). - README section with commands, the seam, and what is not covered. ## Verification - Unit tests: `pnpm exec vitest run --config tests/restart-chaos/vitest.config.ts`. 23 files and 279 tests pass on the final head. The new tests drive the real shim against a stand-in provider for each kind (delay, drop, duplicate, interrupt ack, once per task, held message discarded on exit, a multibyte character split across writes). They also run the resolve hook in a child Node process. - Typecheck: `node scripts/typecheck-test-projects.mjs tests/restart-chaos` passes. - Live waves (real Runner, real Codex 0.160.0, gpt-6.1-sol, macOS arm64): Pre-fix runs use a worktree at the merge parent with `tests/restart-chaos` copied from this branch. Common arguments: `--env local --adapters codex --workloads Q1`. | Scenario | Product code | Wave | Result | | --- | --- | --- | --- | | 1. PV1, terminal 18 s late (`--faults PV1 --provider-fault-delays 18 --tasks 1 --seed 101`) | master (`8004ad069e`) | `w-20261110T000333-s101` | pass: terminal held 18 s and released, asking run `succeeded`, one question, brief has the answer | | 1. same | `34c1e6469c` (parent of #20330 merge) | `w-20261110T014734-s101` | **fail**: Codex stopped at 14.9 s with the terminal still held, asking run `failed native_provider_terminal_failed`, banner "Couldn’t finish this attempt. Your message is saved.", Retry needed (R3) | | 1. sweep 5/14/25/29 s (`--faults PV1 --provider-fault-delays 5,14,25,29 --concurrency 2 --tasks 4 --seed 102`) | master | `w-20261110T015426-s102` | 4/4 pass | | 1. same sweep | `34c1e6469c` | `w-20261110T015708-s102` | 5 s and 14 s pass; **25 s and 29 s fail** (`native_provider_terminal_failed`, Retry needed): the old 15 s drain | | 2. PV2, terminal dropped (`--faults PV2 --tasks 1 --seed 103`) | master | `w-20261110T015239-s103` | pass: terminal dropped, asking run `succeeded` through verified shutdown, one question, nothing ran before the answer | | 2. same | `a67b936885` (parent of #20336 merge) | `w-20261110T015010-s103` | **fail**: asking run `failed native_provider_terminal_failed`, banner "Couldn’t finish this attempt", Retry needed (R3) | The first master wave exited 1 for one cold `agent_boot` speed sample (8.6 s against a 7 s p95 budget), before any fault, while a preflight build ran in parallel. Its task passed. No new product failure was found on master. ## Risks - Test-only. No product code changes. - The shim adds one Node process in front of Codex for PV waves only. Other waves do not change. - The nonce binds a rule to a Codex process. If one Codex process serves two tasks, the second task's traffic can bind it too. Q1 tasks use separate sessions. - Claude is not covered. Claude runs through the ACPX sidecar against a hash-verified provider pack, and its ACP protocol has no interrupt acknowledgement. A Claude shim needs a different seam. ## Model Used Anthropic Claude Opus 5.5 (`claude-opus-5-5`), through Claude Code, with tool use, code edits and local execution. ## 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 - [x] 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 --- tests/restart-chaos/README.md | 64 +++++ tests/restart-chaos/config.ts | 33 ++- tests/restart-chaos/main.ts | 18 +- tests/restart-chaos/observer.ts | 7 + tests/restart-chaos/provider-faults.test.ts | 218 ++++++++++++++++++ tests/restart-chaos/provider-faults.ts | 213 +++++++++++++++++ .../provider-shim/codex-resolve-hook.mjs | 22 ++ .../provider-shim/codex-shim.mjs | 186 +++++++++++++++ tests/restart-chaos/question-handoff.test.ts | 97 ++++++++ tests/restart-chaos/question-handoff.ts | 163 +++++++++++++ tests/restart-chaos/scheduler.ts | 12 +- tests/restart-chaos/types.ts | 24 +- tests/restart-chaos/workloads.ts | 2 + 13 files changed, 1046 insertions(+), 13 deletions(-) create mode 100644 tests/restart-chaos/provider-faults.test.ts create mode 100644 tests/restart-chaos/provider-faults.ts create mode 100644 tests/restart-chaos/provider-shim/codex-resolve-hook.mjs create mode 100644 tests/restart-chaos/provider-shim/codex-shim.mjs create mode 100644 tests/restart-chaos/question-handoff.test.ts create mode 100644 tests/restart-chaos/question-handoff.ts diff --git a/tests/restart-chaos/README.md b/tests/restart-chaos/README.md index 7066ddf8a2e..7466974a64e 100644 --- a/tests/restart-chaos/README.md +++ b/tests/restart-chaos/README.md @@ -293,6 +293,68 @@ pending is retried for the same settle window; if it is still blocked then, the wave fails (not as a harness error) and the instance is kept, because its database owns the cleanup work. The result is in `/sandbox-sweep.json`. +## Provider message faults (PV1–PV3) and the saved question (Q1) + +```sh +# A Codex governed wait whose interruption terminal arrives 18 s late (PR #20330's boundary). +node cli/node_modules/tsx/dist/cli.mjs tests/restart-chaos/main.ts --env local \ + --adapters codex --workloads Q1 --faults PV1 --provider-fault-delays 18 --concurrency 1 --tasks 1 --seed 101 +# Sweep the window around the 15 s / 30 s deadlines, two tasks at a time. +node cli/node_modules/tsx/dist/cli.mjs tests/restart-chaos/main.ts --env local \ + --adapters codex --workloads Q1 --faults PV1 --provider-fault-delays 5,14,25,29 --concurrency 2 --tasks 4 --seed 102 +# The saved question's terminal never arrives during shutdown (PR #20336's boundary). +node cli/node_modules/tsx/dist/cli.mjs tests/restart-chaos/main.ts --env local \ + --adapters codex --workloads Q1 --faults PV2 --concurrency 1 --tasks 1 --seed 103 +``` + +The third fault family: the provider stays alive, but one protocol message between the runner and the +Codex app-server is late, missing or repeated. `--faults` takes them next to restart faults: + +| Fault | The chosen message | +| --- | --- | +| PV1 | is held for the task's delay (`--provider-fault-delays`, seconds, cycled across tasks; default 18) | +| PV2 | is dropped | +| PV3 | is sent twice | + +`--provider-fault-message terminal` (default) picks the `turn/completed` notification that follows a +`turn/interrupt`, which a governed wait (a saved question or approval) or a board Stop sends. +`interrupt_ack` picks the response to that `turn/interrupt`. Each rule applies once per task, to the +first matching message. A PV task fires no restart. Provider faults run only with `--env local --adapters +codex`, only with workloads that interrupt a turn (Q1, S1, G1), and not yet in the same wave as restart +faults: a restart kills the shim with any held message. A PV1 task passes `provider_fault_applied` only when +the held message was released; a provider that exited first turned the delay into a drop. + +**Seam.** Product code finds the local Codex by resolving the installed `@openai/codex/package.json`. +The harness starts the server with `NODE_OPTIONS=--import provider-shim/codex-resolve-hook.mjs`. That +hook answers only this lookup, with a stand-in package in the instance temp root. The stand-in's +`codex` bin is `provider-shim/codex-shim.mjs`, which starts the real Codex with the same arguments and +relays its JSON-lines stdio unchanged. It changes only the message that the task's rule names. The +hook removes itself from `NODE_OPTIONS`, so the server's children do not load it. The stand-in links the real +platform package, so Codex's sandbox still gets its native executable. A process is bound to a task +when the task's nonce appears in its traffic, such as the prompt or a tool result. Product code is +unchanged, so the same harness can run against an older product commit: copy `tests/restart-chaos` +into a worktree at that commit. Every shim decision is logged in `/provider-faults.jsonl`. A PV +task also fails `provider_fault_applied` when its fault never happened. A pass where nothing was +injected proves nothing. + +Claude is not covered. It runs through the ACPX sidecar against a hash-verified provider pack, and it +speaks ACP, where a cancel has no acknowledgement and the prompt response is the terminal. The +same resolve-hook idea would need a different shim and an integrity exception. + +**Q1** asks the agent to save one single-select question with `request_human_input`, end its turn, +and, once answered, write a `brief` containing the board's answer. The harness answers through +`POST /api/issues/:id/interactions/:interactionId/respond`. The nonce fixes the answer as `ALPHA` or +`BRAVO`. The harness answers only after no run of the task is queued or running, so the saved +question must settle on its own first. The oracle checks these points: + +- exactly one question was asked; +- the asking run ended `succeeded`; +- no run of the task failed; +- nothing started between the question and the answer; +- the brief carries the answer. + +Q1 is not a default workload. + ## Fault timing (`--triggers`) Each task samples one trigger with its cell. `time` (default) fires at the @@ -327,6 +389,8 @@ re-arms the task's own triggered faults instead of using their offsets. | `workloads.ts` | A1 brief document, B1 python CLI with tests, B5 long bench, C1 comment while running, M1 delegation, G1 approval across a restart, H1 budget hard stop across a restart, with their T-oracles. | | `evidence.ts`, `ledger.ts` | Evidence bundle for each failure and the JSONL ledger plus `SUMMARY.md`. | | `daytona-image.ts` | Docker-free extraction of the image's runnerd and provider pack, cached per digest, and the controller/image content-ID check. | +| `provider-faults.ts`, `provider-shim/` | PV1–PV3: the resolve hook, the Codex stdio shim, per-task rules and the `provider_fault_applied` check. | +| `question-handoff.ts` | Q1 saved question: the board's answer step and its oracle. | | `daytona-sweep.ts` | Daytona R6: read-only reconciliation of this wave's labelled sandboxes with leases; cleanup only through Paperclip. | ## Output (gitignored) diff --git a/tests/restart-chaos/config.ts b/tests/restart-chaos/config.ts index 72db19379c6..a51e02c61c9 100644 --- a/tests/restart-chaos/config.ts +++ b/tests/restart-chaos/config.ts @@ -2,6 +2,7 @@ import { execFileSync } from "node:child_process"; import path from "node:path"; import { parseArgs } from "node:util"; import { STORAGE_FAULT_KINDS } from "./storage-chaos.js"; +import { PROVIDER_FAULT_KINDS, PROVIDER_FAULT_WORKLOADS, isProviderFault, parseProviderDelays, parseProviderMessage, type ProviderFaultKind } from "./provider-faults.js"; import { ADAPTER_IDS, ENVIRONMENT_KINDS, FAULT_KINDS, TRIGGER_IDS, WORKLOAD_IDS, type AdapterId, type EnvironmentKind, type FaultKind, type TriggerId, type WaveOptions, type WorkloadId, @@ -103,13 +104,15 @@ export function parseCommand(argv: readonly string[], now = Date.now()): ParsedC storage: { type: "string" }, "storage-faults": { type: "string" }, "storage-fault-seconds": { type: "string" }, + "provider-fault-message": { type: "string" }, + "provider-fault-delays": { type: "string" }, }, }); const keep = values.keep === true; const pollMs = integer(values["poll-ms"], "poll-ms", 3_000, 2_000, 5_000); const budgetCents = integer(values["budget-cents"], "budget-cents", 3_000, 1, 100_000); if (values.replay !== undefined) { - for (const key of ["env", "adapters", "workloads", "faults", "triggers", "concurrency", "tasks", "seed", "daytona-image", "daytona-cpu", "daytona-reuse-lease", "storm-minutes", "lifecycle", "k4-hold-max-seconds", "crash-point", "crash-point-hold-ms", "upgrade-from", "workspace-repo", "storage", "storage-faults", "storage-fault-seconds", "fault-after-seconds"] as const) { + for (const key of ["env", "adapters", "workloads", "faults", "triggers", "concurrency", "tasks", "seed", "daytona-image", "daytona-cpu", "daytona-reuse-lease", "storm-minutes", "lifecycle", "k4-hold-max-seconds", "crash-point", "crash-point-hold-ms", "upgrade-from", "workspace-repo", "storage", "storage-faults", "storage-fault-seconds", "fault-after-seconds", "provider-fault-message", "provider-fault-delays"] as const) { if (values[key] !== undefined) throw new Error(`--replay reproduces the saved plan; do not combine it with --${key}`); } return { mode: "replay", ref: parseReplayRef(values.replay), keep, @@ -142,14 +145,19 @@ 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*) and provider message faults (PV*, provider-faults.ts). + const allFaults = list(values.faults, [...FAULT_KINDS, ...PROVIDER_FAULT_KINDS], "faults", + stormMinutes ? STORM_DEFAULT_FAULTS : crashPoint ? ["K1"] : ["K1", "K2", "K4"]); + const providerFaults = allFaults.filter(isProviderFault); const options: WaveOptions = { env, adapters, - // The governance scenarios (G1, H1) are opt-in: H1 needs a wave of its own. - workloads: list(values.workloads, workloadChoices, "workloads", workloadChoices.filter(id => id !== "G1" && id !== "H1" && id !== "R1" && id !== "S1" + // The governance scenarios (G1, H1) and the saved question (Q1) are opt-in: H1 needs a wave of its own. + workloads: list(values.workloads, workloadChoices, "workloads", workloadChoices.filter(id => id !== "G1" && id !== "H1" && id !== "R1" && id !== "S1" && id !== "Q1" && (id !== "M1" || adapters.length >= 2))), // A crash-point wave schedules no fault; K1 is the restart its self-kill gets. - faults: list(values.faults, FAULT_KINDS, "faults", stormMinutes ? STORM_DEFAULT_FAULTS : crashPoint ? ["K1"] : ["K1", "K2", "K4"]), + faults: allFaults.filter((fault): fault is FaultKind => !isProviderFault(fault)), + ...(providerFaults.length ? { providerFaults } : {}), 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), @@ -177,6 +185,23 @@ export function parseCommand(argv: readonly string[], now = Date.now()): ParsedC throw new Error("--fault-after-seconds fires faults by time; use it with --triggers time"); } if (!crashPoint && values["crash-point-hold-ms"] !== undefined) throw new Error("--crash-point-hold-ms applies only to --crash-point"); + if (providerFaults.length) { + if (stormMinutes) throw new Error("Provider message faults (PV*) are planned per task; do not combine them with --storm-minutes"); + if (env !== "local") throw new Error("Provider message faults (PV*) shim the local Codex executable; run them with --env local"); + if (adapters.some(adapter => adapter !== "codex")) throw new Error("Provider message faults (PV*) cover Codex only; pass --adapters codex"); + // A restart kills the shim with any held message and its once-per-task claim, so the two families do not mix yet. + if (options.faults.length) throw new Error("Provider message faults (PV*) do not combine with restart faults (K*) yet; run them in their own wave"); + // Every planned task gets the fault; only these workloads send the turn/interrupt the rules target. + const idle = options.workloads.filter(id => !PROVIDER_FAULT_WORKLOADS.includes(id)); + if (idle.length) throw new Error(`Provider message faults (PV*) need workloads that interrupt a Codex turn (${PROVIDER_FAULT_WORKLOADS.join(", ")}); remove ${idle.join(", ")} from --workloads`); + const message = parseProviderMessage(values["provider-fault-message"]); + if (message) options.providerFaultMessage = message; + const delays = parseProviderDelays(values["provider-fault-delays"]); + if (delays && !providerFaults.includes("PV1")) throw new Error("--provider-fault-delays sets PV1 delays; add PV1 to --faults"); + if (delays) options.providerFaultDelaysMs = delays; + } else if (values["provider-fault-message"] !== undefined || values["provider-fault-delays"] !== undefined) { + throw new Error("--provider-fault-message and --provider-fault-delays need a provider fault (PV1, PV2 or PV3) in --faults"); + } const upgradeFrom = values["upgrade-from"]; if (options.faults.includes("K14") !== (upgradeFrom !== undefined)) { throw new Error("K14 (upgrade mid-run) and --upgrade-from go together: pass both or neither"); diff --git a/tests/restart-chaos/main.ts b/tests/restart-chaos/main.ts index 10d1f767afb..c9e5056c934 100644 --- a/tests/restart-chaos/main.ts +++ b/tests/restart-chaos/main.ts @@ -17,6 +17,7 @@ import { BoardUi } from "./board-ui.js"; import { DEFAULT_MINIO_BINARY, MINIO_RELEASE, minioBinary, objectStoreServerEnvironment, startMinio, type ObjectStore } from "./object-store.js"; import { StorageChaos, planStorageFaults, replayStorageWindows, type StorageFaultReceipt } from "./storage-chaos.js"; import { startStorageFaultProxy } from "./storage-proxy.js"; +import { describeProviderFault, installProviderFaultSeam, resolveRealCodex, withProviderFaultCheck, type ProviderFaultSeam } from "./provider-faults.js"; import { CREDENTIAL_FOR_ADAPTER, parseCommand, replayCommand, waveIdFor } from "./config.js"; import { assertControllerMatchesImage, defaultDaytonaImageCacheRoot, prepareDaytonaImageArtifacts } from "./daytona-image.js"; import { PLUGIN_BUILD_STAMP, pluginBuildState } from "./plugin-build.js"; @@ -224,6 +225,7 @@ async function main(): Promise { let objectStore: ObjectStore | null = null; let storageChaos: StorageChaos | null = null; let storageProxy: Awaited> | null = null; + let providerFaults: ProviderFaultSeam | null = null; let speedSection: string | null = null; let harnessError: unknown; process.on("uncaughtException", error => { @@ -258,6 +260,17 @@ async function main(): Promise { Object.assign(serverEnvironment, objectStoreServerEnvironment(objectStore, storageProxy.endpoint, `chaos/${plan.waveId}/`)); timeline.add("object_store_started", { release: MINIO_RELEASE, endpoint: objectStore.endpoint, proxy: storageProxy.endpoint, bucket: objectStore.bucket }); } + if (plan.tasks.some(task => task.providerFault)) { + // PV1–PV3: route the server's Codex launches through the fault shim (no product change). + const realCodex = resolveRealCodex(repositoryRoot); + providerFaults = await installProviderFaultSeam({ root: supervisor.paths.temporaryRoot, logPath: path.join(waveDir, "provider-faults.jsonl"), + realCommand: realCodex.command, realVersion: realCodex.version, realManifestPath: realCodex.manifestPath, + ...(process.env.NODE_OPTIONS ? { inheritedNodeOptions: process.env.NODE_OPTIONS } : {}) }); + await providerFaults.arm(plan.tasks); + Object.assign(serverEnvironment, providerFaults.serverEnvironment); + timeline.add("provider_faults_installed", { realCodex: realCodex.command, codexVersion: realCodex.version, + rules: plan.tasks.flatMap(task => task.providerFault ? [`${task.taskId}:${describeProviderFault(task.providerFault)}`] : []) }); + } await supervisor.start(serverEnvironment); const api = new ChaosApi(supervisor.baseUrl); const lab = await seedChaosLab({ api, env: plan.options.env, waitForReadyAgents: true, cancelled: () => interrupted, daytonaImage: plan.options.daytonaImage, daytonaCpu: plan.options.daytonaCpu, @@ -485,8 +498,11 @@ async function main(): Promise { } let artifacts: OracleArtifacts | null = null; if (observer.record.issueId) artifacts = await observer.collectArtifacts().catch(() => null); - const oracle = artifacts ? observer.definition.grade(observer.plan.nonce, artifacts, observer.workloadAgents()) + if (artifacts && providerFaults && observer.plan.providerFault) artifacts.providerFaultEvents = await providerFaults.eventsFor(observer.plan.nonce); + const graded = artifacts ? observer.definition.grade(observer.plan.nonce, artifacts, observer.workloadAgents()) : { verdict: "fail" as const, checks: [{ name: "artifacts_collected", ok: false, detail: observer.record.launchError ?? "collection failed" }] }; + // A provider-fault task whose fault never happened tested nothing. + const oracle = withProviderFaultCheck(graded, observer.plan, artifacts?.providerFaultEvents ?? []); const grade = gradeTask(observer.record, faults, observer.definition.budgetMs, oracle); let evidence: string | null = null; if (!grade.pass) { diff --git a/tests/restart-chaos/observer.ts b/tests/restart-chaos/observer.ts index 9cb756a5325..7ed6e1fe0b9 100644 --- a/tests/restart-chaos/observer.ts +++ b/tests/restart-chaos/observer.ts @@ -15,6 +15,7 @@ import type { 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 { captureRemoteWorkspaceFiles, readRemoteWorkspaceFile } from "./remote-workspace-files.js"; import { taskFaultSchedule } from "./scheduler.js"; import { g1ChildTitle, g1Decision, workloadDefinitions, type WorkloadAgents } from "./workloads.js"; @@ -725,6 +726,7 @@ export class TaskObserver { const governance = this.record.governance; if (this.plan.workload === "G1") return Boolean(governance?.approvalSeenAt) && !governance?.decision; if (this.plan.workload === "H1") return Boolean(governance?.budgetPausedAt) && !governance?.resume && !governance?.pauseLostAt; + if (this.plan.workload === "Q1") return awaitingQuestionAnswer(this.record); return false; } @@ -758,6 +760,10 @@ export class TaskObserver { if (!sample.reachable || !this.record.issueId) return; if (this.plan.workload === "G1") await this.approvalStep(sample, now); if (this.plan.workload === "H1") await this.budgetStep(sample, now); + if (this.plan.workload === "Q1") { + await questionStep({ api: this.api, issueId: this.record.issueId, nonce: this.plan.nonce, record: this.record, sample, now, + boardMayAct: this.boardMayAct(sample), note: (kind, data) => this.timeline.add(kind, { taskId: this.plan.taskId, ...data }) }); + } } private async approvalStep(sample: TaskSample, now: number): Promise { @@ -935,6 +941,7 @@ export class TaskObserver { ...(this.plan.workload === "G1" ? { approvals: await this.collectApprovals(issueId) } : {}), ...(this.plan.workload === "H1" ? await this.finalPauseCheck(issueId) : {}), ...(this.record.governance ? { governance: this.record.governance } : {}), + ...(this.plan.workload === "Q1" ? { question: await collectQuestionArtifacts(this.api, issueId, this.record) } : {}), }; } diff --git a/tests/restart-chaos/provider-faults.test.ts b/tests/restart-chaos/provider-faults.test.ts new file mode 100644 index 00000000000..fa884da243b --- /dev/null +++ b/tests/restart-chaos/provider-faults.test.ts @@ -0,0 +1,218 @@ +import { spawn, spawnSync } from "node:child_process"; +import { chmod, mkdtemp, readFile, rm, writeFile } from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { afterEach, describe, expect, it } from "vitest"; +import { parseCommand } from "./config.js"; +import { + RESOLVE_HOOK, installProviderFaultSeam, parseProviderDelays, withProviderFaultCheck, withProviderFaultRules, + type ProviderFaultEvent, type ProviderFaultSeam, +} from "./provider-faults.js"; +import { planWave, taskFaultSchedule } from "./scheduler.js"; +import type { TaskPlan, WaveOptions } from "./types.js"; + +/** A stand-in Codex app-server: answers turn/start, acknowledges turn/interrupt, then sends the interrupted terminal. */ +const FAKE_PROVIDER = `#!/usr/bin/env node +let buffer = ""; +const send = value => process.stdout.write(JSON.stringify(value) + "\\n"); +process.stdin.on("data", chunk => { + buffer += chunk; + let index; + while ((index = buffer.indexOf("\\n")) >= 0) { + const message = JSON.parse(buffer.slice(0, index)); + buffer = buffer.slice(index + 1); + if (message.method === "turn/start") { + // A multibyte character split across two writes must reach the runner intact. + const delta = Buffer.from(JSON.stringify({ method: "item/agentMessage/delta", params: { delta: "h\\u00e9llo \\u2713" } }) + "\\n"); + const cut = delta.indexOf(Buffer.from("\\u2713")) + 1; + process.stdout.write(delta.subarray(0, cut)); + setTimeout(() => { + process.stdout.write(delta.subarray(cut)); + send({ id: message.id, result: { turn: { id: "turn-" + message.id } } }); + send({ method: "turn/started", params: { threadId: message.params.threadId, turn: { id: "turn-" + message.id } } }); + }, 20); + } + if (message.method === "turn/interrupt") { + // After the split delta above has finished, even when both requests arrive in one read. + setTimeout(() => { + send({ id: message.id, result: {} }); + send({ method: "turn/completed", params: { threadId: message.params.threadId, turn: { id: message.params.turnId, status: "interrupted" } } }); + }, 50); + } + } +}); +// Let replies scheduled above go out first. +process.stdin.on("end", () => setTimeout(() => process.exit(0), 100)); +`; + +const roots: string[] = []; +afterEach(async () => { for (const root of roots.splice(0)) await rm(root, { recursive: true, force: true }); }); + +async function seam(): Promise<{ seam: ProviderFaultSeam; root: string }> { + const root = await mkdtemp(path.join(os.tmpdir(), "chaos-provider-faults-")); + roots.push(root); + const provider = path.join(root, "fake-codex.mjs"); + await writeFile(provider, FAKE_PROVIDER); + await chmod(provider, 0o755); + const installed = await installProviderFaultSeam({ root, logPath: path.join(root, "provider-faults.jsonl"), realCommand: provider, realVersion: "0.0.0-test" }); + return { seam: installed, root }; +} + +function task(nonce: string, rule: TaskPlan["providerFault"]): TaskPlan { + return { taskId: "t01", index: 0, workload: "Q1", adapter: "codex", fault: rule!.kind, fireAfterMs: 0, holdMs: 0, followUpAfterMs: null, nonce, providerFault: rule }; +} + +/** Drive one turn through the shim: start (with the prompt), interrupt, and collect what reaches the runner. */ +async function turnThroughShim(root: string, prompt: string, waitMs = 600) { + const shim = path.join(root, "provider-faults", "codex-package", "bin", "codex.js"); + const child = spawn(shim, ["app-server"], { stdio: ["pipe", "pipe", "inherit"] }); + const lines: Array<{ at: number; message: Record }> = []; + let buffer = ""; + child.stdout.setEncoding("utf8"); + child.stdout.on("data", chunk => { + buffer += chunk; + let index; + while ((index = buffer.indexOf("\n")) >= 0) { + const text = buffer.slice(0, index); let parsed; try { parsed = JSON.parse(text); } catch { throw new Error(`bad line: ${JSON.stringify(text)}`); } lines.push({ at: Date.now(), message: parsed }); + buffer = buffer.slice(index + 1); + } + }); + const write = (value: unknown) => child.stdin.write(`${JSON.stringify(value)}\n`); + write({ id: 1, method: "turn/start", params: { threadId: "thread-1", input: [{ type: "text", text: prompt }] } }); + await new Promise(resolve => setTimeout(resolve, 100)); + const interruptedAt = Date.now(); + write({ id: 2, method: "turn/interrupt", params: { threadId: "thread-1", turnId: "turn-1" } }); + await new Promise(resolve => setTimeout(resolve, waitMs)); + child.stdin.end(); + const code = await new Promise(resolve => child.on("close", resolve)); + return { + code, interruptedAt, + acks: lines.filter(line => line.message.id === 2), + terminals: lines.filter(line => line.message.method === "turn/completed"), + starts: lines.filter(line => line.message.id === 1), + deltas: lines.filter(line => line.message.method === "item/agentMessage/delta").map(line => line.message.params.delta as string), + }; +} + +describe("provider message fault shim", () => { + it("relays a turn unchanged when no rule is armed for the task", async () => { + const { root } = await seam(); + const result = await turnThroughShim(root, "Chaos Q1 question abcd1234"); + expect(result.code).toBe(0); + expect(result.starts).toHaveLength(1); + expect(result.acks).toHaveLength(1); + expect(result.terminals).toHaveLength(1); + expect(result.deltas).toEqual(["h\u00e9llo \u2713"]); + }); + + it("PV1 delays only the interrupted turn's terminal, once per task", async () => { + const { seam: installed, root } = await seam(); + await installed.arm([task("abcd1234", { kind: "PV1", message: "terminal", delayMs: 400 })]); + const first = await turnThroughShim(root, "Chaos Q1 question abcd1234", 900); + expect(first.acks).toHaveLength(1); + expect(first.terminals).toHaveLength(1); + expect(first.terminals[0]!.at - first.acks[0]!.at).toBeGreaterThanOrEqual(350); + // A second provider process for the same task is not delayed again. + const second = await turnThroughShim(root, "Chaos Q1 question abcd1234", 200); + expect(second.terminals[0]!.at - second.acks[0]!.at).toBeLessThan(300); + const events = await installed.eventsFor("abcd1234"); + expect(events.filter(event => event.event === "applied")).toMatchObject([{ kind: "PV1", message: "terminal", action: "delay", delayMs: 400, turnStatus: "interrupted" }]); + expect(events.filter(event => event.event === "released")).toHaveLength(1); + }); + + it("PV2 drops the terminal and PV3 duplicates it; other tasks are untouched", async () => { + const { seam: installed, root } = await seam(); + await installed.arm([task("aaaa0000", { kind: "PV2", message: "terminal" }), task("bbbb1111", { kind: "PV3", message: "terminal" })]); + expect((await turnThroughShim(root, "nonce aaaa0000")).terminals).toHaveLength(0); + expect((await turnThroughShim(root, "nonce bbbb1111")).terminals).toHaveLength(2); + expect((await turnThroughShim(root, "nonce cccc2222")).terminals).toHaveLength(1); + }); + + it("targets the interruption acknowledgement", async () => { + const { seam: installed, root } = await seam(); + await installed.arm([task("dddd3333", { kind: "PV2", message: "interrupt_ack" })]); + const result = await turnThroughShim(root, "nonce dddd3333"); + expect(result.acks).toHaveLength(0); + expect(result.starts).toHaveLength(1); + expect(result.terminals).toHaveLength(1); + }); + + it("discards a held message when the provider exits first, as a dead provider would", async () => { + const { seam: installed, root } = await seam(); + await installed.arm([task("eeee4444", { kind: "PV1", message: "terminal", delayMs: 5_000 })]); + const result = await turnThroughShim(root, "nonce eeee4444", 200); + expect(result.terminals).toHaveLength(0); + const events = await installed.eventsFor("eeee4444"); + expect(events.filter(event => event.event === "held_message_discarded")).toMatchObject([{ kind: "PV1", message: "terminal", delayMs: 5_000 }]); + const rule = { kind: "PV1" as const, message: "terminal" as const, delayMs: 5_000 }; + // The runner never saw the late terminal: that was a drop, not a delay. + expect(withProviderFaultCheck({ verdict: "pass", checks: [] }, task("eeee4444", rule), events).checks[0]) + .toMatchObject({ name: "provider_fault_applied", ok: false, detail: expect.stringContaining("discarded=1") }); + }); + + it("the server hook answers only the Codex package lookup and leaves no hook for children", async () => { + const { seam: installed, root } = await seam(); + const probe = path.join(root, "probe.cjs"); + await writeFile(probe, `const { createRequire } = require("node:module"); +console.log(JSON.stringify({ codex: createRequire(__filename).resolve("@openai/codex/package.json"), nodeOptions: process.env.NODE_OPTIONS ?? null }));`); + const result = spawnSync(process.execPath, [probe], { encoding: "utf8", + env: { ...process.env, ...installed.serverEnvironment, NODE_OPTIONS: `--no-warnings ${installed.serverEnvironment.NODE_OPTIONS}` } }); + expect(result.status, result.stderr).toBe(0); + const output = JSON.parse(result.stdout) as { codex: string; nodeOptions: string | null }; + expect(output.codex).toBe(installed.serverEnvironment.RESTART_CHAOS_CODEX_PACKAGE_JSON); + expect(output.nodeOptions).toBe("--no-warnings"); + const manifest = JSON.parse(await readFile(output.codex, "utf8")) as { name: string; bin: { codex: string } }; + expect(manifest).toMatchObject({ name: "@openai/codex", bin: { codex: "bin/codex.js" } }); + expect(installed.serverEnvironment.NODE_OPTIONS).toContain(RESOLVE_HOOK); + }); +}); + +describe("provider fault planning", () => { + const base: WaveOptions = { env: "local", adapters: ["codex"], workloads: ["Q1"], faults: [], providerFaults: ["PV1"], concurrency: 2, tasks: 4, + seed: 7, budgetCents: 3000, pollMs: 3000 }; + + it("parses PV kinds from --faults and sweeps PV1 delays across tasks", () => { + const command = parseCommand(["--adapters", "codex", "--workloads", "Q1", "--faults", "PV1", "--provider-fault-delays", "5,14,18,25", + "--tasks", "4", "--concurrency", "2", "--seed", "3"]); + if (command.mode !== "wave") throw new Error("expected a wave"); + expect(command.options).toMatchObject({ faults: [], providerFaults: ["PV1"], providerFaultDelaysMs: [5_000, 14_000, 18_000, 25_000] }); + const plan = planWave("w-20261109T000000-s3", command.options); + expect(plan.tasks.map(task => task.providerFault?.delayMs)).toEqual([5_000, 14_000, 18_000, 25_000]); + expect(plan.tasks.every(task => task.fault === "PV1" && taskFaultSchedule(task).length === 0)).toBe(true); + }); + + it("parses provider faults and rejects combinations that cannot exercise them", () => { + const command = parseCommand(["--adapters", "codex", "--workloads", "Q1,S1", "--faults", "PV2", "--tasks", "2", "--seed", "1"]); + if (command.mode !== "wave") throw new Error("expected a wave"); + expect(command.options).toMatchObject({ faults: [], providerFaults: ["PV2"], workloads: ["Q1", "S1"] }); + expect(() => parseCommand(["--adapters", "codex", "--workloads", "Q1", "--faults", "K1,PV2"])).toThrow(/restart faults/); + expect(() => parseCommand(["--adapters", "codex", "--workloads", "Q1,A1", "--faults", "PV1"])).toThrow(/remove A1/); + expect(() => parseCommand(["--adapters", "codex", "--faults", "PV1"])).toThrow(/interrupt a Codex turn/); + expect(() => parseCommand(["--adapters", "codex,claude", "--workloads", "Q1", "--faults", "PV1"])).toThrow(/Codex only/); + expect(() => parseCommand(["--adapters", "codex", "--workloads", "Q1", "--faults", "PV2", "--provider-fault-delays", "18"])).toThrow(/PV1/); + expect(() => parseCommand(["--adapters", "codex", "--workloads", "Q1", "--provider-fault-message", "terminal"])).toThrow(/provider fault/); + expect(() => parseCommand(["--adapters", "codex", "--workloads", "Q1", "--faults", "PV1", "--provider-fault-message", "result"])).toThrow(/terminal/); + expect(() => parseProviderDelays("18,abc")).toThrow(/not a number/); + expect(() => parseProviderDelays("601")).toThrow(/PV2/); + }); + + it("defaults PV1 to the 18 s late terminal and leaves restart tasks alone", () => { + const tasks = withProviderFaultRules([ + { ...task("aa", { kind: "PV1", message: "terminal" }), providerFault: undefined, fireAfterMs: 30_000 }, + { ...task("bb", { kind: "PV1", message: "terminal" }), fault: "K1", providerFault: undefined, fireAfterMs: 30_000 }, + ], base); + expect(tasks[0]).toMatchObject({ providerFault: { kind: "PV1", message: "terminal", delayMs: 18_000 }, fireAfterMs: 0 }); + expect(tasks[1]!.providerFault).toBeUndefined(); + expect(tasks[1]!.fireAfterMs).toBe(30_000); + }); + + it("fails a provider-fault task whose fault never happened", () => { + const rule = { kind: "PV1" as const, message: "terminal" as const, delayMs: 18_000 }; + const pass = { verdict: "pass" as const, checks: [{ name: "issue_done", ok: true }] }; + const applied: ProviderFaultEvent[] = [{ at: 1, pid: 1, event: "applied", nonce: "aa", kind: "PV1", message: "terminal", action: "delay" }, + { at: 2, pid: 1, event: "released", nonce: "aa", kind: "PV1", message: "terminal" }]; + expect(withProviderFaultCheck(pass, task("aa", rule), applied).verdict).toBe("pass"); + expect(withProviderFaultCheck(pass, task("aa", rule), []).checks.at(-1)).toMatchObject({ name: "provider_fault_applied", ok: false }); + expect(withProviderFaultCheck(pass, { ...task("aa", rule), providerFault: undefined }, [])).toBe(pass); + }); +}); diff --git a/tests/restart-chaos/provider-faults.ts b/tests/restart-chaos/provider-faults.ts new file mode 100644 index 00000000000..1d362fa0624 --- /dev/null +++ b/tests/restart-chaos/provider-faults.ts @@ -0,0 +1,213 @@ +/** + * Provider message faults (PV1 delay, PV2 drop, PV3 duplicate): the third fault + * family, next to control-plane restarts (K*) and storage faults (ST*). The + * provider stays alive; one chosen protocol message on the real path between the + * Paperclip runner and the Codex app-server is late, missing or repeated. + * + * Seam: `provider-shim/codex-resolve-hook.mjs` is loaded into the wave's server + * through NODE_OPTIONS. It answers the server's lookup of + * `@openai/codex/package.json` (how product code finds the local Codex) with a + * stand-in package in the instance temp root whose `codex` bin is + * `provider-shim/codex-shim.mjs`. The shim runs the real Codex and relays its + * stdio, applying the rule armed for the task's nonce. No product code changes, + * so the same harness runs against older product commits. + * + * Codex only: Claude runs through the ACPX sidecar with a hash-verified provider + * pack, so the same seam would need a different shim (see README). + */ +import { chmod, copyFile, mkdir, readFile, rename, symlink, writeFile } from "node:fs/promises"; +import { readFileSync, realpathSync } from "node:fs"; +import { createRequire } from "node:module"; +import path from "node:path"; +import type { OracleResult, TaskPlan, WaveOptions, WorkloadId } from "./types.js"; + +export const PROVIDER_FAULT_KINDS = ["PV1", "PV2", "PV3"] as const; +export type ProviderFaultKind = (typeof PROVIDER_FAULT_KINDS)[number]; +export const PROVIDER_FAULT_ACTIONS: Record = { PV1: "delay", PV2: "drop", PV3: "duplicate" }; + +/** `terminal`: the turn/completed that follows a turn/interrupt. `interrupt_ack`: the response to that turn/interrupt. */ +export const PROVIDER_MESSAGES = ["terminal", "interrupt_ack"] as const; +export type ProviderMessage = (typeof PROVIDER_MESSAGES)[number]; + +/** PV1 default: the late terminal of PR #20330 (the old 15 s governed-wait drain against Codex's 30 s command window). */ +export const DEFAULT_PROVIDER_DELAY_MS = 18_000; +/** Bound on a PV1 delay: beyond 10 minutes a message is effectively dropped (PV2). */ +export const MAX_PROVIDER_DELAY_MS = 600_000; + +export interface ProviderFaultRule { + kind: ProviderFaultKind; + message: ProviderMessage; + /** PV1 only. */ + delayMs?: number; +} + +/** One shim log line (JSONL, `provider-faults.jsonl` in the wave directory). */ +export interface ProviderFaultEvent { + at: number; + pid: number; + event: "spawned" | "bound" | "interrupt_requested" | "applied" | "released" | "held_message_discarded" | "exited" | "spawn_failed"; + nonce?: string | null; + kind?: ProviderFaultKind; + message?: ProviderMessage; + action?: "delay" | "drop" | "duplicate"; + delayMs?: number; + turnStatus?: string | null; + [key: string]: unknown; +} + +/** Workloads whose run sends turn/interrupt: a saved question (Q1), a board Stop (S1), an approval wait (G1). */ +export const PROVIDER_FAULT_WORKLOADS: readonly WorkloadId[] = ["Q1", "S1", "G1"]; + +export function isProviderFault(value: string): value is ProviderFaultKind { + return (PROVIDER_FAULT_KINDS as readonly string[]).includes(value); +} + +/** + * `--provider-fault-delays 5,14,18` (seconds) for PV1: task i gets the i-th delay, cycling, so a + * small wave sweeps the window around a deadline. + */ +export function parseProviderDelays(raw: string | undefined): number[] | undefined { + if (raw === undefined) return undefined; + const values = raw.split(",").map(value => value.trim()).filter(Boolean); + if (!values.length) throw new Error("--provider-fault-delays must not be empty"); + return values.map(value => { + if (!/^[0-9]+(\.[0-9]+)?$/.test(value)) throw new Error(`--provider-fault-delays: "${value}" is not a number of seconds`); + const ms = Math.round(Number(value) * 1000); + if (ms > MAX_PROVIDER_DELAY_MS) throw new Error(`--provider-fault-delays: ${value}s exceeds ${MAX_PROVIDER_DELAY_MS / 1000}s; use PV2 to drop a message`); + return ms; + }); +} + +export function parseProviderMessage(raw: string | undefined): ProviderMessage | undefined { + if (raw === undefined) return undefined; + if (!(PROVIDER_MESSAGES as readonly string[]).includes(raw)) throw new Error(`--provider-fault-message must be one of ${PROVIDER_MESSAGES.join(", ")}`); + return raw as ProviderMessage; +} + +/** Attach each provider-fault task's rule. Restart-fault tasks are unchanged. */ +export function withProviderFaultRules(tasks: TaskPlan[], options: Pick): TaskPlan[] { + const delays = options.providerFaultDelaysMs?.length ? options.providerFaultDelaysMs : [DEFAULT_PROVIDER_DELAY_MS]; + let pv1 = 0; + return tasks.map(task => { + if (!isProviderFault(task.fault)) return task; + const rule: ProviderFaultRule = { kind: task.fault, message: options.providerFaultMessage ?? "terminal", + ...(task.fault === "PV1" ? { delayMs: delays[pv1++ % delays.length]! } : {}) }; + return { ...task, providerFault: rule, fireAfterMs: 0, holdMs: 0 }; + }); +} + +export function describeProviderFault(rule: ProviderFaultRule): string { + return `${rule.kind} ${PROVIDER_FAULT_ACTIONS[rule.kind]} ${rule.message}${rule.delayMs !== undefined ? ` ${rule.delayMs}ms` : ""}`; +} + +/** The Codex executable product code would resolve for this checkout (mirrors codex-command.ts). */ +export function resolveRealCodex(repositoryRoot: string): { command: string; version: string; manifestPath: string } { + const runnerRequire = createRequire(path.join(repositoryRoot, "packages/paperclip-runner/package.json")); + let manifestPath: string; + try { + const bridge = runnerRequire.resolve("@agentclientprotocol/codex-acp/package.json"); + manifestPath = createRequire(bridge).resolve("@openai/codex/package.json"); + } catch { + try { + const adapter = runnerRequire.resolve("@paperclipai/adapter-codex-local/server"); + manifestPath = createRequire(createRequire(adapter).resolve("@agentclientprotocol/codex-acp/package.json")).resolve("@openai/codex/package.json"); + } catch { + manifestPath = runnerRequire.resolve("@openai/codex/package.json"); + } + } + manifestPath = realpathSync(manifestPath); + const manifest = JSON.parse(readFileSync(manifestPath, "utf8")) as { version?: string; bin?: string | Record }; + const bin = typeof manifest.bin === "string" ? manifest.bin : manifest.bin?.codex; + if (!bin) throw new Error(`Installed Codex at ${manifestPath} has no codex bin`); + return { command: realpathSync(path.resolve(path.dirname(manifestPath), bin)), version: String(manifest.version ?? "0.0.0"), manifestPath }; +} + +export interface ProviderFaultSeam { + /** Merge into the server environment before it starts (kept across restarts). */ + serverEnvironment: Record; + controlPath: string; + logPath: string; + realCommand: string; + /** Arm the rules of these tasks (idempotent; replaces the armed set). */ + arm(tasks: readonly TaskPlan[]): Promise; + /** The shim's log lines for one task. */ + eventsFor(nonce: string): Promise; +} + +const SHIM_SOURCE = path.join(import.meta.dirname, "provider-shim", "codex-shim.mjs"); +export const RESOLVE_HOOK = path.join(import.meta.dirname, "provider-shim", "codex-resolve-hook.mjs"); + +/** + * Install the stand-in Codex package under `root` (the instance temp root) and return the server + * environment that routes product Codex launches through the shim. + */ +export async function installProviderFaultSeam(input: { root: string; logPath: string; realCommand: string; realVersion: string; + /** The real `@openai/codex/package.json`: its platform package stays reachable from the stand-in. */ + realManifestPath?: string; inheritedNodeOptions?: string }): Promise { + const base = path.join(input.root, "provider-faults"); + const packageRoot = path.join(base, "codex-package"); + const stateDirectory = path.join(base, "state"); + const controlPath = path.join(base, "control.json"); + await mkdir(path.join(packageRoot, "bin"), { recursive: true, mode: 0o700 }); + await mkdir(stateDirectory, { recursive: true, mode: 0o700 }); + // "type": "module" makes the copied .js shim an ES module, like the real codex.js. The real + // package's platform dependency is linked in, so product code still grants Codex's own + // native executable to its sandbox (codexExecutableReadOnlyRoots). + const real = input.realManifestPath ? JSON.parse(await readFile(input.realManifestPath, "utf8")) as { optionalDependencies?: Record } : {}; + await writeFile(path.join(packageRoot, "package.json"), `${JSON.stringify({ name: "@openai/codex", version: input.realVersion, + type: "module", bin: { codex: "bin/codex.js" }, ...(real.optionalDependencies ? { optionalDependencies: real.optionalDependencies } : {}) }, null, 2)}\n`); + const platformPackage = `@openai/codex-${process.platform}-${process.arch}`; + if (input.realManifestPath && real.optionalDependencies?.[platformPackage]) { + try { + const platformRoot = path.dirname(createRequire(input.realManifestPath).resolve(`${platformPackage}/package.json`)); + await mkdir(path.join(packageRoot, "node_modules", "@openai"), { recursive: true }); + await symlink(platformRoot, path.join(packageRoot, "node_modules", platformPackage), "dir").catch(error => { + if ((error as NodeJS.ErrnoException).code !== "EEXIST") throw error; + }); + } catch { /* No platform package: Codex runs as a standalone executable. */ } + } + const shimPath = path.join(packageRoot, "bin", "codex.js"); + await copyFile(SHIM_SOURCE, shimPath); + await chmod(shimPath, 0o755); + await writeFile(path.join(packageRoot, "shim-config.json"), `${JSON.stringify({ realCommand: input.realCommand, + controlPath, logPath: input.logPath, stateDirectory }, null, 2)}\n`); + await writeFile(controlPath, `${JSON.stringify({ rules: [] })}\n`); + await writeFile(input.logPath, "", { flag: "a", mode: 0o600 }); + const nodeOptions = [input.inheritedNodeOptions, `--import=${RESOLVE_HOOK}`].filter(Boolean).join(" "); + return { + serverEnvironment: { NODE_OPTIONS: nodeOptions, RESTART_CHAOS_CODEX_PACKAGE_JSON: path.join(packageRoot, "package.json") }, + controlPath, logPath: input.logPath, realCommand: input.realCommand, + async arm(tasks) { + const rules = tasks.flatMap(task => task.providerFault ? [{ nonce: task.nonce, ...task.providerFault }] : []); + const temporary = `${controlPath}.${process.pid}.tmp`; + await writeFile(temporary, `${JSON.stringify({ rules }, null, 2)}\n`); + await rename(temporary, controlPath); + }, + eventsFor: nonce => readProviderFaultEvents(input.logPath, nonce), + }; +} + +export async function readProviderFaultEvents(logPath: string, nonce?: string): Promise { + const text = await readFile(logPath, "utf8").catch(() => ""); + const events = text.split("\n").filter(Boolean).flatMap(line => { try { return [JSON.parse(line) as ProviderFaultEvent]; } catch { return []; } }); + return nonce === undefined ? events : events.filter(event => event.nonce === nonce); +} + +/** + * A provider-fault task passes only if its fault really happened: otherwise the scenario was not + * exercised (for example the turn ended before any interrupt), and a pass would prove nothing. + */ +export function withProviderFaultCheck(oracle: OracleResult, plan: TaskPlan, events: readonly ProviderFaultEvent[]): OracleResult { + if (!plan.providerFault) return oracle; + const rule = plan.providerFault; + const applied = events.filter(event => event.event === "applied" && event.kind === rule.kind && event.message === rule.message); + const released = events.filter(event => event.event === "released" && event.kind === rule.kind); + // A held message the provider took with it when it exited never reached the runner: that tested a + // drop (PV2), not a late message. It stays in the detail for diagnosis but does not count. + const discarded = events.filter(event => event.event === "held_message_discarded" && event.kind === rule.kind); + const exercised = applied.length === 1 && (rule.kind !== "PV1" || released.length === 1); + const checks = [...oracle.checks, { name: "provider_fault_applied", ok: exercised, + detail: `${describeProviderFault(rule)}: applied=${applied.length}${rule.kind === "PV1" ? ` released=${released.length} discarded=${discarded.length}` : ""}` + + (applied[0]?.turnStatus ? ` turnStatus=${applied[0].turnStatus}` : "") }]; + return { verdict: checks.every(check => check.ok) ? "pass" : "fail", checks }; +} diff --git a/tests/restart-chaos/provider-shim/codex-resolve-hook.mjs b/tests/restart-chaos/provider-shim/codex-resolve-hook.mjs new file mode 100644 index 00000000000..cbfdafef307 --- /dev/null +++ b/tests/restart-chaos/provider-shim/codex-resolve-hook.mjs @@ -0,0 +1,22 @@ +// Restart chaos provider-message faults: loaded into the wave's server only +// (`NODE_OPTIONS=--import `, set by the harness). Product code +// resolves the local Codex executable through the installed +// `@openai/codex/package.json`; this hook answers that one lookup with the +// harness's stand-in package, whose `codex` bin is the fault shim in front of +// the real Codex. Nothing else resolves differently, and product code is unchanged. +import { registerHooks } from "node:module"; +import { pathToFileURL } from "node:url"; + +const target = process.env.RESTART_CHAOS_CODEX_PACKAGE_JSON; +if (target) { + registerHooks({ + resolve(specifier, context, next) { + if (specifier === "@openai/codex/package.json") return { url: pathToFileURL(target).href, format: "json", shortCircuit: true }; + return next(specifier, context); + }, + }); +} +// The server's children (runner, providers, agent tools) must not load this hook: +// remove only its own option and keep anything else the harness inherited. +const remaining = (process.env.NODE_OPTIONS ?? "").split(/\s+/).filter(option => option && !option.includes("codex-resolve-hook")).join(" "); +if (remaining) process.env.NODE_OPTIONS = remaining; else delete process.env.NODE_OPTIONS; diff --git a/tests/restart-chaos/provider-shim/codex-shim.mjs b/tests/restart-chaos/provider-shim/codex-shim.mjs new file mode 100644 index 00000000000..dad34e4a612 --- /dev/null +++ b/tests/restart-chaos/provider-shim/codex-shim.mjs @@ -0,0 +1,186 @@ +#!/usr/bin/env node +// Restart chaos provider-message fault shim (PV1 delay, PV2 drop, PV3 duplicate). +// +// The harness installs this file as the `codex` executable of a stand-in +// `@openai/codex` package (see ../provider-faults.ts). It starts the real Codex +// command with the same arguments and relays its JSON-lines stdio unchanged, +// except for the one provider message an armed rule names: +// - "terminal": the `turn/completed` notification that follows a `turn/interrupt` +// request (a governed wait or a board Stop); +// - "interrupt_ack": the response to that `turn/interrupt` request. +// Rules are armed per task in the control file, by the task's nonce. A process is +// bound to a nonce once the nonce appears in its traffic (the task's prompt or a +// tool result). Every rule applies at most once per task (an exclusive marker file +// survives provider restarts). Every decision is appended to the log file. +// Plain Node built-ins only: this runs in the provider process, not the harness. +import { spawn } from "node:child_process"; +import { appendFileSync, readFileSync, statSync, writeFileSync } from "node:fs"; +import path from "node:path"; +import { StringDecoder } from "node:string_decoder"; +import { fileURLToPath } from "node:url"; + +const configPath = path.join(path.dirname(fileURLToPath(import.meta.url)), "..", "shim-config.json"); +const config = JSON.parse(readFileSync(configPath, "utf8")); +const startedAt = Date.now(); + +function log(entry) { + try { appendFileSync(config.logPath, `${JSON.stringify({ at: Date.now(), pid: process.pid, ...entry })}\n`); } catch { /* best effort */ } +} + +let rules = []; +let rulesMtimeMs = -1; +function armedRules() { + try { + const mtimeMs = statSync(config.controlPath).mtimeMs; + if (mtimeMs !== rulesMtimeMs) { + rulesMtimeMs = mtimeMs; + const parsed = JSON.parse(readFileSync(config.controlPath, "utf8")); + rules = Array.isArray(parsed.rules) ? parsed.rules : []; + } + } catch { /* keep the last good rules */ } + return rules; +} + +let boundNonce = null; +const nonceByThread = new Map(); +const interruptRequestIds = new Map(); +const interruptedThreads = new Set(); +let interruptSeen = false; + +function findNonce(line) { + for (const rule of armedRules()) if (typeof rule.nonce === "string" && rule.nonce && line.includes(rule.nonce)) return rule.nonce; + return null; +} + +function bindFrom(line, message) { + const nonce = findNonce(line); + if (!nonce) return; + const threadId = message?.params?.threadId; + if (typeof threadId === "string") nonceByThread.set(threadId, nonce); + if (boundNonce !== nonce) { boundNonce = nonce; log({ event: "bound", nonce }); } +} + +function parse(line) { + if (!line.startsWith("{")) return null; + try { return JSON.parse(line); } catch { return null; } +} + +function claim(rule) { + const marker = path.join(config.stateDirectory, `${rule.nonce}.${rule.kind}.${rule.message}.applied`); + try { writeFileSync(marker, String(process.pid), { flag: "wx" }); return true; } catch { return false; } +} + +const child = spawn(config.realCommand, process.argv.slice(2), { stdio: ["pipe", "pipe", "inherit"], env: process.env }); +log({ event: "spawned", childPid: child.pid ?? null, args: process.argv.slice(2).filter(arg => arg === "app-server" || arg === "--version") }); + +// Parent -> provider: relayed byte for byte; read only to bind and to see interrupts. +// A decoder per stream keeps a multibyte character that spans two chunks intact. +const inputDecoder = new StringDecoder("utf8"); +let inputBuffer = ""; +process.stdin.on("data", chunk => { + child.stdin.write(chunk); + inputBuffer += inputDecoder.write(chunk); + let newline; + while ((newline = inputBuffer.indexOf("\n")) >= 0) { + const line = inputBuffer.slice(0, newline); + inputBuffer = inputBuffer.slice(newline + 1); + const message = parse(line); + bindFrom(line, message); + if (message?.method === "turn/start") { + // A new turn: only an interrupt of this turn makes its terminal a target. + interruptSeen = false; + if (typeof message.params?.threadId === "string") interruptedThreads.delete(message.params.threadId); + } + if (message?.method === "turn/interrupt") { + interruptSeen = true; + if (message.id !== undefined) interruptRequestIds.set(JSON.stringify(message.id), message.params?.threadId ?? null); + if (typeof message.params?.threadId === "string") interruptedThreads.add(message.params.threadId); + log({ event: "interrupt_requested", nonce: boundNonce, requestId: message.id ?? null }); + } + } +}); +process.stdin.on("end", () => child.stdin.end()); +child.stdin.on("error", () => undefined); + +/** Held PV1 messages: timer -> the rule it belongs to. */ +const pendingTimers = new Map(); +function emit(line) { process.stdout.write(`${line}\n`); } + +function classify(message) { + if (!message) return null; + if (message.method === "turn/completed") { + const threadId = message.params?.threadId; + const afterInterrupt = typeof threadId === "string" && interruptedThreads.size ? interruptedThreads.has(threadId) : interruptSeen; + return { message: "terminal", afterInterrupt, threadId: typeof threadId === "string" ? threadId : null, + turnStatus: message.params?.turn?.status ?? null }; + } + if (message.method === undefined && message.id !== undefined && interruptRequestIds.has(JSON.stringify(message.id))) { + return { message: "interrupt_ack", afterInterrupt: true, threadId: interruptRequestIds.get(JSON.stringify(message.id)), turnStatus: null }; + } + return null; +} + +function ruleFor(target) { + const nonce = (target.threadId && nonceByThread.get(target.threadId)) || boundNonce; + if (!nonce) return null; + return armedRules().find(rule => rule.nonce === nonce && rule.message === target.message + && (target.afterInterrupt || rule.afterInterrupt === false)) ?? null; +} + +function relay(line) { + const message = parse(line); + if (message) bindFrom(line, message); + const target = classify(message); + const rule = target ? ruleFor(target) : null; + if (!target || !rule || !claim(rule)) { emit(line); return; } + const base = { event: "applied", nonce: rule.nonce, kind: rule.kind, message: target.message, turnStatus: target.turnStatus, + requestId: message.id ?? null, sinceSpawnMs: Date.now() - startedAt }; + if (rule.kind === "PV2") { log({ ...base, action: "drop" }); return; } + if (rule.kind === "PV3") { emit(line); emit(line); log({ ...base, action: "duplicate" }); return; } + const delayMs = Number(rule.delayMs ?? 0); + log({ ...base, action: "delay", delayMs }); + const timer = setTimeout(() => { + pendingTimers.delete(timer); + emit(line); + log({ event: "released", nonce: rule.nonce, kind: rule.kind, message: target.message, delayMs }); + }, delayMs); + pendingTimers.set(timer, { nonce: rule.nonce, kind: rule.kind, message: target.message, delayMs }); +} + +const outputDecoder = new StringDecoder("utf8"); +let outputBuffer = ""; +child.stdout.on("data", chunk => { + outputBuffer += outputDecoder.write(chunk); + let newline; + while ((newline = outputBuffer.indexOf("\n")) >= 0) { + const line = outputBuffer.slice(0, newline); + outputBuffer = outputBuffer.slice(newline + 1); + relay(line); + } +}); +child.stdout.on("end", () => { + outputBuffer += outputDecoder.end(); + if (outputBuffer) { process.stdout.write(outputBuffer); outputBuffer = ""; } +}); + +for (const signal of ["SIGTERM", "SIGINT", "SIGHUP"]) { + process.on(signal, () => { try { child.kill(signal); } catch { /* already gone */ } }); +} + +child.on("error", error => { + log({ event: "spawn_failed", error: String(error) }); + process.stderr.write(`restart-chaos codex shim: ${error.message}\n`); + process.exit(127); +}); +// "close": the provider exited and its stdout is fully relayed. +child.on("close", (code, signal) => { + // A provider that exits takes its held messages with it, as a dead provider would. + for (const [timer, held] of pendingTimers) { + clearTimeout(timer); + log({ event: "held_message_discarded", ...held }); + } + pendingTimers.clear(); + log({ event: "exited", code, signal }); + const finish = () => process.exit(code ?? (signal ? 128 + (signal === "SIGKILL" ? 9 : signal === "SIGINT" ? 2 : 15) : 1)); + if (process.stdout.writableNeedDrain) process.stdout.once("drain", finish); else finish(); +}); diff --git a/tests/restart-chaos/question-handoff.test.ts b/tests/restart-chaos/question-handoff.test.ts new file mode 100644 index 00000000000..ade2c8df09a --- /dev/null +++ b/tests/restart-chaos/question-handoff.test.ts @@ -0,0 +1,97 @@ +import { describe, expect, it } from "vitest"; +import { parseCommand } from "./config.js"; +import { awaitingQuestionAnswer, gradeQuestionHandoff, q1AnswerMarker, q1Code, questionStep, type QuestionArtifacts } from "./question-handoff.js"; +import type { OracleArtifacts, TaskRecord, TaskSample } from "./types.js"; +import { workloadDefinitions } from "./workloads.js"; + +const NONCE = "0bcd1234"; +const record = () => ({ plan: {}, issueId: "i1" } as unknown as TaskRecord); +const sample = (status: string): TaskSample => ({ at: 0, reachable: true, serverReady: true, + runs: [{ id: "run-1", status, createdAt: null, startedAt: null, finishedAt: null, errorCode: null, retryOfRunId: null, wakeCommentIds: [], livenessState: null }] }); + +function fakeApi(interactions: Array>) { + const posts: Array<{ route: string; body: unknown }> = []; + return { posts, api: { + async get(route: string) { return (route.endsWith("/interactions") ? interactions : []) as T; }, + async post(route: string, body?: unknown) { posts.push({ route, body }); return {} as T; }, + } }; +} + +const pendingQuestion = { id: "q1", kind: "ask_user_questions", status: "pending", sourceRunId: "run-1", + payload: { questions: [{ id: "codename", options: [{ id: "opt-a", label: "ALPHA" }, { id: "opt-b", label: "BRAVO" }] }] } }; + +describe("Q1 saved question", () => { + it("is registered as an opt-in workload", () => { + expect(workloadDefinitions.Q1.id).toBe("Q1"); + const command = parseCommand(["--adapters", "codex,claude"]); + if (command.mode !== "wave") throw new Error("expected a wave"); + expect(command.options.workloads).not.toContain("Q1"); + expect(workloadDefinitions.Q1.description(NONCE)).toContain("request_human_input"); + }); + + it("waits for the asking run to settle, then answers once with the nonce's option", async () => { + const task = record(); + const { api, posts } = fakeApi([pendingQuestion]); + 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) }); + await step("running", 1_000); + expect(task.question).toMatchObject({ seenAt: 1_000, askedByRunId: "run-1", pendingNow: true }); + expect(awaitingQuestionAnswer(task)).toBe(true); + expect(posts).toHaveLength(0); + await step("succeeded", 2_000); + expect(posts).toEqual([{ route: "/api/issues/i1/interactions/q1/respond", + body: { answers: [{ questionId: "codename", optionIds: [q1Code(NONCE) === "ALPHA" ? "opt-a" : "opt-b"] }] } }]); + expect(awaitingQuestionAnswer(task)).toBe(false); + await step("succeeded", 3_000); + expect(posts).toHaveLength(1); + expect(notes).toEqual(["question_pending", "question_answered"]); + }); + + it("does not answer while the board may not act (fault in progress)", async () => { + const task = record(); + const { api, posts } = fakeApi([pendingQuestion]); + await questionStep({ api, issueId: "i1", nonce: NONCE, record: task, sample: sample("succeeded"), now: 1, boardMayAct: false, note: () => undefined }); + expect(posts).toHaveLength(0); + expect(awaitingQuestionAnswer(task)).toBe(true); + }); + + const question = (overrides: Partial = {}): QuestionArtifacts => ({ interactionIds: ["q1"], seenAt: 1_000, askedByRunId: "run-1", + pendingNow: false, answer: { at: 5_000, interactionId: "q1", code: q1Code(NONCE), httpStatus: 200, error: null }, statuses: { q1: "answered" }, + runs: [{ id: "run-1", status: "succeeded", errorCode: null, createdAt: new Date(500).toISOString() }, + { id: "run-2", status: "succeeded", errorCode: null, createdAt: new Date(6_000).toISOString() }], ...overrides }); + const artifacts = (overrides: Partial = {}): OracleArtifacts => ({ issueStatus: "done", + documents: { brief: `${q1AnswerMarker(NONCE, q1Code(NONCE))}\n- a\n- b\n- c\n` }, agentComments: [], files: {}, commands: {}, workspacePath: null, + question: question(overrides) }); + + it("passes a saved question that settled once and was answered", () => { + expect(gradeQuestionHandoff(NONCE, artifacts())).toMatchObject({ verdict: "pass" }); + }); + + it("fails a failed attempt, a repeated question or work before the answer", () => { + const failed = gradeQuestionHandoff(NONCE, artifacts({ runs: [{ id: "run-1", status: "failed", errorCode: "governed_wait_provider_terminal_unproven", createdAt: null }] })); + expect(failed.verdict).toBe("fail"); + expect(failed.checks.filter(check => !check.ok).map(check => check.name)).toEqual(expect.arrayContaining(["asking_run_settled", "no_failed_attempt"])); + expect(gradeQuestionHandoff(NONCE, artifacts({ interactionIds: ["q1", "q2"] })).checks.find(check => check.name === "one_question_asked")?.ok).toBe(false); + const between = gradeQuestionHandoff(NONCE, artifacts({ runs: [...question().runs, { id: "run-3", status: "succeeded", errorCode: null, createdAt: new Date(3_000).toISOString() }] })); + expect(between.checks.find(check => check.name === "nothing_ran_before_answer")?.ok).toBe(false); + }); + + it("measures the wait from when the question was saved, and counts a queued run that starts during it", () => { + const check = (overrides: Partial) => gradeQuestionHandoff(NONCE, artifacts(overrides)).checks + .find(candidate => candidate.name === "nothing_ran_before_answer")?.ok; + // Created after the save but before the first poll saw the question. + expect(check({ savedAt: 800, runs: [...question().runs, { id: "run-3", status: "succeeded", errorCode: null, createdAt: new Date(900).toISOString() }] })).toBe(false); + // Queued before the save, started while the question waited. + expect(check({ savedAt: 800, runs: [...question().runs, { id: "run-3", status: "succeeded", errorCode: null, + createdAt: new Date(100).toISOString(), startedAt: new Date(2_000).toISOString() }] })).toBe(false); + expect(check({ savedAt: 800 })).toBe(true); + }); + + it("records when the question was saved", async () => { + const task = record(); + const { api } = fakeApi([{ ...pendingQuestion, createdAt: new Date(400).toISOString() }]); + 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 }); + }); +}); diff --git a/tests/restart-chaos/question-handoff.ts b/tests/restart-chaos/question-handoff.ts new file mode 100644 index 00000000000..fbb93a9ac1d --- /dev/null +++ b/tests/restart-chaos/question-handoff.ts @@ -0,0 +1,163 @@ +/** + * Q1: the agent saves a question for the board (a governed wait), the runner + * interrupts the provider turn, the turn settles, and the board answers. With a + * provider message fault (provider-faults.ts) the interruption's terminal arrives + * late (PV1) or never (PV2); the saved question must survive and settle once, + * without a failed attempt, a Retry or repeated work. + * + * The board's side uses the same routes as the board UI: it lists the task's + * interactions and answers the pending question once the run that asked it has + * settled (no run of the task is queued or running) with the server ready. + */ +import type { ChaosApi } from "./api.js"; +import type { OracleArtifacts, OracleResult, TaskRecord, TaskSample } from "./types.js"; +import type { WorkloadDefinition } from "./workloads.js"; + +type Json = Record; + +/** What the observer saw of the task's question and what the board did. */ +export interface QuestionRecord { + /** Every ask_user_questions interaction on the task (a second one is repeated work). */ + interactionIds: string[]; + /** First poll that saw the question pending. */ + seenAt: number | null; + /** When the question was saved (the interaction's createdAt), else the first poll that saw it. */ + savedAt?: number | null; + /** The run that saved the question. */ + askedByRunId: string | null; + /** The question is pending at the latest poll. */ + pendingNow: boolean; + answer?: { at: number; interactionId: string; code: string; httpStatus: number | null; error: string | null }; +} + +export interface QuestionArtifacts extends QuestionRecord { + runs: Array<{ id: string; status: string; errorCode: string | null; createdAt: string | null; startedAt?: string | null }>; + statuses: Record; +} + +const ACTIVE = new Set(["queued", "running", "scheduled_retry"]); +export const Q1_CODES = ["ALPHA", "BRAVO"] as const; + +/** The board's answer for this task, fixed by its nonce. */ +export function q1Code(nonce: string): (typeof Q1_CODES)[number] { + return Number.parseInt(nonce.slice(-1), 16) % 2 === 0 ? "ALPHA" : "BRAVO"; +} +export const q1Question = (nonce: string) => `Which codename should the Q1-${nonce} brief use?`; +export const q1AnswerMarker = (nonce: string, code: string) => `ANSWER-${nonce}-${code}`; + +const description = (nonce: string) => [ + "This task needs one answer from the board before you write anything.", + `1. Ask exactly one question with your \`request_human_input\` tool: a questions card (continuationPolicy \`wake_assignee\`,`, + ` idempotencyKey \`q1-${nonce}\`) with one required single-select question "${q1Question(nonce)}"`, + ` and exactly two options labelled \`${Q1_CODES[0]}\` and \`${Q1_CODES[1]}\`.`, + "2. Then end your turn and wait for the answer. Do not guess, do not ask again, and do not create documents, files or tasks yet.", + "3. When you are woken with the answer, save this issue's document with key `brief` (title \"Brief\"), in markdown.", + ` It must contain the line \`${q1AnswerMarker(nonce, "")}\` exactly, on its own line, with replaced by the chosen option,`, + " and at least three bullet points starting with \"- \" on why saved questions must survive a provider that stops answering.", + "4. When the document is saved, mark this task done.", +].join("\n"); + +function questionInteractions(rows: Json[] | null | undefined): Json[] { + return (rows ?? []).filter(row => row.kind === "ask_user_questions"); +} + +/** Q1 board step, called on every poll. */ +export async function questionStep(input: { api: Pick; issueId: string; nonce: string; record: TaskRecord; + sample: TaskSample; now: number; boardMayAct: boolean; note: (kind: string, data: Record) => void }): Promise { + const question = input.record.question ??= { interactionIds: [], seenAt: null, askedByRunId: null, pendingNow: false }; + const rows = await input.api.get(`/api/issues/${input.issueId}/interactions`, 5_000).catch(() => null); + if (rows === null) return; + const interactions = questionInteractions(rows); + for (const interaction of interactions) if (!question.interactionIds.includes(String(interaction.id))) question.interactionIds.push(String(interaction.id)); + const pending = interactions.find(interaction => interaction.status === "pending"); + question.pendingNow = Boolean(pending) && !question.answer; + if (!pending || question.answer) return; + if (question.seenAt === null) { + question.seenAt = input.now; + const created = Date.parse(String(pending.createdAt ?? "")); + question.savedAt = Number.isFinite(created) ? Math.min(created, input.now) : input.now; + question.askedByRunId = typeof pending.sourceRunId === "string" ? pending.sourceRunId + : (input.sample.runs ?? []).find(run => run.status === "running")?.id ?? null; + input.note("question_pending", { interactionId: pending.id, askedByRunId: question.askedByRunId }); + } + // 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; + const code = q1Code(input.nonce); + const questions = (pending.payload?.questions ?? []) as Json[]; + const target = questions[0]; + const option = ((target?.options ?? []) as Json[]).find(candidate => String(candidate.label ?? "").toUpperCase().includes(code)); + const answer: NonNullable = { at: input.now, interactionId: String(pending.id), code, httpStatus: null, error: null }; + try { + if (!target) throw new Error("the pending question card has no questions"); + await input.api.post(`/api/issues/${input.issueId}/interactions/${pending.id}/respond`, { + answers: [{ questionId: String(target.id), optionIds: option ? [String(option.id)] : [], ...(option ? {} : { otherText: code }) }], + }); + answer.httpStatus = 200; + } catch (error) { + answer.httpStatus = typeof (error as { status?: unknown }).status === "number" ? (error as { status: number }).status : null; + answer.error = error instanceof Error ? error.message : String(error); + } + question.answer = answer; + question.pendingNow = false; + input.note("question_answered", { ...answer }); +} + +/** The task waits on the board while its question is pending and unanswered. */ +export function awaitingQuestionAnswer(record: TaskRecord): boolean { + return Boolean(record.question?.pendingNow) && !record.question?.answer; +} + +export async function collectQuestionArtifacts(api: Pick, issueId: string, record: TaskRecord): Promise { + 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 }; + 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 })), + }; +} + +export function gradeQuestionHandoff(nonce: string, artifacts: OracleArtifacts): OracleResult { + const question = artifacts.question; + const code = q1Code(nonce); + const brief = artifacts.documents.brief ?? ""; + const bullets = brief.split(/\r?\n/).filter(line => /^\s*[-*] \S/.test(line)).length; + const runs = question?.runs ?? []; + const failed = runs.filter(run => run.status === "failed" || run.status === "timed_out"); + const asked = question?.askedByRunId ? runs.find(run => run.id === question.askedByRunId) : undefined; + const answeredAt = question?.answer?.at ?? null; + // Repeated work: another run that started (or was created) after the question was saved and + // before the board answered. A run queued earlier that starts during the wait counts too. + const savedAt = question?.savedAt ?? question?.seenAt ?? null; + const between = savedAt !== null && answeredAt !== null ? runs.filter(run => run.id !== question?.askedByRunId + && [run.startedAt, run.createdAt].some(at => at && Date.parse(at) > savedAt && Date.parse(at) < answeredAt)) : []; + const checks = [ + { name: "issue_done", ok: artifacts.issueStatus === "done", detail: `status=${artifacts.issueStatus ?? "unknown"}` }, + { name: "one_question_asked", ok: question?.interactionIds.length === 1, detail: `questions=${question?.interactionIds.length ?? 0}` }, + { name: "question_answered_by_board", ok: question?.answer?.httpStatus === 200, + detail: question?.answer ? `${question.answer.code} http=${question.answer.httpStatus}${question.answer.error ? ` ${question.answer.error}` : ""}` : "never pending" }, + { name: "asking_run_settled", ok: asked?.status === "succeeded", + 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" }, + { 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}` }, + ]; + return { verdict: checks.every(check => check.ok) ? "pass" : "fail", checks }; +} + +export const questionHandoffWorkload: WorkloadDefinition = { + id: "Q1", + label: "Saved question settles and is answered", + usesRepository: false, + // Provider-fault tasks fire no restart; with --faults K* the restart lands around the question. + fireWindowMs: [20_000, 60_000], + budgetMs: 15 * 60_000, + title: nonce => `Chaos Q1 question ${nonce}`, + description, + collect: { documents: ["brief"], files: () => [], commands: [] }, + grade: gradeQuestionHandoff, +}; diff --git a/tests/restart-chaos/scheduler.ts b/tests/restart-chaos/scheduler.ts index fce5c82f9cc..d9acb1fef9a 100644 --- a/tests/restart-chaos/scheduler.ts +++ b/tests/restart-chaos/scheduler.ts @@ -1,3 +1,4 @@ +import { isProviderFault, withProviderFaultRules } from "./provider-faults.js"; import { workloadDefinitions } from "./workloads.js"; import type { AdapterId, FaultKind, ScheduledFault, StormFault, StormPlan, TaskFaultKind, TaskPlan, TriggerId, WavePlan, WaveOptions, WorkloadId } from "./types.js"; import type { ReplayRef } from "./config.js"; @@ -67,8 +68,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; +function stormCellFaults(options: Pick, trigger: TriggerId): readonly TaskFaultKind[] { + if (!options.stormMinutes) return [...options.faults, ...(options.providerFaults ?? [])]; 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)); @@ -84,7 +85,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) { @@ -136,7 +137,7 @@ function replayMasks(source: readonly string[] | undefined, added: readonly stri export function planWave(waveId: string, options: WaveOptions, createdAt = new Date()): WavePlan { return { schema: "paperclip.restart-chaos.wave-plan/v1", waveId, seed: options.seed, - createdAt: createdAt.toISOString(), options, tasks: planTasks(options), + createdAt: createdAt.toISOString(), options, tasks: withProviderFaultRules(planTasks(options), options), ...(options.stormMinutes ? { storm: planStorm(options.seed, options.stormMinutes, options.faults, options.k4HoldMaxMs) } : {}) }; } @@ -273,7 +274,8 @@ export function observedFaultSchedule(createdAt: number | null, faultIds: readon /** The faults a task schedules: its replayed sequence, or its one planned fault. */ export function taskFaultSchedule(task: TaskPlan): ScheduledFault[] { if (task.replayFaults?.length) return task.replayFaults; - if (task.fault === "K13") return []; + // A storm crosses K13 tasks; a provider fault (PV*) restarts nothing. + if (task.fault === "K13" || isProviderFault(task.fault)) return []; const planned: ScheduledFault = { kind: task.fault, holdMs: task.holdMs, atMs: task.fireAfterMs }; const trigger = task.trigger ?? "time"; if (trigger === "time") return [planned]; diff --git a/tests/restart-chaos/types.ts b/tests/restart-chaos/types.ts index bf6ba389093..5e5def6b5c1 100644 --- a/tests/restart-chaos/types.ts +++ b/tests/restart-chaos/types.ts @@ -1,13 +1,18 @@ /** Shared types for the control-plane restart chaos harness (plan §5). */ +import type { ProviderFaultEvent, ProviderFaultKind, ProviderFaultRule, ProviderMessage } from "./provider-faults.js"; +import type { QuestionArtifacts, QuestionRecord } from "./question-handoff.js"; /** K14 (upgrade mid-run): a graceful stop, then the replacement starts from the current build (`--upgrade-from`). */ export const FAULT_KINDS = ["K1", "K2", "K4", "K6", "K14"] as const; export type FaultKind = (typeof FAULT_KINDS)[number]; -/** A task's planned fault. "K13" (restart storm) tasks have none of their own; the wave's storm crosses them. */ -export type TaskFaultKind = FaultKind | "K13"; +/** + * 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. + */ +export type TaskFaultKind = FaultKind | "K13" | ProviderFaultKind; /** G1: an approval-gated action across a restart. H1: a budget hard stop crossed mid-run with a restart (its own wave). */ -export const WORKLOAD_IDS = ["A1", "B1", "B5", "C1", "M1", "G1", "H1", "R1", "S1"] as const; +export const WORKLOAD_IDS = ["A1", "B1", "B5", "C1", "M1", "G1", "H1", "R1", "S1", "Q1"] as const; export type WorkloadId = (typeof WORKLOAD_IDS)[number]; export const ADAPTER_IDS = ["codex", "claude", "opencode"] as const; @@ -80,6 +85,11 @@ export interface WaveOptions { storageWindows?: import("./storage-chaos.js").StorageFaultWindow[]; workspaceCommit?: string; workspaceGitUrl?: string; + /** Provider message faults from `--faults` (PV1 delay, PV2 drop, PV3 duplicate); see provider-faults.ts. */ + providerFaults?: ProviderFaultKind[]; + providerFaultMessage?: ProviderMessage; + /** PV1 delays, cycled across tasks (`--provider-fault-delays`). */ + providerFaultDelaysMs?: number[]; } export interface StormFault { @@ -115,6 +125,8 @@ export interface TaskPlan { * its offset from the task's launch. Replaces the single planned fault. */ replayFaults?: ScheduledFault[]; + /** PV1–PV3 tasks: the provider message rule armed for this task's nonce. */ + providerFault?: ProviderFaultRule; } export interface ScheduledFault { @@ -240,6 +252,10 @@ export interface OracleArtifacts { approvals?: Array<{ id: string; status: string; title: string | null }>; /** G1 and H1: the board's actions and what the observer saw around them. */ governance?: GovernanceRecord; + /** Q1: the saved question and the task's runs. */ + question?: QuestionArtifacts; + /** PV1–PV3: the shim's log lines for this task. */ + providerFaultEvents?: ProviderFaultEvent[]; } /** What the board did for a governance task (G1, H1), and what the observer saw. */ @@ -323,6 +339,8 @@ export interface TaskRecord { governance?: GovernanceRecord; /** K14 waves: the build serving when the task launched. */ serverBuildAtLaunch?: "upgrade_from" | "current"; + /** Q1: the saved question and the board's answer. */ + question?: QuestionRecord; } export interface TaskGrade { diff --git a/tests/restart-chaos/workloads.ts b/tests/restart-chaos/workloads.ts index 64c376aefd1..fd8b60a7605 100644 --- a/tests/restart-chaos/workloads.ts +++ b/tests/restart-chaos/workloads.ts @@ -1,3 +1,4 @@ +import { questionHandoffWorkload } from "./question-handoff.js"; import type { OracleArtifacts, OracleResult, WorkloadId } from "./types.js"; /** Deterministic bench output: sum of squares 1..240. */ @@ -166,6 +167,7 @@ function budgetChecks(artifacts: OracleArtifacts) { } export const workloadDefinitions: Record = { + Q1: questionHandoffWorkload, G1: { id: "G1", label: "Approval-gated action across a restart",