From 2c2186706ef4bb352b662f9d8faebefe47fb52ac Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Mon, 9 Nov 2026 22:37:24 -0600 Subject: [PATCH] test(restart-chaos): human-in-the-loop journeys (Q2, Q3), journey triggers and a journey x fault matrix (#20448) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - The restart chaos harness (`tests/restart-chaos/`) restarts the control plane while real agents work, and grades each task. > - Almost every harness workload is "the agent does a file task and finishes". Few workloads have a person in the loop. > - On 2026-11-09, production failures came from journeys with a person in the loop: a saved answer that could not reach a dependency-blocked task (#20363), and an app-action approval that a routine OAuth refresh or a one-hour deadline made invalid (#20340). > - No workload exercised these journeys, so no wave could catch them. > - This pull request adds journey workloads with deterministic oracles and journey-specific fault triggers. It also adds a journey × fault matrix and a way to run a wave against an older build. > - The benefit: one command crosses every journey with every fault kind, and the same wave proves red before a fix and green after it. ## Linked Issues or Issue Description Refs #20363, Refs #20340, Refs #20336, Refs #20432 ## What Changed - New workload **Q2** (`journeys.ts`, `journey-driver.ts`): the agent saves a question with its native `request_human_input` tool. The board adds an unfinished `backlog` prerequisite, marks the task `blocked` by it, and answers through the task-thread respond route. The oracle requires that the answer reaches the agent in bounded answer mode while the task stays blocked, that the prerequisite relation stays, that no run has `issue_dependencies_blocked`, and that no stale Retry notice stays at rest. Then the board finishes the prerequisite and the agent finishes the task. - New workload **Q3**: a harness-owned OAuth 2.1 + MCP app on loopback (`journey-oauth-app.ts`, discovery, dynamic client registration, PKCE, rotating refresh tokens, one write tool). The board connects it through the Apps routes and sets the tool to Ask first for a scripted `process` agent. When the review is pending, the harness moves the review's saved timestamps and the token expiry back 61 minutes in the wave's own database, runs Check connection (which refreshes the token through the app), and approves. The oracle requires one review, a refresh between request and decision, an accepted approval, exactly one `tools/call` with the reviewed arguments, an `executed` request, and an agent report that carries a receipt only the provider result contains. - New triggers `asked` and `answered` fire a restart fault 0–3 s after the saved request is first seen and after the board's response. They work for the existing **Q1** saved question (#20432) too. The board answers before an `answered` fault fires and after every other planned fault. - New `--matrix` option: one task per workload × adapter × (a `K0` no-fault baseline, each restart fault at each of the workload's own triggers, each provider message fault). `K0` is a new task fault kind with no fault. - New `--server-from ` option: every server start and restart of the wave runs another prepared checkout, for example the parent of a fix, with the same harness code. - Q3 setup allows unpriced usage on the wave's company budget policy, because a scripted process run records usage with no price and the default policy pauses the whole company. - README documents the journeys, the triggers, the matrix and the red-before-fix recipe. Unit tests cover the oracles (pass and the pre-fix failure signatures), the driver's ordering, the matrix plan, option validation and the OAuth app. ## Verification - `pnpm exec vitest run --config tests/restart-chaos/vitest.config.ts`: 295 passed. - `node scripts/typecheck-test-projects.mjs tests/restart-chaos`: passed. - Live local waves, codex, one wave at a time (shared machine lock), `--concurrency 2`, at most 3 tasks: | Wave | Server build | Command | Result | | --- | --- | --- | --- | | `w-20261110T024308-s106` | this branch (master `5570debc5b` base) | `--workloads Q3 --faults K1 --matrix --seed 106` | 3/3 pass (K0, K1^asked, K1^answered); the app saw one refresh before each decision and one `tools/call` | | `w-20261110T024447-s106` | `--server-from` parent of #20340 (`0e325e8d52`) | same | 3/3 fail as expected: approve returns 409 `action_expired` ("Tool review has expired"), no `tools/call`, task stays `in_review` | | `w-20261110T024644-s107` | this branch | `--workloads Q2 --faults K1 --matrix --seed 107` | K1^asked and K1^answered pass. The K0 cell failed only because the oracle then required a comment; the agent had reported `ANSWER-…` in its bounded turn's run result 14 s after the answer, while blocked. The oracle now accepts either. | | `w-20261110T030341-s107` | `--server-from` parent of #20363 (`eebdfe71b1`) | `--workloads Q2 --faults K1 --matrix --seed 107` | 3/3 fail as expected: `answer_reached_agent_while_blocked`. No run starts between the answer and the finished prerequisite (6 minutes of ready server). | | `w-20261110T033426-s108` | this branch, rebased onto #20432 | `--workloads Q1,Q2 --faults K1 --triggers asked,answered --tasks 4 --seed 108` | Q1^asked and Q2^asked pass. Q1^answered and Q2^answered fail on a product gap (below). | | `w-20261110T020037-s102` | this branch, before the rebase | `--workloads Q1,Q2,Q3 --faults K1 --triggers asked,answered --tasks 3 --seed 102` | Q1 with `asked` and `answered` K1 restarts: both done with the answer. | Waves `…-s107` (pre-fix) and `…-s106` (pre-fix) used the same harness code as the green waves. Only the server build changed. **New product failure (not fixed here).** A K1 restart 0.3–3 s after the board's answer can lose the answer turn's native run: - `w-20261110T033426-s108/t04` (Q1, K1 2.9 s after the answer): run `a40cc8f7` resumes with `resume_dead_runner`, then fails `runner_state_identity_mismatch: authority_indeterminate` (`adapter_failed`). The task shows "Couldn't finish this attempt. Your message is saved." and needs Retry. Retry works. - `w-20261110T033426-s108/t03` (Q2, blocked task, K1 0.27 s after the answer): run `5b5355de` is `process_lost`. Its bounded retry `92a57f50` fails `native_session_recovery_failed` (`session.snapshot` failed). The Retry click starts `0e6aa9d1`, which fails `runner_state_identity_mismatch: prior_owner_terminal_state_indeterminate; retirement_refused=journal_commands_unsettled(turn.start.failed)`. The task then waits with `retryWait.reason=process_identity_missing` and stays blocked without the answer. - Suspected path: native runner restart reconciliation of the runner state identity (`resume_dead_runner`, journal command settlement) in `server/src/services/native-runtime/` and `packages/paperclip-runner/`. Host load was 8–14 during the wave. ## Risks - Test harness only. No product code changes. - Q2 depends on the model following its instructions. The oracle accepts the answer in either a comment or the bounded turn's run result, because both were seen. - Q3 shifts persisted timestamps only in the wave's isolated database. It does not shift unrelated tables, so it simulates elapsed time for the review and the token, not for the whole instance. - `K0` and the new triggers add values to shared registries (`types.ts`, `scheduler.ts`). Existing plans keep their meaning. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, used through Claude Code with tool use and code execution. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-authored-by: Claude --- tests/restart-chaos/README.md | 122 ++++++ tests/restart-chaos/config.ts | 14 +- tests/restart-chaos/journey-driver.ts | 508 +++++++++++++++++++++++ tests/restart-chaos/journey-matrix.ts | 86 ++++ tests/restart-chaos/journey-oauth-app.ts | 157 +++++++ tests/restart-chaos/journeys.test.ts | 314 ++++++++++++++ tests/restart-chaos/journeys.ts | 237 +++++++++++ tests/restart-chaos/main.ts | 15 +- tests/restart-chaos/observer.ts | 35 +- tests/restart-chaos/scheduler.ts | 11 +- tests/restart-chaos/types.ts | 23 +- tests/restart-chaos/workloads.ts | 2 + 12 files changed, 1501 insertions(+), 23 deletions(-) create mode 100644 tests/restart-chaos/journey-driver.ts create mode 100644 tests/restart-chaos/journey-matrix.ts create mode 100644 tests/restart-chaos/journey-oauth-app.ts create mode 100644 tests/restart-chaos/journeys.test.ts create mode 100644 tests/restart-chaos/journeys.ts diff --git a/tests/restart-chaos/README.md b/tests/restart-chaos/README.md index 23ac4751495..59dd2eef70c 100644 --- a/tests/restart-chaos/README.md +++ b/tests/restart-chaos/README.md @@ -27,6 +27,8 @@ default 3000), `--poll-ms` (2000–5000), `--keep` (keep the temporary instance) `--k4-hold-max-seconds` (raise the seeded K4 hold upper end from 300 s, up to the 600 s outage cap). `--fault-after-seconds ` fires every planned fault `n` seconds after its task launches (5–3600; with `--triggers time`), keeping the rest of the seeded plan, for example to restart a Daytona R1 task after its workspace preparation. +`--matrix` crosses every selected workload with every fault kind (below), and `--server-from ` +runs every server of the wave from another prepared checkout, for example a commit before a fix. Each wave's temporary instance (embedded Postgres, workspaces, often 5–13 GB) is deleted when the wave ends, pass or fail, including after an interrupt. Evidence @@ -132,6 +134,123 @@ run already finished when the stop window opens, the harness records a missed St (`task_stop_missed`), skips the stop-triggered fault, still posts the continue comment, and the oracle checks only the continue. S1 is not a default workload. +## Human-in-the-loop journeys (Q1, Q2, Q3) + +```sh +node cli/node_modules/tsx/dist/cli.mjs tests/restart-chaos/main.ts --env local \ + --adapters codex,claude --workloads Q1,Q2,Q3 --faults K1,K2,K4 --triggers asked,answered --concurrency 2 --tasks 4 --seed 41 +``` + +Each journey starts from a user entry point in +[`maps/feature-map/questions-and-approvals.md`](../../maps/feature-map/questions-and-approvals.md). +The harness acts as the board through the routes the board UI uses. It acts only +when the server is ready, no fault is in progress, and every fault the task planned +before the response has fired. Two triggers fire at the journey's own boundaries: + +- `asked`: 0–3 s after the observer first sees the saved request pending. +- `answered`: 0–3 s after the board's response returned. + +Until the board acts, the task waits on the board, as G1 does: this is not a rest, +not a stall, and gets no Retry. The code lives in `journeys.ts` (definitions and +oracles of Q2 and Q3), `journey-driver.ts` (their board) and `journey-oauth-app.ts` (Q3's app). + +**Q1, a question** (`questions`, `continuation`) is the saved question described +under "Provider message faults" below (`question-handoff.ts`). With restart faults +it takes the same two triggers: `asked` arms on its first pending poll and +`answered` on the board's answer. The board answers before an `answered` fault +fires and after every other planned fault. + +**Q2, a question on a blocked task** (#20363). The agent asks one single-select +question with its native `request_human_input` tool and ends its turn. The board +reads the card and answers with the option named by the task's nonce (`amber` or +`cobalt`), on the task thread +(`POST /api/issues/:id/interactions/:interactionId/respond`). Before it answers, the board +adds an unassigned `backlog` prerequisite. It sets the task `blocked` by it +(`blockedByIssueIds`), and then answers. The answer must reach the agent in the +bounded answer mode, while the task stays blocked. The agent replies with the +answer marker, in a comment or in that turn's accepted run result (a bounded turn +may report through either). The board waits up to 6 minutes of ready +server for that reply. Then it marks the prerequisite `done`. The +blockers-resolved wake lets the agent write a brief that carries +`ANSWER--` and finish. The oracle checks these points: + +- exactly one question was saved (a restart that made the agent ask twice fails); +- the question was still pending when the board answered; +- with `asked`, the board answered after the restart; +- the brief carries the answer marker and the task rests `done`; +- the reply came while the task was `blocked`; +- the prerequisite relation was kept through the reply; +- the reply came before the prerequisite was finished; +- no run of the task carries `issue_dependencies_blocked`; +- no stale Retry notice at rest. + +**Q3, an app action reviewed late** (`tool-review`, #20340). The harness starts its +own OAuth 2.1 + MCP app on loopback. The app has discovery, dynamic client +registration, PKCE, rotating refresh tokens, one-hour access tokens and one write +tool, `create_page`. No third-party account is used. Before launch, the board +connects the app (`POST /api/companies/:id/tools/apps/connect`). It follows the +app's authorize redirect back to `/api/tools/oauth/callback`. Then it finishes setup +for one scripted `process` agent with `create_page` set to Ask first. That agent +is adapted from `tests/e2e/connection-reviews.spec.ts`. Q3 grades the review path, +not a model, so `--adapters` only labels the cell. A process run records usage with no price, so Q3 setup also sets the company budget policy to allow unpriced usage (the amount and hard stop still apply). + +The agent's call waits for review. Then the board does three things: + +1. It moves that review's saved timestamps and the grant's access-token expiry back + by 61 minutes in the wave's own database (`shiftReviewClock`); a shared connection keeps that expiry on the connection. This is the + `GRACE_SHIFT_MS` approach used by `invariants.ts`. +2. It runs **Check connection** (`POST /api/tool-connections/:id/health-check`). + That check renews the now-expiring token through the app. +3. It approves (`POST /api/tool-gateway/action-requests/:id/approve`). + +The oracle checks these points: + +- exactly one review was requested; +- the review aged more than an hour; +- the app saw a refresh between the request and the decision; +- the approval was accepted; +- the app saw exactly one `tools/call` with the reviewed title; +- the request ended `executed`; +- the agent reported the provider's one-time receipt, which only the tool result carries; +- the task rests `done`. + +## Journey × fault matrix (`--matrix`) + +```sh +node cli/node_modules/tsx/dist/cli.mjs tests/restart-chaos/main.ts --env local \ + --adapters codex --workloads Q1,Q2,Q3 --faults K1,K2,K4 --matrix --concurrency 2 --seed 43 +``` + +`--matrix` plans one task per cell, with a fixed order and a seeded nonce. +The cells are workload × adapter × one of these: + +- `K0`: the same journey with no fault, as a baseline; +- each restart fault in `--faults` at each of the workload's own triggers; +- each provider message fault in `--faults` (PV1–PV3), armed once per workload. + +For a journey workload, those triggers are `asked` and `answered`. The example +plans 3 × (1 + 3 × 2) = 21 tasks. Any workload can join the matrix, with its own +boundary triggers: G1 at `approval`, H1 at `paused`, S1 at `stop`, C1 at +`comment` and `result`, B5 at `tool`, and the rest at `time` (`WORKLOAD_TRIGGERS` +in `journeys.ts`). G1 and H1 get no baseline, because their oracles grade a board +decision made after a restart. `--fault-after-seconds` pins the time-fired cells. +The matrix sets `--tasks` itself. It does not combine with `--tasks`, `--triggers`, +`--storm-minutes`, `--crash-point` or K14. + +A restart hits every task in flight. So the baselines come first, and the wave +never runs a baseline beside a fault cell: fault cells start only after the last +baseline rests. + +To prove that a journey fails before a fix, prepare a checkout of the fix's parent. +Then run the same wave against it with `--server-from`. The harness code stays the +same; every server start and restart runs the older build: + +```sh +git worktree add --detach ../paperclip-pre20363 3cc1032f6a^ && (cd ../paperclip-pre20363 && node scripts/agent-preflight.mjs) +node cli/node_modules/tsx/dist/cli.mjs tests/restart-chaos/main.ts --env local --adapters codex --workloads Q2 \ + --faults K1 --triggers answered --concurrency 1 --tasks 1 --seed 44 --server-from ../paperclip-pre20363 +``` + ## Code change in a real repository (R1, `--workspace-repo`) ```sh @@ -502,6 +621,8 @@ the observer, after a seeded delay: | `retry` | 0–2 s after an accepted Retry click, as an extra K1 after the planned fault | K8 | | `queued` | on the first poll whose run is still `queued` with no start | B1 | | `tool` | 15–60 s after `bench.sh` progress first appears (B5 only) | B6 | +| `asked` | 0–3 s after the saved question or tool review is first seen pending (Q1–Q3) | journeys | +| `answered` | 0–3 s after the board's answer or approval returned (Q1–Q3) | journeys | A trigger whose condition can no longer hold (comment posted at rest, run already started, no Retry offered) is recorded as `triggerMissed` in the @@ -524,6 +645,7 @@ re-arms the task's own triggered faults instead of using their offsets. | `universal.ts`, `universal-audit.ts` | U1–U4 on every task, and their audit over saved waves. | | `contention.ts` | L1 workload, its accounting oracle and the `--db-pressure` writers. | | `workloads.ts` | A1 brief document, B1 python CLI with tests, B5 long bench, C1 comment while running, M1 delegation, G1 approval across a restart, H1 budget hard stop across a restart, with their T-oracles. | +| `journeys.ts`, `journey-driver.ts`, `journey-oauth-app.ts`, `journey-matrix.ts` | Q2 and Q3 definitions and oracles, the board's side of each, Q3's loopback OAuth/MCP app, and the `--matrix` planner. | | `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. | diff --git a/tests/restart-chaos/config.ts b/tests/restart-chaos/config.ts index 9de23edefb1..5ede7a843a8 100644 --- a/tests/restart-chaos/config.ts +++ b/tests/restart-chaos/config.ts @@ -1,6 +1,8 @@ import { execFileSync } from "node:child_process"; import path from "node:path"; import { parseArgs } from "node:util"; +import { applyJourneyOptions } from "./journey-matrix.js"; +import { takesJourneyTriggers } from "./journeys.js"; import { STORAGE_FAULT_KINDS } from "./storage-chaos.js"; import { PROVIDER_FAULT_KINDS, PROVIDER_FAULT_WORKLOADS, isProviderFault, parseProviderDelays, parseProviderMessage, type ProviderFaultKind } from "./provider-faults.js"; import { @@ -107,13 +109,15 @@ export function parseCommand(argv: readonly string[], now = Date.now()): ParsedC "provider-fault-message": { type: "string" }, "provider-fault-delays": { type: "string" }, "db-pressure": { type: "string" }, + matrix: { type: "boolean" }, + "server-from": { 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"] 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"] 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, @@ -153,9 +157,9 @@ export function parseCommand(argv: readonly string[], now = Date.now()): ParsedC const options: WaveOptions = { env, adapters, - // The governance scenarios (G1, H1), the saved question (Q1) and the contention run (L1) are opt-in: H1 needs a wave of its own. - workloads: list(values.workloads, workloadChoices, "workloads", workloadChoices.filter(id => id !== "G1" && id !== "H1" && id !== "R1" && id !== "S1" && id !== "Q1" && id !== "L1" - && (id !== "M1" || adapters.length >= 2))), + // The governance scenarios (G1, H1), the journeys (Q1–Q3) and the contention run (L1) are opt-in: H1 needs a wave of its own. + workloads: list(values.workloads, workloadChoices, "workloads", workloadChoices.filter(id => id !== "G1" && id !== "H1" && id !== "R1" && id !== "S1" && id !== "L1" + && !takesJourneyTriggers(id) && (id !== "M1" || adapters.length >= 2))), // A crash-point wave schedules no fault; K1 is the restart its self-kill gets. faults: allFaults.filter((fault): fault is FaultKind => !isProviderFault(fault)), ...(providerFaults.length ? { providerFaults } : {}), @@ -213,6 +217,8 @@ export function parseCommand(argv: readonly string[], now = Date.now()): ParsedC if (crashPoint) throw new Error("--upgrade-from does not combine with --crash-point"); options.upgradeFrom = path.resolve(upgradeFrom); } + applyJourneyOptions(options, { matrix: values.matrix === true, serverFrom: values["server-from"], tasksGiven: values.tasks !== undefined, + triggersGiven: values.triggers !== undefined }); const storage = values.storage ?? "local"; const workspaceRepo = values["workspace-repo"]; if (options.workloads.includes("R1") !== (workspaceRepo !== undefined)) { diff --git a/tests/restart-chaos/journey-driver.ts b/tests/restart-chaos/journey-driver.ts new file mode 100644 index 00000000000..d11d111e968 --- /dev/null +++ b/tests/restart-chaos/journey-driver.ts @@ -0,0 +1,508 @@ +/** + * The board's side of the journey workloads (journeys.ts). Each poll the observer + * hands the driver its sample; the driver acts through the routes the board UI + * uses, only while the server is ready and no fault is in progress, and records + * what it did and saw on `record.journey`. + */ +import { readFileSync } from "node:fs"; +import { connectionGrants, toolActionRequests, toolConnections } from "../../packages/db/src/schema/index.js"; +import { ChaosApiError, type ChaosApi } from "./api.js"; +import { ACTIVE_RUN_STATUSES } from "./grader.js"; +import { startOAuthApp, TOOL_NAME, type OAuthApp } from "./journey-oauth-app.js"; +import { + ANSWER_TURN_WAIT_MS, answerMarker, isJourneyWorkload, journeyChoice, Q3_REVIEW_AGE_MS, q3ToolArgument, + type JourneyArtifacts, type JourneyRecord, type JourneyWorkload, +} from "./journeys.js"; +import type { ChaosLab } from "./seed.js"; +import type { Timeline } from "./timeline.js"; +import type { ScheduledFault, TaskPlan, TaskRecord, TaskSample } from "./types.js"; + +type Json = Record; + +export interface JourneyHost { + plan: TaskPlan; + api: ChaosApi; + lab: ChaosLab; + timeline: Timeline; + record: TaskRecord; + /** The wave database (Q3 moves the review's saved timestamps). */ + databaseUrl: () => string | null; +} + +/** What the observer knows each poll that the board's decision depends on. */ +export interface JourneyTick { + sample: TaskSample; + now: number; + /** Server ready and no fault in progress. */ + quiet: boolean; + /** The task's faults that have not fired yet. */ + pendingFaults: readonly ScheduledFault[]; +} + +const errorText = (error: unknown) => error instanceof Error ? error.message : String(error); +const httpStatus = (error: unknown) => error instanceof ChaosApiError ? error.status : null; +const hasActiveRun = (sample: TaskSample) => (sample.runs ?? []).some(run => ACTIVE_RUN_STATUSES.has(run.status)); + +/** Apps → Review rows: `{ actionRequests: [{ request: {...}, ... }] }`; each row's request carries id, issueId and status. */ +export function actionRequestRows(body: unknown): Json[] { + const rows = Array.isArray(body) ? body : (body as { actionRequests?: unknown } | null)?.actionRequests; + return (Array.isArray(rows) ? rows as Json[] : []).map(row => (row?.request && typeof row.request === "object" ? row.request : row) as Json); +} + +/** + * The board's answer as the question form sends it: the option whose id or label is the chosen + * code word. The agent wrote the card with its own ids, so the board reads them, as a person would. + */ +export function questionAnswer(interaction: Json, choice: string): { questionId: string; optionIds: string[] } | null { + const payload = (interaction.payload ?? {}) as Json; + const questions = [...(payload.questionSet?.questions ?? []), ...(payload.questions ?? [])] as Json[]; + for (const question of questions) { + const option = ((question.options ?? []) as Json[]).find(candidate => [candidate.id, candidate.label] + .some(value => typeof value === "string" && value.trim().toLowerCase() === choice)); + if (option && typeof question.id === "string") return { questionId: question.id, optionIds: [String(option.id)] }; + } + return null; +} + +/** The scripted Q3 agent: one connected-app call, then report the reviewed result. Adapted from tests/e2e/connection-reviews.spec.ts. */ +export function q3AgentScript(connectionId: string, title: string, serverBaseUrl: string): string { + // The wave's own loopback server. On a tailnet host, PAPERCLIP_API_URL names the machine's + // public URL, which reaches a different instance. + return ` +const base = ${JSON.stringify(serverBaseUrl)} + "/api"; +const headers = { authorization: "Bearer " + process.env.PAPERCLIP_API_KEY, "content-type": "application/json", "x-paperclip-run-id": process.env.PAPERCLIP_RUN_ID }; +const api = async (path, method = "GET", body) => { + const response = await fetch(base + path, { method, headers, body: body === undefined ? undefined : JSON.stringify(body) }); + if (!response.ok) throw new Error(method + " " + path + " " + response.status + ": " + await response.text()); + return await response.json(); +}; +let issueId = process.env.PAPERCLIP_TASK_ID; +try { +// This agent is hired for one task: find it by its own assignment. +const assigned = await api("/companies/" + process.env.PAPERCLIP_COMPANY_ID + "/issues?assigneeAgentId=" + process.env.PAPERCLIP_AGENT_ID); +issueId ??= (Array.isArray(assigned) ? assigned : assigned.issues ?? []).find(item => item.title?.includes(${JSON.stringify(title)}))?.id; +if (!issueId) throw new Error("No assigned task names " + ${JSON.stringify(title)}); +const review = (await api("/issues/" + issueId + "/interactions")).find(item => item.payload?.toolAction); +if (review?.status === "pending") { console.log("Waiting for the saved review; no retry."); process.exit(0); } +if (review) { + const result = review.result?.toolAction; + if (review.status === "accepted" && (!result || ["approved", "executing"].includes(result.status))) { console.log("Approved; waiting for the stored result."); process.exit(0); } + const executed = review.status === "accepted" && result?.status === "executed"; + await api("/issues/" + issueId + "/comments", "POST", { body: (executed ? "Q3 result: " : "Q3 review ended " + review.status + ": ") + JSON.stringify(result ?? null) }); + if (executed) await api("/issues/" + issueId, "PATCH", { status: "done" }); + process.exit(0); +} +const session = await api("/tool-gateway/sessions", "POST", { runId: process.env.PAPERCLIP_RUN_ID, ttlMs: 60000 }); +const gatewayHeaders = { "x-paperclip-tool-gateway-token": session.token, "content-type": "application/json" }; +const tools = await (await fetch(base + "/tool-gateway/tools", { headers: gatewayHeaders })).json(); +const tool = tools.find(t => t.connectionId === ${JSON.stringify(connectionId)}); +if (!tool) throw new Error("The chaos app tool is not available to this agent"); +const call = await fetch(base + "/tool-gateway/tools/call", { method: "POST", headers: gatewayHeaders, body: JSON.stringify({ tool: tool.name, parameters: { title: ${JSON.stringify(title)} } }) }); +const outcome = await call.json(); +if (!call.ok) { + if (!JSON.stringify(outcome).includes("approval_required")) throw new Error(JSON.stringify(outcome)); + await api("/issues/" + issueId, "PATCH", { status: "in_review" }); + console.log("Waiting for the board's review. No retry."); + process.exit(0); +} +await api("/issues/" + issueId + "/comments", "POST", { body: "Q3 ran without review: " + JSON.stringify(outcome) }); +} catch (error) { + // Leave the failure on the task, where the evidence bundle collects it. + console.error(error); + if (issueId) await api("/issues/" + issueId + "/comments", "POST", { body: "Q3 agent error: " + String(error?.stack ?? error).slice(0, 2000) }).catch(() => {}); + process.exit(1); +} +`; +} + +/** Drivers with a running Q3 app; the wave closes any still open at teardown. */ +const openDrivers = new Set(); +export async function closeJourneyApps(): Promise { + await Promise.all([...openDrivers].map(driver => driver.close())); +} + +export class JourneyDriver { + private app: OAuthApp | null = null; + /** Q2: ready time accumulated since the answer, for the answer-turn wait. */ + private quietSinceAnswerMs = 0; + private lastTickAt: number | null = null; + + constructor(readonly workload: JourneyWorkload, private readonly host: JourneyHost) { + const own = host.plan.trigger && host.plan.trigger !== "time" ? host.plan.trigger : null; + this.state.faultTrigger = own; + } + + get state(): JourneyRecord { return this.host.record.journey ??= {}; } + + /** Q3: the scripted agent that calls the connected app; others use the seeded agent. */ + get assigneeAgentId(): string | null { return this.state.app?.agentId ?? null; } + + /** Q3: start the harness's OAuth app, connect it as the board, and hire the scripted agent. */ + async prepare(): Promise { + if (this.workload !== "Q3" || this.state.app) return; + const { api, lab, plan, timeline } = this.host; + const app = this.app = await startOAuthApp(); + openDrivers.add(this); + const state: NonNullable = this.state.app = { connectionId: null, agentId: null, receipt: app.receipt, setupError: null }; + try { + const connected = await api.post(`/api/companies/${lab.companyId}/tools/apps/connect`, + { link: app.mcpUrl, name: `Chaos pages ${plan.nonce}`, grantKind: "organization" }, 30_000); + state.connectionId = String(connected.connectionId ?? connected.connection?.id); + const startUrl = connected.auth?.startUrl; + if (typeof startUrl !== "string") throw new Error(`connect returned no OAuth start URL: ${JSON.stringify(connected).slice(0, 300)}`); + // The browser visits the app, consents, and is sent back to Paperclip's callback. + const authorize = await fetch(startUrl, { redirect: "manual" }); + const location = authorize.headers.get("location"); + if (!location) throw new Error(`authorize returned ${authorize.status} without a redirect`); + const back = new URL(location); + const callback = await fetch(`${api.baseUrl}${back.pathname}${back.search}`, + { headers: { origin: api.baseUrl, accept: "application/json" }, redirect: "manual", signal: AbortSignal.timeout(30_000) }); + if (callback.status >= 400) throw new Error(`OAuth callback returned ${callback.status}: ${(await callback.text()).slice(0, 300)}`); + const { catalog } = await api.get<{ catalog: Json[] }>(`/api/tool-connections/${state.connectionId}/catalog`); + const entry = (catalog ?? []).find(row => row.toolName === TOOL_NAME || String(row.upstreamToolName ?? "").endsWith(TOOL_NAME)); + if (!entry) throw new Error(`catalog has no ${TOOL_NAME}: ${JSON.stringify(catalog).slice(0, 300)}`); + const agent = await api.post(`/api/companies/${lab.companyId}/agents`, { + name: `Chaos pages agent ${plan.nonce}`, role: "engineer", title: "Connected-app journey agent", + capabilities: "Creates one page with the chaos app after review.", adapterType: "process", + adapterConfig: { command: process.execPath, args: ["--input-type=module", "-e", q3AgentScript(state.connectionId, q3ToolArgument(plan.nonce), api.baseUrl)] }, + }); + state.agentId = String(agent.id); + // A scripted process run records usage with no price, which the company's default budget + // policy blocks (pausing the whole company). The operator allows unpriced usage, as the + // budget page offers; the amount and hard stop stay in force for priced model usage. + await api.post(`/api/companies/${lab.companyId}/budgets/policies`, + { scopeType: "company", scopeId: lab.companyId, unpricedUsagePolicy: "allow" }); + // Apps → finish: this agent may use the app, and the write tool asks first. + await api.post(`/api/companies/${lab.companyId}/tools/apps/${state.connectionId}/finish`, { + enabledCatalogEntryIds: [entry.id], askFirstCatalogEntryIds: [entry.id], access: { agentIds: [state.agentId] } }); + timeline.add("journey_app_connected", { taskId: plan.taskId, connectionId: state.connectionId, agentId: state.agentId, origin: app.origin }); + } catch (error) { + state.setupError = errorText(error); + timeline.add("journey_app_setup_failed", { taskId: plan.taskId, error: state.setupError }); + // No issue exists yet, so artifact collection (which closes the app) never runs for this task. + await this.close(); + throw new Error(`Q3 app setup failed: ${state.setupError}`); + } + } + + armedAt(trigger: "asked" | "answered"): number | null { + return trigger === "asked" ? this.state.requestSeenAt ?? null : this.state.answer?.doneAt ?? null; + } + + /** The task waits on the board: not a rest, not a stall, no Retry. */ + awaitingBoard(): boolean { + const state = this.state; + if (!state.requestSeenAt) return false; + if (this.workload === "Q2") return !state.blockerResolved && (!state.answer || state.answer.error === null); + return !state.answer; + } + + /** One poll. Returns true when the board acted (the observer then needs fresh samples before rest). */ + async step(tick: JourneyTick): Promise { + const { sample, now, quiet } = tick; + const previous = this.lastTickAt; + this.lastTickAt = now; + if (quiet && previous !== null && this.state.answer) this.quietSinceAnswerMs += now - previous; + if (!sample.reachable || !this.host.record.issueId) return false; + if (this.workload === "Q3" && this.state.app?.setupError) return false; + if (!this.state.requestSeenAt && !(await this.findRequest(now))) return false; + // The board answers once every fault the task planned before the answer has fired. + const answerMayGo = quiet && tick.pendingFaults.every(fault => fault.trigger === "answered"); + if (this.workload === "Q2" && !this.state.blocker) { + if (!quiet || hasActiveRun(sample)) return false; + await this.addPrerequisite(now); + return true; + } + if (!this.state.answer) { + if (!answerMayGo) return false; + if (this.workload === "Q2" && this.state.blocker?.error) return false; + if (this.workload === "Q3") await this.ageAndRefresh(now); + await this.respond(now); + return true; + } + if (this.workload !== "Q2" || this.state.answer.error !== null) return false; + if (!this.state.answerTurn) { + if (!quiet) return false; + if (await this.observeAnswerTurn(now, sample)) return false; + if (this.quietSinceAnswerMs < ANSWER_TURN_WAIT_MS) return false; + this.state.answerTurn = { missedAt: now }; + this.host.timeline.add("journey_answer_turn_missed", { taskId: this.host.plan.taskId, waitedMs: this.quietSinceAnswerMs }); + } + if (!this.state.blockerResolved && quiet && tick.pendingFaults.length === 0 && !hasActiveRun(sample)) { + await this.finishPrerequisite(now); + return true; + } + return false; + } + + private async findRequest(now: number): Promise { + const { api, lab, record, timeline, plan } = this.host; + let id: string | null = null; + if (this.workload === "Q3") { + const rows = await api.get(`/api/companies/${lab.companyId}/tools/action-requests?status=pending`, 5_000).catch(() => null); + id = actionRequestRows(rows).find(row => row.issueId === record.issueId)?.id ?? null; + } else { + const rows = await api.get(`/api/issues/${record.issueId}/interactions`, 5_000).catch(() => null); + id = (rows ?? []).find(row => row.kind === "ask_user_questions" && row.status === "pending")?.id ?? null; + } + if (!id) return false; + this.state.requestSeenAt = now; + this.state.requestId = String(id); + timeline.add("journey_request_pending", { taskId: plan.taskId, workload: this.workload, requestId: id }); + return true; + } + + /** Q2: the board adds an unfinished, unassigned prerequisite and marks the task blocked by it. */ + private async addPrerequisite(now: number) { + const { api, lab, record, plan, timeline } = this.host; + const blocker: NonNullable = { issueId: null, at: now, error: null }; + try { + const created = await api.post(`/api/companies/${lab.companyId}/issues`, + { title: `Chaos Q2 prerequisite ${plan.nonce}`, status: "backlog", priority: "medium" }); + blocker.issueId = String(created.id); + await api.patch(`/api/issues/${record.issueId}`, { status: "blocked", blockedByIssueIds: [blocker.issueId] }); + } catch (error) { + blocker.error = errorText(error); + } + this.state.blocker = blocker; + timeline.add("journey_prerequisite_added", { taskId: plan.taskId, ...blocker }); + } + + /** Q3: the review waits more than an hour. Move its saved timestamps and the access token's expiry back. */ + private async ageAndRefresh(now: number) { + const { api, plan, timeline } = this.host; + if (!this.state.timeShift) { + const shift: NonNullable = { at: now, ms: Q3_REVIEW_AGE_MS, requests: 0, grants: 0, error: null }; + try { Object.assign(shift, await shiftReviewClock(this.host.databaseUrl(), this.state.requestId!, this.state.app!.connectionId!, Q3_REVIEW_AGE_MS)); } + catch (error) { shift.error = errorText(error); } + this.state.timeShift = shift; + timeline.add("journey_review_aged", { taskId: plan.taskId, ...shift }); + } + if (!this.state.refresh) { + // Apps → connection → Check connection: a routine use that renews the expiring token. + const refresh: NonNullable = { at: Date.now(), httpStatus: null, error: null }; + try { await api.post(`/api/tool-connections/${this.state.app!.connectionId}/health-check`, {}, 30_000); refresh.httpStatus = 200; } + catch (error) { refresh.httpStatus = httpStatus(error); refresh.error = errorText(error); } + this.state.refresh = refresh; + timeline.add("journey_connection_checked", { taskId: plan.taskId, ...refresh, refreshes: this.app?.record.refreshes.length ?? null }); + } + } + + private async respond(now: number) { + const { api, lab, record, plan, timeline } = this.host; + const answer: NonNullable = { at: now, doneAt: now, httpStatus: null, error: null, status: null, + pendingWhenAnswered: false, faultsBefore: record.faultIds.length }; + try { + const request = await this.currentRequest(); + const status = request?.status ?? null; + answer.pendingWhenAnswered = status === "pending"; + if (status !== "pending") throw new Error(`the saved request is no longer pending (status ${status ?? "missing"})`); + const choice = this.workload === "Q3" ? null : questionAnswer(request!, journeyChoice(plan.nonce)); + if (this.workload !== "Q3" && !choice) throw new Error(`the saved question offers no option "${journeyChoice(plan.nonce)}"`); + answer.at = Date.now(); + const response = this.workload === "Q3" + ? await api.post(`/api/tool-gateway/action-requests/${this.state.requestId}/approve`, { companyId: lab.companyId }, 60_000) + : await api.post(`/api/issues/${record.issueId}/interactions/${this.state.requestId}/respond`, { answers: [choice] }, 30_000); + answer.httpStatus = 200; + answer.status = typeof response?.status === "string" ? response.status : null; + } catch (error) { + answer.httpStatus = httpStatus(error); + answer.error = errorText(error); + } + answer.doneAt = Math.max(Date.now(), now); + this.state.answer = answer; + timeline.add("journey_board_responded", { taskId: plan.taskId, workload: this.workload, requestId: this.state.requestId, ...answer }); + } + + /** The saved request as the board sees it now (Q3: listed only while pending). */ + private async currentRequest(): Promise { + const { api, lab, record } = this.host; + if (this.workload === "Q3") { + const rows = await api.get(`/api/companies/${lab.companyId}/tools/action-requests?status=pending`, 10_000); + return actionRequestRows(rows).some(row => row.id === this.state.requestId) ? { status: "pending" } : { status: "not pending" }; + } + const rows = await api.get(`/api/issues/${record.issueId}/interactions`, 10_000); + return rows.find(row => row.id === this.state.requestId) ?? null; + } + + /** Q2: the agent's reply to the answer, while the prerequisite is still unfinished. */ + private async observeAnswerTurn(now: number, sample: TaskSample): Promise { + const { api, record, plan, timeline } = this.host; + if (hasActiveRun(sample)) return false; + const comments = await api.get(`/api/issues/${record.issueId}/comments`, 5_000).catch(() => null); + const since = this.state.answer!.at; + const marker = answerMarker(plan.nonce); + const reply = (comments ?? []).find(comment => comment.authorAgentId && String(comment.body ?? "").includes(marker) + && Date.parse(comment.createdAt) >= since - 1_000); + // A bounded answer turn may report the answer in its accepted run result instead of a comment. + let at = reply ? Date.parse(reply.createdAt) : null; + if (at === null) { + for (const run of (sample.runs ?? []).filter(run => run.createdAt && Date.parse(run.createdAt) >= since - 1_000 && run.finishedAt)) { + const events = await api.get(`/api/heartbeat-runs/${run.id}/events?limit=500`, 5_000).catch(() => null); + // Only an accepted result counts; a rejected one still carries the proposed body. + if ((events ?? []).some(event => event.eventType === "run.result.accepted" && JSON.stringify(event.payload ?? {}).includes(marker))) { + at = Date.parse(run.finishedAt!); + break; + } + } + } + if (at === null) return false; + const issue = await api.get(`/api/issues/${record.issueId}`, 5_000).catch(() => null); + const blockerId = this.state.blocker?.issueId; + const blockedBy = Array.isArray(issue?.blockedBy) ? issue.blockedBy as Json[] : []; + this.state.answerTurn = { at, issueStatus: issue?.status ?? null, + blockerKept: Boolean(blockerId) && (blockedBy.some(row => row.id === blockerId) || (issue?.blockedByIssueIds ?? []).includes(blockerId)) }; + timeline.add("journey_answer_turn", { taskId: plan.taskId, observedAt: now, ...this.state.answerTurn }); + return true; + } + + private async finishPrerequisite(now: number) { + const { api, plan, timeline } = this.host; + const resolved: NonNullable = { at: now, error: null }; + try { await api.patch(`/api/issues/${this.state.blocker!.issueId}`, { status: "done" }); } + catch (error) { resolved.error = errorText(error); } + this.state.blockerResolved = resolved; + timeline.add("journey_prerequisite_finished", { taskId: plan.taskId, ...resolved }); + } + + async collect(issueId: string): Promise { + const { api, lab } = this.host; + const [issue, interactions, comments, runs] = await Promise.all([ + api.get(`/api/issues/${issueId}`).catch(() => null), + api.get(`/api/issues/${issueId}/interactions`).catch(() => []), + api.get(`/api/issues/${issueId}/comments`).catch(() => []), + api.get(`/api/issues/${issueId}/runs`).catch(() => []), + ]); + const interruption = issue?.interruption ? { state: issue.interruption.state ?? null, canRetry: issue.interruption.canRetry === true } : null; + const artifacts: JourneyArtifacts = { + record: this.state, + questions: (interactions ?? []).filter(row => row.kind === "ask_user_questions") + .map(row => ({ id: String(row.id), status: String(row.status), createdAt: row.createdAt ?? null })), + agentComments: (comments ?? []).filter(comment => comment.authorAgentId) + .map(comment => ({ createdAt: Date.parse(comment.createdAt), body: String(comment.body ?? "") })).sort((a, b) => a.createdAt - b.createdAt), + runErrorCodes: (runs ?? []).map(run => run.errorCode).filter((code): code is string => typeof code === "string"), + interruption, + }; + if (this.workload === "Q3") { + const rows = await Promise.all(["pending", "approved", "executing", "rejected", "expired", "cancelled", "executed", "failed"] + .map(status => api.get(`/api/companies/${lab.companyId}/tools/action-requests?status=${status}`).catch(() => []))); + const requests = new Map(); + for (const page of rows) for (const row of actionRequestRows(page)) { + if (row.issueId === issueId) requests.set(String(row.id), String(row.status)); + } + artifacts.actionRequests = [...requests].map(([id, status]) => ({ id, status })); + // The scripted agent's own output for every run that did not succeed. + for (const run of (runs ?? []).filter(row => row.status !== "succeeded").slice(-5)) { + const id = String(run.runId ?? run.id); + const log = await api.get(`/api/heartbeat-runs/${id}/log?limitBytes=8000`).catch(error => ({ error: errorText(error) })); + (artifacts.runLogs ??= {})[id] = JSON.stringify(log).slice(-4000); + } + artifacts.provider = this.app ? { toolCalls: [...this.app.record.toolCalls], refreshes: [...this.app.record.refreshes] } : { toolCalls: [], refreshes: [] }; + } + return artifacts; + } + + async close(): Promise { openDrivers.delete(this); await this.app?.close().catch(() => undefined); this.app = null; } +} + +/** + * Q3: the honest way to let more than an hour pass for one review in an isolated wave database. + * The pending request's created/updated/deadline timestamps and the OAuth grant's access-token + * expiry (and last refresh) move back by `ms`, as GRACE_SHIFT_MS does for the invariant pass. + * Nothing else changes; the product's own code decides what that elapsed time means. + */ +export async function shiftReviewClock(databaseUrl: string | null, actionRequestId: string, connectionId: string, ms: number) { + if (!databaseUrl) throw new Error("the wave database URL is unknown"); + const { createDb } = await import("../../packages/db/src/index.js"); + const db = createDb(databaseUrl); + const client = (db as unknown as { $client: SqlClient & { end(options?: { timeout?: number }): Promise } }).$client; + try { + return await shiftReviewClockWith(client, reviewClockNames(), actionRequestId, connectionId, ms); + } finally { + await client.end({ timeout: 5 }).catch(() => undefined); + } +} + +export interface SqlClient { unsafe(query: string, params?: unknown[]): Promise } + +type SchemaTable = Record; +const tableName = (table: SchemaTable) => String(table[Symbol.for("drizzle:Name") as unknown as string]); +const columnName = (table: SchemaTable, key: string) => { + const column = table[key] as { name?: unknown } | undefined; + if (typeof column?.name !== "string") throw new Error(`schema table ${tableName(table)} has no column ${key}`); + return column.name; +}; + +/** Table and column names from the Drizzle schema, so a rename fails here, not as a vague live-wave ageing error. */ +export function reviewClockNames() { + const requests = toolActionRequests as unknown as SchemaTable, grants = connectionGrants as unknown as SchemaTable; + const connections = toolConnections as unknown as SchemaTable; + return { + requests: tableName(requests), requestId: columnName(requests, "id"), requestStatus: columnName(requests, "status"), + requestTimes: ["createdAt", "updatedAt", "expiresAt"].map(key => columnName(requests, key)), + grants: tableName(grants), grantId: columnName(grants, "id"), grantConnection: columnName(grants, "connectionId"), + grantTenant: columnName(grants, "providerTenant"), + connections: tableName(connections), connectionId: columnName(connections, "id"), + connectionConfig: columnName(connections, "config"), connectionTransport: columnName(connections, "transportConfig"), + }; +} + +export async function shiftReviewClockWith(client: SqlClient, names: ReturnType, actionRequestId: string, + connectionId: string, ms: number) { + { + const n = names; + const interval = `${Math.round(ms / 1000)} seconds`; + const requests = await client.unsafe(`UPDATE ${n.requests} SET ${n.requestTimes.map(column => `${column} = ${column} - $2::interval`).join(", ")} + WHERE ${n.requestId} = $1 AND ${n.requestStatus} = 'pending' RETURNING ${n.requestId}`, [actionRequestId, interval]); + const grants = await client.unsafe(`SELECT ${n.grantId} AS id, ${n.grantTenant} AS provider_tenant FROM ${n.grants} WHERE ${n.grantConnection} = $1`, [connectionId]); + let shiftedGrants = 0; + for (const grant of grants) { + const tenant = (grant.provider_tenant ?? {}) as Json; + const oauth = { ...(tenant.oauth ?? {}) } as Json; + let changed = false; + for (const key of ["accessTokenExpiresAt", "refreshedAt", "expiresAt"]) { + const at = typeof oauth[key] === "string" ? Date.parse(oauth[key]) : Number.NaN; + if (Number.isFinite(at)) { oauth[key] = new Date(at - ms).toISOString(); changed = true; } + } + if (!changed) continue; + await client.unsafe(`UPDATE ${n.grants} SET ${n.grantTenant} = $2::jsonb WHERE ${n.grantId} = $1`, [grant.id, JSON.stringify({ ...tenant, oauth })]); + shiftedGrants += 1; + } + // A shared (organization) connection keeps its token expiry on the connection itself. + const connections = await client.unsafe(`SELECT ${n.connectionId} AS id, ${n.connectionConfig} AS config, ${n.connectionTransport} AS transport_config + FROM ${n.connections} WHERE ${n.connectionId} = $1`, [connectionId]); + for (const connection of connections) { + const shifted: Record = {}; + for (const column of ["config", "transport_config"]) { + const value = connection[column] as Json | null; + const oauth = value?.oauth as Json | undefined; + if (!oauth) continue; + const next = { ...oauth }; + let changed = false; + for (const key of ["expiresAt", "refreshedAt", "accessTokenExpiresAt"]) { + const at = typeof next[key] === "string" ? Date.parse(next[key]) : Number.NaN; + if (Number.isFinite(at)) { next[key] = new Date(at - ms).toISOString(); changed = true; } + } + if (changed) shifted[column] = { ...value, oauth: next }; + } + if (!Object.keys(shifted).length) continue; + await client.unsafe(`UPDATE ${n.connections} SET ${n.connectionConfig} = COALESCE($2::jsonb, ${n.connectionConfig}), + ${n.connectionTransport} = COALESCE($3::jsonb, ${n.connectionTransport}) WHERE ${n.connectionId} = $1`, + [connection.id, shifted.config ? JSON.stringify(shifted.config) : null, shifted.transport_config ? JSON.stringify(shifted.transport_config) : null]); + shiftedGrants += 1; + } + return { requests: requests.length, grants: shiftedGrants }; + } +} + +/** The wave database URL from the instance config (as main.ts reads it for the crash-point suite). */ +export function databaseUrlFromConfig(configPath: string | undefined): string | null { + if (!configPath) return null; + try { + return (JSON.parse(readFileSync(configPath, "utf8")) as { database?: { connectionString?: string } }).database?.connectionString ?? null; + } catch { return null; } +} + +export function journeyDriverFor(host: JourneyHost): JourneyDriver | null { + return isJourneyWorkload(host.plan.workload) ? new JourneyDriver(host.plan.workload, host) : null; +} diff --git a/tests/restart-chaos/journey-matrix.ts b/tests/restart-chaos/journey-matrix.ts new file mode 100644 index 00000000000..7ea15b1958a --- /dev/null +++ b/tests/restart-chaos/journey-matrix.ts @@ -0,0 +1,86 @@ +/** + * `--matrix`: one task per cell of workload × adapter × (no fault, or each + * `--faults` kind at each of the workload's own fault boundaries). A journey + * workload (Q1–Q3) gets a K0 baseline plus every fault at `asked` and at + * `answered`; G1 at `approval`, C1 at `comment` and `result`, and so on + * (journeys.ts WORKLOAD_TRIGGERS). The cells run in a fixed order so a seed + * always means the same matrix; `--tasks` is the cell count. + * + * A restart hits every task in flight, so the K0 baselines run first and never + * share the server with a fault cell (`matrixMayLaunch`). + */ +import path from "node:path"; +import { WORKLOAD_TRIGGERS } from "./journeys.js"; +import { withProviderFaultRules } from "./provider-faults.js"; +import { createRng, k4HoldRange, TRIGGER_DELAY_RANGE_MS } from "./scheduler.js"; +import { workloadDefinitions } from "./workloads.js"; +import type { AdapterId, TaskFaultKind, TaskPlan, TriggerId, WaveOptions, WavePlan, WorkloadId } from "./types.js"; + +export interface MatrixCell { workload: WorkloadId; adapter: AdapterId; fault: TaskFaultKind; trigger: TriggerId } + +/** G1 and H1 grade the board's decision after a restart, so a no-fault run cannot pass them: they get no baseline. */ +export const NO_BASELINE_WORKLOADS: readonly WorkloadId[] = ["G1", "H1"]; + +export function matrixCells(options: Pick): MatrixCell[] { + const baselines: MatrixCell[] = []; + const faulted: MatrixCell[] = []; + for (const workload of options.workloads) for (const adapter of options.adapters) { + // The baseline: the same journey with no fault, so a fault cell's failure is attributable. + if (!NO_BASELINE_WORKLOADS.includes(workload)) baselines.push({ workload, adapter, fault: "K0", trigger: "time" }); + for (const fault of options.faults) for (const trigger of WORKLOAD_TRIGGERS[workload]) faulted.push({ workload, adapter, fault, trigger }); + // A provider message fault is armed per task in the provider shim; it has no trigger of its own. + for (const fault of options.providerFaults ?? []) faulted.push({ workload, adapter, fault, trigger: "time" }); + } + return [...baselines, ...faulted]; +} + +/** A baseline launches only beside other baselines, and a fault cell only once no baseline is in flight. */ +export function matrixMayLaunch(next: Pick, inFlight: ReadonlyArray>): boolean { + const baseline = next.fault === "K0"; + return inFlight.every(task => (task.fault === "K0") === baseline); +} + +export function planMatrixTasks(options: Pick): TaskPlan[] { + const rng = createRng((options.seed ^ 0x6a6d7478) >>> 0); + return matrixCells(options).map((cell, index) => { + const definition = workloadDefinitions[cell.workload]; + const [fireMin, fireMax] = definition.fireWindowMs; + const drawn = rng.int(fireMin, fireMax); + return { + taskId: `t${String(index + 1).padStart(2, "0")}`, + index, + workload: cell.workload, + adapter: cell.adapter, + fault: cell.fault, + ...(cell.trigger === "time" ? {} : { trigger: cell.trigger, triggerDelayMs: rng.int(...TRIGGER_DELAY_RANGE_MS[cell.trigger]) }), + // `--fault-after-seconds` pins time-fired restarts, as it does outside the matrix. + fireAfterMs: cell.fault === "K0" || cell.fault.startsWith("PV") ? 0 : cell.trigger === "time" ? options.faultAfterMs ?? drawn : drawn, + holdMs: cell.fault === "K4" ? rng.int(...k4HoldRange(options.k4HoldMaxMs)) : 0, + followUpAfterMs: definition.followUpWindowMs ? rng.int(definition.followUpWindowMs[0], definition.followUpWindowMs[1]) : null, + nonce: rng.hex(8), + }; + }); +} + +/** `--matrix` and `--server-from`, validated against the rest of the parsed options. */ +export function applyJourneyOptions(options: WaveOptions, input: { matrix: boolean; serverFrom: string | undefined; tasksGiven: boolean; triggersGiven: boolean }) { + if (input.matrix) { + if (options.stormMinutes || options.crashPoint) throw new Error("--matrix plans its own cells; it does not combine with --storm-minutes or --crash-point"); + if (input.tasksGiven) throw new Error("--matrix runs one task per cell; do not pass --tasks"); + if (input.triggersGiven) throw new Error("--matrix uses each workload's own triggers; do not pass --triggers"); + if (options.faults.includes("K14")) throw new Error("--matrix does not cross K14 (an upgrade happens once per wave)"); + options.matrix = true; + options.tasks = matrixCells(options).length; + } + if (input.serverFrom !== undefined) { + if (options.env !== "local") throw new Error("--server-from runs only with --env local"); + if (options.faults.includes("K14")) throw new Error("--server-from runs every server from one checkout; use --upgrade-from for K14"); + options.serverFrom = path.resolve(input.serverFrom); + } +} + +export function planMatrixWave(waveId: string, options: WaveOptions, createdAt = new Date()): WavePlan { + const tasks = withProviderFaultRules(planMatrixTasks(options), options); + return { schema: "paperclip.restart-chaos.wave-plan/v1", waveId, seed: options.seed, createdAt: createdAt.toISOString(), + options: { ...options, tasks: tasks.length }, tasks }; +} diff --git a/tests/restart-chaos/journey-oauth-app.ts b/tests/restart-chaos/journey-oauth-app.ts new file mode 100644 index 00000000000..2f238302854 --- /dev/null +++ b/tests/restart-chaos/journey-oauth-app.ts @@ -0,0 +1,157 @@ +/** + * Q3: a harness-owned OAuth 2.1 + MCP app on loopback, so no real third-party + * account is involved. It supports discovery, dynamic client registration, PKCE + * authorization codes and rotating refresh tokens, and serves one write tool, + * `create_page`. It records every token exchange and tool call for the oracle. + * + * Access tokens last one hour, a common provider default. A shared connection + * renews its token only once it has (nearly) expired, so routine use during the + * test does not renew it. Once the harness moves the saved expiry back 61 minutes, + * the next use (the board's connection check) refreshes it. + */ +import { createHash, randomUUID } from "node:crypto"; +import { createServer, type IncomingMessage, type ServerResponse } from "node:http"; +import type { AddressInfo } from "node:net"; + +export const TOOL_NAME = "create_page"; +export const ACCESS_TOKEN_SECONDS = 60 * 60; + +export interface OAuthAppRecord { + registrations: number; + codeExchanges: number; + /** Authorization codes refused: unknown, reused, or a missing or wrong PKCE verifier. */ + rejectedCodes: number; + /** Epoch ms of each successful refresh-token exchange. */ + refreshes: number[]; + rejectedRefreshes: number; + toolCalls: Array<{ at: number; title: unknown }>; +} + +export interface OAuthApp { + origin: string; + mcpUrl: string; + /** Only the provider's tool result carries this; an agent that reports it saw the real result. */ + receipt: string; + record: OAuthAppRecord; + close(): Promise; +} + +/** RFC 7636 S256: BASE64URL(SHA256(verifier)). */ +export function pkceChallenge(verifier: string): string { + return createHash("sha256").update(verifier).digest("base64url"); +} + +async function readBody(req: IncomingMessage): Promise { + let body = ""; + for await (const chunk of req) body += chunk; + return body; +} + +export async function startOAuthApp(options: { host?: string } = {}): Promise { + const host = options.host ?? "127.0.0.1"; + const receipt = `receipt-${randomUUID()}`; + const record: OAuthAppRecord = { registrations: 0, codeExchanges: 0, rejectedCodes: 0, refreshes: [], rejectedRefreshes: 0, toolCalls: [] }; + const codes = new Map(); + let generation = 0; + let accessToken = ""; + let refreshToken = ""; + let origin = ""; + const issue = () => { + generation += 1; + accessToken = `chaos-access-${generation}-${randomUUID()}`; + refreshToken = `chaos-refresh-${generation}-${randomUUID()}`; + return { access_token: accessToken, refresh_token: refreshToken, expires_in: ACCESS_TOKEN_SECONDS, + token_type: "Bearer", scope: "pages.write offline_access" }; + }; + const handle = async (req: IncomingMessage, res: ServerResponse) => { + const body = await readBody(req); + const url = new URL(req.url ?? "/", origin); + const json = (value: unknown, status = 200) => { + res.writeHead(status, { "content-type": "application/json" }); + res.end(JSON.stringify(value)); + }; + if (url.pathname.startsWith("/.well-known/oauth-protected-resource")) { + return json({ resource: `${origin}/mcp`, authorization_servers: [origin], scopes_supported: ["pages.write", "offline_access"] }); + } + if (url.pathname === "/.well-known/oauth-authorization-server" || url.pathname === "/.well-known/openid-configuration") { + return json({ issuer: origin, authorization_endpoint: `${origin}/authorize`, token_endpoint: `${origin}/token`, + registration_endpoint: `${origin}/register`, code_challenge_methods_supported: ["S256"], + token_endpoint_auth_methods_supported: ["none"], response_types_supported: ["code"], + scopes_supported: ["pages.write", "offline_access"], grant_types_supported: ["authorization_code", "refresh_token"] }); + } + if (url.pathname === "/register" && req.method === "POST") { + record.registrations += 1; + return json({ ...(JSON.parse(body || "{}") as Record), client_id: "chaos-client", token_endpoint_auth_method: "none" }, 201); + } + if (url.pathname === "/authorize") { + // The person consents at once: send the browser back with a code. + const redirectUri = url.searchParams.get("redirect_uri"); + const challenge = url.searchParams.get("code_challenge"); + if (!redirectUri || !challenge || url.searchParams.get("code_challenge_method") !== "S256") return json({ error: "invalid_request" }, 400); + const code = `chaos-code-${randomUUID()}`; + codes.set(code, { redirectUri, challenge }); + const back = new URL(redirectUri); + back.searchParams.set("code", code); + back.searchParams.set("state", url.searchParams.get("state") ?? ""); + back.searchParams.set("iss", origin); + res.writeHead(302, { location: back.toString() }); + res.end(); + return; + } + if (url.pathname === "/token" && req.method === "POST") { + const form = new URLSearchParams(body); + if (form.get("grant_type") === "refresh_token") { + if (form.get("refresh_token") !== refreshToken) { record.rejectedRefreshes += 1; return json({ error: "invalid_grant" }, 400); } + record.refreshes.push(Date.now()); + return json(issue()); + } + if (form.get("grant_type") === "authorization_code") { + const code = form.get("code") ?? ""; + const saved = codes.get(code); + codes.delete(code); + // PKCE S256: the verifier must hash to the challenge sent with the authorization request. + const verifier = form.get("code_verifier") ?? ""; + if (!saved || !verifier || pkceChallenge(verifier) !== saved.challenge + || (form.has("redirect_uri") && form.get("redirect_uri") !== saved.redirectUri)) { + record.rejectedCodes += 1; + return json({ error: "invalid_grant" }, 400); + } + record.codeExchanges += 1; + return json(issue()); + } + return json({ error: "unsupported_grant_type" }, 400); + } + if (url.pathname === "/mcp") { + if (!accessToken || req.headers.authorization !== `Bearer ${accessToken}`) { + res.setHeader("www-authenticate", `Bearer resource_metadata="${origin}/.well-known/oauth-protected-resource/mcp"`); + return json({ error: "invalid_token" }, 401); + } + if (req.method !== "POST") return json({}, 405); + const rpc = JSON.parse(body || "{}") as { id?: unknown; method?: string; params?: { arguments?: { title?: unknown } } }; + if (rpc.method?.startsWith("notifications/")) { res.writeHead(202).end(); return; } + if (rpc.method === "initialize") { + return json({ jsonrpc: "2.0", id: rpc.id, result: { protocolVersion: "2025-06-18", capabilities: { tools: {} }, + serverInfo: { name: "chaos-pages", version: "1" } } }); + } + if (rpc.method === "tools/list") { + return json({ jsonrpc: "2.0", id: rpc.id, result: { tools: [{ name: TOOL_NAME, description: "Create a page in the chaos workspace", + inputSchema: { type: "object", properties: { title: { type: "string" } }, required: ["title"] }, + annotations: { readOnlyHint: false, destructiveHint: false } }] } }); + } + if (rpc.method === "tools/call") { + const title = rpc.params?.arguments?.title; + record.toolCalls.push({ at: Date.now(), title }); + return json({ jsonrpc: "2.0", id: rpc.id, result: { content: [{ type: "text", text: `Created page ${String(title)}. ${receipt}` }] } }); + } + return json({ jsonrpc: "2.0", id: rpc.id, error: { code: -32601, message: "Method not found" } }); + } + json({ error: "not_found" }, 404); + }; + const server = createServer((req, res) => { handle(req, res).catch(error => { res.writeHead(500).end(String(error)); }); }); + await new Promise(resolve => server.listen(0, host, resolve)); + origin = `http://${host}:${(server.address() as AddressInfo).port}`; + return { + origin, mcpUrl: `${origin}/mcp`, receipt, record, + close: () => new Promise(resolve => { server.closeAllConnections(); server.close(() => resolve()); }), + }; +} diff --git a/tests/restart-chaos/journeys.test.ts b/tests/restart-chaos/journeys.test.ts new file mode 100644 index 00000000000..86ab338dcad --- /dev/null +++ b/tests/restart-chaos/journeys.test.ts @@ -0,0 +1,314 @@ +import { describe, expect, it } from "vitest"; +import { parseCommand } from "./config.js"; +import { actionRequestRows, JourneyDriver, questionAnswer, reviewClockNames, shiftReviewClockWith, type JourneyHost } from "./journey-driver.js"; +import { matrixCells, matrixMayLaunch, planMatrixWave } from "./journey-matrix.js"; +import { pkceChallenge, startOAuthApp } from "./journey-oauth-app.js"; +import { + answerMarker, gradeBlockedQuestionJourney, gradeToolReviewJourney, journeyChoice, q3ToolArgument, Q3_REVIEW_AGE_MS, + type JourneyArtifacts, type JourneyRecord, +} from "./journeys.js"; +import { planTasks, taskFaultSchedule } from "./scheduler.js"; +import { Timeline } from "./timeline.js"; +import type { OracleArtifacts, ScheduledFault, TaskPlan, TaskRecord, TaskSample } from "./types.js"; +import { workloadDefinitions } from "./workloads.js"; + +const nonce = "a1b2c3d4"; + +function artifacts(journey: Partial & { record: JourneyRecord }, overrides: Partial = {}): OracleArtifacts { + return { issueStatus: "done", documents: { brief: `CHAOS-${nonce}\n${answerMarker(nonce)}\n- one\n- two\n- three\n` }, agentComments: [], + files: {}, commands: {}, workspacePath: null, + journey: { questions: [{ id: "q", status: "answered", createdAt: null }], agentComments: [], runErrorCodes: [], interruption: null, ...journey }, + ...overrides }; +} + +const answered = (extra: Partial> = {}) => ({ at: 2_000, doneAt: 2_100, httpStatus: 200, error: null, + status: "answered", pendingWhenAnswered: true, faultsBefore: 1, ...extra }); + +const failed = (result: ReturnType) => result.checks.filter(check => !check.ok).map(check => check.name); + +const q2Record = (extra: Partial = {}): JourneyRecord => ({ requestSeenAt: 1_000, blocker: { issueId: "b", at: 1_500, error: null }, + answer: answered(), answerTurn: { at: 60_000, issueStatus: "blocked", blockerKept: true }, blockerResolved: { at: 70_000, error: null }, ...extra }); + +describe("journey oracles", () => { + it("passes Q2 when the answer turn ran while the task stayed blocked by its prerequisite", () => { + expect(failed(gradeBlockedQuestionJourney(nonce, artifacts({ record: q2Record({ faultTrigger: "asked" }) })))).toEqual([]); + }); + + it("fails Q2 when a restart made the agent ask twice, the board answered before the restart, or the answer missed the brief", () => { + const result = gradeBlockedQuestionJourney(nonce, artifacts({ record: q2Record({ faultTrigger: "asked", answer: answered({ faultsBefore: 0 }) }), + questions: [{ id: "a", status: "answered", createdAt: null }, { id: "b", status: "pending", createdAt: null }] }, + { documents: { brief: `CHAOS-${nonce}\n- one\n- two\n- three` } })); + expect(failed(result)).toEqual(["one_question_saved", "response_after_restart", "brief_contains_answer_marker"]); + }); + + it("fails Q2 on a stale Retry notice left after the answered work finished", () => { + const result = gradeBlockedQuestionJourney(nonce, artifacts({ record: q2Record(), interruption: { state: "failed", canRetry: true } })); + expect(failed(result)).toEqual(["no_stale_retry_notice"]); + }); + + it("fails Q2 the way #20363's parent does: the answer never reaches the blocked task", () => { + const record = q2Record({ answerTurn: { missedAt: 400_000 }, blockerResolved: { at: 400_100, error: null } }); + expect(failed(gradeBlockedQuestionJourney(nonce, artifacts({ record, runErrorCodes: ["issue_dependencies_blocked"] })))) + .toEqual(["answer_reached_agent_while_blocked", "prerequisite_kept_through_answer", "answer_turn_before_prerequisite_finished", + "no_dependency_blocked_failure"]); + }); + + const q3Record = (extra: Partial = {}): JourneyRecord => ({ + requestSeenAt: 1_000, app: { connectionId: "c", agentId: "a", receipt: "receipt-x", setupError: null }, + timeShift: { at: 1_100, ms: Q3_REVIEW_AGE_MS, requests: 1, grants: 1, error: null }, + refresh: { at: 1_200, httpStatus: 200, error: null }, answer: answered({ status: "executed" }), ...extra }); + + it("passes Q3 when an aged review approved after a token refresh runs the reviewed action once", () => { + const result = gradeToolReviewJourney(nonce, artifacts({ record: q3Record(), actionRequests: [{ id: "r", status: "executed" }], + provider: { toolCalls: [{ at: 2_050, title: q3ToolArgument(nonce) }], refreshes: [1_250] }, + agentComments: [{ createdAt: 3_000, body: "Q3 result: Created page. receipt-x" }] }, { documents: {} })); + expect(failed(result)).toEqual([]); + }); + + it("fails Q3 the way #20340's parent does: the refresh invalidates the approval", () => { + const record = q3Record({ answer: answered({ httpStatus: 409, error: "approved_tool_target_changed", status: null }) }); + const result = gradeToolReviewJourney(nonce, artifacts({ record, actionRequests: [{ id: "r", status: "pending" }], + provider: { toolCalls: [], refreshes: [1_250] } }, { issueStatus: "in_review", documents: {} })); + expect(failed(result)).toEqual(["issue_done", "board_approved", "reviewed_action_ran_once", "action_executed", "agent_reported_result"]); + }); + + it("does not credit a refresh that happened only during the approval itself", () => { + const result = gradeToolReviewJourney(nonce, artifacts({ record: q3Record(), actionRequests: [{ id: "r", status: "executed" }], + provider: { toolCalls: [{ at: 2_050, title: q3ToolArgument(nonce) }], refreshes: [2_010] }, + agentComments: [{ createdAt: 3_000, body: "receipt-x" }] }, { documents: {} })); + expect(failed(result)).toEqual(["token_refreshed_before_decision"]); + }); +}); + +describe("journey planning", () => { + it("fires journey faults at their own boundaries", () => { + const tasks = planTasks({ seed: 5, tasks: 4, workloads: ["Q1", "Q2"], adapters: ["codex"], faults: ["K1"], triggers: ["asked", "answered"] }); + expect(new Set(tasks.map(task => task.trigger))).toEqual(new Set(["asked", "answered"])); + for (const task of tasks) expect(taskFaultSchedule(task)).toEqual([expect.objectContaining({ kind: "K1", trigger: task.trigger })]); + expect(() => planTasks({ seed: 5, tasks: 1, workloads: ["A1"], adapters: ["codex"], faults: ["K1"], triggers: ["asked"] })) + .toThrow(/Q1, Q2 or Q3/); + }); + + it("crosses every journey with a baseline and every fault at each of its triggers", () => { + const cells = matrixCells({ workloads: ["Q1", "Q2", "Q3"], adapters: ["codex"], faults: ["K1", "K2", "K4"] }); + expect(cells).toHaveLength(3 * (1 + 3 * 2)); + expect(cells.filter(cell => cell.workload === "Q3").map(cell => `${cell.fault}^${cell.trigger}`)).toEqual([ + "K0^time", "K1^asked", "K1^answered", "K2^asked", "K2^answered", "K4^asked", "K4^answered"]); + // Baselines come first, so they finish before any fault cell shares the server. + expect(cells.slice(0, 3).map(cell => cell.fault)).toEqual(["K0", "K0", "K0"]); + expect(cells.slice(3).every(cell => cell.fault !== "K0")).toBe(true); + const command = parseCommand(["--workloads", "Q1,Q2,Q3,G1", "--faults", "K1,K2,K4", "--adapters", "codex", "--matrix", "--seed", "9"]); + if (command.mode !== "wave") throw new Error("expected a wave"); + expect(command.options).toMatchObject({ matrix: true, tasks: 21 + 3 }); + const plan = planMatrixWave("w-20261109T000000-s9", command.options); + expect(plan.tasks).toHaveLength(24); + const baseline = plan.tasks.find(task => task.fault === "K0")!; + expect(taskFaultSchedule(baseline)).toEqual([]); + // G1 grades a decision made after a restart, so it has no no-fault baseline. + expect(plan.tasks.filter(task => task.workload === "G1").map(task => task.trigger ?? "time")).toEqual(["approval", "approval", "approval"]); + expect(planMatrixWave("w-20261109T000000-s9", command.options).tasks).toEqual(plan.tasks); + }); + + it("never runs a baseline beside a fault cell", () => { + expect(matrixMayLaunch({ fault: "K0" }, [])).toBe(true); + expect(matrixMayLaunch({ fault: "K0" }, [{ fault: "K0" }])).toBe(true); + expect(matrixMayLaunch({ fault: "K1" }, [{ fault: "K0" }])).toBe(false); + expect(matrixMayLaunch({ fault: "K0" }, [{ fault: "K4" }])).toBe(false); + expect(matrixMayLaunch({ fault: "K2" }, [{ fault: "K1" }])).toBe(true); + }); + + it("honours --fault-after-seconds for time-fired matrix cells", () => { + const command = parseCommand(["--workloads", "A1,Q1", "--faults", "K1", "--adapters", "codex", "--matrix", "--fault-after-seconds", "60", "--seed", "4"]); + if (command.mode !== "wave") throw new Error("expected a wave"); + const tasks = planMatrixWave("w-20261109T000000-s4", command.options).tasks; + expect(tasks.find(task => task.workload === "A1" && task.fault === "K1")!.fireAfterMs).toBe(60_000); + expect(tasks.filter(task => task.fault === "K0").every(task => task.fireAfterMs === 0)).toBe(true); + }); + + it("validates --matrix and --server-from", () => { + expect(() => parseCommand(["--workloads", "Q1", "--matrix", "--tasks", "3"])).toThrow(/do not pass --tasks/); + expect(() => parseCommand(["--workloads", "Q1", "--matrix", "--triggers", "asked"])).toThrow(/own triggers/); + const command = parseCommand(["--workloads", "Q2", "--server-from", "../old", "--faults", "K1"]); + if (command.mode !== "wave") throw new Error("expected a wave"); + expect(command.options.serverFrom).toMatch(/\/old$/); + expect(() => parseCommand(["--workloads", "Q2", "--server-from", "../old", "--faults", "K14", "--upgrade-from", "../older"])).toThrow(/upgrade-from/); + // Provider message faults cross the matrix too, each armed once per workload. + const provider = parseCommand(["--workloads", "Q1", "--faults", "PV1,PV2", "--adapters", "codex", "--matrix", "--seed", "3"]); + if (provider.mode !== "wave") throw new Error("expected a wave"); + const tasks = planMatrixWave("w-20261109T000000-s3", provider.options).tasks; + expect(tasks.map(task => task.fault)).toEqual(["K0", "PV1", "PV2"]); + expect(tasks.filter(task => task.providerFault).map(task => task.providerFault!.kind)).toEqual(["PV1", "PV2"]); + }); + + it("keeps journeys out of the default workloads and registers their definitions", () => { + const command = parseCommand(["--adapters", "codex"]); + if (command.mode !== "wave") throw new Error("expected a wave"); + expect(command.options.workloads).not.toContain("Q1"); + expect(workloadDefinitions.Q2.description(nonce)).toContain(`ANSWER-${nonce}-`); + expect(journeyChoice(nonce)).toMatch(/^(amber|cobalt)$/); + }); +}); + +describe("harness OAuth app", () => { + it("runs a PKCE authorization, rotates tokens on refresh and records tool calls", async () => { + const app = await startOAuthApp(); + try { + const unauthenticated = await fetch(app.mcpUrl, { method: "POST", body: "{}" }); + expect(unauthenticated.status).toBe(401); + expect(unauthenticated.headers.get("www-authenticate")).toContain("oauth-protected-resource"); + await fetch(`${app.origin}/register`, { method: "POST", body: JSON.stringify({ redirect_uris: ["http://127.0.0.1:1/cb"] }) }); + const verifier = "chaos-verifier-0123456789-abcdefghijklmnopqrstuvwxyz"; + const authorizeUrl = (state: string) => `${app.origin}/authorize?redirect_uri=${encodeURIComponent("http://127.0.0.1:1/cb")}` + + `&state=${state}&code_challenge=${pkceChallenge(verifier)}&code_challenge_method=S256`; + expect((await fetch(`${app.origin}/authorize?redirect_uri=${encodeURIComponent("http://127.0.0.1:1/cb")}&state=x`, { redirect: "manual" })).status).toBe(400); + const codeFor = async (state: string) => { + const back = new URL((await fetch(authorizeUrl(state), { redirect: "manual" })).headers.get("location")!); + expect(back.searchParams.get("state")).toBe(state); + return back.searchParams.get("code")!; + }; + const token = async (form: Record) => (await fetch(`${app.origin}/token`, { method: "POST", body: new URLSearchParams(form) })).json() as Promise>; + // A missing or wrong verifier is refused, and the refused code cannot be reused. + expect((await token({ grant_type: "authorization_code", code: await codeFor("s0") })).error).toBe("invalid_grant"); + const wrong = await codeFor("s2"); + expect((await token({ grant_type: "authorization_code", code: wrong, code_verifier: `${verifier}x` })).error).toBe("invalid_grant"); + expect((await token({ grant_type: "authorization_code", code: wrong, code_verifier: verifier })).error).toBe("invalid_grant"); + const first = await token({ grant_type: "authorization_code", code: await codeFor("s1"), code_verifier: verifier, + redirect_uri: "http://127.0.0.1:1/cb" }); + const second = await token({ grant_type: "refresh_token", refresh_token: first.refresh_token! }); + expect(second.access_token).not.toBe(first.access_token); + expect((await token({ grant_type: "refresh_token", refresh_token: first.refresh_token! })).error).toBe("invalid_grant"); + const call = await fetch(app.mcpUrl, { method: "POST", headers: { authorization: `Bearer ${second.access_token}` }, + body: JSON.stringify({ jsonrpc: "2.0", id: 1, method: "tools/call", params: { name: "create_page", arguments: { title: "T" } } }) }); + expect(JSON.stringify(await call.json())).toContain(app.receipt); + expect(app.record).toMatchObject({ registrations: 1, codeExchanges: 1, rejectedCodes: 3, rejectedRefreshes: 1, toolCalls: [{ title: "T" }] }); + expect(app.record.refreshes).toHaveLength(1); + } finally { await app.close(); } + }); +}); + +describe("journey driver", () => { + function harness(workload: "Q2", trigger: TaskPlan["trigger"] = "answered") { + const plan: TaskPlan = { taskId: "t01", index: 0, workload, adapter: "codex", fault: "K1", fireAfterMs: 0, holdMs: 0, + trigger, triggerDelayMs: 0, followUpAfterMs: null, nonce }; + const record = { plan, issueId: "issue-1", faultIds: [] } as unknown as TaskRecord; + const calls: string[] = []; + let interaction = { id: "i1", kind: "ask_user_questions", status: "pending", payload: { version: 1, questions: [{ id: "codeword", + prompt: "Which code word?", options: [{ id: "opt-a", label: "Amber" }, { id: "opt-c", label: "Cobalt" }] }] } }; + const api = { + baseUrl: "http://x", + async get(route: string) { + if (route.endsWith("/interactions")) return [interaction]; + if (route.endsWith("/comments")) return [{ authorAgentId: "agent", body: `ok ${answerMarker(nonce)}`, createdAt: new Date(Date.now() + 60_000).toISOString() }]; + if (route === "/api/issues/issue-1") return { status: "blocked", blockedBy: [{ id: "blocker-1" }] }; + throw new Error(`unexpected GET ${route}`); + }, + async post(route: string, body: unknown) { + calls.push(`POST ${route} ${JSON.stringify(body)}`); + if (route.endsWith("/respond")) { interaction = { ...interaction, status: "answered" }; return interaction; } + if (route.endsWith("/issues")) return { id: "blocker-1" }; + throw new Error(`unexpected POST ${route}`); + }, + async patch(route: string, body: unknown) { calls.push(`PATCH ${route} ${JSON.stringify(body)}`); return {}; }, + }; + const host: JourneyHost = { plan, api: api as unknown as JourneyHost["api"], lab: { companyId: "c" } as JourneyHost["lab"], + timeline: new Timeline(null), record, databaseUrl: () => null }; + return { driver: new JourneyDriver(workload, host), calls, record }; + } + it("reads Apps → Review rows in the route's nested shape", () => { + expect(actionRequestRows({ actionRequests: [{ request: { id: "r1", issueId: "i", status: "pending" }, connection: {} }] })) + .toEqual([{ id: "r1", issueId: "i", status: "pending" }]); + expect(actionRequestRows(null)).toEqual([]); + }); + + it("answers with the agent's own question and option ids, matched by the chosen code word", () => { + const card = { payload: { questionSet: { questions: [{ id: "q-7", options: [{ id: "x", label: " Cobalt " }, { id: "amber", label: "Warm" }] }] } } }; + expect(questionAnswer(card, "cobalt")).toEqual({ questionId: "q-7", optionIds: ["x"] }); + expect(questionAnswer(card, "amber")).toEqual({ questionId: "q-7", optionIds: ["amber"] }); + expect(questionAnswer({ payload: {} }, "amber")).toBeNull(); + }); + + const sample = (runs: TaskSample["runs"] = []): TaskSample => ({ at: 0, reachable: true, serverReady: true, runs }); + const answeredFault: ScheduledFault = { kind: "K1", holdMs: 0, atMs: 0, trigger: "answered", delayMs: 0 }; + + it("answers only once the planned faults before the answer fired, then arms the answered trigger", async () => { + const { driver, calls, record } = harness("Q2", "asked"); + const askedFault: ScheduledFault = { kind: "K1", holdMs: 0, atMs: 0, trigger: "asked", delayMs: 0 }; + expect(await driver.step({ sample: sample(), now: 1_000, quiet: true, pendingFaults: [askedFault] })).toBe(true); + expect(driver.armedAt("asked")).toBe(1_000); + expect(driver.awaitingBoard()).toBe(true); + expect(await driver.step({ sample: sample(), now: 2_000, quiet: true, pendingFaults: [askedFault] })).toBe(false); + expect(calls.some(call => call.includes("/respond"))).toBe(false); + record.faultIds.push("f1"); + expect(await driver.step({ sample: sample(), now: 5_000, quiet: true, pendingFaults: [] })).toBe(true); + const optionId = journeyChoice(nonce) === "amber" ? "opt-a" : "opt-c"; + expect(calls.at(-1)).toBe(`POST /api/issues/issue-1/interactions/i1/respond {"answers":[{"questionId":"codeword","optionIds":["${optionId}"]}]}`); + expect(driver.armedAt("answered")).toBeGreaterThanOrEqual(5_000); + expect(record.journey!.answer).toMatchObject({ pendingWhenAnswered: true, faultsBefore: 1, error: null }); + }); + + it("Q2: adds the prerequisite, answers, waits for the answer turn, then finishes the prerequisite", async () => { + const { driver, calls, record } = harness("Q2"); + const running = sample([{ id: "r1", status: "running", createdAt: null, startedAt: null, finishedAt: null, errorCode: null, + retryOfRunId: null, wakeCommentIds: [], livenessState: null }]); + expect(await driver.step({ sample: running, now: 1_000, quiet: true, pendingFaults: [answeredFault] })).toBe(false); + expect(await driver.step({ sample: sample(), now: 2_000, quiet: true, pendingFaults: [answeredFault] })).toBe(true); + expect(calls.at(-1)).toBe('PATCH /api/issues/issue-1 {"status":"blocked","blockedByIssueIds":["blocker-1"]}'); + expect(await driver.step({ sample: sample(), now: 3_000, quiet: true, pendingFaults: [answeredFault] })).toBe(true); + expect(calls.at(-1)).toContain("/respond"); + // The answered fault is still pending: the turn is observed, the prerequisite stays until the fault fired. + expect(await driver.step({ sample: sample(), now: 4_000, quiet: true, pendingFaults: [answeredFault] })).toBe(false); + expect(record.journey!.answerTurn).toMatchObject({ issueStatus: "blocked", blockerKept: true }); + expect(driver.awaitingBoard()).toBe(true); + expect(await driver.step({ sample: sample(), now: 5_000, quiet: true, pendingFaults: [] })).toBe(true); + expect(calls.at(-1)).toBe('PATCH /api/issues/blocker-1 {"status":"done"}'); + expect(driver.awaitingBoard()).toBe(false); + }); + + it("Q2: counts the answer only from a comment or an accepted run result, never a rejected one", async () => { + const runAfterAnswer = { id: "r2", status: "succeeded", createdAt: new Date(10_000).toISOString(), startedAt: null, + finishedAt: new Date(20_000).toISOString(), errorCode: null, retryOfRunId: null, wakeCommentIds: [], livenessState: null }; + for (const eventType of ["run.result.rejected", "run.result.accepted"]) { + const plan: TaskPlan = { taskId: "t01", index: 0, workload: "Q2", adapter: "codex", fault: "K0", fireAfterMs: 0, holdMs: 0, + followUpAfterMs: null, nonce }; + const record = { plan, issueId: "issue-1", faultIds: [], journey: { requestSeenAt: 1, requestId: "i1", + blocker: { issueId: "blocker-1", at: 2, error: null }, + answer: { at: 5_000, doneAt: 5_000, httpStatus: 200, error: null, status: "answered", pendingWhenAnswered: true, faultsBefore: 0 } } } as unknown as TaskRecord; + const api = { baseUrl: "http://x", async get(route: string) { + if (route.endsWith("/comments")) return []; + if (route.startsWith("/api/heartbeat-runs/r2/events")) return [{ eventType, payload: { result: { summary: answerMarker(nonce) } } }]; + if (route === "/api/issues/issue-1") return { status: "blocked", blockedBy: [{ id: "blocker-1" }] }; + throw new Error(`unexpected GET ${route}`); + }, async post() { return {}; }, async patch() { return {}; } }; + const driver = new JourneyDriver("Q2", { plan, api: api as unknown as JourneyHost["api"], lab: { companyId: "c" } as JourneyHost["lab"], + timeline: new Timeline(null), record, databaseUrl: () => null }); + await driver.step({ sample: { at: 0, reachable: true, serverReady: true, runs: [runAfterAnswer] }, now: 30_000, quiet: true, pendingFaults: [] }); + expect(record.journey!.answerTurn ?? null).toEqual(eventType === "run.result.accepted" + ? { at: 20_000, issueStatus: "blocked", blockerKept: true } : null); + } + }); +}); + +describe("Q3 review clock", () => { + it("moves the pending review and the token expiry back by the given time, using the schema's names", async () => { + const queries: Array<{ query: string; params: unknown[] }> = []; + const hour = 61 * 60_000; + const client = { async unsafe(query: string, params: unknown[] = []) { + queries.push({ query, params }); + if (query.startsWith("UPDATE tool_action_requests")) return [{ id: "req" }]; + if (query.startsWith("SELECT") && query.includes("FROM connection_grants")) return [{ id: "g", provider_tenant: { oauth: { accessTokenExpiresAt: "2026-11-10T03:00:00.000Z" } } }]; + if (query.startsWith("SELECT") && query.includes("FROM tool_connections")) return [{ id: "c", config: { oauth: { expiresAt: "2026-11-10T03:00:00.000Z" } }, transport_config: null }]; + return []; + } }; + const names = reviewClockNames(); + expect(names).toMatchObject({ requests: "tool_action_requests", requestTimes: ["created_at", "updated_at", "expires_at"], + grants: "connection_grants", grantTenant: "provider_tenant", connections: "tool_connections", connectionTransport: "transport_config" }); + expect(await shiftReviewClockWith(client, names, "req", "c", hour)).toEqual({ requests: 1, grants: 2 }); + expect(queries[0]!.params).toEqual(["req", "3660 seconds"]); + expect(queries[0]!.query).toContain("expires_at = expires_at - $2::interval"); + const grantUpdate = queries.find(entry => entry.query.startsWith("UPDATE connection_grants"))!; + expect(JSON.parse(String(grantUpdate.params[1])).oauth.accessTokenExpiresAt).toBe("2026-11-10T01:59:00.000Z"); + const connectionUpdate = queries.find(entry => entry.query.startsWith("UPDATE tool_connections"))!; + expect(JSON.parse(String(connectionUpdate.params[1])).oauth.expiresAt).toBe("2026-11-10T01:59:00.000Z"); + expect(connectionUpdate.params[2]).toBeNull(); + }); +}); diff --git a/tests/restart-chaos/journeys.ts b/tests/restart-chaos/journeys.ts new file mode 100644 index 00000000000..e2b0199ec0c --- /dev/null +++ b/tests/restart-chaos/journeys.ts @@ -0,0 +1,237 @@ +/** + * Human-in-the-loop journey workloads: a person answers or approves what an agent + * saved, through the same board routes the UI uses, with a restart fault at the + * journey's own boundaries (`asked`, `answered`). + * + * Q1 questions-and-approvals/questions + continuation: the plain saved question + * (question-handoff.ts). It takes the `asked` and `answered` triggers too. + * Q2 a question on a dependency-blocked task: the board adds an unfinished + * prerequisite before answering. The answer must reach the agent in bounded + * answer mode while the task stays blocked (#20363), then the board finishes + * the prerequisite and the agent finishes the task. + * Q3 questions-and-approvals/tool-review: a connected OAuth app action waits for + * review. Before the board decides, more than an hour passes and the access + * token is refreshed (#20340). The reviewed action must still run exactly once. + * + * This module holds only definitions and pure oracles; journey-driver.ts acts as + * the board, journey-oauth-app.ts is the harness-owned OAuth/MCP app and + * journey-matrix.ts crosses every journey with every fault kind. + */ +import type { OracleArtifacts, OracleResult, TriggerId, WorkloadId } from "./types.js"; +import type { WorkloadAgents, WorkloadDefinition } from "./workloads.js"; + +/** Journeys whose board side is journey-driver.ts. */ +export const JOURNEY_WORKLOADS = ["Q2", "Q3"] as const; +export type JourneyWorkload = (typeof JOURNEY_WORKLOADS)[number]; +export const isJourneyWorkload = (workload: string): workload is JourneyWorkload => + (JOURNEY_WORKLOADS as readonly string[]).includes(workload); +/** Workloads with a saved request and a board response, so `asked` and `answered` can fire. */ +export const takesJourneyTriggers = (workload: string): boolean => workload === "Q1" || isJourneyWorkload(workload); + +/** `asked`: the request (question or tool review) is saved. `answered`: the board's response was posted. */ +export const JOURNEY_TRIGGERS = ["asked", "answered"] as const satisfies readonly TriggerId[]; +export type JourneyTrigger = (typeof JOURNEY_TRIGGERS)[number]; + +/** Each workload's own fault boundaries, for the journey × fault matrix. */ +export const WORKLOAD_TRIGGERS: Record = { + A1: ["time"], B1: ["time"], B5: ["tool"], C1: ["comment", "result"], M1: ["time"], G1: ["approval"], + H1: ["paused"], R1: ["time"], S1: ["stop"], L1: ["time"], Q1: JOURNEY_TRIGGERS, Q2: JOURNEY_TRIGGERS, Q3: JOURNEY_TRIGGERS, +}; + +/** Q3 reviews must survive a delay longer than the implicit deadline #20340 removed. */ +export const Q3_REVIEW_AGE_MS = 61 * 60_000; +/** Q2: how long the board waits (server ready) for the answer turn before it finishes the prerequisite anyway. */ +export const ANSWER_TURN_WAIT_MS = 6 * 60_000; + +export const CODE_WORDS = ["amber", "cobalt"] as const; +/** The board's answer, fixed by the nonce so the agent cannot guess it. */ +export function journeyChoice(nonce: string): (typeof CODE_WORDS)[number] { + return CODE_WORDS[Number.parseInt(nonce.slice(1, 2), 16) % 2]!; +} +export const answerMarker = (nonce: string, choice = journeyChoice(nonce)) => `ANSWER-${nonce}-${choice}`; +export const questionTitle = (nonce: string) => `Code word for ${nonce}`; +export const q3ToolArgument = (nonce: string) => `Q3-${nonce}`; +const briefMarker = (nonce: string) => `CHAOS-${nonce}`; + +/** What the board did and saw during a journey. Persisted on the task record. */ +export interface JourneyRecord { + /** The trigger the task's own fault waits for (absent: time-fired or no fault). */ + faultTrigger?: TriggerId | null; + /** The saved question (Q2) or tool review (Q3), first seen pending. */ + requestSeenAt?: number | null; + requestId?: string | null; + /** Q2: the prerequisite the board added before answering. */ + blocker?: { issueId: string | null; at: number; error: string | null }; + /** The board's answer (Q2) or approval (Q3). `doneAt` arms the `answered` trigger. */ + answer?: { at: number; doneAt: number; httpStatus: number | null; error: string | null; status: string | null; + pendingWhenAnswered: boolean; faultsBefore: number }; + /** Q2: the agent's reply to the answer while the prerequisite was unfinished. */ + answerTurn?: { at: number; issueStatus: string | null; blockerKept: boolean } | { missedAt: number }; + /** Q2: the board finished the prerequisite. */ + blockerResolved?: { at: number; error: string | null }; + /** Q3: the harness-owned OAuth app, its connection and the scripted agent that uses it. */ + app?: { connectionId: string | null; agentId: string | null; receipt: string; setupError: string | null }; + /** Q3: persisted review and token timestamps moved back, as if the review waited this long. */ + timeShift?: { at: number; ms: number; requests: number; grants: number; error: string | null }; + /** Q3: the board's connection check, which renews the expiring access token. */ + refresh?: { at: number; httpStatus: number | null; error: string | null }; +} + +export interface JourneyArtifacts { + record: JourneyRecord; + /** Q2: every question card on the task. */ + questions: Array<{ id: string; status: string; createdAt: string | null }>; + /** Agent comments, oldest first. */ + agentComments: Array<{ createdAt: number; body: string }>; + /** Every run's error code (cancelled and failed runs included). */ + runErrorCodes: string[]; + /** The task's interruption notice at rest. */ + interruption: { state: string | null; canRetry: boolean } | null; + /** Q3: every action request on the task and what the provider saw. */ + actionRequests?: Array<{ id: string; status: string }>; + provider?: { toolCalls: Array<{ at: number; title: unknown }>; refreshes: number[] }; + /** Q3: the scripted agent's output for its runs that did not succeed (evidence only). */ + runLogs?: Record; +} + +function result(checks: OracleResult["checks"]): OracleResult { + return { verdict: checks.every(check => check.ok) ? "pass" : "fail", checks }; +} + +function briefChecks(body: string | null | undefined, markers: string[]): OracleResult["checks"] { + const text = body ?? ""; + const bullets = text.split(/\r?\n/).filter(line => /^\s*[-*] \S/.test(line)).length; + return [ + { name: "brief_document_present", ok: text.trim().length > 0 }, + ...markers.map(marker => ({ name: `brief_contains_${marker.split("-")[0]!.toLowerCase()}_marker`, ok: text.includes(marker), detail: marker })), + { name: "brief_has_three_bullets", ok: bullets >= 3, detail: `bullets=${bullets}` }, + ]; +} + +/** Checks every journey shares: the board's response landed after the planned fault, and no stale Retry is left. */ +function responseChecks(journey: JourneyArtifacts, label: string): OracleResult["checks"] { + const answer = journey.record.answer; + return [ + { name: `board_${label}`, ok: answer !== undefined && answer.error === null, + detail: answer ? `status=${answer.httpStatus} ${answer.error ?? answer.status ?? ""}`.trim() : "board never acted" }, + { name: "request_pending_until_response", ok: answer?.pendingWhenAnswered === true }, + // An `asked` fault must have hit the saved request before the board answered it. + ...(journey.record.faultTrigger === "asked" + ? [{ name: "response_after_restart", ok: (answer?.faultsBefore ?? 0) > 0, detail: `faults_before=${answer?.faultsBefore ?? 0}` }] : []), + { name: "no_stale_retry_notice", ok: !journey.interruption?.canRetry, + detail: journey.interruption ? `state=${journey.interruption.state} canRetry=${journey.interruption.canRetry}` : "none" }, + ]; +} + +/** Q2: a saved question answered while the task is blocked by an unfinished prerequisite. */ +export function gradeBlockedQuestionJourney(nonce: string, artifacts: OracleArtifacts): OracleResult { + const journey = artifacts.journey; + if (!journey) return result([{ name: "journey_recorded", ok: false }]); + const marker = answerMarker(nonce); + const checks: OracleResult["checks"] = [ + { name: "issue_done", ok: artifacts.issueStatus === "done", detail: `status=${artifacts.issueStatus ?? "unknown"}` }, + // A restart that made the agent ask again is a finding: the saved question is the durable wait. + { name: "one_question_saved", ok: journey.questions.length === 1, detail: journey.questions.map(question => question.status).join(",") || "none" }, + ...responseChecks(journey, "answered"), + ...briefChecks(artifacts.documents.brief, [briefMarker(nonce), marker]), + ]; + { + const record = journey.record; + const turn = record.answerTurn; + const reached = turn !== undefined && "at" in turn; + checks.push( + { name: "prerequisite_added", ok: record.blocker !== undefined && record.blocker.error === null, detail: record.blocker?.error ?? undefined }, + { name: "answer_reached_agent_while_blocked", ok: reached && turn.issueStatus === "blocked", + detail: !turn ? "board never waited for it" : reached ? `status=${turn.issueStatus}` : "no answer turn within the wait" }, + { name: "prerequisite_kept_through_answer", ok: reached && turn.blockerKept }, + { name: "answer_turn_before_prerequisite_finished", ok: reached && record.blockerResolved !== undefined && turn.at <= record.blockerResolved.at }, + { name: "no_dependency_blocked_failure", ok: !journey.runErrorCodes.includes("issue_dependencies_blocked"), + detail: journey.runErrorCodes.join(",") || "none" }, + ); + } + return result(checks); +} + +export function gradeToolReviewJourney(nonce: string, artifacts: OracleArtifacts): OracleResult { + const journey = artifacts.journey; + if (!journey) return result([{ name: "journey_recorded", ok: false }]); + const record = journey.record; + const requests = journey.actionRequests ?? []; + const calls = journey.provider?.toolCalls ?? []; + const seen = record.requestSeenAt ?? null; + const decided = record.answer?.at ?? null; + const refreshes = (journey.provider?.refreshes ?? []).filter(at => seen !== null && decided !== null && at > seen && at < decided); + const receipt = record.app?.receipt ?? ""; + return result([ + { name: "oauth_app_connected", ok: Boolean(record.app?.connectionId) && !record.app?.setupError, detail: record.app?.setupError ?? undefined }, + { name: "issue_done", ok: artifacts.issueStatus === "done", detail: `status=${artifacts.issueStatus ?? "unknown"}` }, + { name: "one_review_requested", ok: requests.length === 1, detail: requests.map(request => request.status).join(",") || "none" }, + { name: "review_aged_past_one_hour", ok: record.timeShift !== undefined && record.timeShift.error === null + && record.timeShift.ms > 60 * 60_000 && record.timeShift.requests === 1, detail: JSON.stringify(record.timeShift ?? null) }, + { name: "token_refreshed_before_decision", ok: refreshes.length > 0, + detail: `refreshes_between=${refreshes.length} check=${record.refresh?.httpStatus ?? "none"}` }, + ...responseChecks(journey, "approved"), + { name: "reviewed_action_ran_once", ok: calls.length === 1 && calls[0]!.title === q3ToolArgument(nonce), + detail: `calls=${calls.length}${calls.length ? ` title=${JSON.stringify(calls[0]!.title)}` : ""}` }, + { name: "action_executed", ok: requests.length === 1 && requests[0]!.status === "executed" }, + // Only the provider's response carries the receipt: the agent saw the real result. + { name: "agent_reported_result", ok: receipt.length > 0 && journey.agentComments.some(comment => comment.body.includes(receipt)) }, + ]); +} + +// The agent saves the question through its native `request_human_input` tool, the governed wait +// whose saved handoff #20336 keeps across a provider shutdown. +const askSteps = (nonce: string) => [ + "Before writing anything, you need one answer from the board.", + "1. Ask exactly one question with your `request_human_input` tool: a questions card (continuationPolicy `wake_assignee`,", + ` idempotencyKey \`journey-${nonce}\`, title \"${questionTitle(nonce)}\") with one required single-select question`, + ` "Which code word should the brief use?" and exactly two options labelled \`${CODE_WORDS[0]}\` and \`${CODE_WORDS[1]}\`.`, + "2. Then end your turn and wait for the answer. Do not guess, do not ask again, and do not write the brief yet.", +]; + +const briefSteps = (nonce: string) => [ + "Save this issue's document with key `brief` (title \"Brief\"), in markdown, on why idempotent retries matter for background job queues.", + `It must contain the line \`${briefMarker(nonce)}\` exactly, on its own line, and the line \`ANSWER-${nonce}-\` on its own line,`, + "where is the option the board chose, in lowercase, and at least three bullet points that start with \"- \".", + "Then mark this task done. Do not create files or other tasks.", +]; + +function q2Description(nonce: string) { + return [ + ...askSteps(nonce), + "3. The board may add a prerequisite task that blocks this one. If you are woken with the answer while this task is still blocked:", + ` post exactly one comment containing the line \`ANSWER-${nonce}-\` and end your turn.`, + " Do not change this task's status or its blockers, and do not work on the prerequisite.", + "4. When you are woken after the prerequisite is done (no unfinished blockers remain):", + ...briefSteps(nonce).map(line => ` ${line}`), + ].join("\n"); +} + +const collectBrief = { documents: ["brief"], files: () => [], commands: [] }; + +export const journeyWorkloadDefinitions: Record = { + Q2: { + id: "Q2", + label: "Board answers a question on a dependency-blocked task", + usesRepository: false, + fireWindowMs: [20_000, 90_000], + budgetMs: 25 * 60_000, + title: nonce => `Chaos Q2 blocked question ${nonce}`, + description: (nonce: string, _agents?: WorkloadAgents) => q2Description(nonce), + collect: collectBrief, + grade: (nonce, artifacts) => gradeBlockedQuestionJourney(nonce, artifacts), + }, + Q3: { + id: "Q3", + label: "Connected-app action approved after token refresh and a long delay", + usesRepository: false, + fireWindowMs: [10_000, 40_000], + budgetMs: 12 * 60_000, + // The scripted agent finds its task by this title. + title: nonce => `Chaos ${q3ToolArgument(nonce)} app review`, + // A scripted process agent does the work (journey-driver.ts); the description is for people. + description: nonce => `Create the page \`${q3ToolArgument(nonce)}\` with the connected chaos app, wait for the board's review, then report the result.`, + collect: { documents: [], files: () => [], commands: [] }, + grade: (nonce, artifacts) => gradeToolReviewJourney(nonce, artifacts), + }, +}; diff --git a/tests/restart-chaos/main.ts b/tests/restart-chaos/main.ts index 7d5da9dd9e1..4040920f6f6 100644 --- a/tests/restart-chaos/main.ts +++ b/tests/restart-chaos/main.ts @@ -31,6 +31,8 @@ import { createDb } from "../../packages/db/src/index.js"; import { appendLedger, ledgerLine, writeSummary, type LedgerLine } from "./ledger.js"; import { collectWaveSpeed, speedMarkdown } from "./wave-speed.js"; import { TaskObserver, childSpeedTasksInTurn } from "./observer.js"; +import { closeJourneyApps } from "./journey-driver.js"; +import { matrixMayLaunch, planMatrixWave } from "./journey-matrix.js"; import { StormRun, TASK_FAULT_LOG, observedFaultSchedule, planWave, replayOnCurrentBuild, replayRecordFor, replayWave, type TaskFaultLog } from "./scheduler.js"; import { chaosLifecycleOverrides, seedChaosLab } from "./seed.js"; import { ChaosSupervisor } from "./supervisor.js"; @@ -76,7 +78,8 @@ async function loadPlan(): Promise<{ plan: WavePlan; keep: boolean }> { } return { plan, keep: command.keep }; } - return { plan: planWave(waveIdFor(command.options.seed), command.options), keep: command.keep }; + const waveId = waveIdFor(command.options.seed); + return { plan: command.options.matrix ? planMatrixWave(waveId, command.options) : planWave(waveId, command.options), keep: command.keep }; } function run(command: string, args: string[], cwd: string, label: string) { @@ -133,7 +136,8 @@ function preflight(plan: WavePlan) { ?? path.join(repositoryRoot, "packages/paperclip-runner/runner/target/debug/paperclip-runnerd"); if (!existsSync(runner)) throw new Error(`Runner binary missing at ${runner}. Run: pnpm --filter @paperclipai/paperclip-runner run build:runner-binaries`); // K14: the older checkout must be prepared (installed and built) before the wave, like this one. - const upgradeFrom = plan.options.upgradeFrom; + // `--server-from` runs the whole wave on another prepared checkout, prepared like an upgrade's older build. + const upgradeFrom = plan.options.upgradeFrom ?? plan.options.serverFrom; // One runner binary for both builds would not test the older runner's saved state. if (upgradeFrom && process.env.PAPERCLIP_RUNNER_BINARY) { throw new Error("--upgrade-from runs each build's own runner; unset PAPERCLIP_RUNNER_BINARY"); @@ -188,7 +192,7 @@ async function main(): Promise { const batchSize = crashPoints.length ? Math.max(1, Math.floor(plan.tasks.length / crashPoints.length)) : 0; if (crashPoint && plan.tasks.length !== batchSize * crashPoints.length) throw new Error(`--tasks must be a multiple of the ${crashPoints.length} crash point(s)`); const supervisor = await ChaosSupervisor.create({ repositoryRoot, waveDir, timeline, - ...(plan.options.upgradeFrom ? { upgradeFrom: plan.options.upgradeFrom } : {}) }); + ...(plan.options.upgradeFrom ?? plan.options.serverFrom ? { upgradeFrom: plan.options.upgradeFrom ?? plan.options.serverFrom } : {}) }); // Crash-point suite: one booted server, one point at a time. Each point gets // its own batch of tasks; the point is armed once the batch launched and the // server kills itself at its nth hit (restarted like K1). Invariants are @@ -429,7 +433,9 @@ async function main(): Promise { } // 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 - && (!storm || storm.mayLaunch(Date.now()))) { + && (!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(); @@ -601,6 +607,7 @@ async function main(): Promise { // Best effort for an interrupted or failed wave; the server may already be down. if (supervisor.serverReady) await recordWaveSpend().catch(() => undefined); await board?.close().catch(() => undefined); + await closeJourneyApps(); // A failure here (for example writing the report) must not skip stopping the controller and Postgres. try { if (sweeper) try { const owned = sweeper as DaytonaWaveSweeper; diff --git a/tests/restart-chaos/observer.ts b/tests/restart-chaos/observer.ts index 0b3a27c5abf..b395697c16b 100644 --- a/tests/restart-chaos/observer.ts +++ b/tests/restart-chaos/observer.ts @@ -10,7 +10,7 @@ import type { ChaosLab } from "./seed.js"; import type { Timeline } from "./timeline.js"; import type { AdapterId, GovernanceRecord, InterruptionSample, OracleArtifacts, RestSweep, RetryAttempt, RunSample, ScheduledFault, TaskPlan, TaskRecord, - TaskSample, + TaskSample, TriggerId, } from "./types.js"; import { restartProcessStartMatches } from "../runner-e2e/process-tree-owner.js"; import type { UiRetryClick } from "./board-ui.js"; @@ -19,6 +19,7 @@ import { awaitingQuestionAnswer, collectQuestionArtifacts, questionStep } from " import { captureRemoteWorkspaceFiles, readRemoteWorkspaceFile } from "./remote-workspace-files.js"; import { taskFaultSchedule } from "./scheduler.js"; import { g1ChildTitle, g1Decision, workloadDefinitions, type WorkloadAgents } from "./workloads.js"; +import { databaseUrlFromConfig, journeyDriverFor, type JourneyDriver } from "./journey-driver.js"; const IGNORED_DIRECTORIES = new Set([".git", ".bench", "__pycache__", "node_modules"]); const MAX_SNAPSHOT_FILES = 2_000; @@ -193,7 +194,7 @@ export class TaskObserver { constructor(readonly plan: TaskPlan, private readonly api: ChaosApi, private readonly lab: ChaosLab, private readonly timeline: Timeline, - private readonly supervisor: { serverReady: boolean; faultInProgress: boolean; serverBuild?: "upgrade_from" | "current" }, + private readonly supervisor: { serverReady: boolean; faultInProgress: boolean; serverBuild?: "upgrade_from" | "current"; paths?: { configPath: string } }, /** Paths oracle commands may not read: the operator's home, the instance root (secrets key, database) and results. */ private readonly protectedRoots: readonly string[] = [os.homedir()], private readonly board: { clickRetry(identifier: string, issueId: string): Promise } | null = null) { @@ -202,8 +203,13 @@ export class TaskObserver { preFaultFiles: null, lostFiles: null }; this.faultSchedule = taskFaultSchedule(plan); this.pendingFaults = [...this.faultSchedule]; + this.journey = journeyDriverFor({ plan, api, lab, timeline, record: this.record, + databaseUrl: () => databaseUrlFromConfig(supervisor.paths?.configPath) }); } + /** Q1–Q3: the board's side of the journey (journey-driver.ts). */ + readonly journey: JourneyDriver | null; + get definition() { return workloadDefinitions[this.plan.workload]; } /** @@ -222,7 +228,7 @@ export class TaskObserver { const until = Date.now() + deadlineMs; for (;;) { try { - await this.api.patch(`/api/issues/${this.record.issueId}`, { assigneeAgentId: this.lab.agents[this.plan.adapter].id }); + await this.api.patch(`/api/issues/${this.record.issueId}`, { assigneeAgentId: this.journey?.assigneeAgentId ?? this.lab.agents[this.plan.adapter].id }); break; } catch (error) { const transient = error instanceof ChaosApiError && (error.status === null || error.status >= 500); @@ -245,12 +251,14 @@ export class TaskObserver { const assign = options.assign !== false; if (this.plan.workload === "H1") await this.lowerBudgetToSpend(); try { + // Q3: connect the harness's app and hire its agent; a failed setup is a failed launch. + await this.journey?.prepare(); const issue = await this.api.post(`/api/companies/${this.lab.companyId}/issues`, { title: definition.title(this.plan.nonce), description: definition.description(this.plan.nonce, this.workloadAgents()), status: "todo", priority: "medium", - ...(assign ? { assigneeAgentId: this.lab.agents[this.plan.adapter].id } : {}), + ...(assign ? { assigneeAgentId: this.journey?.assigneeAgentId ?? this.lab.agents[this.plan.adapter].id } : {}), ...(definition.usesRepository ? { projectId: this.lab.projectId } : {}), }); this.record.issueId = String(issue.id); @@ -357,6 +365,13 @@ export class TaskObserver { case "tool": return this.record.benchFirstSeenAt ?? 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. + case "asked": return this.journey ? this.journey.armedAt("asked") : this.record.question?.seenAt ?? null; + case "answered": { + if (this.journey) return this.journey.armedAt("answered"); + const answer = this.record.question?.answer; + return answer ? (answer.error ? "missed" : answer.at) : null; + } case "stop": { const stop = this.record.stop; if (!stop) return null; @@ -434,6 +449,8 @@ export class TaskObserver { this.readySince = sample.reachable && sample.serverReady ? (this.readySince ?? now) : null; await this.observeBenchProgress(); await this.governanceStep(sample, now); + if (this.journey && await this.journey.step({ sample, now, pendingFaults: this.pendingFaults, + quiet: sample.reachable && sample.serverReady && !this.supervisor.faultInProgress })) this.restCandidateSince = null; if (await this.observeChildStalls(now)) { this.restCandidateSince = null; this.childrenSettledAt = null; @@ -735,6 +752,7 @@ export class TaskObserver { /** G1 before the board's decision, or H1 while its company is paused and not yet resumed. */ private awaitingBoard(): boolean { const governance = this.record.governance; + if (this.journey) return this.journey.awaitingBoard(); if (this.plan.workload === "G1") return Boolean(governance?.approvalSeenAt) && !governance?.decision; if (this.plan.workload === "H1") return Boolean(governance?.budgetPausedAt) && !governance?.resume && !governance?.pauseLostAt; if (this.plan.workload === "Q1") return awaitingQuestionAnswer(this.record); @@ -742,8 +760,9 @@ export class TaskObserver { } /** The board acts only after every fault this task planned has fired (or was missed), with the server ready. */ - private boardMayAct(sample: TaskSample): boolean { - return sample.reachable && sample.serverReady && !this.supervisor.faultInProgress && this.pendingFaults.length === 0; + private boardMayAct(sample: TaskSample, waitsForThisAction?: TriggerId): boolean { + return sample.reachable && sample.serverReady && !this.supervisor.faultInProgress + && this.pendingFaults.every(fault => waitsForThisAction !== undefined && fault.trigger === waitsForThisAction); } /** @@ -773,7 +792,8 @@ export class TaskObserver { if (this.plan.workload === "H1") await this.budgetStep(sample, now); if (this.plan.workload === "Q1") { await questionStep({ api: this.api, issueId: this.record.issueId, nonce: this.plan.nonce, record: this.record, sample, now, - boardMayAct: this.boardMayAct(sample), note: (kind, data) => this.timeline.add(kind, { taskId: this.plan.taskId, ...data }) }); + // An `answered` fault waits for the answer, so the board may answer before it fires. + boardMayAct: this.boardMayAct(sample, "answered"), note: (kind, data) => this.timeline.add(kind, { taskId: this.plan.taskId, ...data }) }); } } @@ -953,6 +973,7 @@ export class TaskObserver { ...(this.plan.workload === "H1" ? await this.finalPauseCheck(issueId) : {}), ...(this.record.governance ? { governance: this.record.governance } : {}), ...(this.plan.workload === "Q1" ? { question: await collectQuestionArtifacts(this.api, issueId, this.record) } : {}), + ...(this.journey ? { journey: await this.journey.collect(issueId).finally(() => this.journey!.close()) } : {}), }; } diff --git a/tests/restart-chaos/scheduler.ts b/tests/restart-chaos/scheduler.ts index d9acb1fef9a..c3383562fb9 100644 --- a/tests/restart-chaos/scheduler.ts +++ b/tests/restart-chaos/scheduler.ts @@ -1,4 +1,5 @@ import { isProviderFault, withProviderFaultRules } from "./provider-faults.js"; +import { takesJourneyTriggers } from "./journeys.js"; import { workloadDefinitions } from "./workloads.js"; import type { AdapterId, FaultKind, ScheduledFault, StormFault, StormPlan, TaskFaultKind, TaskPlan, TriggerId, WavePlan, WaveOptions, WorkloadId } from "./types.js"; import type { ReplayRef } from "./config.js"; @@ -46,6 +47,8 @@ export const TRIGGER_DELAY_RANGE_MS: Record, readonly approval: [0, 3_000], paused: [0, 3_000], stop: [0, 2_000], + asked: [0, 3_000], + answered: [0, 3_000], }; export function triggerFitsWorkload(trigger: TriggerId, workload: WorkloadId): boolean { @@ -54,11 +57,13 @@ export function triggerFitsWorkload(trigger: TriggerId, workload: WorkloadId): b if (trigger === "approval") return workload === "G1"; if (trigger === "paused") return workload === "H1"; if (trigger === "stop") return workload === "S1"; + if (trigger === "asked" || trigger === "answered") return takesJourneyTriggers(workload); return true; } /** The workload a trigger needs, for the error when none is selected. */ -const TRIGGER_WORKLOAD: Partial> = { comment: "C1", result: "C1", tool: "B5", approval: "G1", paused: "H1", stop: "S1" }; +const TRIGGER_WORKLOAD: Partial> = { comment: "C1", result: "C1", tool: "B5", approval: "G1", paused: "H1", stop: "S1", + asked: "Q1, Q2 or Q3", answered: "Q1, Q2 or Q3" }; /** Faults a `result` task may fire inside a storm: a fast restart that lands on the boundary. */ const STORM_RESULT_FAULTS: readonly FaultKind[] = ["K1", "K2"]; @@ -274,8 +279,8 @@ export function observedFaultSchedule(createdAt: number | null, faultIds: readon /** The faults a task schedules: its replayed sequence, or its one planned fault. */ export function taskFaultSchedule(task: TaskPlan): ScheduledFault[] { if (task.replayFaults?.length) return task.replayFaults; - // A storm crosses K13 tasks; a provider fault (PV*) restarts nothing. - if (task.fault === "K13" || isProviderFault(task.fault)) return []; + // A storm crosses K13 tasks; a provider fault (PV*) and a matrix baseline (K0) restart nothing. + 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"; if (trigger === "time") return [planned]; diff --git a/tests/restart-chaos/types.ts b/tests/restart-chaos/types.ts index e16036657f4..e6bec5e3acf 100644 --- a/tests/restart-chaos/types.ts +++ b/tests/restart-chaos/types.ts @@ -8,18 +8,23 @@ export type FaultKind = (typeof FAULT_KINDS)[number]; /** * A task's planned fault. "K13" (restart storm) tasks have none of their own; the wave's storm crosses them. * PV1–PV3 (provider message faults, provider-faults.ts) restart nothing: the task's rule is armed in the Codex shim. + * "K0" is a `--matrix` baseline cell: the same journey with no fault. */ -export type TaskFaultKind = FaultKind | "K13" | ProviderFaultKind; +export type TaskFaultKind = FaultKind | "K13" | "K0" | ProviderFaultKind; -/** G1: an approval-gated action across a restart. H1: a budget hard stop crossed mid-run with a restart (its own wave). */ -/** L1: high-output, high-accounting run for database contention (contention.ts). */ -export const WORKLOAD_IDS = ["A1", "B1", "B5", "C1", "M1", "G1", "H1", "R1", "S1", "Q1", "L1"] as const; +/** + * G1: an approval-gated action across a restart. H1: a budget hard stop crossed mid-run with a restart (its own wave). + * Q1: a saved question (question-handoff.ts). Q2, Q3: journeys.ts. + * L1: high-output, high-accounting run for database contention (contention.ts). + */ +export const WORKLOAD_IDS = ["A1", "B1", "B5", "C1", "M1", "G1", "H1", "R1", "S1", "Q1", "L1", "Q2", "Q3"] as const; export type WorkloadId = (typeof WORKLOAD_IDS)[number]; export const ADAPTER_IDS = ["codex", "claude", "opencode"] as const; export type AdapterId = (typeof ADAPTER_IDS)[number]; -export const TRIGGER_IDS = ["time", "comment", "retry", "queued", "tool", "result", "approval", "paused", "stop"] as const; +/** `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]; export const ENVIRONMENT_KINDS = ["local", "daytona"] as const; @@ -93,6 +98,10 @@ export interface WaveOptions { providerFaultDelaysMs?: number[]; /** `--db-pressure `: harness-owned writers touching the wave's running run rows back to back (contention.ts). */ dbPressure?: number; + /** `--matrix`: one task per workload × adapter × (K0, or each fault at each of the workload's triggers). */ + matrix?: boolean; + /** `--server-from `: every server of the wave runs from this prepared checkout (for example a pre-fix commit). */ + serverFrom?: string; } export interface StormFault { @@ -261,6 +270,8 @@ export interface OracleArtifacts { providerFaultEvents?: ProviderFaultEvent[]; /** L1: accounting and run-log evidence read from the wave database at rest. */ contention?: import("./contention.js").ContentionEvidence; + /** Q2, Q3: what the board did and what the task, its runs and the harness's app saw. */ + journey?: import("./journeys.js").JourneyArtifacts; } /** What the board did for a governance task (G1, H1), and what the observer saw. */ @@ -347,6 +358,8 @@ export interface TaskRecord { serverBuildAtLaunch?: "upgrade_from" | "current"; /** Q1: the saved question and the board's answer. */ question?: QuestionRecord; + /** Q2, Q3: the board's journey actions (journey-driver.ts). */ + journey?: import("./journeys.js").JourneyRecord; } export interface TaskGrade { diff --git a/tests/restart-chaos/workloads.ts b/tests/restart-chaos/workloads.ts index 6cc01a57211..4e01eb612bb 100644 --- a/tests/restart-chaos/workloads.ts +++ b/tests/restart-chaos/workloads.ts @@ -1,5 +1,6 @@ import { questionHandoffWorkload } from "./question-handoff.js"; import { l1Workload } from "./contention.js"; +import { journeyWorkloadDefinitions } from "./journeys.js"; import type { OracleArtifacts, OracleResult, WorkloadId } from "./types.js"; /** Deterministic bench output: sum of squares 1..240. */ @@ -170,6 +171,7 @@ function budgetChecks(artifacts: OracleArtifacts) { export const workloadDefinitions: Record = { Q1: questionHandoffWorkload, L1: l1Workload, + ...journeyWorkloadDefinitions, G1: { id: "G1", label: "Approval-gated action across a restart",