diff --git a/tests/restart-chaos/README.md b/tests/restart-chaos/README.md index 160e8e23ef8..394f83f292c 100644 --- a/tests/restart-chaos/README.md +++ b/tests/restart-chaos/README.md @@ -481,6 +481,12 @@ node cli/node_modules/tsx/dist/cli.mjs tests/restart-chaos/main.ts --env local \ # 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 same for Claude (ACPX): its interrupted prompt's response is held 18 s. +node cli/node_modules/tsx/dist/cli.mjs tests/restart-chaos/main.ts --env local \ + --adapters claude --workloads Q1 --faults PV1 --concurrency 1 --tasks 1 --seed 104 +# A dropped terminal straddles a control-plane kill -9: K1 fires 3 s after the drop. +node cli/node_modules/tsx/dist/cli.mjs tests/restart-chaos/main.ts --env local \ + --adapters codex --workloads Q1 --faults PV2 --provider-fault-restart K1 --concurrency 1 --tasks 1 --seed 105 ``` The third fault family: the provider stays alive, but one protocol message between the runner and the @@ -492,13 +498,41 @@ Codex app-server is late, missing or repeated. `--faults` takes them next to res | 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. +`--provider-fault-message terminal` (default) picks the turn's end after an interruption, which a +governed wait (a saved question or approval) or a board Stop sends. `interrupt_ack` picks the +acknowledgement of that interruption. Each rule applies once per task, to the first matching message. +Provider faults run only with `--env local --adapters codex,claude` (either or both) and only with +workloads that interrupt a turn (Q1, S1, G1). 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. +For Claude, the runner settles a cancelled turn about 3–4 s after the cancel without waiting for the +provider's prompt response, and it later closes the sidecar with the provider. A Claude PV1 delay +longer than that is therefore a drop (wave w-20261110T040202-s202: 18 s held, never released, the task +still passed its Q1 oracle). Sweep Claude PV1 below that window, for example +`--provider-fault-delays 1,3`. + +G1 grades the board decision after a restart, so G1 with a provider fault needs +`--provider-fault-restart K1`. On a board Stop (S1) the runner ends Codex without waiting for its terminal: +in wave w-20261110T050407-s206 Codex exited 1.5 s after `turn/interrupt` with no `turn/completed`, so a +`terminal` rule never applies there and the task fails `provider_fault_applied`. Use `interrupt_ack` for S1. + +| Adapter | `terminal` | `interrupt_ack` | +| --- | --- | --- | +| Codex | the `turn/completed` after a `turn/interrupt` | the response to that `turn/interrupt` | +| Claude (ACPX) | the provider's ACP `session/prompt` response after a `session/cancel` | the sidecar's response to the runner's `turn.cancel` (ACP's `session/cancel` is a notification with no acknowledgement) | + +**Across a restart.** A PV task fires no restart unless `--provider-fault-restart K1` is given. Then each +PV1 or PV2 task also kills -9 the control plane `--provider-fault-restart-after-seconds` (default 3) +after its fault applied, through the `provider` trigger. That delay must be shorter than every PV1 delay, +so the message is still held, and the wave runs with `--concurrency 1`, because every K1 kills the whole control plane. K1 kills the runner with its provider and the held message, so the late +or missing terminal straddles the restart: the replacement control plane must settle the saved wait +without it. Planned restart faults (`--faults K1,PV2`) are still rejected, and PV3 does not combine with +a restart, since a duplicate holds nothing. Such a task passes when: + +- `provider_fault_applied`: the fault applied once, the K1 started after it, and for PV1 no release + came before the K1; +- the ordinary grade passes: R1–R6 for a restart-crossing task and the workload's T-oracle. For Q1 + that means one question, the asking run `succeeded`, no failed run, nothing started before the + answer, and the brief carries the answer. **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 @@ -513,9 +547,21 @@ into a worktree at that commit. Every shim decision is logged in `/provide 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. +**Claude seam** (`acpx-provider-faults.ts`, `provider-shim/acpx-sidecar-hook.mjs`, +`provider-shim/acpx-fault-tap.cjs`). Claude runs through the ACPX sidecar. Its provider package and +runtime are digest-verified and run from private snapshots with a scrubbed environment, so the Codex +approach of replacing the executable cannot work. The sidecar itself is the build-owned +`dist/cli/acpx-runtime-sidecar.cjs`, which product code hashes and the runner verifies at launch. The +harness builds a stand-in runner package under `/node_modules/.restart-chaos-/`. It is all +symlinks to the real package, except that `dist/cli/acpx-runtime-sidecar.cjs` is a generated stand-in. +It lives under the workspace `node_modules`, so the provider package root that product code derives from +it is the same workspace root. A server hook loads only `runnerd-codex-transport` (`src/live/*.ts` from source, `dist/live/*.js` when built), the +module that computes the local sidecar path from its own URL, from the stand-in package. That module's +imports still resolve to the real package. The runner then verifies and launches the stand-in. The +stand-in installs the tap and runs the real sidecar bundle in the same process, as the runner's +verified CommonJS loader does. The tap wraps the stdio of the provider child that product code spawns +and admits, and the sidecar's own frames to the runner. Product code is unchanged. The stand-in +package is removed at the end of the wave. **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 @@ -754,6 +800,7 @@ re-arms the task's own triggered faults instead of using their offsets. | `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. | +| `acpx-provider-faults.ts`, `provider-shim/acpx-*` | PV1–PV3 for Claude (ACPX): the stand-in runner package, the server hook, the sidecar/provider tap, and `--provider-fault-restart`. | | `question-handoff.ts` | Q1 saved question: the board's answer step and its oracle. | | `time-faults.ts`, `time-faults-config.ts` | TM1–TM3: process freezing, database aging with recorded rows, and the time-corrected API the grader reads through. | | `daytona-sweep.ts` | Daytona R6: read-only reconciliation of this wave's labelled sandboxes with leases; cleanup only through Paperclip. | diff --git a/tests/restart-chaos/acpx-provider-faults.test.ts b/tests/restart-chaos/acpx-provider-faults.test.ts new file mode 100644 index 00000000000..ae27be62b10 --- /dev/null +++ b/tests/restart-chaos/acpx-provider-faults.test.ts @@ -0,0 +1,265 @@ +import { spawn, spawnSync } from "node:child_process"; +import { mkdir, mkdtemp, readFile, readlink, rm, writeFile } from "node:fs/promises"; +import { createRequire } from "node:module"; +import os from "node:os"; +import path from "node:path"; +import { pathToFileURL } from "node:url"; +import { afterEach, describe, expect, it } from "vitest"; +import { ACPX_FAULT_TAP, ACPX_SIDECAR_HOOK, installAcpxSidecarSeam, parseProviderFaultRestart, standInSidecarSource, withAcpxSidecarHook } from "./acpx-provider-faults.js"; +import { parseCommand } from "./config.js"; +import { readProviderFaultEvents, withProviderFaultCheck, type ProviderFaultEvent, type ProviderFaultRule } from "./provider-faults.js"; +import { planWave, taskFaultSchedule } from "./scheduler.js"; +import type { TaskPlan } from "./types.js"; + +const tap = createRequire(import.meta.url)(ACPX_FAULT_TAP) as { + PROVIDER_BOOTSTRAP_MARKER: string; + createAcpTap(runtime: unknown): { toProvider(line: string): void; fromProvider(line: string, emit: (line: string) => void): void; ended(): void }; + createSidecarTap(runtime: unknown): { fromRunner(line: string): void; toRunner(line: string, emit: (line: string) => void): void; ended(): void }; + createFaultRuntime(config: { controlPath: string; logPath: string; stateDirectory: string }): unknown; +}; + +const roots: string[] = []; +afterEach(async () => { for (const root of roots.splice(0)) await rm(root, { recursive: true, force: true }); }); + +async function temporary(prefix: string): Promise { + const root = await mkdtemp(path.join(os.tmpdir(), prefix)); + roots.push(root); + return root; +} + +/** Rules, log and markers in a temp root, as the Codex seam lays them out. */ +async function faultFiles(rules: Array) { + const root = await temporary("chaos-acpx-tap-"); + const files = { controlPath: path.join(root, "control.json"), logPath: path.join(root, "provider-faults.jsonl"), stateDirectory: path.join(root, "state") }; + await mkdir(files.stateDirectory); + await writeFile(files.controlPath, JSON.stringify({ rules })); + await writeFile(files.logPath, ""); + return { root, files }; +} + +const line = (value: unknown) => JSON.stringify(value); +const prompt = (id: number, sessionId: string, text: string) => line({ jsonrpc: "2.0", id, method: "session/prompt", params: { sessionId, prompt: [{ type: "text", text }] } }); +const cancel = (sessionId: string) => line({ jsonrpc: "2.0", method: "session/cancel", params: { sessionId } }); +const promptResponse = (id: number, stopReason: string) => line({ jsonrpc: "2.0", id, result: { stopReason } }); + +describe("ACPX provider fault tap (ACP pipe)", () => { + it("relays everything unchanged when no rule names the session's task", async () => { + const { files } = await faultFiles([{ nonce: "other123", kind: "PV2", message: "terminal" }]); + const acp = tap.createAcpTap(tap.createFaultRuntime(files)); + const out: string[] = []; + acp.toProvider(prompt(1, "s1", "task aaaa1111")); + acp.toProvider(cancel("s1")); + acp.fromProvider(promptResponse(1, "cancelled"), value => out.push(value)); + expect(out).toEqual([promptResponse(1, "cancelled")]); + }); + + it("drops (PV2) only the prompt response that follows a cancel, once per task", async () => { + const { files } = await faultFiles([{ nonce: "aaaa1111", kind: "PV2", message: "terminal" }]); + const acp = tap.createAcpTap(tap.createFaultRuntime(files)); + const out: string[] = []; + const update = line({ jsonrpc: "2.0", method: "session/update", params: { sessionId: "s1", update: { sessionUpdate: "agent_message_chunk" } } }); + acp.toProvider(prompt(1, "s1", "Chaos Q1 aaaa1111")); + acp.fromProvider(update, value => out.push(value)); + acp.fromProvider(promptResponse(1, "end_turn"), value => out.push(value)); + // Not after a cancel: not the target. + expect(out).toEqual([update, promptResponse(1, "end_turn")]); + acp.toProvider(prompt(2, "s1", "continue")); + acp.toProvider(cancel("s1")); + acp.fromProvider(promptResponse(2, "cancelled"), value => out.push(value)); + expect(out).toHaveLength(2); + acp.toProvider(prompt(3, "s1", "again")); + acp.toProvider(cancel("s1")); + acp.fromProvider(promptResponse(3, "cancelled"), value => out.push(value)); + expect(out.at(-1)).toBe(promptResponse(3, "cancelled")); + const events = await readProviderFaultEvents(files.logPath, "aaaa1111"); + expect(events.filter(event => event.event === "applied")).toEqual([expect.objectContaining({ kind: "PV2", message: "terminal", action: "drop", turnStatus: "cancelled", layer: "acp" })]); + expect(events.some(event => event.event === "interrupt_requested")).toBe(true); + }); + + it("duplicates (PV3) and delays (PV1) the terminal; a provider exit discards a held one", async () => { + const { files } = await faultFiles([{ nonce: "bbbb2222", kind: "PV3", message: "terminal" }, { nonce: "cccc3333", kind: "PV1", message: "terminal", delayMs: 30 }, + { nonce: "dddd4444", kind: "PV1", message: "terminal", delayMs: 60_000 }]); + const runtime = tap.createFaultRuntime(files); + const duplicated: string[] = []; + const first = tap.createAcpTap(runtime); + first.toProvider(prompt(7, "s7", "bbbb2222")); + first.toProvider(cancel("s7")); + first.fromProvider(promptResponse(7, "cancelled"), value => duplicated.push(value)); + expect(duplicated).toEqual([promptResponse(7, "cancelled"), promptResponse(7, "cancelled")]); + + const delayed: string[] = []; + const second = tap.createAcpTap(runtime); + second.toProvider(prompt(1, "s8", "cccc3333")); + second.toProvider(cancel("s8")); + second.fromProvider(promptResponse(1, "cancelled"), value => delayed.push(value)); + expect(delayed).toEqual([]); + await new Promise(resolve => setTimeout(resolve, 80)); + expect(delayed).toEqual([promptResponse(1, "cancelled")]); + + const third = tap.createAcpTap(runtime); + third.toProvider(prompt(1, "s9", "dddd4444")); + third.toProvider(cancel("s9")); + third.fromProvider(promptResponse(1, "cancelled"), () => { throw new Error("must stay held"); }); + third.ended(); + const events = await readProviderFaultEvents(files.logPath); + expect(events.filter(event => event.nonce === "cccc3333").map(event => event.event)).toEqual(expect.arrayContaining(["applied", "released"])); + expect(events.filter(event => event.nonce === "dddd4444").map(event => event.event)).toEqual(expect.arrayContaining(["applied", "held_message_discarded"])); + }); +}); + +describe("ACPX provider fault tap (runner pipe)", () => { + it("targets the sidecar's turn.cancel acknowledgement, never events", async () => { + const { files } = await faultFiles([{ nonce: "eeee5555", kind: "PV2", message: "interrupt_ack" }]); + const sidecar = tap.createSidecarTap(tap.createFaultRuntime(files)); + const out: string[] = []; + sidecar.fromRunner(line({ protocolVersion: 1, id: 4, command: "turn.start", params: { turnId: "t1", input: "eeee5555" } })); + sidecar.toRunner(line({ protocolVersion: 1, id: 4, ok: true, result: { turnId: "t1" } }), value => out.push(value)); + sidecar.fromRunner(line({ protocolVersion: 1, id: 5, command: "turn.cancel", params: { turnId: "t1" } })); + const event = line({ protocolVersion: 1, sequence: 9, eventType: "turn.completed", runId: "r", turnId: "t1", payload: {} }); + sidecar.toRunner(event, value => out.push(value)); + sidecar.toRunner(line({ protocolVersion: 1, id: 5, ok: true, result: { cancelled: true, sessionClosed: false } }), value => out.push(value)); + expect(out).toEqual([line({ protocolVersion: 1, id: 4, ok: true, result: { turnId: "t1" } }), event]); + expect(await readProviderFaultEvents(files.logPath, "eeee5555")).toEqual(expect.arrayContaining([ + expect.objectContaining({ event: "applied", kind: "PV2", message: "interrupt_ack", turnStatus: "cancelled", layer: "sidecar" })])); + }); +}); + +/** A stand-in for the real ACPX sidecar bundle (runner frames in, ACP to a marked "provider"). */ +const FAKE_SIDECAR = path.join(import.meta.dirname, "fixtures", "fake-acpx-sidecar.cjs"); + +async function driveStandIn(rules: Array) { + const { root, files } = await faultFiles(rules); + const realSidecar = FAKE_SIDECAR; + const standIn = path.join(root, "stand-in.cjs"); + await writeFile(standIn, standInSidecarSource({ tapModule: ACPX_FAULT_TAP, realSidecar, ...files })); + const child = spawn(process.execPath, [standIn], { stdio: ["pipe", "pipe", "inherit"] }); + const frames: Array> = []; + let buffer = ""; + child.stdout.on("data", chunk => { + buffer += chunk; + let index; + while ((index = buffer.indexOf("\n")) >= 0) { frames.push(JSON.parse(buffer.slice(0, index))); buffer = buffer.slice(index + 1); } + }); + const send = (value: unknown) => child.stdin.write(`${JSON.stringify(value)}\n`); + const until = async (predicate: () => boolean, ms = 3_000) => { + const end = Date.now() + ms; + while (!predicate()) { if (Date.now() > end) return false; await new Promise(resolve => setTimeout(resolve, 10)); } + return true; + }; + return { child, frames, send, until, files }; +} + +describe("ACPX stand-in sidecar", () => { + it("loads the real sidecar with the tap: a dropped provider terminal never reaches the runner", async () => { + const { child, frames, send, until, files } = await driveStandIn([{ nonce: "ffff6666", kind: "PV2", message: "terminal" }]); + send({ protocolVersion: 1, id: 1, command: "turn.start", params: { turnId: "t1", input: "Chaos Q1 ffff6666" } }); + expect(await until(() => frames.some(frame => frame.id === 1))).toBe(true); + send({ protocolVersion: 1, id: 2, command: "turn.cancel", params: { turnId: "t1" } }); + expect(await until(() => frames.some(frame => frame.id === 2))).toBe(true); + await new Promise(resolve => setTimeout(resolve, 200)); + expect(frames.some(frame => frame.eventType === "turn.completed")).toBe(false); + send({ protocolVersion: 1, id: 3, command: "session.close", params: {} }); + child.stdin.end(); + await new Promise(resolve => child.once("exit", resolve)); + const events = await readProviderFaultEvents(files.logPath, "ffff6666"); + expect(events).toEqual(expect.arrayContaining([expect.objectContaining({ event: "applied", layer: "acp", action: "drop", turnStatus: "cancelled" })])); + }); + + it("without a rule the stand-in relays the turn end; with an ack rule it delays only the ack", async () => { + const { child, frames, send, until, files } = await driveStandIn([{ nonce: "abab7777", kind: "PV1", message: "interrupt_ack", delayMs: 300 }]); + send({ protocolVersion: 1, id: 1, command: "turn.start", params: { turnId: "t1", input: "Chaos S1 abab7777" } }); + expect(await until(() => frames.some(frame => frame.id === 1))).toBe(true); + send({ protocolVersion: 1, id: 2, command: "turn.cancel", params: { turnId: "t1" } }); + expect(await until(() => frames.some(frame => frame.eventType === "turn.completed"))).toBe(true); + expect(frames.some(frame => frame.id === 2)).toBe(false); + expect(await until(() => frames.some(frame => frame.id === 2))).toBe(true); + send({ protocolVersion: 1, id: 3, command: "session.close", params: {} }); + child.stdin.end(); + await new Promise(resolve => child.once("exit", resolve)); + const events = await readProviderFaultEvents(files.logPath, "abab7777"); + expect(events.map(event => event.event)).toEqual(expect.arrayContaining(["applied", "released"])); + }); +}); + +describe("ACPX sidecar hook", () => { + it("loads only the sidecar-path module from the stand-in package; its imports stay real", async () => { + const root = await temporary("chaos-acpx-hook-"); + const runner = path.join(root, "packages", "paperclip-runner"); + await mkdir(path.join(runner, "dist", "live"), { recursive: true }); + await mkdir(path.join(runner, "dist", "cli"), { recursive: true }); + await mkdir(path.join(root, "node_modules")); + await writeFile(path.join(runner, "package.json"), JSON.stringify({ name: "@paperclipai/paperclip-runner", type: "module" })); + await writeFile(path.join(runner, "dist", "live", "helper.js"), "export const helperUrl = import.meta.url;\n"); + await writeFile(path.join(runner, "dist", "live", "runnerd-codex-transport.js"), [ + `import { helperUrl } from "./helper.js";`, + `import { fileURLToPath } from "node:url";`, + `export const sidecar = fileURLToPath(new URL("../cli/acpx-runtime-sidecar.cjs", import.meta.url));`, + `export { helperUrl };`, ""].join("\n")); + await writeFile(path.join(runner, "dist", "cli", "acpx-runtime-sidecar.cjs"), "// real\n"); + await writeFile(path.join(runner, "dist", "cli", "codex.js"), "// real\n"); + const seam = await installAcpxSidecarSeam({ repositoryRoot: root, waveId: "w-test-s1", controlPath: "/c", logPath: "/l", stateDirectory: "/s" }); + expect(seam.standInRoot).toBe(path.join(await import("node:fs/promises").then(fs => fs.realpath(root)), "node_modules", ".restart-chaos-w-test-s1", "paperclip-runner")); + expect(await readlink(path.join(seam.standInRoot, "package.json"))).toBe(path.join(seam.realSidecar, "../../..", "package.json")); + expect(await readlink(path.join(seam.standInRoot, "dist", "cli", "codex.js"))).toContain(path.join("packages", "paperclip-runner", "dist", "cli", "codex.js")); + const realTransport = path.join(seam.serverEnvironment.RESTART_CHAOS_ACPX_RUNNER_REAL!, "dist", "live", "runnerd-codex-transport.js"); + const script = `const m = await import(${JSON.stringify(pathToFileURL(realTransport).href)}); console.log(JSON.stringify({ sidecar: m.sidecar, helperUrl: m.helperUrl, nodeOptions: process.env.NODE_OPTIONS ?? null }));`; + const result = spawnSync(process.execPath, ["--input-type=module", "--eval", script], { encoding: "utf8", + env: { ...process.env, ...seam.serverEnvironment, NODE_OPTIONS: withAcpxSidecarHook("--no-warnings") } }); + expect(result.status, result.stderr).toBe(0); + const output = JSON.parse(result.stdout) as { sidecar: string; helperUrl: string; nodeOptions: string | null }; + expect(output.sidecar).toBe(path.join(seam.standInRoot, "dist", "cli", "acpx-runtime-sidecar.cjs")); + expect(output.helperUrl).toBe(pathToFileURL(path.join(path.dirname(realTransport), "helper.js")).href); + expect(output.nodeOptions).toBe("--no-warnings"); + const standIn = await readFile(output.sidecar, "utf8"); + expect(standIn).toContain(ACPX_FAULT_TAP); + expect(standIn).toContain(seam.realSidecar); + expect(withAcpxSidecarHook(undefined)).toBe(`--import=${ACPX_SIDECAR_HOOK}`); + await seam.remove(); + }); +}); + +function task(nonce: string, rule: ProviderFaultRule): TaskPlan { + return { taskId: "t01", index: 0, workload: "Q1", adapter: "claude", fault: rule.kind, fireAfterMs: 0, holdMs: 0, followUpAfterMs: null, nonce, providerFault: rule }; +} + +describe("provider faults across a restart", () => { + it("plans Claude PV tasks and a K1 armed by the provider trigger", () => { + const command = parseCommand(["--adapters", "claude", "--workloads", "Q1", "--faults", "PV1,PV2", "--provider-fault-restart", "K1", + "--provider-fault-restart-after-seconds", "4", "--tasks", "2", "--concurrency", "1", "--seed", "5"]); + if (command.mode !== "wave") throw new Error("expected a wave"); + expect(command.options).toMatchObject({ adapters: ["claude"], providerFaultRestart: "K1", providerFaultRestartAfterMs: 4_000, faults: [] }); + const plan = planWave("w-20261109T000000-s5", command.options); + for (const planned of plan.tasks) { + expect(planned.providerFault?.restart).toEqual({ kind: "K1", afterAppliedMs: 4_000 }); + expect(taskFaultSchedule(planned)).toEqual([{ kind: "K1", holdMs: 0, atMs: 0, trigger: "provider", delayMs: 4_000 }]); + } + }); + + it("rejects restarts that cannot straddle the fault", () => { + expect(() => parseCommand(["--adapters", "claude", "--workloads", "Q1", "--faults", "PV3", "--provider-fault-restart", "K1"])).toThrow(/PV1 or PV2/); + expect(() => parseCommand(["--adapters", "codex", "--workloads", "Q1", "--faults", "PV1", "--provider-fault-delays", "5", + "--provider-fault-restart", "K1", "--provider-fault-restart-after-seconds", "6"])).toThrow(/shorter than every PV1 delay/); + expect(() => parseCommand(["--adapters", "codex", "--workloads", "Q1", "--faults", "PV2", "--provider-fault-restart", "K2"])).toThrow(/K1/); + expect(() => parseCommand(["--adapters", "codex", "--workloads", "Q1", "--faults", "K1", "--provider-fault-restart", "K1"])).toThrow(/provider fault/); + expect(() => parseCommand(["--adapters", "codex", "--workloads", "G1", "--faults", "PV2"])).toThrow(/G1 grades the board decision after a restart/); + expect(parseCommand(["--adapters", "codex", "--workloads", "G1", "--faults", "PV2", "--provider-fault-restart", "K1", "--concurrency", "1"]).mode).toBe("wave"); + expect(() => parseCommand(["--adapters", "codex", "--workloads", "Q1", "--faults", "PV2", "--provider-fault-restart", "K1", "--concurrency", "2"])).toThrow(/--concurrency 1/); + expect(parseProviderFaultRestart(undefined, undefined, ["PV2"], undefined)).toEqual({}); + expect(() => parseProviderFaultRestart(undefined, "3", ["PV2"], undefined)).toThrow(/needs --provider-fault-restart/); + }); + + it("passes only when the restart came after the fault and before a PV1 release", () => { + const pass = { verdict: "pass" as const, checks: [] }; + const rule: ProviderFaultRule = { kind: "PV1", message: "terminal", delayMs: 18_000, restart: { kind: "K1", afterAppliedMs: 3_000 } }; + const applied: ProviderFaultEvent = { at: 1_000, pid: 1, event: "applied", nonce: "aa", kind: "PV1", message: "terminal", action: "delay" }; + expect(withProviderFaultCheck(pass, task("aa", rule), [applied], 4_000).verdict).toBe("pass"); + expect(withProviderFaultCheck(pass, task("aa", rule), [applied], null).checks.at(-1)).toMatchObject({ ok: false, detail: expect.stringContaining("restart=none") }); + expect(withProviderFaultCheck(pass, task("aa", rule), [applied, { ...applied, event: "released", at: 2_000 }], 4_000).checks.at(-1)) + .toMatchObject({ ok: false, detail: expect.stringContaining("releasedBeforeRestart") }); + expect(withProviderFaultCheck(pass, task("aa", rule), [applied], 500).verdict).toBe("fail"); + // A provider that closed first took the held message with it before the K1: nothing straddled the restart. + expect(withProviderFaultCheck(pass, task("aa", rule), [applied, { ...applied, event: "held_message_discarded", at: 3_000 }], 4_000).checks.at(-1)) + .toMatchObject({ ok: false, detail: expect.stringContaining("discardedBeforeRestart") }); + expect(withProviderFaultCheck(pass, task("aa", rule), [applied, { ...applied, event: "held_message_discarded", at: 5_000 }], 4_000).verdict).toBe("pass"); + }); +}); diff --git a/tests/restart-chaos/acpx-provider-faults.ts b/tests/restart-chaos/acpx-provider-faults.ts new file mode 100644 index 00000000000..c004cd85812 --- /dev/null +++ b/tests/restart-chaos/acpx-provider-faults.ts @@ -0,0 +1,141 @@ +/** + * Provider message faults (PV1–PV3) for ACPX agents (Claude). The Codex seam + * (provider-faults.ts) replaces the Codex executable; that does not work for + * ACPX, whose provider package, its runtime and the sidecar are digest-verified + * and run from private snapshots with a scrubbed environment. + * + * Seam, no product change: + * 1. The harness builds a stand-in of the runner package under + * `/node_modules/.restart-chaos-/paperclip-runner`: symlinks to + * the real package, except `dist/cli/acpx-runtime-sidecar.cjs`, which is a + * generated stand-in sidecar. It sits under the workspace `node_modules` so + * that the provider package root product code derives from it is the same + * workspace root it derives from the real package. + * 2. `provider-shim/acpx-sidecar-hook.mjs`, loaded into the server through + * NODE_OPTIONS, loads the one module that computes the local sidecar path + * (`runnerd-codex-transport`, from `src/live` or `dist/live`) from the stand-in package. Product + * code then hashes the stand-in, and the runner verifies and launches it + * exactly as it would the real sidecar. + * 3. The stand-in installs `provider-shim/acpx-fault-tap.cjs` and then loads the + * real sidecar bundle in the same process, the way the runner's verified + * CommonJS loader does. The tap relays the sidecar's ACP pipe to the + * verified provider (still launched and admitted by product code) and the + * sidecar's frames to the runner, changing only the message a task's rule names. + * + * Rules, the log and the once-per-task markers are shared with the Codex seam. + */ +import { chmod, mkdir, readdir, realpath, rm, symlink, writeFile } from "node:fs/promises"; +import { existsSync } from "node:fs"; +import path from "node:path"; +import { DEFAULT_PROVIDER_DELAY_MS, DEFAULT_PROVIDER_FAULT_RESTART_AFTER_MS, type ProviderFaultKind } from "./provider-faults.js"; +import type { AdapterId, WaveOptions } from "./types.js"; + +export const ACPX_SIDECAR_HOOK = path.join(import.meta.dirname, "provider-shim", "acpx-sidecar-hook.mjs"); +export const ACPX_FAULT_TAP = path.join(import.meta.dirname, "provider-shim", "acpx-fault-tap.cjs"); + +/** Adapters whose provider messages the harness can fault: Codex through its shim, Claude through the ACPX tap. */ +export const PROVIDER_FAULT_ADAPTERS: readonly AdapterId[] = ["codex", "claude"]; + +export interface AcpxSidecarSeam { + /** Merge into the server environment before it starts (kept across restarts). */ + serverEnvironment: Record; + standInRoot: string; + realSidecar: string; + /** Remove the stand-in package (after the wave). */ + remove(): Promise; +} + +/** The stand-in sidecar: install the tap, then run the real bundle as the runner's verified loader would. */ +export function standInSidecarSource(config: { tapModule: string; realSidecar: string; controlPath: string; logPath: string; stateDirectory: string }): string { + return `#!/usr/bin/env node +// Restart chaos stand-in ACPX sidecar (generated by tests/restart-chaos/acpx-provider-faults.ts). +"use strict"; +const config = ${JSON.stringify(config)}; +require(config.tapModule).installAcpxFaultTap(config); +const fs = require("node:fs"); +const Module = require("node:module"); +const source = fs.readFileSync(config.realSidecar, "utf8").replace(/^#![^\\r\\n]*(?:\\r?\\n|$)/, ""); +const sidecar = new Module(config.realSidecar); +sidecar.filename = config.realSidecar; +sidecar.paths = []; +sidecar._compile(source, config.realSidecar); +`; +} + +async function mirror(source: string, target: string, replace: Record): Promise { + await mkdir(target, { recursive: true, mode: 0o755 }); + for (const entry of await readdir(source)) { + if (entry in replace) continue; + await symlink(path.join(source, entry), path.join(target, entry)); + } +} + +/** + * Build the stand-in runner package for `waveId` and return the server environment that loads the + * sidecar-path module from it. + */ +export async function installAcpxSidecarSeam(input: { repositoryRoot: string; waveId: string; controlPath: string; logPath: string; + stateDirectory: string }): Promise { + const realRunner = await realpath(path.join(input.repositoryRoot, "packages", "paperclip-runner")); + const realSidecar = path.join(realRunner, "dist", "cli", "acpx-runtime-sidecar.cjs"); + for (const required of [realSidecar]) { + if (!existsSync(required)) throw new Error(`ACPX provider faults need the built runner (${path.relative(input.repositoryRoot, required)} is missing); run node scripts/agent-preflight.mjs`); + } + if (!/^[A-Za-z0-9._-]+$/.test(input.waveId)) throw new Error(`Unexpected wave id ${input.waveId}`); + const container = path.join(await realpath(path.join(input.repositoryRoot, "node_modules")), `.restart-chaos-${input.waveId}`); + const standInRoot = path.join(container, "paperclip-runner"); + await rm(container, { recursive: true, force: true }); + // package.json and node_modules resolve to the real package; dist/cli, dist/live and src/live are + // real directories so the stand-in sidecar can replace one file and the hook can load one module + // (from src/ when the server runs from source, from dist/ when it runs built). + await mirror(realRunner, standInRoot, { dist: null, src: null }); + await mirror(path.join(realRunner, "dist"), path.join(standInRoot, "dist"), { cli: null, live: null }); + await mirror(path.join(realRunner, "dist", "live"), path.join(standInRoot, "dist", "live"), {}); + if (existsSync(path.join(realRunner, "src", "live"))) { + await mirror(path.join(realRunner, "src"), path.join(standInRoot, "src"), { live: null }); + await mirror(path.join(realRunner, "src", "live"), path.join(standInRoot, "src", "live"), {}); + } + await mirror(path.join(realRunner, "dist", "cli"), path.join(standInRoot, "dist", "cli"), { "acpx-runtime-sidecar.cjs": null }); + const standInSidecar = path.join(standInRoot, "dist", "cli", "acpx-runtime-sidecar.cjs"); + await writeFile(standInSidecar, standInSidecarSource({ tapModule: ACPX_FAULT_TAP, realSidecar, + controlPath: input.controlPath, logPath: input.logPath, stateDirectory: input.stateDirectory }), { mode: 0o755 }); + await chmod(standInSidecar, 0o755); + return { + serverEnvironment: { RESTART_CHAOS_ACPX_RUNNER_REAL: realRunner, RESTART_CHAOS_ACPX_RUNNER_SHADOW: standInRoot }, + standInRoot, + realSidecar, + remove: () => rm(container, { recursive: true, force: true }), + }; +} + +/** NODE_OPTIONS for the server: what it already has plus the ACPX hook. */ +export function withAcpxSidecarHook(nodeOptions: string | undefined): string { + return [nodeOptions, `--import=${ACPX_SIDECAR_HOOK}`].filter(Boolean).join(" "); +} + +/** + * `--provider-fault-restart K1 [--provider-fault-restart-after-seconds n]`: each PV1/PV2 task also + * kills -9 the control plane n seconds (default 3) after its provider fault applied. Only K1, and + * not PV3: a duplicate has nothing in flight for a restart to cross. + */ +export function parseProviderFaultRestart(raw: string | undefined, afterRaw: string | undefined, kinds: readonly ProviderFaultKind[], + delaysMs: readonly number[] | undefined): Pick { + if (raw === undefined) { + if (afterRaw !== undefined) throw new Error("--provider-fault-restart-after-seconds needs --provider-fault-restart K1"); + return {}; + } + if (raw !== "K1") throw new Error("--provider-fault-restart supports K1 (kill -9 the control plane) only"); + if (kinds.includes("PV3")) throw new Error("--provider-fault-restart combines with PV1 or PV2; a PV3 duplicate holds nothing across a restart"); + let afterMs = DEFAULT_PROVIDER_FAULT_RESTART_AFTER_MS; + if (afterRaw !== undefined) { + if (!/^[0-9]+(\.[0-9]+)?$/.test(afterRaw)) throw new Error("--provider-fault-restart-after-seconds must be a number of seconds"); + afterMs = Math.round(Number(afterRaw) * 1000); + if (afterMs > 300_000) throw new Error("--provider-fault-restart-after-seconds must be at most 300"); + } + // PV1: the restart must come while the message is still held, or it straddles nothing. + const shortest = kinds.includes("PV1") ? Math.min(...(delaysMs?.length ? delaysMs : [DEFAULT_PROVIDER_DELAY_MS])) : null; + if (shortest !== null && afterMs >= shortest) { + throw new Error(`--provider-fault-restart-after-seconds (${afterMs / 1000}s) must be shorter than every PV1 delay (shortest ${shortest / 1000}s)`); + } + return { providerFaultRestart: "K1", providerFaultRestartAfterMs: afterMs }; +} diff --git a/tests/restart-chaos/config.ts b/tests/restart-chaos/config.ts index 806bed199c3..2a77eb86271 100644 --- a/tests/restart-chaos/config.ts +++ b/tests/restart-chaos/config.ts @@ -6,6 +6,7 @@ import { takesJourneyTriggers } from "./journeys.js"; import { STORAGE_FAULT_KINDS, type StorageFaultKind } from "./storage-chaos.js"; import { parseTimeFaultOptions } from "./time-faults-config.js"; import { PROVIDER_FAULT_KINDS, PROVIDER_FAULT_WORKLOADS, isProviderFault, parseProviderDelays, parseProviderMessage, type ProviderFaultKind } from "./provider-faults.js"; +import { PROVIDER_FAULT_ADAPTERS, parseProviderFaultRestart } from "./acpx-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, @@ -113,13 +114,15 @@ export function parseCommand(argv: readonly string[], now = Date.now()): ParsedC matrix: { type: "boolean" }, "server-from": { type: "string" }, "time-age-hours": { type: "string" }, + "provider-fault-restart": { type: "string" }, + "provider-fault-restart-after-seconds": { 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", "provider-fault-message", "provider-fault-delays", "db-pressure", "matrix", "server-from", "time-age-hours"] 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", "db-pressure", "matrix", "server-from", "time-age-hours", "provider-fault-restart", "provider-fault-restart-after-seconds"] 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, @@ -194,20 +197,27 @@ export function parseCommand(argv: readonly string[], now = Date.now()): ParsedC 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"); + if (env !== "local") throw new Error("Provider message faults (PV*) shim the local provider; run them with --env local"); + if (adapters.some(adapter => !PROVIDER_FAULT_ADAPTERS.includes(adapter))) throw new Error(`Provider message faults (PV*) cover ${PROVIDER_FAULT_ADAPTERS.join(" and ")}; remove the other adapters from --adapters`); + // Every PV task gets its own restart through --provider-fault-restart; a planned K cell would be a different task. + if (options.faults.length) throw new Error("Provider message faults (PV*) do not share a wave with planned restart faults (K*); use --provider-fault-restart K1 to restart after each provider fault"); // 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`); + if (idle.length) throw new Error(`Provider message faults (PV*) need workloads that interrupt a provider 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"); + Object.assign(options, parseProviderFaultRestart(values["provider-fault-restart"], values["provider-fault-restart-after-seconds"], providerFaults, delays)); + // Each K1 kills the whole control plane: a second task in flight would lose its held message to the + // first task's restart and miss its own, so a restart wave runs one task at a time. + if (options.providerFaultRestart && options.concurrency !== 1) throw new Error("--provider-fault-restart kills the whole control plane per task; run it with --concurrency 1"); + // G1's oracle requires the board's decision to come after a restart (board_decided_after_restart). + if (options.workloads.includes("G1") && !options.providerFaultRestart) throw new Error("G1 grades the board decision after a restart; with provider faults add --provider-fault-restart K1"); + } else if (values["provider-fault-message"] !== undefined || values["provider-fault-delays"] !== undefined + || values["provider-fault-restart"] !== undefined || values["provider-fault-restart-after-seconds"] !== undefined) { + throw new Error("--provider-fault-message, --provider-fault-delays and --provider-fault-restart need a provider fault (PV1, PV2 or PV3) in --faults"); } Object.assign(options, parseTimeFaultOptions(options, values["time-age-hours"])); const upgradeFrom = values["upgrade-from"]; diff --git a/tests/restart-chaos/fixtures/fake-acpx-sidecar.cjs b/tests/restart-chaos/fixtures/fake-acpx-sidecar.cjs new file mode 100644 index 00000000000..e312b1e953f --- /dev/null +++ b/tests/restart-chaos/fixtures/fake-acpx-sidecar.cjs @@ -0,0 +1,50 @@ +// Test fixture for acpx-provider-faults.test.ts: a stand-in for the real ACPX sidecar bundle. It +// speaks the runner frame protocol on stdio, spawns a "provider" whose arguments carry the verified +// bootstrap marker, and speaks ACP to it: the provider answers the active prompt after a cancel. +"use strict"; +const { spawn } = require("node:child_process"); +const readline = require("node:readline"); + +const PROVIDER = ` +let buffer = ""; +let active = null; +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 === "session/prompt") active = message.id; + if (message.method === "session/cancel" && active !== null) setTimeout(() => send({ jsonrpc: "2.0", id: active, result: { stopReason: "cancelled" } }), 20); + } +}); +process.stdin.on("end", () => process.exit(0)); +`; +const MARKER = 'throw new Error("ACPX provider requires verified package snapshots")'; + +let child = null; +let turnId = null; +const frame = value => process.stdout.write(`${JSON.stringify(value)}\n`); +const requests = readline.createInterface({ input: process.stdin }); +requests.on("line", text => { + const request = JSON.parse(text); + if (request.command === "turn.start") { + turnId = request.params.turnId; + child = spawn(process.execPath, ["--eval", PROVIDER, MARKER], { stdio: ["pipe", "pipe", "inherit"] }); + readline.createInterface({ input: child.stdout }).on("line", reply => { + frame({ protocolVersion: 1, sequence: 1, eventType: "turn.completed", runId: "r", turnId, payload: JSON.parse(reply).result }); + }); + child.stdin.write(`${JSON.stringify({ jsonrpc: "2.0", id: 1, method: "session/prompt", + params: { sessionId: "s1", prompt: [{ type: "text", text: request.params.input }] } })}\n`); + frame({ protocolVersion: 1, id: request.id, ok: true, result: { turnId } }); + } + if (request.command === "turn.cancel") { + child.stdin.write(`${JSON.stringify({ jsonrpc: "2.0", method: "session/cancel", params: { sessionId: "s1" } })}\n`); + frame({ protocolVersion: 1, id: request.id, ok: true, result: { cancelled: true } }); + } + if (request.command === "session.close") { + child?.stdin.end(); + requests.close(); + } +}); diff --git a/tests/restart-chaos/main.ts b/tests/restart-chaos/main.ts index 906a157d9bc..f67193a1454 100644 --- a/tests/restart-chaos/main.ts +++ b/tests/restart-chaos/main.ts @@ -19,6 +19,7 @@ import { StorageChaos, planStorageFaults, replayStorageWindows, type StorageFaul import { startStorageFaultProxy } from "./storage-proxy.js"; import { needsCredentialRotation, startCredentialRotation, type CredentialRotation } from "./storage-credentials.js"; import { describeProviderFault, installProviderFaultSeam, resolveRealCodex, withProviderFaultCheck, type ProviderFaultSeam } from "./provider-faults.js"; +import { installAcpxSidecarSeam, withAcpxSidecarHook, type AcpxSidecarSeam } from "./acpx-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"; @@ -251,6 +252,7 @@ async function main(): Promise { let storageProxy: Awaited> | null = null; let credentialRotation: CredentialRotation | null = null; let providerFaults: ProviderFaultSeam | null = null; + let acpxSidecarSeam: AcpxSidecarSeam | null = null; let speedSection: string | null = null; let harnessError: unknown; process.on("uncaughtException", error => { @@ -292,12 +294,22 @@ async function main(): Promise { } 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); + // The provider the server actually runs: `--server-from` serves another checkout. + const realCodex = resolveRealCodex(plan.options.serverFrom ?? 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); + if (plan.tasks.some(task => task.providerFault && task.adapter === "claude")) { + // Claude (ACPX): the stand-in sidecar taps the verified provider's pipe (acpx-provider-faults.ts). + // The seam must mirror the runner the server actually loads (`--server-from` runs another checkout). + acpxSidecarSeam = await installAcpxSidecarSeam({ repositoryRoot: plan.options.serverFrom ?? repositoryRoot, waveId: plan.waveId, controlPath: providerFaults.controlPath, + logPath: providerFaults.logPath, stateDirectory: providerFaults.stateDirectory }); + Object.assign(serverEnvironment, acpxSidecarSeam.serverEnvironment, + { NODE_OPTIONS: withAcpxSidecarHook(serverEnvironment.NODE_OPTIONS) }); + timeline.add("acpx_provider_faults_installed", { standInRoot: acpxSidecarSeam.standInRoot, realSidecar: acpxSidecarSeam.realSidecar }); + } timeline.add("provider_faults_installed", { realCodex: realCodex.command, codexVersion: realCodex.version, rules: plan.tasks.flatMap(task => task.providerFault ? [`${task.taskId}:${describeProviderFault(task.providerFault)}`] : []) }); } @@ -452,6 +464,16 @@ async function main(): Promise { if (plan.options.storageWindows && storageChaos && launched.length === 1) storageChaos.start(); if (!next.done) active.push(next); } + // `--provider-fault-restart`: a task's K1 is armed once the shim logged its fault. + if (providerFaults) { + for (const observer of active) { + if (!observer.plan.providerFault?.restart || observer.record.providerFaultAppliedAt !== undefined) continue; + const rule = observer.plan.providerFault; + const applied = (await providerFaults.eventsFor(observer.plan.nonce)) + .find(event => event.event === "applied" && event.kind === rule.kind && event.message === rule.message); + if (applied) observer.record.providerFaultAppliedAt = applied.at; + } + } await Promise.all(active.filter(observer => now - (lastPoll.get(observer) ?? 0) >= plan.options.pollMs).map(async observer => { lastPoll.set(observer, now); await observer.poll(now); @@ -550,7 +572,8 @@ async function main(): Promise { 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 restart = faults.find(receipt => receipt.requestedBy === observer.plan.taskId && receipt.trigger === "provider"); + const oracle = withProviderFaultCheck(graded, observer.plan, artifacts?.providerFaultEvents ?? [], restart?.startedAt ?? null); const grade = gradeTask(observer.record, faults, observer.definition.budgetMs, oracle, universalByIssue ? { runMessages: allRunMessages, recoveryActions: universalByIssue.get(observer.record.issueId ?? "")?.recoveryActions ?? [] } : undefined); let evidence: string | null = null; @@ -665,6 +688,7 @@ async function main(): Promise { await storageProxy.close().catch(() => undefined); } await objectStore?.stop().catch(error => timeline.add("object_store_stop_failed", { error: String(error) })); + await acpxSidecarSeam?.remove().catch(error => timeline.add("acpx_stand_in_cleanup_failed", { error: String(error) })); const retention = await releaseInstance(supervisor.paths.temporaryRoot, { keep, stopFailed, remoteCleanupPending, owner: plan.waveId }) .catch(error => { timeline.add("instance_cleanup_failed", { error: String(error) }); return null; }); if (retention?.kept) timeline.add("instance_kept", { temporaryRoot: supervisor.paths.temporaryRoot, reason: retention.reason }); diff --git a/tests/restart-chaos/observer.ts b/tests/restart-chaos/observer.ts index b395697c16b..c76446bf4ea 100644 --- a/tests/restart-chaos/observer.ts +++ b/tests/restart-chaos/observer.ts @@ -363,6 +363,7 @@ export class TaskObserver { return queued && !started ? first.at : "missed"; } case "tool": return this.record.benchFirstSeenAt ?? null; + case "provider": return this.record.providerFaultAppliedAt ?? null; case "approval": return this.record.governance?.approvalSeenAt ?? null; case "paused": return this.record.governance?.budgetPausedAt ?? null; // Q1 keeps its own record (question-handoff.ts); Q2 and Q3 use the journey driver. diff --git a/tests/restart-chaos/provider-faults.test.ts b/tests/restart-chaos/provider-faults.test.ts index fa884da243b..c0b04c469f5 100644 --- a/tests/restart-chaos/provider-faults.test.ts +++ b/tests/restart-chaos/provider-faults.test.ts @@ -187,8 +187,8 @@ describe("provider fault planning", () => { 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", "--faults", "PV1"])).toThrow(/interrupt a provider turn/); + expect(() => parseCommand(["--adapters", "codex,opencode", "--workloads", "Q1", "--faults", "PV1"])).toThrow(/cover codex and claude/); 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/); diff --git a/tests/restart-chaos/provider-faults.ts b/tests/restart-chaos/provider-faults.ts index 1d362fa0624..17b0f210527 100644 --- a/tests/restart-chaos/provider-faults.ts +++ b/tests/restart-chaos/provider-faults.ts @@ -12,8 +12,8 @@ * 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). + * Claude runs through the ACPX sidecar with a hash-verified provider pack: its + * seam is acpx-provider-faults.ts, which shares this file's rules, log and check. */ import { chmod, copyFile, mkdir, readFile, rename, symlink, writeFile } from "node:fs/promises"; import { readFileSync, realpathSync } from "node:fs"; @@ -39,8 +39,16 @@ export interface ProviderFaultRule { message: ProviderMessage; /** PV1 only. */ delayMs?: number; + /** + * `--provider-fault-restart K1` (PV1, PV2): kill -9 the control plane this long after the fault + * applied, so the held or dropped message straddles a restart. + */ + restart?: { kind: "K1"; afterAppliedMs: number }; } +/** Default gap between a provider fault applying and its K1: well inside the default 18 s PV1 hold. */ +export const DEFAULT_PROVIDER_FAULT_RESTART_AFTER_MS = 3_000; + /** One shim log line (JSONL, `provider-faults.jsonl` in the wave directory). */ export interface ProviderFaultEvent { at: number; @@ -85,19 +93,23 @@ export function parseProviderMessage(raw: string | undefined): ProviderMessage | } /** Attach each provider-fault task's rule. Restart-fault tasks are unchanged. */ -export function withProviderFaultRules(tasks: TaskPlan[], options: Pick): TaskPlan[] { +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]! } : {}) }; + ...(task.fault === "PV1" ? { delayMs: delays[pv1++ % delays.length]! } : {}), + ...(options.providerFaultRestart && task.fault !== "PV3" + ? { restart: { kind: options.providerFaultRestart, afterAppliedMs: options.providerFaultRestartAfterMs ?? DEFAULT_PROVIDER_FAULT_RESTART_AFTER_MS } } : {}) }; 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` : ""}`; + return `${rule.kind} ${PROVIDER_FAULT_ACTIONS[rule.kind]} ${rule.message}${rule.delayMs !== undefined ? ` ${rule.delayMs}ms` : ""}` + + (rule.restart ? ` +${rule.restart.kind}@${rule.restart.afterAppliedMs}ms` : ""); } /** The Codex executable product code would resolve for this checkout (mirrors codex-command.ts). */ @@ -127,6 +139,8 @@ export interface ProviderFaultSeam { serverEnvironment: Record; controlPath: string; logPath: string; + /** Once-per-task markers, shared with the ACPX tap. */ + stateDirectory: string; realCommand: string; /** Arm the rules of these tasks (idempotent; replaces the armed set). */ arm(tasks: readonly TaskPlan[]): Promise; @@ -176,7 +190,7 @@ export async function installProviderFaultSeam(input: { root: string; logPath: s 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, + controlPath, logPath: input.logPath, stateDirectory, realCommand: input.realCommand, async arm(tasks) { const rules = tasks.flatMap(task => task.providerFault ? [{ nonce: task.nonce, ...task.providerFault }] : []); const temporary = `${controlPath}.${process.pid}.tmp`; @@ -196,8 +210,14 @@ export async function readProviderFaultEvents(logPath: string, nonce?: string): /** * 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. + * + * With `--provider-fault-restart K1` the fault must also straddle the restart: the K1 started after + * the fault applied and, for PV1, before the held message was released or discarded. The K1 kills the runner + * with the held message, so a PV1 task then needs no release. Whether the task itself recovered + * is the ordinary grade (R1–R6 and its T-oracle), which this check never replaces. */ -export function withProviderFaultCheck(oracle: OracleResult, plan: TaskPlan, events: readonly ProviderFaultEvent[]): OracleResult { +export function withProviderFaultCheck(oracle: OracleResult, plan: TaskPlan, events: readonly ProviderFaultEvent[], + restartStartedAt: number | null = null): 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); @@ -205,9 +225,23 @@ export function withProviderFaultCheck(oracle: OracleResult, plan: TaskPlan, eve // 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); + let exercised: boolean; + let restartDetail = ""; + if (rule.restart) { + const appliedAt = applied[0]?.at ?? null; + // Released or discarded before the K1: the message was already gone, so nothing straddled the restart. + const releasedBeforeRestart = restartStartedAt !== null && released.some(event => event.at < restartStartedAt); + const discardedBeforeRestart = restartStartedAt !== null && discarded.some(event => event.at < restartStartedAt); + exercised = applied.length === 1 && appliedAt !== null && restartStartedAt !== null && restartStartedAt >= appliedAt + && !releasedBeforeRestart && !discardedBeforeRestart; + restartDetail = restartStartedAt === null ? " restart=none" + : ` restartAfterAppliedMs=${appliedAt === null ? "n/a" : restartStartedAt - appliedAt}` + + (releasedBeforeRestart ? " releasedBeforeRestart" : "") + (discardedBeforeRestart ? " discardedBeforeRestart" : ""); + } else { + 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}` : "") }]; + + (applied[0]?.turnStatus ? ` turnStatus=${applied[0].turnStatus}` : "") + restartDetail }]; return { verdict: checks.every(check => check.ok) ? "pass" : "fail", checks }; } diff --git a/tests/restart-chaos/provider-shim/acpx-fault-tap.cjs b/tests/restart-chaos/provider-shim/acpx-fault-tap.cjs new file mode 100644 index 00000000000..508c3522dad --- /dev/null +++ b/tests/restart-chaos/provider-shim/acpx-fault-tap.cjs @@ -0,0 +1,268 @@ +// Restart chaos provider-message faults for ACPX agents (Claude): PV1 delay, PV2 drop, PV3 duplicate. +// +// Loaded by the harness's stand-in ACPX sidecar (see ../acpx-provider-faults.ts) inside the real +// sidecar process, before the real sidecar bundle runs. It taps two JSON-lines streams and changes +// only the one message an armed rule names: +// - "terminal": the provider's ACP `session/prompt` response that follows a `session/cancel` for +// the same session (the turn's end after a governed wait or a board Stop). This is a real +// provider message on the sidecar <-> provider pipe. +// - "interrupt_ack": ACP has no acknowledgement for `session/cancel` (a notification), so this is +// the sidecar's response to the runner's `turn.cancel` request: the acknowledgement the runner +// waits for. +// The provider process itself is not changed: it is still launched and admitted by product code. +// Rules, the log and the once-per-task markers are the same files the Codex shim uses. +// Plain Node built-ins only: this runs in the sidecar process, not in the harness. +"use strict"; +const { appendFileSync, readFileSync, statSync, writeFileSync } = require("node:fs"); +const path = require("node:path"); +const { PassThrough } = require("node:stream"); +const { StringDecoder } = require("node:string_decoder"); + +/** The marker only the verified ACPX provider bootstrap carries (installation-integrity.ts snapshotBootstrap). */ +const PROVIDER_BOOTSTRAP_MARKER = "ACPX provider requires verified package snapshots"; + +function createFaultRuntime(config, now = Date.now) { + let rules = []; + let rulesMtimeMs = -1; + return { + log(entry) { + try { appendFileSync(config.logPath, `${JSON.stringify({ at: now(), pid: process.pid, agent: "acpx", ...entry })}\n`); } catch { /* best effort */ } + }, + 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; + }, + /** Every rule applies at most once per task, across sidecar and provider restarts. */ + 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; } + }, + }; +} + +function parse(line) { + if (!line.startsWith("{")) return null; + try { return JSON.parse(line); } catch { return null; } +} + +function findNonce(runtime, line) { + for (const rule of runtime.armedRules()) if (typeof rule.nonce === "string" && rule.nonce && line.includes(rule.nonce)) return rule.nonce; + return null; +} + +function matchingRule(runtime, nonce, message, afterInterrupt) { + if (!nonce) return null; + return runtime.armedRules().find(rule => rule.nonce === nonce && rule.message === message + && (afterInterrupt || rule.afterInterrupt === false)) ?? null; +} + +/** Apply PV1/PV2/PV3 to one line. Returns true when the rule took the line. */ +function createApplier(runtime, layer) { + const pending = new Map(); + return { + apply(rule, target, line, emit, requestId) { + if (!rule || !runtime.claim(rule)) return false; + const base = { event: "applied", layer, nonce: rule.nonce, kind: rule.kind, message: target.message, + turnStatus: target.turnStatus ?? null, requestId: requestId ?? null }; + if (rule.kind === "PV2") { runtime.log({ ...base, action: "drop" }); return true; } + if (rule.kind === "PV3") { emit(line); emit(line); runtime.log({ ...base, action: "duplicate" }); return true; } + const delayMs = Number(rule.delayMs ?? 0); + runtime.log({ ...base, action: "delay", delayMs }); + const timer = setTimeout(() => { + pending.delete(timer); + emit(line); + runtime.log({ event: "released", layer, nonce: rule.nonce, kind: rule.kind, message: target.message, delayMs }); + }, delayMs); + pending.set(timer, { layer, nonce: rule.nonce, kind: rule.kind, message: target.message, delayMs }); + return true; + }, + /** The stream ended: a held message goes with it, as it would with a dead process. */ + discard() { + for (const [timer, held] of pending) { + clearTimeout(timer); + runtime.log({ event: "held_message_discarded", ...held }); + } + pending.clear(); + }, + }; +} + +/** + * The sidecar <-> provider ACP pipe. `toProvider` sees every line the sidecar sends (prompt, cancel); + * `fromProvider` decides each line the provider sends. + */ +function createAcpTap(runtime) { + const applier = createApplier(runtime, "acp"); + const nonceBySession = new Map(); + const prompts = new Map(); + const cancelledSessions = new Set(); + let boundNonce = null; + const bind = (line, sessionId) => { + const nonce = findNonce(runtime, line); + if (!nonce) return; + if (typeof sessionId === "string") nonceBySession.set(sessionId, nonce); + if (boundNonce !== nonce) { boundNonce = nonce; runtime.log({ event: "bound", layer: "acp", nonce }); } + }; + const nonceFor = sessionId => (typeof sessionId === "string" && nonceBySession.get(sessionId)) || boundNonce; + return { + toProvider(line) { + const message = parse(line); + const sessionId = message?.params?.sessionId; + bind(line, sessionId); + if (message?.method === "session/prompt" && message.id !== undefined) { + // A new prompt: only a cancel of this prompt makes its response a target. + if (typeof sessionId === "string") cancelledSessions.delete(sessionId); + prompts.set(JSON.stringify(message.id), typeof sessionId === "string" ? sessionId : null); + } + if (message?.method === "session/cancel") { + if (typeof sessionId === "string") cancelledSessions.add(sessionId); + runtime.log({ event: "interrupt_requested", layer: "acp", nonce: nonceFor(sessionId), sessionId: sessionId ?? null }); + } + }, + fromProvider(line, emit) { + const message = parse(line); + if (message) bind(line, message.params?.sessionId); + const key = message && message.method === undefined && message.id !== undefined ? JSON.stringify(message.id) : null; + if (key === null || !prompts.has(key)) { emit(line); return; } + const sessionId = prompts.get(key); + prompts.delete(key); + const afterInterrupt = sessionId === null ? cancelledSessions.size > 0 : cancelledSessions.has(sessionId); + const turnStatus = message.result?.stopReason ?? (message.error ? "error" : null); + const rule = matchingRule(runtime, nonceFor(sessionId), "terminal", afterInterrupt); + if (!applier.apply(rule, { message: "terminal", turnStatus }, line, emit, message.id)) emit(line); + }, + ended() { applier.discard(); }, + }; +} + +/** + * The runner <-> sidecar pipe. `fromRunner` sees each request frame; `toRunner` decides each frame + * the sidecar writes. + */ +function createSidecarTap(runtime) { + const applier = createApplier(runtime, "sidecar"); + const cancelRequests = new Set(); + let boundNonce = null; + return { + fromRunner(line) { + const nonce = findNonce(runtime, line); + if (nonce && nonce !== boundNonce) { boundNonce = nonce; runtime.log({ event: "bound", layer: "sidecar", nonce }); } + const message = parse(line); + if (message?.command === "turn.cancel" && message.id !== undefined) { + cancelRequests.add(JSON.stringify(message.id)); + runtime.log({ event: "interrupt_requested", layer: "sidecar", nonce: boundNonce, requestId: message.id }); + } + }, + toRunner(line, emit) { + const message = parse(line); + const key = message && message.command === undefined && message.eventType === undefined && message.id !== undefined + ? JSON.stringify(message.id) : null; + if (key === null || !cancelRequests.has(key)) { emit(line); return; } + cancelRequests.delete(key); + const rule = matchingRule(runtime, boundNonce, "interrupt_ack", true); + const turnStatus = message.ok === true ? (message.result?.cancelled === true ? "cancelled" : "acknowledged") : "error"; + if (!applier.apply(rule, { message: "interrupt_ack", turnStatus }, line, emit, message.id)) emit(line); + }, + ended() { applier.discard(); }, + }; +} + +/** Split a byte or text stream into lines; a decoder keeps a multibyte character that spans two chunks intact. */ +function lineSplitter(onLine) { + const decoder = new StringDecoder("utf8"); + let buffer = ""; + return { + push(chunk) { + buffer += typeof chunk === "string" ? chunk : decoder.write(chunk); + let newline; + while ((newline = buffer.indexOf("\n")) >= 0) { + const line = buffer.slice(0, newline); + buffer = buffer.slice(newline + 1); + onLine(line); + } + }, + /** Whatever is left without a newline. */ + rest() { const value = buffer + decoder.end(); buffer = ""; return value; }, + }; +} + +function isVerifiedProviderLaunch(args) { + return Array.isArray(args) && args.includes("--eval") && args.some(arg => typeof arg === "string" && arg.includes(PROVIDER_BOOTSTRAP_MARKER)); +} + +/** Put the ACP tap between the sidecar and one provider child. The child's process is untouched. */ +function wrapProviderChild(child, tap, runtime) { + if (!child.stdout || !child.stdin) return; + const realOutput = child.stdout; + const output = new PassThrough(); + child.stdout = output; + if (Array.isArray(child.stdio)) child.stdio[1] = output; + const outgoing = lineSplitter(line => tap.fromProvider(line, value => { if (!output.destroyed && !output.writableEnded) output.write(`${value}\n`); })); + realOutput.on("data", chunk => outgoing.push(chunk)); + realOutput.once("end", () => { + const rest = outgoing.rest(); + tap.ended(); + if (rest && !output.destroyed) output.write(rest); + output.end(); + }); + realOutput.once("error", error => { tap.ended(); output.destroy(error); }); + // The sidecar closing its side closes the real pipe too. + output.once("close", () => { if (!realOutput.destroyed) realOutput.destroy(); }); + const incoming = lineSplitter(line => tap.toProvider(line)); + const write = child.stdin.write; + child.stdin.write = function tappedWrite(chunk, ...rest) { + try { if (typeof chunk === "string" || chunk instanceof Uint8Array) incoming.push(chunk); } catch { /* observing never blocks a write */ } + return write.call(this, chunk, ...rest); + }; + runtime.log({ event: "spawned", layer: "acp", childPid: child.pid ?? null }); +} + +/** Install both taps in this (sidecar) process, before the real sidecar runs. */ +function installAcpxFaultTap(config) { + const runtime = createFaultRuntime(config); + const childProcess = require("node:child_process"); + const spawn = childProcess.spawn; + childProcess.spawn = function tappedSpawn(command, args, ...rest) { + const child = spawn.call(this, command, args, ...rest); + if (isVerifiedProviderLaunch(args)) { + try { wrapProviderChild(child, createAcpTap(runtime), runtime); } + catch (error) { runtime.log({ event: "spawn_failed", layer: "acp", error: String(error) }); } + } + return child; + }; + const sidecar = createSidecarTap(runtime); + const stdoutWrite = process.stdout.write; + process.stdout.write = function tappedStdoutWrite(chunk, ...rest) { + // The sidecar writes one whole frame per call (writeFrame); anything else passes through. + if (typeof chunk !== "string" || !chunk.endsWith("\n") || chunk.indexOf("\n") !== chunk.length - 1) return stdoutWrite.call(this, chunk, ...rest); + let result = true; + let wrote = false; + sidecar.toRunner(chunk.slice(0, -1), line => { + const value = `${line}\n`; + if (!wrote) { wrote = true; result = stdoutWrite.call(process.stdout, value, ...rest); } + else stdoutWrite.call(process.stdout, value); + }); + // A held or dropped frame was accepted, as a slow peer would accept it. + if (!wrote && typeof rest[rest.length - 1] === "function") process.nextTick(rest[rest.length - 1]); + return result; + }; + // Observe the runner's requests where the stdin stream receives them, without adding a reader: + // a 'data' listener would start the flow before the sidecar's own reader is attached. + const requests = lineSplitter(line => sidecar.fromRunner(line)); + const push = process.stdin.push; + process.stdin.push = function tappedPush(chunk, ...rest) { + try { if (chunk !== null && chunk !== undefined) requests.push(chunk); } catch { /* observing never blocks a read */ } + return push.call(this, chunk, ...rest); + }; + process.once("exit", () => sidecar.ended()); + runtime.log({ event: "spawned", layer: "sidecar" }); +} + +module.exports = { PROVIDER_BOOTSTRAP_MARKER, createAcpTap, createFaultRuntime, createSidecarTap, installAcpxFaultTap, isVerifiedProviderLaunch, lineSplitter }; diff --git a/tests/restart-chaos/provider-shim/acpx-sidecar-hook.mjs b/tests/restart-chaos/provider-shim/acpx-sidecar-hook.mjs new file mode 100644 index 00000000000..e7da829354e --- /dev/null +++ b/tests/restart-chaos/provider-shim/acpx-sidecar-hook.mjs @@ -0,0 +1,40 @@ +// Restart chaos provider-message faults for ACPX agents: loaded into the wave's server only +// (`NODE_OPTIONS=--import `, set by the harness). Product code launches the local ACPX +// sidecar from the runner package that owns the module computing the path +// (`runnerd-codex-transport`, from its own `import.meta.url`), and the runner verifies that file's +// digest at launch. This hook loads that one module from the harness's stand-in package (a mirror +// of the runner package whose `dist/cli/acpx-runtime-sidecar.cjs` is the fault stand-in), so the +// digest product code computes and the runner verifies is the stand-in's. A source server loads +// `src/live/runnerd-codex-transport.ts`, a built one `dist/live/runnerd-codex-transport.js`; both +// are covered. Every import the module makes still resolves to the real package, so there is one +// copy of every other module. Product code is unchanged. +import { registerHooks } from "node:module"; +import { pathToFileURL } from "node:url"; + +const realRoot = process.env.RESTART_CHAOS_ACPX_RUNNER_REAL; +const shadowRoot = process.env.RESTART_CHAOS_ACPX_RUNNER_SHADOW; +const MODULES = ["src/live/runnerd-codex-transport.ts", "dist/live/runnerd-codex-transport.js"]; +if (realRoot && shadowRoot) { + const realPrefix = `${pathToFileURL(realRoot).href}/`; + const shadowPrefix = `${pathToFileURL(shadowRoot).href}/`; + const targets = new Set(MODULES.map(module => `${realPrefix}${module}`)); + let announced = false; + registerHooks({ + resolve(specifier, context, next) { + // The stand-in module's own imports resolve as the real module's would. + const parentURL = context.parentURL?.startsWith(shadowPrefix) ? `${realPrefix}${context.parentURL.slice(shadowPrefix.length)}` : context.parentURL; + const resolved = next(specifier, parentURL === context.parentURL ? context : { ...context, parentURL }); + const url = resolved.url.split("?")[0]; + if (!targets.has(url)) return resolved; + if (!announced) { + announced = true; + process.stderr.write(`[restart-chaos] acpx-sidecar-hook: pid ${process.pid} loads ${url.slice(realPrefix.length)} from the stand-in package\n`); + } + return { ...resolved, url: `${shadowPrefix}${resolved.url.slice(realPrefix.length)}`, shortCircuit: true }; + }, + }); +} +// 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("acpx-sidecar-hook")).join(" "); +if (remaining) process.env.NODE_OPTIONS = remaining; else delete process.env.NODE_OPTIONS; diff --git a/tests/restart-chaos/scheduler.ts b/tests/restart-chaos/scheduler.ts index 3f33112c573..a1fdca3d39b 100644 --- a/tests/restart-chaos/scheduler.ts +++ b/tests/restart-chaos/scheduler.ts @@ -50,6 +50,8 @@ export const TRIGGER_DELAY_RANGE_MS: Record, readonly stop: [0, 2_000], asked: [0, 3_000], answered: [0, 3_000], + // Not a `--triggers` choice: `--provider-fault-restart-after-seconds` sets its delay. + provider: [0, 0], }; export function triggerFitsWorkload(trigger: TriggerId, workload: WorkloadId): boolean { @@ -280,7 +282,10 @@ 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; - // A storm crosses K13 tasks; a provider fault (PV*) and a matrix baseline (K0) restart nothing. + // A storm crosses K13 tasks; a provider fault (PV*) and a matrix baseline (K0) restart nothing, unless + // `--provider-fault-restart` asked for a K1 once the provider fault applied (the `provider` trigger). + const restart = task.providerFault?.restart; + if (restart) return [{ kind: restart.kind, holdMs: 0, atMs: 0, trigger: "provider", delayMs: restart.afterAppliedMs }]; if (task.fault === "K13" || task.fault === "K0" || isProviderFault(task.fault)) return []; const planned: ScheduledFault = { kind: task.fault, holdMs: task.holdMs, atMs: task.fireAfterMs }; const trigger = task.trigger ?? "time"; diff --git a/tests/restart-chaos/types.ts b/tests/restart-chaos/types.ts index d8e4a349f86..0791fb889c0 100644 --- a/tests/restart-chaos/types.ts +++ b/tests/restart-chaos/types.ts @@ -29,7 +29,8 @@ export type AdapterId = (typeof ADAPTER_IDS)[number]; /** `asked` and `answered`: the journey boundaries of Q1–Q3 (journeys.ts, question-handoff.ts). */ export const TRIGGER_IDS = ["time", "comment", "retry", "queued", "tool", "result", "approval", "paused", "stop", "asked", "answered"] as const; -export type TriggerId = (typeof TRIGGER_IDS)[number]; +/** `provider` is not a `--triggers` choice: `--provider-fault-restart` schedules it (K1 after a provider fault applied). */ +export type TriggerId = (typeof TRIGGER_IDS)[number] | "provider"; export const ENVIRONMENT_KINDS = ["local", "daytona"] as const; export type EnvironmentKind = (typeof ENVIRONMENT_KINDS)[number]; @@ -108,6 +109,9 @@ export interface WaveOptions { serverFrom?: string; /** TM2/TM3: how far every database timestamp is shifted back (`--time-age-hours`, default 26 h). */ timeAgeMs?: number; + /** `--provider-fault-restart K1`: each PV1/PV2 task also kills -9 the control plane after its fault applied. */ + providerFaultRestart?: "K1"; + providerFaultRestartAfterMs?: number; } export interface StormFault { @@ -368,6 +372,8 @@ export interface TaskRecord { question?: QuestionRecord; /** Q2, Q3: the board's journey actions (journey-driver.ts). */ journey?: import("./journeys.js").JourneyRecord; + /** PV with `--provider-fault-restart`: when the shim logged the task's fault (arms the `provider` trigger). */ + providerFaultAppliedAt?: number; } export interface TaskGrade {