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