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",