test(chaos): fire a planned restart while the server is mid S3 transfer (#20451)

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work
> - The restart-chaos harness restarts the control plane while agent
runs are in flight, and `--storage s3` runs the server against a
loopback MinIO behind a fault proxy
> - Planned restarts fire by time or by run events. A time-based restart
seldom lands while the server is moving bytes to or from the store
> - The riskiest storage moments are a workspace snapshot upload, a
snapshot restore, and a run-log journal write. A restart in the middle
of one must not lose or corrupt the workspace or the log
> - This pull request adds a storage-triggered restart: the planned
fault fires the moment the proxy sees a matching S3 request, and the
proxy holds that request so the restart interrupts it
> - The benefit is that restart-during-transfer becomes a repeatable
wave with receipts that name the interrupted request

## Linked Issues or Issue Description

**Subsystem affected**
Restart-chaos test harness (`tests/restart-chaos`), cloud storage mode.

**Problem or motivation**
A K1/K2/K4 restart fires on a timer or on a run trigger. With `--storage
s3`, the timer almost never lands during a snapshot upload, a snapshot
download, or a run-log write. So the harness does not prove that a
restart in the middle of a transfer is safe.

**Proposed solution**
Add `--storage-trigger <target>`. At its planned time, each task's fault
is armed. The next server request that matches the target fires it. The
proxy holds the request open (up to 60 s) while the harness snapshots
the tasks and restarts the server.

**Alternatives considered**
A new `--triggers storage` id. It needs edits to the observer (near its
1,000-line limit), the trigger registry and the delay table. The storage
trigger is a wave-level mode in a new module instead.

## What Changed

- New `tests/restart-chaos/storage-trigger.ts`: targets
`snapshot-upload`, `snapshot-part` (multipart `UploadPart`),
`snapshot-download` and `run-log`; the `StorageTrigger` that arms
faults, fires on a match, holds the request and records receipts.
- `storage-proxy.ts`: an `observe(listener)` hook. The listener sees
every request and can hold it (the existing latency path).
- `main.ts`, `config.ts`, `types.ts`: registry lines only.
`--storage-trigger` (needs `--storage s3` and `--triggers time`); a due
fault is armed instead of fired; the wave writes `storage-trigger.json`.
- `storage-trigger.json` records the matching request (method, path with
upload ids hidden, declared bytes, outcome `aborted` or `completed`,
status), every other request in flight, and the fault id. A fault that
never matched is recorded with `missedReason`.
- README: "Restart during a transfer".

## Verification

- `PAPERCLIP_IN_WORKTREE=false npx vitest run --config
tests/restart-chaos/vitest.config.ts` (25 files, 304 tests pass).
- Red/green: `storage-trigger.test.ts` against the master
`storage-proxy.ts` fails 3 of 6 tests (no hook, so nothing is held or
fired); it passes with this change.
- `node scripts/typecheck-test-projects.mjs tests/restart-chaos` passes.
- Live waves (local run-log, Daytona snapshot upload and download) are
recorded in a PR comment as they finish.

## Risks

- Low risk. Test harness only. Without `--storage-trigger`, the proxy
has no listener and behaves as before.
- A replay re-arms its faults at their observed offsets. It fires on the
next matching request, not on the exact same request.

## Model Used

- Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, with tool use and code
execution in Claude Code.

## Checklist

- [x] I have included a thinking path that traces from project context
to this change
- [x] I have specified the model used (with version and capability
details)
- [x] I have checked ROADMAP.md and confirmed this PR does not duplicate
planned core work
- [x] I have searched GitHub for duplicate or related PRs and linked
them above
- [x] I have either (a) linked existing issues with `Fixes: #` / `Closes
#` / `Refs #` OR (b) described the issue in-PR following the relevant
issue template
- [x] I have not referenced internal/instance-local Paperclip issues or
links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip`
URLs)
- [x] My branch name describes the change (e.g. `docs/...`, `fix/...`)
and contains no internal Paperclip ticket id or instance-derived details
- [x] I have run tests locally and they pass
- [x] I have added or updated tests where applicable
- [x] I have updated relevant documentation to reflect my changes
- [x] I have considered and documented any risks above
- [ ] All Paperclip CI gates are green
- [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups
- [x] I will address all Greptile and reviewer comments before
requesting merge

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

tests/restart-chaos/README.md

@@ -385,6 +385,30 @@ GitHub `origin` over https at its `origin/master` commit, so the sandbox clones
 snapshots are Paperclip-sized. Its oracle reads the task's files through the board API, retrying for up
 to 4 minutes while the closing save makes the workspace readable, and runs its commands on a local copy.
 
+### Restart during a transfer (`--storage-trigger`)
+
+A time-based fault rarely lands while the server is moving bytes. `--storage-trigger <target>` turns
+each task's planned K fault into an armed fault: at its planned time (`--fault-after-seconds` pins it)
+the fault is armed, and it fires the moment the proxy sees the next matching server request. The
+proxy holds that request (up to 60 s) while the harness takes its pre-fault snapshot and restarts the
+server, so the restart provably interrupts the transfer.
+
+| Target | Matching request |
+| --- | --- |
+| `snapshot-upload` | `PUT` with a body under `workspace-snapshots/` (single PUT or multipart part; not create or complete) |
+| `snapshot-part` | multipart `UploadPart` (`PUT …?partNumber=`) under `workspace-snapshots/` |
+| `snapshot-download` | `GET` under `workspace-snapshots/` (restore) |
+| `run-log` | `PUT`/`POST` under `run-logs/` (run-log journal segments) |
+
+`storage-trigger.json` records, per armed fault, the request that fired it (method, path with upload
+ids hidden, declared bytes, outcome `aborted`/`completed`), every request in flight at that
+moment (the held one included), and the fault id it produced. A fault whose task rested first, or that saw no matching request
+before the wave ended, is recorded with `missedReason` (`task_rested` or `no_matching_request`). The wave
+summary adds a "Storage trigger" section with each armed fault's result and the count not reached. It
+fails when no restart fired, or when a fired restart has no fault id or its held request was not
+`aborted` (the hold expired first). It needs `--storage s3` and `--triggers time`, and does not take
+G1, H1 or Q1: their board actions would run while the fault waits for a request.
+
 Every Retry the grader needs is clicked in the real board UI: the harness opens the issue in a headless
 Chromium, waits for the visible Retry button and clicks it. A task whose Retry button is missing fails
 R3 as `ui_retry_button_missing (user saw: …)`; a click that sends no retry request fails as

tests/restart-chaos/config.ts

@@ -5,6 +5,7 @@ import { applyJourneyOptions } from "./journey-matrix.js";
 import { takesJourneyTriggers } from "./journeys.js";
 import { STORAGE_FAULT_KINDS, type StorageFaultKind } from "./storage-chaos.js";
 import { parseTimeFaultOptions } from "./time-faults-config.js";
+import { parseStorageTrigger } from "./storage-trigger.js";
 import { parseRepeatedRestartOptions } from "./repeated-restarts.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";
@@ -109,6 +110,7 @@ export function parseCommand(argv: readonly string[], now = Date.now()): ParsedC
       storage: { type: "string" },
       "storage-faults": { type: "string" },
       "storage-fault-seconds": { type: "string" },
+      "storage-trigger": { type: "string" },
       "provider-fault-message": { type: "string" },
       "provider-fault-delays": { type: "string" },
       "db-pressure": { type: "string" },
@@ -124,7 +126,7 @@ export function parseCommand(argv: readonly string[], now = Date.now()): ParsedC
   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", "provider-fault-restart", "provider-fault-restart-after-seconds", "rr-gap-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", "storage-trigger", "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", "rr-gap-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,
@@ -286,6 +288,7 @@ export function parseCommand(argv: readonly string[], now = Date.now()): ParsedC
     if (!match || min < 5 || max < min || max > 900) throw new Error("--storage-fault-seconds takes <min>-<max> seconds, 5 to 900, min not above max");
     options.storageFaultMs = [min * 1000, max * 1000];
   }
+  if (values["storage-trigger"] !== undefined) options.storageTrigger = parseStorageTrigger(values["storage-trigger"], options);
   return { mode: "wave", options, keep };
 }
 

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 { exposedAfterLaunch, StorageTrigger, storageTriggerMarkdown, storageTriggerShortfall } from "./storage-trigger.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";
@@ -250,11 +251,13 @@ async function main(): Promise<number> {
   let board: BoardUi | null = null;
   let objectStore: ObjectStore | null = null;
   let storageChaos: StorageChaos | null = null;
+  let storageTrigger: StorageTrigger | null = null;
   let storageProxy: Awaited<ReturnType<typeof startStorageFaultProxy>> | null = null;
   let credentialRotation: CredentialRotation | null = null;
   let providerFaults: ProviderFaultSeam | null = null;
   let acpxSidecarSeam: AcpxSidecarSeam | null = null;
   let speedSection: string | null = null;
+  let storageSection: string | null = null;
   let harnessError: unknown;
   process.on("uncaughtException", error => {
     if (isSocketTypeOfServiceRace(error)) { timeline.add("harness_socket_race_ignored", { error: String(error) }); return; }
@@ -385,6 +388,28 @@ async function main(): Promise<number> {
       }));
     };
 
+    let launching: Promise<void> | null = null;
+    if (storageProxy && plan.options.storageTrigger) {
+      storageTrigger = new StorageTrigger(plan.options.storageTrigger, {
+        canFire: () => !interrupted && !faultsQueued && !supervisor.faultInProgress && supervisor.serverReady,
+        isOpen: taskId => observers.some(observer => observer.plan.taskId === taskId && !observer.done && observer.record.restAt === null),
+        record: (kind, data) => timeline.add(kind, data),
+        fire: armed => {
+          faultsQueued += 1;
+          const work = (async () => {
+            const exposed = await exposedAfterLaunch(launching, launched);
+            await Promise.all(exposed.map(candidate => candidate.snapshotBeforeFault()));
+            const queued = faultWork.length;
+            fireFault(armed.kind, armed.holdMs, armed.taskId, exposed, undefined, undefined, armed.restartSteps);
+            await faultWork[queued];
+          })().finally(() => { faultsQueued -= 1; });
+          faultWork.push(work);
+          return work;
+        },
+      });
+      storageProxy.observe(storageTrigger.listener);
+    }
+
     supervisor.onCrashPoint = receipt => {
       const owner = pointResults.find(result => result.token === receipt.crashPoint?.token);
       if (owner) owner.fault = receipt;
@@ -437,7 +462,7 @@ async function main(): Promise<number> {
         errors: check.errors.map(error => redactKnownSecrets(error, knownSecretValues())) });
     };
 
-    while (!interrupted && (pending.length || active.length)) {
+    while (!interrupted && (pending.length || active.length || storageTrigger?.busy)) {
       const now = Date.now();
       if (storm?.endIfDue(now)) {
         // Unlaunched pool tasks are not part of the wave; the in-flight ones drain without new faults.
@@ -455,13 +480,14 @@ async function main(): Promise<number> {
         }
       }
       // Recheck the storm clock per launch: ensureServing or a slow launch can cross the storm's end.
-      while (!crashPoint && active.length < plan.options.concurrency && pending.length && supervisor.serverReady && !supervisor.faultInProgress
+      while (!crashPoint && active.length < plan.options.concurrency && pending.length && supervisor.serverReady && !supervisor.faultInProgress && !storageTrigger?.busy
         && (!storm || storm.mayLaunch(Date.now()))
         // `--matrix`: no-fault baselines never share the server with a fault cell.
         && (!plan.options.matrix || matrixMayLaunch(pending[0]!.plan, active.filter(observer => !observer.done).map(observer => observer.plan)))) {
         const next = pending.shift()!;
         launched.push(next);
-        await next.launch();
+        launching = next.launch().finally(() => { launching = null; });
+        await launching;
         if (plan.options.storageWindows && storageChaos && launched.length === 1) storageChaos.start();
         if (!next.done) active.push(next);
       }
@@ -488,6 +514,7 @@ async function main(): Promise<number> {
         const due = observer.faultDue(Date.now());
         if (!due) continue;
         if (observer.record.restAt !== null) { timeline.add("fault_skipped_task_resting", { taskId: observer.plan.taskId, fault: due.kind }); continue; }
+        if (storageTrigger) { storageTrigger.arm({ taskId: observer.plan.taskId, kind: due.kind, holdMs: due.holdMs, armedAt: Date.now(), ...(due.restartSteps ? { restartSteps: due.restartSteps } : {}) }); continue; }
         const exposed = active.filter(candidate => !candidate.done);
         await Promise.all(exposed.map(candidate => candidate.snapshotBeforeFault()));
         if (!(await observer.confirmTrigger(due, Date.now()))) continue;
@@ -519,6 +546,12 @@ async function main(): Promise<number> {
       await delay(1_000);
     }
     await Promise.all(faultWork);
+    if (storageTrigger) {
+      const shortfall = storageTriggerShortfall(storageTrigger.finish(faults));
+      storageSection = storageTriggerMarkdown(storageTrigger.receipts);
+      timeline.add("storage_trigger_settled", { shortfall, receipts: storageTrigger.receipts.length });
+      if (shortfall) harnessError ??= new Error(shortfall);
+    }
     await stopPressure();
     checkWorktree("wave_end");
     if (worktreeChanged) harnessError ??= new Error("The worktree changed during the wave (build, install or branch switch); later restarts may have run different code");
@@ -684,6 +717,8 @@ async function main(): Promise<number> {
     await credentialRotation?.settled();
     if (credentialRotation?.failures.length) harnessError ??= new Error(`ST5R credential rotation failed: ${credentialRotation.failures.join("; ")}`);
     if (storageProxy) {
+      storageProxy.observe(null);
+      if (storageTrigger) await writeFile(path.join(waveDir, "storage-trigger.json"), `${JSON.stringify(storageTrigger.finish(faults), null, 2)}\n`, { mode: 0o600 }).catch(() => undefined);
       if (storageChaos) await writeFile(path.join(waveDir, "storage-faults.json"), `${JSON.stringify(storageChaos.receipts, null, 2)}\n`, { mode: 0o600 }).catch(() => undefined);
       await writeFile(path.join(waveDir, "storage-proxy.json"), `${JSON.stringify(storageProxy.counts(), null, 2)}\n`, { mode: 0o600 }).catch(() => undefined);
       await storageProxy.close().catch(() => undefined);
@@ -696,7 +731,7 @@ async function main(): Promise<number> {
     if (lines.length) {
       await appendLedger(resultsRoot, lines);
       const summary = await writeSummary(resultsRoot, plan.waveId);
-      await writeFile(path.join(waveDir, "SUMMARY.md"), speedSection ? `${summary}\n${speedSection}` : summary, { mode: 0o600 });
+      await writeFile(path.join(waveDir, "SUMMARY.md"), [summary, speedSection, storageSection].filter(Boolean).join("\n"), { mode: 0o600 });
       console.info(`\n${summary}`);
     }
   }

tests/restart-chaos/storage-proxy.ts

@@ -22,6 +22,8 @@ const CREDENTIAL_ERROR_MESSAGES: Record<CredentialErrorCode, string> = {
 
 export interface StorageFaultRule { fault: StorageFault; methods?: readonly string[]; pathIncludes?: string; remaining?: number }
 
+export type StorageRequestListener = (incoming: IncomingMessage, outgoing: ServerResponse) => number | void;
+
 export interface StorageProxyCounts { requests: number; injected: number; byMethod: Record<string, number>; byStatus: Record<string, number> }
 
 const XML = "application/xml";
@@ -53,6 +55,7 @@ export async function startStorageFaultProxy(endpoint: string, random: () => num
     throw new Error("Storage fault injection requires a loopback store");
   }
   let rules: StorageFaultRule[] = [];
+  let listener: StorageRequestListener | null = null;
   const counts: StorageProxyCounts = { requests: 0, injected: 0, byMethod: {}, byStatus: {} };
   const sockets = new Set<Socket>();
   const delayed = new Set<ReturnType<typeof setTimeout>>();
@@ -83,9 +86,10 @@ export async function startStorageFaultProxy(endpoint: string, random: () => num
   };
   const server = createServer((incoming, outgoing) => {
     const method = incoming.method ?? "GET";
-    const rule = rules.find(candidate => matches(candidate, method, incoming.url ?? "/"));
+    const heldMs = listener?.(incoming, outgoing) || 0;
+    const rule = heldMs ? undefined : rules.find(candidate => matches(candidate, method, incoming.url ?? "/"));
     if (rule?.remaining !== undefined) rule.remaining -= 1;
-    const fault = rule?.fault ?? { kind: "pass" as const };
+    const fault: StorageFault = heldMs ? { kind: "latency", ms: heldMs } : rule?.fault ?? { kind: "pass" };
     if (fault.kind === "latency") {
       const delay = fault.ms + Math.round((fault.jitterMs ?? 0) * random());
       if (fault.bytesPerSecond && !links.has(rule!)) links.set(rule!, linkRates(fault.bytesPerSecond));
@@ -119,6 +123,7 @@ export async function startStorageFaultProxy(endpoint: string, random: () => num
     counts: () => structuredClone(counts),
     setRules(next: readonly StorageFaultRule[]) { endLinks(); rules = next.map(rule => ({ ...rule })); },
     clear() { endLinks(); rules = []; },
+    observe(next: StorageRequestListener | null) { listener = next; },
     async close() {
       for (const timer of delayed) clearTimeout(timer);
       delayed.clear();

tests/restart-chaos/storage-trigger.test.ts

@@ -0,0 +1,200 @@
+import { createServer, request, type Server } from "node:http";
+import { once } from "node:events";
+import { afterEach, describe, expect, it } from "vitest";
+import { parseCommand } from "./config.js";
+import { startStorageFaultProxy } from "./storage-proxy.js";
+import { exposedAfterLaunch, redactStorageUrl, storageTriggerMarkdown, storageTriggerMatches, storageTriggerShortfall, StorageTrigger, type StorageTriggerReceipt, type ArmedStorageFault, type StorageTriggerHooks } from "./storage-trigger.js";
+import type { FaultReceipt } from "./types.js";
+
+const closers: Array<() => Promise<unknown>> = [];
+afterEach(async () => { for (const close of closers.splice(0).reverse()) await close(); });
+
+async function upstream() {
+  const seen: string[] = [];
+  const server: Server = createServer((incoming, outgoing) => {
+    incoming.resume();
+    incoming.on("end", () => { seen.push(`${incoming.method} ${incoming.url}`); outgoing.writeHead(200).end("ok"); });
+  });
+  server.listen(0, "127.0.0.1"); await once(server, "listening");
+  closers.push(async () => { server.closeAllConnections(); server.close(); });
+  return { endpoint: `http://127.0.0.1:${(server.address() as { port: number }).port}`, seen };
+}
+
+function hooks(overrides: Partial<StorageTriggerHooks> = {}) {
+  const fired: Array<{ armed: ArmedStorageFault; path: string }> = [];
+  const events: string[] = [];
+  const value: StorageTriggerHooks = {
+    canFire: () => true, isOpen: () => true,
+    fire: async (armed, held) => { fired.push({ armed, path: held.path }); },
+    record: kind => { events.push(kind); },
+    ...overrides,
+  };
+  return { value, fired, events };
+}
+
+function send(endpoint: string, method: string, route: string, body = "bytes") {
+  const outgoing = request(`${endpoint}${route}`, { method, headers: { "content-length": Buffer.byteLength(body) } });
+  outgoing.on("error", () => undefined);
+  const response = new Promise<number>(resolve => outgoing.on("response", incoming => { incoming.resume(); resolve(incoming.statusCode ?? 0); }));
+  outgoing.end(body);
+  return { outgoing, response };
+}
+
+const until = async (check: () => boolean) => { for (let index = 0; index < 200 && !check(); index += 1) await new Promise(resolve => setTimeout(resolve, 10)); };
+const snapshotPut = "/bucket/chaos/w/c1/workspace-snapshots/ws1/s1/u1.pcws";
+
+describe("storage trigger targets", () => {
+  it("matches snapshot uploads, parts, downloads and run-log writes only", () => {
+    expect(storageTriggerMatches("snapshot-upload", "PUT", snapshotPut, 4096)).toBe(true);
+    expect(storageTriggerMatches("snapshot-upload", "PUT", `${snapshotPut}?partNumber=1&uploadId=u`, 4096)).toBe(true);
+    expect(storageTriggerMatches("snapshot-upload", "PUT", snapshotPut, 0)).toBe(false);
+    expect(storageTriggerMatches("snapshot-upload", "POST", `${snapshotPut}?uploads`, 0)).toBe(false);
+    expect(storageTriggerMatches("snapshot-upload", "POST", `${snapshotPut}?uploadId=u`, 120)).toBe(false);
+    expect(storageTriggerMatches("snapshot-upload", "GET", snapshotPut)).toBe(false);
+    expect(storageTriggerMatches("snapshot-part", "PUT", `${snapshotPut}?partNumber=2&uploadId=abc`)).toBe(true);
+    expect(storageTriggerMatches("snapshot-part", "PUT", snapshotPut)).toBe(false);
+    expect(storageTriggerMatches("snapshot-download", "GET", snapshotPut)).toBe(true);
+    expect(storageTriggerMatches("snapshot-download", "HEAD", snapshotPut)).toBe(false);
+    expect(storageTriggerMatches("run-log", "PUT", "/bucket/chaos/w/run-logs/c1/r1/000001.ndjson")).toBe(true);
+    expect(storageTriggerMatches("run-log", "GET", "/bucket/chaos/w/run-logs/c1/r1/000001.ndjson")).toBe(false);
+    expect(storageTriggerMatches("run-log", "PUT", snapshotPut)).toBe(false);
+  });
+
+  it("keeps part numbers but hides upload ids and signatures in receipts", () => {
+    expect(redactStorageUrl(`${snapshotPut}?partNumber=2&uploadId=abc&X-Amz-Signature=s`)).toBe(`${snapshotPut}?partNumber=2&uploadId=…`);
+    expect(redactStorageUrl(`${snapshotPut}?uploads`)).toBe(`${snapshotPut}?uploads`);
+    expect(redactStorageUrl(`${snapshotPut}?x-id=PutObject`)).toBe(`${snapshotPut}?x-id=PutObject`);
+  });
+
+  it("parses --storage-trigger only with s3 storage and time triggers", () => {
+    const base = ["--env", "local", "--adapters", "codex", "--workloads", "A1", "--faults", "K1"];
+    const parsed = parseCommand([...base, "--storage", "s3", "--storage-trigger", "run-log"]);
+    expect(parsed.mode === "wave" && parsed.options.storageTrigger).toBe("run-log");
+    expect(() => parseCommand([...base, "--storage-trigger", "run-log"])).toThrow(/needs --storage s3/);
+    expect(() => parseCommand([...base, "--storage", "s3", "--storage-trigger", "everything"])).toThrow(/takes one of/);
+    expect(() => parseCommand([...base, "--workloads", "C1", "--triggers", "comment", "--storage", "s3", "--storage-trigger", "run-log"])).toThrow(/--triggers time/);
+    expect(() => parseCommand([...base, "--workloads", "A1,G1", "--storage", "s3", "--storage-trigger", "run-log"])).toThrow(/board-acting workloads \(G1\)/);
+    expect(() => parseCommand([...base, "--workloads", "C1", "--matrix", "--storage", "s3", "--storage-trigger", "run-log"])).toThrow(/--matrix/);
+  });
+});
+
+describe("storage-triggered restart", () => {
+  it("fires an armed fault on the matching request, holds it open and records what was in flight", async () => {
+    const store = await upstream();
+    const proxy = await startStorageFaultProxy(store.endpoint);
+    closers.push(() => proxy.close());
+    const { value, fired, events } = hooks();
+    const trigger = new StorageTrigger("snapshot-upload", value, 5_000);
+    proxy.observe(trigger.listener);
+
+    const before = send(proxy.endpoint, "PUT", snapshotPut);
+    expect(await before.response).toBe(200);
+    expect(fired).toEqual([]);
+
+    trigger.arm({ taskId: "t01", kind: "K1", holdMs: 0, armedAt: Date.now(), restartSteps: [{ gapMs: 5_000, signal: "SIGKILL" }] });
+    const other = send(proxy.endpoint, "PUT", "/bucket/chaos/w/run-logs/r1/1.ndjson");
+    expect(await other.response).toBe(200);
+    const held = send(proxy.endpoint, "PUT", `${snapshotPut}?partNumber=1&uploadId=u`, "x".repeat(1024));
+    await until(() => fired.length > 0);
+    expect(fired).toEqual([{ armed: expect.objectContaining({ taskId: "t01", kind: "K1", restartSteps: [{ gapMs: 5_000, signal: "SIGKILL" }] }), path: `${snapshotPut}?partNumber=1&uploadId=…` }]);
+    await new Promise(resolve => setTimeout(resolve, 100));
+    expect(store.seen.filter(line => line.includes("partNumber"))).toEqual([]);
+
+    held.outgoing.destroy();
+    await until(() => trigger.receipts[0]?.request?.outcome !== "in_flight");
+    const faults = [{ faultId: "f01", requestedBy: "t01", requestedAt: Date.now() } as FaultReceipt];
+    const [receipt] = trigger.finish(faults);
+    expect(receipt).toMatchObject({ target: "snapshot-upload", taskId: "t01", faultId: "f01", missedAt: null,
+      request: { method: "PUT", held: true, declaredBytes: 1024, outcome: "aborted", status: null } });
+    expect(receipt!.inFlight.map(entry => entry.method)).toEqual(["PUT"]);
+    expect(events).toEqual(["storage_trigger_armed", "storage_trigger_fired"]);
+    expect(proxy.counts().byMethod.PUT).toBe(2);
+  });
+
+  it("skips a match while a fault cannot fire and records armed faults that never matched", async () => {
+    const store = await upstream();
+    const proxy = await startStorageFaultProxy(store.endpoint);
+    closers.push(() => proxy.close());
+    let ready = false;
+    const { value, fired } = hooks({ canFire: () => ready, isOpen: taskId => taskId !== "t02" });
+    const trigger = new StorageTrigger("snapshot-download", value, 5_000);
+    proxy.observe(trigger.listener);
+    trigger.arm({ taskId: "t02", kind: "K2", holdMs: 0, armedAt: 1 });
+    trigger.arm({ taskId: "t03", kind: "K4", holdMs: 30_000, armedAt: 2 });
+    trigger.arm({ taskId: "t04", kind: "K1", holdMs: 0, armedAt: 3 });
+
+    expect(await send(proxy.endpoint, "GET", snapshotPut, "").response).toBe(200);
+    expect(fired).toEqual([]);
+    ready = true;
+    const held = send(proxy.endpoint, "GET", snapshotPut, "");
+    await until(() => fired.length > 0);
+    expect(fired.map(entry => entry.armed.taskId)).toEqual(["t03"]);
+    held.outgoing.destroy();
+
+    trigger.arm({ taskId: "t02", kind: "K1", holdMs: 0, armedAt: 4 });
+    const receipts = trigger.finish([]);
+    expect(receipts.map(receipt => [receipt.taskId, receipt.missedReason ?? null, receipt.firedAt !== null])).toEqual([
+      ["t02", "task_rested", false], ["t03", null, true], ["t04", "no_matching_request", false], ["t02", "task_rested", false]]);
+  });
+
+  it("forwards a held request after the hold when the restart never came", async () => {
+    const store = await upstream();
+    const proxy = await startStorageFaultProxy(store.endpoint);
+    closers.push(() => proxy.close());
+    const { value } = hooks();
+    const trigger = new StorageTrigger("run-log", value, 50);
+    proxy.observe(trigger.listener);
+    trigger.arm({ taskId: "t01", kind: "K1", holdMs: 0, armedAt: 1 });
+    expect(await send(proxy.endpoint, "PUT", "/bucket/chaos/w/run-logs/r1/1.ndjson").response).toBe(200);
+    await until(() => trigger.receipts[0]?.request?.outcome === "completed");
+    expect(trigger.receipts[0]!.request).toMatchObject({ held: true, outcome: "completed", status: 200 });
+  });
+});
+
+describe("storage trigger receipts", () => {
+  const receipt = (taskId: string, firedAt: number | null, outcome: "aborted" | "completed" = "aborted", faultId?: string): StorageTriggerReceipt => ({
+    target: "snapshot-upload", taskId, kind: "K1", armedAt: 0, firedAt, missedAt: firedAt === null ? 1 : null, inFlight: [],
+    request: firedAt === null ? null : { id: 1, method: "PUT", path: snapshotPut, startedAt: firedAt, declaredBytes: 1, held: true, outcome, status: null, settledAt: firedAt + 1 },
+    ...(faultId ? { faultId } : {}) });
+
+  it("links each fired receipt to the first fault its task requested at or after the fire", () => {
+    const trigger = new StorageTrigger("snapshot-upload", hooks().value);
+    trigger.receipts.push(receipt("t01", 100), receipt("t01", 500), receipt("t02", 300));
+    const faults = [{ faultId: "f03", requestedBy: "t01", requestedAt: 520 }, { faultId: "f01", requestedBy: "t01", requestedAt: 110 },
+      { faultId: "f00", requestedBy: "t01", requestedAt: 50 }, { faultId: "f02", requestedBy: "t02", requestedAt: 305 }] as FaultReceipt[];
+    expect(trigger.finish(faults).map(entry => entry.faultId)).toEqual(["f01", "f03", "f02"]);
+  });
+
+  it("fails a wave whose restarts did not provably interrupt a held transfer", () => {
+    expect(storageTriggerShortfall([receipt("t01", 10, "aborted", "f01"), receipt("t02", null)])).toBeNull();
+    expect(storageTriggerShortfall([receipt("t01", null), receipt("t02", null)])).toMatch(/no restart interrupted a matching transfer \(2 armed, none fired\)/);
+    expect(storageTriggerShortfall([])).toMatch(/0 armed/);
+    expect(storageTriggerShortfall([receipt("t01", 10, "completed", "f01")])).toMatch(/t01 fault=f01 outcome=completed/);
+    expect(storageTriggerShortfall([receipt("t01", 10, "aborted")])).toMatch(/t01 fault=none outcome=aborted/);
+  });
+
+  it("summarises interrupted and unreached faults for the wave summary", () => {
+    const text = storageTriggerMarkdown([receipt("t01", 10, "aborted", "f01"), { ...receipt("t02", null), missedReason: "task_rested" }]);
+    expect(text).toContain("1 of 2 armed faults interrupted a transfer; 1 not reached.");
+    expect(text).toContain("| t01 | K1 | f01 | aborted | `PUT ws1/s1/u1.pcws` |");
+    expect(text).toContain("| t02 | K1 | - | not reached (task_rested) | - |");
+  });
+});
+
+describe("storage-triggered restart exposure", () => {
+  it("waits for a launch in flight and exposes the task it launched", async () => {
+    const launched = [{ id: "t01", done: false }, { id: "t02", done: true }];
+    let finish!: () => void;
+    const launching = new Promise<void>(resolve => { finish = resolve; });
+    launched.push({ id: "t03", done: false });
+    const exposed = exposedAfterLaunch(launching, launched);
+    let settled = false;
+    void exposed.then(() => { settled = true; });
+    await new Promise(resolve => setTimeout(resolve, 10));
+    expect(settled).toBe(false);
+    launched.push({ id: "t04", done: false });
+    finish();
+    expect((await exposed).map(task => task.id)).toEqual(["t01", "t03", "t04"]);
+    expect((await exposedAfterLaunch(null, launched)).map(task => task.id)).toEqual(["t01", "t03", "t04"]);
+  });
+});

tests/restart-chaos/storage-trigger.ts

@@ -0,0 +1,167 @@
+import type { IncomingMessage, ServerResponse } from "node:http";
+import type { FaultKind, FaultReceipt, WaveOptions, WorkloadId } from "./types.js";
+import type { StorageRequestListener } from "./storage-proxy.js";
+import type { RestartStep } from "./repeated-restarts.js";
+
+export const STORAGE_TRIGGER_TARGETS = ["snapshot-upload", "snapshot-part", "snapshot-download", "run-log"] as const;
+export type StorageTriggerTarget = typeof STORAGE_TRIGGER_TARGETS[number];
+export const STORAGE_TRIGGER_HOLD_MS = 60_000;
+
+export const STORAGE_TRIGGER_BOARD_WORKLOADS: readonly WorkloadId[] = ["G1", "H1", "Q1"];
+
+export function storageTriggerMatches(target: StorageTriggerTarget, method: string, url: string, declaredBytes: number | null = null): boolean {
+  const [route = "", query = ""] = url.split("?", 2);
+  switch (target) {
+    case "snapshot-upload": return method === "PUT" && route.includes("/workspace-snapshots/") && (declaredBytes ?? 0) > 0;
+    case "snapshot-part": return method === "PUT" && route.includes("/workspace-snapshots/") && /(^|&)partNumber=/.test(query);
+    case "snapshot-download": return method === "GET" && route.includes("/workspace-snapshots/");
+    case "run-log": return (method === "PUT" || method === "POST") && route.includes("/run-logs/");
+  }
+}
+
+export function parseStorageTrigger(value: string, options: Pick<WaveOptions, "storage" | "stormMinutes" | "crashPoint" | "triggers" | "workloads" | "matrix">): StorageTriggerTarget {
+  if (options.storage !== "s3") throw new Error("--storage-trigger needs --storage s3");
+  if (options.stormMinutes || options.crashPoint) throw new Error("--storage-trigger arms each task's planned fault; a storm or crash-point wave has none");
+  if (options.matrix) throw new Error("--storage-trigger arms faults by time; --matrix plans trigger cells whose windows it would skip");
+  if ((options.triggers ?? []).some(trigger => trigger !== "time")) throw new Error("--storage-trigger replaces the fault's trigger; use it with --triggers time");
+  const board = options.workloads.filter(workload => STORAGE_TRIGGER_BOARD_WORKLOADS.includes(workload));
+  if (board.length) throw new Error(`--storage-trigger keeps a fault armed after its planned time; board-acting workloads (${board.join(", ")}) would act before the restart`);
+  if (!(STORAGE_TRIGGER_TARGETS as readonly string[]).includes(value)) {
+    throw new Error(`--storage-trigger takes one of ${STORAGE_TRIGGER_TARGETS.join(", ")}`);
+  }
+  return value as StorageTriggerTarget;
+}
+
+export function redactStorageUrl(url: string): string {
+  const [route = "", query] = url.split("?", 2);
+  if (query === undefined) return route;
+  const kept = query.split("&").filter(part => part && !/^x-amz-/i.test(part)).map(part => {
+    const [name = "", raw] = part.split("=", 2);
+    return name === "partNumber" || name === "x-id" || raw === undefined ? part : `${name}=…`;
+  });
+  return kept.length ? `${route}?${kept.join("&")}` : route;
+}
+
+export type StorageRequestOutcome = "in_flight" | "completed" | "aborted";
+
+export interface StorageRequestRecord {
+  id: number; method: string; path: string; startedAt: number; declaredBytes: number | null;
+  held: boolean; outcome: StorageRequestOutcome; status: number | null; settledAt: number | null;
+}
+
+export interface ArmedStorageFault { taskId: string; kind: FaultKind; holdMs: number; armedAt: number; restartSteps?: RestartStep[] }
+
+export interface StorageTriggerReceipt {
+  target: StorageTriggerTarget; taskId: string; kind: FaultKind; armedAt: number;
+  firedAt: number | null; missedAt: number | null; missedReason?: string;
+  request: StorageRequestRecord | null; inFlight: StorageRequestRecord[]; faultId?: string;
+}
+
+export interface StorageTriggerHooks {
+  canFire(): boolean;
+  isOpen(taskId: string): boolean;
+  fire(armed: ArmedStorageFault, request: StorageRequestRecord): Promise<void>;
+  record(kind: string, data: Record<string, unknown>): void;
+}
+
+export class StorageTrigger {
+  readonly receipts: StorageTriggerReceipt[] = [];
+  private readonly armed: Array<{ fault: ArmedStorageFault; receipt: StorageTriggerReceipt }> = [];
+  private readonly open = new Map<number, StorageRequestRecord>();
+  private nextId = 1;
+  private firing = false;
+
+  constructor(readonly target: StorageTriggerTarget, private readonly hooks: StorageTriggerHooks, private readonly holdMs = STORAGE_TRIGGER_HOLD_MS) {}
+
+  get busy(): boolean { return this.firing; }
+
+  arm(fault: ArmedStorageFault): void {
+    const receipt: StorageTriggerReceipt = { target: this.target, taskId: fault.taskId, kind: fault.kind, armedAt: fault.armedAt,
+      firedAt: null, missedAt: null, request: null, inFlight: [] };
+    this.receipts.push(receipt);
+    this.armed.push({ fault, receipt });
+    this.hooks.record("storage_trigger_armed", { taskId: fault.taskId, fault: fault.kind, target: this.target });
+  }
+
+  readonly listener: StorageRequestListener = (incoming: IncomingMessage, outgoing: ServerResponse) => {
+    const method = incoming.method ?? "GET";
+    const url = incoming.url ?? "/";
+    const length = Number(incoming.headers["content-length"]);
+    const request: StorageRequestRecord = { id: this.nextId++, method, path: redactStorageUrl(url), startedAt: Date.now(),
+      declaredBytes: Number.isFinite(length) ? length : null, held: false, outcome: "in_flight", status: null, settledAt: null };
+    this.open.set(request.id, request);
+    outgoing.on("close", () => {
+      this.open.delete(request.id);
+      request.outcome = outgoing.writableFinished ? "completed" : "aborted";
+      request.status = outgoing.headersSent ? outgoing.statusCode : null;
+      request.settledAt = Date.now();
+    });
+    if (!storageTriggerMatches(this.target, method, url, request.declaredBytes)) return;
+    const next = this.take(request.startedAt);
+    if (!next) return;
+    request.held = true;
+    this.firing = true;
+    next.receipt.firedAt = request.startedAt;
+    next.receipt.request = request;
+    next.receipt.inFlight = [...this.open.values()];
+    this.hooks.record("storage_trigger_fired", { taskId: next.fault.taskId, fault: next.fault.kind, target: this.target,
+      request: `${method} ${request.path}`, inFlight: next.receipt.inFlight.length });
+    void this.hooks.fire(next.fault, request).catch(error => {
+      this.hooks.record("storage_trigger_fire_failed", { taskId: next.fault.taskId, error: String(error) });
+    }).finally(() => { this.firing = false; });
+    return this.holdMs;
+  };
+
+  private take(now: number) {
+    if (this.firing || !this.armed.length || !this.hooks.canFire()) return null;
+    while (this.armed.length) {
+      const next = this.armed.shift()!;
+      if (this.hooks.isOpen(next.fault.taskId)) return next;
+      this.miss(next.receipt, now, "task_rested");
+    }
+    return null;
+  }
+
+  private miss(receipt: StorageTriggerReceipt, at: number, reason: string) {
+    receipt.missedAt = at;
+    receipt.missedReason = reason;
+    this.hooks.record("storage_trigger_missed", { taskId: receipt.taskId, fault: receipt.kind, target: this.target, reason });
+  }
+
+  finish(faults: readonly FaultReceipt[], now = Date.now()): StorageTriggerReceipt[] {
+    for (const { receipt } of this.armed.splice(0)) this.miss(receipt, now, this.hooks.isOpen(receipt.taskId) ? "no_matching_request" : "task_rested");
+    for (const receipt of this.receipts) {
+      if (receipt.firedAt === null) continue;
+      const fault = faults.filter(candidate => candidate.requestedBy === receipt.taskId && candidate.requestedAt >= receipt.firedAt!)
+        .sort((left, right) => left.requestedAt - right.requestedAt)[0];
+      if (fault) receipt.faultId = fault.faultId;
+    }
+    return this.receipts;
+  }
+}
+
+export function storageTriggerShortfall(receipts: readonly StorageTriggerReceipt[]): string | null {
+  const fired = receipts.filter(receipt => receipt.firedAt !== null);
+  if (!fired.length) return `--storage-trigger: no restart interrupted a matching transfer (${receipts.length} armed, none fired)`;
+  const unproven = fired.filter(receipt => !receipt.faultId || receipt.request?.outcome !== "aborted");
+  if (!unproven.length) return null;
+  return `--storage-trigger: ${unproven.length} fired restart(s) did not provably interrupt the held request: ${unproven
+    .map(receipt => `${receipt.taskId} fault=${receipt.faultId ?? "none"} outcome=${receipt.request?.outcome ?? "none"}`).join(", ")}`;
+}
+
+export function storageTriggerMarkdown(receipts: readonly StorageTriggerReceipt[]): string {
+  const fired = receipts.filter(receipt => receipt.firedAt !== null);
+  const proven = fired.filter(receipt => receipt.faultId && receipt.request?.outcome === "aborted");
+  const lines = [`## Storage trigger (${receipts[0]?.target ?? "none"})`, "",
+    `${proven.length} of ${receipts.length} armed faults interrupted a transfer; ${receipts.length - fired.length} not reached.`, "",
+    "| Task | Fault | Fault id | Result | Request |", "| --- | --- | --- | --- | --- |",
+    ...receipts.map(receipt => `| ${receipt.taskId} | ${receipt.kind} | ${receipt.faultId ?? "-"} | ${receipt.firedAt === null
+      ? `not reached (${receipt.missedReason ?? "unknown"})` : receipt.request?.outcome ?? "unknown"} | ${receipt.request
+      ? `\`${receipt.request.method} ${receipt.request.path.split("/").slice(-3).join("/")}\`` : "-"} |`)];
+  return `${lines.join("\n")}\n`;
+}
+
+export async function exposedAfterLaunch<T extends { done: boolean }>(launching: Promise<unknown> | null, launched: readonly T[]): Promise<T[]> {
+  await launching;
+  return launched.filter(candidate => !candidate.done);
+}

tests/restart-chaos/types.ts

@@ -96,6 +96,7 @@ export interface WaveOptions {
   storageFaults?: Array<import("./storage-chaos.js").StorageFaultKind>;
   storageFaultMs?: [number, number];
   storageWindows?: import("./storage-chaos.js").StorageFaultWindow[];
+  storageTrigger?: import("./storage-trigger.js").StorageTriggerTarget;
   workspaceCommit?: string;
   workspaceGitUrl?: string;
   /** Provider message faults from `--faults` (PV1 delay, PV2 drop, PV3 duplicate); see provider-faults.ts. */