diff --git a/tests/restart-chaos/README.md b/tests/restart-chaos/README.md index 994805d9ce5..23ac4751495 100644 --- a/tests/restart-chaos/README.md +++ b/tests/restart-chaos/README.md @@ -22,7 +22,7 @@ node scripts/typecheck-test-projects.mjs tests/restart-chaos # typecheck ``` Options: `--triggers time,comment,retry,queued,tool,result` (fault timing, below), -`--workloads A1,B1,B5,C1`, `--budget-cents` (company hard cap, +`--workloads A1,B1,B5,C1` (opt-in: G1, H1, L1, R1, S1), `--db-pressure ` (below), `--budget-cents` (company hard cap, 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 @@ -403,6 +403,92 @@ question must settle on its own first. The oracle checks these points: Q1 is not a default workload. +## Database contention (L1, `--db-pressure`) + +```sh +node cli/node_modules/tsx/dist/cli.mjs tests/restart-chaos/main.ts --env local \ + --adapters codex,claude --workloads L1 --faults K1 --concurrency 4 --tasks 4 --seed 20385 --db-pressure 4 +``` + +Usage receipts and streamed run-log appends update the same `heartbeat_runs` row. In #20385 a +receipt's repeatable-read snapshot went stale when a log append committed between its company check +and its run-row lock; four retries ran out with SQLSTATE 40001, the event stayed in the runner's +retained outbox, and every later terminal event was blocked (`native_event_replay_conflict: +runner_event_commit_stalled ... last cause: Error`). + +**L1** asks the agent to run five shell commands as five tool calls, each printing 3,000 lines +(about 300 KB), then save a `brief` naming each command's END line. Each run emits many log +events and many usage reports. L1 is opt-in (not a default workload). + +**`--db-pressure `** (1–8) starts harness-owned writers against the wave's own database. +Each one runs `update heartbeat_runs set updated_at = updated_at` on the wave company's `running` +rows back to back. No value changes, but every statement writes a new row version, which is what a +fast stream does to a run row. Locally the race window between a snapshot and its row lock is well +under a millisecond, so without pressure the race almost never fires; the writers stand in for a +hosted database whose round trip widens that window. The writers stop before the R6 sweep; +`db_pressure_started` / `db_pressure_stopped` (with update and error counts) are in the timeline. + +At rest the harness reads each L1 task's runs from the wave database, and the T-oracle adds: + +| Check | Fails when | +| --- | --- | +| `no_event_commit_stall` | a run ended `native_event_replay_conflict` or `runner_event_commit_stalled` | +| `usage_receipts_committed_once` | one producer event was journaled twice for a receipt source | +| `terminal_receipts_committed` | a succeeded native run has no terminal (`turn.*`) receipt: its terminal event never left the outbox | +| `finished_runs_accounted` | a finished native run has no `cost_accounted_at` | +| `run_spend_matches_cost_events` | a run's recorded cost differs from its cost events (0.5% or 0.01 cent) | +| `company_spend_settled` | the board's cost summary differs from the cost-event sum, or runs still wait for accounting | +| `contention_produced` | the task produced fewer than 40 log events or 6 receipts (it did not test anything) | + +`contention-db.test.ts` checks the knob itself on real PostgreSQL: with four writers, the +pre-#20385 repeatable-read receipt write exhausted its four retries with 40001 in 8 of 8 attempts +locally, and the current READ COMMITTED writer (`persistUsageReceipt`) committed all 8. A live L1 +wave needs the runner stall (20 failed deliveries over at least 45 s) to show +`native_event_replay_conflict`; run it against the parent of the #20385 merge (red) and against +master (green) with the command above. + +## Universal checks (U1–U4) + +Every task of every wave is also graded on four scenario-independent rules (`universal.ts`), whatever +its workload. They catch a stuck or misleading task that no scenario anticipated. + +| Rule | Fails when | Bound (env override) | +| --- | --- | --- | +| U1 liveness | a non-terminal task has neither a live owner (a running run with recent evidence, a run queued within the queue bound, a retry timer, a system recovery action) nor a visible, actionable item (Retry, a question or `userAction`, a failed notice, a board recovery action, an execution blocker, `blocked` or `in_review`) for longer than the bound, counting only time with the server ready | 3 min ownerless (`RESTART_CHAOS_U1_ORPHAN_SECONDS`); a run counts as an owner for 5 min of queueing (`RESTART_CHAOS_U1_QUEUED_SECONDS`) | +| U2 retry loop | more than N consecutive failed runs with the same status and code while the issue status stays the same | 3 (`RESTART_CHAOS_U2_SAME_FAILURE_RUNS`) | +| U3 diagnosable | a failed run has no code or a generic one (`adapter_failed`, `unknown`, `error`, ...), its stored error is empty or names only a bare cause (`last cause: Error`), or a recovery action's cause or next action is empty or bare | none | +| U4 stale notice | a Retry or failed notice stays on a task after its latest run succeeded, longer than the bound or at rest | 60 s (`RESTART_CHAOS_U4_STALE_SECONDS`) | + +U1, U2 and U4 fail the task (signature `U:...`). U3 is a warning: it is reported in each grade +(`grade.universal`), the ledger (`universal`, `universalWarnings`) and the wave's `SUMMARY.md`, and +never fails a task. U3 reads run error text and recovery-action causes from the wave database at +wave end, because the board's run list hides most error text. + +The rules and bounds come from an audit of the saved evidence of 74 waves (1,592 tasks, 1,465 of +them passing R1–R6 and T; 28 waves where every task passed), run with: + +```sh +node cli/node_modules/tsx/dist/cli.mjs tests/restart-chaos/universal-audit.ts [...] +``` + +| Rule | Flagged tasks | On passing tasks | On all-pass waves | +| --- | --- | --- | --- | +| U1 | 7 | 0 | 0 | +| U2 | 1 | 0 | 0 | +| U3 | 22 | 3 | 0 | +| U4 | 0 | 0 | 0 | + +On passing tasks the longest ownerless stretch was 19 s and the longest queue wait 136 s (p99 40 s), +and no failure repeated more than twice; the bounds leave about 9×, 2× and one more repeat of +headroom. U1 flags a C1 task that sat 181 s in `todo` with a resolved interruption (it also failed +R5), and six R1 tasks in three waves whose run stayed `queued` for 8 minutes while the issue still +showed a retry row: the #20361 shape, which those waves graded only as a late R1. A retry row counts +as a live timer only until its run is dispatched. U2 flags a claude task that failed four times in a +row with `adapter_failed`. Every U3 +flag is `adapter_failed`, including three tasks that passed after a retry, which is why U3 stays a +warning until `adapter_failed` names its cause. The audit saw only failure codes; the error-text +part of U3 is checked only in waves run with this harness. + ## Fault timing (`--triggers`) Each task samples one trigger with its cell. `time` (default) fires at the @@ -433,8 +519,10 @@ re-arms the task's own triggered faults instead of using their offsets. | `seed.ts` | Chaos Lab company, encrypted provider secrets, native Codex / ACPX Claude / OpenCode agents from the runner-e2e catalog, and the `chaos-sandbox` project (git repo: python CLI, one failing test, `bench.sh` ≈ 4 min) with isolated git-worktree workspaces. It first turns on the instance `enableIsolatedWorkspaces` setting; without it the server ignores the project policy and every repo task shares the primary checkout. | | `scheduler.ts` | Seeded plan: workload × adapter × fault cells and a random fire time per task. Saved to `results//plan.json`. `--replay` reruns one task alone with the same choices and fires the recorded sequence of faults that crossed it (other tasks' faults included) at the same offsets from launch. | | `observer.ts` | Polls each issue and its runs. When a task rests needing Retry it calls `POST /api/issues/:id/retry` once (once more only if a later fault started after that click), with the body `useTaskRetry.ts` sends. Collects T-oracle artifacts and the R6 sweep. | -| `grader.ts` | Pure R1–R6 grading over the samples. | | `config-faults.ts`, `config-wave.ts` | Config faults: the reference catalog, seeded plan and oracle (pure), and the wave runner that breaks, restarts, repairs and grades. | +| `grader.ts` | Pure R1–R6 grading over the samples, plus the universal checks. | +| `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. | | `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. | diff --git a/tests/restart-chaos/config.ts b/tests/restart-chaos/config.ts index a51e02c61c9..9de23edefb1 100644 --- a/tests/restart-chaos/config.ts +++ b/tests/restart-chaos/config.ts @@ -106,13 +106,14 @@ export function parseCommand(argv: readonly string[], now = Date.now()): ParsedC "storage-fault-seconds": { type: "string" }, "provider-fault-message": { type: "string" }, "provider-fault-delays": { type: "string" }, + "db-pressure": { 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"] 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"] 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, @@ -152,8 +153,8 @@ export function parseCommand(argv: readonly string[], now = Date.now()): ParsedC const options: WaveOptions = { env, adapters, - // The governance scenarios (G1, H1) and the saved question (Q1) are opt-in: H1 needs a wave of its own. - workloads: list(values.workloads, workloadChoices, "workloads", workloadChoices.filter(id => id !== "G1" && id !== "H1" && id !== "R1" && id !== "S1" && id !== "Q1" + // 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))), // A crash-point wave schedules no fault; K1 is the restart its self-kill gets. faults: allFaults.filter((fault): fault is FaultKind => !isProviderFault(fault)), @@ -249,6 +250,7 @@ export function parseCommand(argv: readonly string[], now = Date.now()): ParsedC + "for chaos tasks. Lower one of them (RESTART_CHAOS_DAYTONA_VCPU_BUDGET sets the budget)."); } } + if (values["db-pressure"] !== undefined) options.dbPressure = integer(values["db-pressure"], "db-pressure", 0, 1, 8); if (storage !== "local" && storage !== "s3") throw new Error("--storage must be local or s3"); if (storage === "s3") options.storage = "s3"; if (values["storage-faults"] !== undefined) { diff --git a/tests/restart-chaos/contention-db.test.ts b/tests/restart-chaos/contention-db.test.ts new file mode 100644 index 00000000000..b8c41c69392 --- /dev/null +++ b/tests/restart-chaos/contention-db.test.ts @@ -0,0 +1,72 @@ +/** + * Harness-level check that `--db-pressure` reproduces the #20385 race against + * real PostgreSQL: under the pressure writers, a usage receipt written with the + * pre-#20385 repeatable-read transaction exhausts its retries with 40001, + * while the current READ COMMITTED writer (`persistUsageReceipt`) commits. + */ +import { randomUUID } from "node:crypto"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { agents, companies, createDb, heartbeatRuns, type Db } from "../../packages/db/src/index.js"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../server/src/__tests__/helpers/embedded-postgres.js"; +import { withOperationAccountingTransaction } from "../../server/src/services/accounting-transaction.js"; +import { persistUsageReceipt, persistUsageReceiptInTransaction, type UsageReceiptEnvelope } from "../../server/src/services/usage-receipts.js"; +import { RunRowPressure } from "./contention.js"; + +const support = await getEmbeddedPostgresTestSupport(); +const RECEIPTS = 8; + +describe.skipIf(!support.supported)("--db-pressure against real PostgreSQL (#20385)", () => { + let database: Awaited>; + let db: Db; + beforeAll(async () => { + database = await startEmbeddedPostgresTestDatabase("restart-chaos-contention-"); + db = createDb(database.connectionString); + }, 120_000); + afterAll(async () => { await db?.$client.end(); await database?.cleanup(); }); + + async function runningRun() { + const companyId = randomUUID(), agentId = randomUUID(), runId = randomUUID(), sourceId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Contention fixture", issuePrefix: companyId.slice(0, 8) }); + await db.insert(agents).values({ id: agentId, companyId, name: "Contention agent", adapterType: "process", status: "active" }); + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, status: "running", + usageJson: { accountingReceiptSourceId: sourceId, accountingReceiptSequence: 0 } }); + const receipt = (sequence: number): UsageReceiptEnvelope => ({ schema: "paperclip/accounting-receipt/v1", + id: randomUUID(), companyId, runId, sourceId, sequence, receivedAt: new Date().toISOString(), adapterType: "process", + receipt: { complete: false, usageBasis: "per_run", costUsd: sequence / 100, billingType: "metered_api" } }); + return { companyId, runId, receipt }; + } + + async function underPressure(companyId: string, work: () => Promise) { + const pressure = new RunRowPressure(database.connectionString, companyId, 4).start(); + // Let the writers reach steady state. + while (pressure.stats.rowsTouched < 50) await new Promise(resolve => setTimeout(resolve, 10)); + try { return await work(); } finally { + const stats = await pressure.stop(); + expect(stats.errors).toBe(0); + } + } + + it("makes the pre-#20385 repeatable-read receipt writer exhaust its retries with 40001", async () => { + const f = await runningRun(); + const outcomes = await underPressure(f.companyId, async () => { + const results: string[] = []; + for (let sequence = 1; sequence <= RECEIPTS; sequence += 1) { + // The pre-#20385 persistUsageReceipt: the same work in the default repeatable-read transaction. + results.push(await withOperationAccountingTransaction(db, f.companyId, tx => persistUsageReceiptInTransaction(tx, f.receipt(sequence))) + .then(() => "committed", (error: { code?: string; cause?: { code?: string } }) => error.code ?? error.cause?.code ?? "error")); + } + return results; + }); + // Locally every one exhausts (8/8); at least half leaves room for a slow CI host. + expect(outcomes.filter(outcome => outcome === "40001").length).toBeGreaterThanOrEqual(RECEIPTS / 2); + }, 60_000); + + it("commits every receipt with the current READ COMMITTED writer", async () => { + const f = await runningRun(); + await underPressure(f.companyId, async () => { + for (let sequence = 1; sequence <= RECEIPTS; sequence += 1) await persistUsageReceipt(db, f.receipt(sequence)); + }); + const [row] = await db.$client.unsafe("select count(*)::int as n from run_usage_receipts where run_id = $1", [f.runId]) as Array<{ n: number }>; + expect(row?.n).toBe(RECEIPTS); + }, 60_000); +}); diff --git a/tests/restart-chaos/contention.test.ts b/tests/restart-chaos/contention.test.ts new file mode 100644 index 00000000000..1dc41c7b253 --- /dev/null +++ b/tests/restart-chaos/contention.test.ts @@ -0,0 +1,54 @@ +import { describe, expect, it } from "vitest"; +import { parseCommand } from "./config.js"; +import { contentionChecks, l1Workload, L1_CHUNKS, type ContentionEvidence, type ContentionRun } from "./contention.js"; +import type { OracleArtifacts } from "./types.js"; + +const goodRun = (extra: Partial = {}): ContentionRun => ({ id: "r1", status: "succeeded", errorCode: null, error: null, + logEvents: 120, logBytes: 900_000, receipts: 14, duplicateReceipts: [], terminalReceipt: true, nativeProducer: true, costAccounted: true, + recordedCostCents: 12.5, costEventCents: 12.5, ...extra }); +const evidence = (runs: ContentionRun[], extra: Partial = {}): ContentionEvidence => ({ runs, + companyCostEventCents: 12.5, summary: { spendCents: 12.5, pendingRunCount: 0 }, error: null, ...extra }); +const failing = (result: ReturnType) => result.filter(check => !check.ok).map(check => check.name); + +describe("L1 contention oracle", () => { + it("passes settled accounting", () => { + expect(failing(contentionChecks(evidence([goodRun()])))).toEqual([]); + }); + + it("fails the #20385 shape: a commit stall that left the terminal receipt uncommitted", () => { + const stalled = goodRun({ id: "r2", status: "failed", errorCode: "native_event_replay_conflict", terminalReceipt: false, costAccounted: false, + error: "native_event_replay_conflict: runner_event_commit_stalled: runner events kept failing to commit without progress (last cause: Error at source seq 9)" }); + expect(failing(contentionChecks(evidence([goodRun(), stalled])))).toEqual(["no_event_commit_stall", "finished_runs_accounted"]); + expect(failing(contentionChecks(evidence([goodRun({ terminalReceipt: false })])))).toEqual(["terminal_receipts_committed"]); + }); + + it("fails duplicated receipts and spend that does not add up", () => { + expect(failing(contentionChecks(evidence([goodRun({ duplicateReceipts: ["s:evt-3"] })])))).toEqual(["usage_receipts_committed_once"]); + expect(failing(contentionChecks(evidence([goodRun({ costEventCents: 25 })])))).toEqual(["run_spend_matches_cost_events"]); + expect(failing(contentionChecks(evidence([goodRun()], { summary: { spendCents: 12.5, pendingRunCount: 1 } })))).toEqual(["company_spend_settled"]); + }); + + it("fails a task that did not produce the contention it tests", () => { + expect(failing(contentionChecks(evidence([goodRun({ logEvents: 5, receipts: 2 })])))).toEqual(["contention_produced"]); + expect(failing(contentionChecks(undefined))).toEqual(["contention_evidence_collected"]); + }); + + it("grades the brief and the accounting together", () => { + const nonce = "abcd1234"; + const brief = [...Array.from({ length: L1_CHUNKS }, (_, index) => `- L1-${nonce}-c${index + 1}-END`), `L1-${nonce}-DONE`].join("\n"); + const artifacts: OracleArtifacts = { issueStatus: "done", documents: { brief }, agentComments: [], files: {}, commands: {}, workspacePath: null, + contention: evidence([goodRun()]) }; + expect(l1Workload.grade(nonce, artifacts).verdict).toBe("pass"); + const missing = { ...artifacts, documents: { brief: brief.replace(`L1-${nonce}-c3-END`, "") } }; + expect(l1Workload.grade(nonce, missing).checks.find(check => check.name === "brief_has_every_chunk_end")?.ok).toBe(false); + expect(l1Workload.description(nonce)).toContain(`seq -f "L1-${nonce}-c1 row`); + }); + + it("parses --db-pressure and keeps L1 out of the default workloads", () => { + const parsed = parseCommand(["--workloads", "L1", "--db-pressure", "4", "--seed", "1"]); + expect(parsed.mode === "wave" && parsed.options.dbPressure).toBe(4); + const defaults = parseCommand(["--seed", "1"]); + expect(defaults.mode === "wave" && defaults.options.workloads.includes("L1")).toBe(false); + expect(() => parseCommand(["--db-pressure", "9"])).toThrow(/between 1 and 8/); + }); +}); diff --git a/tests/restart-chaos/contention.ts b/tests/restart-chaos/contention.ts new file mode 100644 index 00000000000..688785a8be8 --- /dev/null +++ b/tests/restart-chaos/contention.ts @@ -0,0 +1,219 @@ +/** + * Database contention: L1 (high-output, high-accounting) and the harness-side + * run-row pressure knob (`--db-pressure `). + * + * Usage receipts and streamed run-log appends update the same `heartbeat_runs` + * row. Before #20385 a receipt's repeatable-read snapshot went stale when a + * run-log append committed between its company check and its run-row lock; + * four retries ran out with SQLSTATE 40001, the event stayed in the runner's + * retained outbox, and every later terminal event was blocked behind it + * (`native_event_replay_conflict: runner_event_commit_stalled ... last cause: Error`). + * + * L1 makes each task stream several hundred kilobytes of tool output across + * several tool calls, so a run emits many log events and many usage reports. + * `--db-pressure` adds harness-owned writers that touch the wave's running run + * rows back to back, the way a fast stream does against a hosted database + * whose round trip widens the race window. The writers change no value + * (`updated_at = updated_at`) and only touch the wave's own isolated database. + */ +import { setTimeout as delay } from "node:timers/promises"; +import { createDb, type Db } from "../../packages/db/src/index.js"; +import type { OracleArtifacts, OracleResult } from "./types.js"; +import type { WorkloadDefinition } from "./workloads.js"; + +export const L1_CHUNKS = 5; +export const L1_LINES = 3_000; +/** Proof that the task actually streamed and accounted: minimum log events and usage receipts across its runs. */ +export const L1_MIN_LOG_EVENTS = 40; +export const L1_MIN_RECEIPTS = 6; + +const chunkEnd = (nonce: string, chunk: number) => `L1-${nonce}-c${chunk}-END`; +const doneMarker = (nonce: string) => `L1-${nonce}-DONE`; +const chunkCommand = (nonce: string, chunk: number) => + `seq -f "L1-${nonce}-c${chunk} row %05g ${"x".repeat(72)}" 1 ${L1_LINES}; echo "${chunkEnd(nonce, chunk)}"`; + +const TERMINAL_EVENTS = ["turn.completed", "turn.failed", "turn.cancelled", "turn.interrupted"]; +const COMMIT_STALL = /runner_event_commit_stalled|native_event_replay_conflict/; + +export interface ContentionRun { + id: string; status: string; errorCode: string | null; error: string | null; + logEvents: number; logBytes: number | null; + receipts: number; + /** Producer events journaled more than once for the same receipt source (must be none). */ + duplicateReceipts: string[]; + terminalReceipt: boolean; + nativeProducer: boolean; + costAccounted: boolean; + /** Cost the run's accounting recorded (usage_json), and what its cost events add up to, in cents. */ + recordedCostCents: number | null; + costEventCents: number; +} + +export interface ContentionEvidence { + runs: ContentionRun[]; + /** Sum of the wave company's cost events in cents, and the board's all-time cost summary. */ + companyCostEventCents: number; + summary: CostSummary | null; + error: string | null; +} + +/** The fields of `GET /api/companies/:id/costs/summary?period=all` the oracle compares. */ +export interface CostSummary { spendCents: number; pendingRunCount: number } + +const num = (value: unknown) => (value === null || value === undefined || value === "" ? null : Number(value)); + +/** Accounting and run-log evidence for one task's runs (the ids the observer saw), read from the wave database at rest. */ +export async function collectContentionEvidence(db: Db, companyId: string, runIds: readonly string[], summary: CostSummary | null): Promise { + try { + const rows = await db.$client.unsafe(` + select r.id::text as id, r.status, r.error_code as "errorCode", left(r.error, 400) as error, r.log_bytes as "logBytes", + r.cost_accounted_at is not null as "costAccounted", + coalesce(r.usage_json->>'accountingReceiptOwner', '') = 'native_producer' as "nativeProducer", + coalesce(r.usage_json->>'cacheAdjustedCostUsd', r.usage_json->>'costUsdExact', r.usage_json->>'costUsd') as "recordedCostUsd", + (select count(*) from heartbeat_run_events e where e.run_id = r.id)::int as "logEvents", + (select count(*) from run_usage_receipts u where u.run_id = r.id)::int as receipts, + (select coalesce(array_agg(dup.k), '{}') from ( + select u.source_id::text || ':' || (u.receipt_json->'producerEvent'->>'sourceEventId') as k + from run_usage_receipts u where u.run_id = r.id and u.receipt_json->'producerEvent' ? 'sourceEventId' + group by 1 having count(*) > 1) dup) as "duplicateReceipts", + exists (select 1 from run_usage_receipts u where u.run_id = r.id + and u.receipt_json->'producerEvent'->>'eventType' = any($3::text[])) as "terminalReceipt", + (select coalesce(sum(c.cost_cents), 0) from cost_events c where c.heartbeat_run_id = r.id)::float8 as "costEventCents" + from heartbeat_runs r + where r.company_id::text = $1 and r.id::text = any($2::text[]) + order by r.created_at`, [companyId, [...runIds], TERMINAL_EVENTS]) as Array>; + const [company] = await db.$client.unsafe( + "select coalesce(sum(cost_cents), 0)::float8 as cents from cost_events where company_id::text = $1", [companyId]) as Array<{ cents: number }>; + return { + runs: rows.map(row => ({ + id: String(row.id), status: String(row.status), errorCode: (row.errorCode as string | null) ?? null, error: (row.error as string | null) ?? null, + logEvents: Number(row.logEvents), logBytes: num(row.logBytes), receipts: Number(row.receipts), + duplicateReceipts: (row.duplicateReceipts as string[] | null) ?? [], terminalReceipt: row.terminalReceipt === true, + nativeProducer: row.nativeProducer === true, costAccounted: row.costAccounted === true, + recordedCostCents: row.recordedCostUsd === null || row.recordedCostUsd === undefined ? null : Number(row.recordedCostUsd) * 100, + costEventCents: Number(row.costEventCents), + })), + companyCostEventCents: Number(company?.cents ?? 0), summary, error: null, + }; + } catch (error) { + return { runs: [], companyCostEventCents: 0, summary, error: error instanceof Error ? error.message : String(error) }; + } +} + +const FINISHED = new Set(["succeeded", "failed", "timed_out", "cancelled", "interrupted"]); +const centsMatch = (a: number, b: number) => Math.abs(a - b) <= Math.max(0.01, Math.abs(b) * 0.005); + +/** The accounting oracle shared by L1 and any wave with contention evidence. */ +export function contentionChecks(evidence: ContentionEvidence | undefined): OracleResult["checks"] { + if (!evidence) return [{ name: "contention_evidence_collected", ok: false, detail: "no database evidence" }]; + if (evidence.error) return [{ name: "contention_evidence_collected", ok: false, detail: evidence.error }]; + const runs = evidence.runs; + const stalled = runs.filter(run => COMMIT_STALL.test(`${run.errorCode ?? ""} ${run.error ?? ""}`)); + const duplicates = runs.filter(run => run.duplicateReceipts.length); + const native = runs.filter(run => run.nativeProducer); + // A succeeded run proves its terminal event was delivered; its receipt must be journaled. + const missingTerminal = native.filter(run => run.status === "succeeded" && !run.terminalReceipt); + const unaccounted = native.filter(run => FINISHED.has(run.status) && !run.costAccounted); + const mismatched = runs.filter(run => run.costAccounted && run.recordedCostCents !== null && !centsMatch(run.costEventCents, run.recordedCostCents)); + const logEvents = runs.reduce((sum, run) => sum + run.logEvents, 0); + const receipts = runs.reduce((sum, run) => sum + run.receipts, 0); + const list = (items: ContentionRun[], show: (run: ContentionRun) => string) => items.map(show).join("; ") || "none"; + return [ + { name: "no_event_commit_stall", ok: !stalled.length, + detail: list(stalled, run => `${run.id} ${run.errorCode}: ${(run.error ?? "").slice(0, 200)}`) }, + { name: "usage_receipts_committed_once", ok: !duplicates.length, detail: list(duplicates, run => `${run.id}: ${run.duplicateReceipts.join(",")}`) }, + { name: "terminal_receipts_committed", ok: !missingTerminal.length, detail: list(missingTerminal, run => run.id) }, + { name: "finished_runs_accounted", ok: !unaccounted.length, detail: list(unaccounted, run => `${run.id}(${run.status})`) }, + { name: "run_spend_matches_cost_events", ok: !mismatched.length, + detail: list(mismatched, run => `${run.id} recorded=${run.recordedCostCents} events=${run.costEventCents}`) }, + // The board's spend is the cost-event sum; at rest no finished run may still wait for accounting. + { name: "company_spend_settled", + ok: evidence.summary !== null && centsMatch(evidence.summary.spendCents, evidence.companyCostEventCents) && evidence.summary.pendingRunCount === 0, + detail: evidence.summary ? `summary=${evidence.summary.spendCents} events=${evidence.companyCostEventCents} pending_runs=${evidence.summary.pendingRunCount}` : "summary unread" }, + // Not a product property: proof that this task produced the contention it is meant to test. + { name: "contention_produced", ok: logEvents >= L1_MIN_LOG_EVENTS && receipts >= L1_MIN_RECEIPTS, + detail: `log_events=${logEvents} receipts=${receipts} log_bytes=${runs.reduce((sum, run) => sum + (run.logBytes ?? 0), 0)}` }, + ]; +} + +export const l1Workload: WorkloadDefinition = { + id: "L1", + label: "High-output, high-accounting run (database contention)", + usesRepository: false, + fireWindowMs: [30_000, 120_000], + budgetMs: 15 * 60_000, + title: nonce => `Chaos L1 loud run ${nonce}`, + description: nonce => [ + `Run each of these ${L1_CHUNKS} shell commands, one at a time, as ${L1_CHUNKS} separate tool calls, in order.`, + "Let each print its full output to the terminal: do not pipe, redirect, truncate or combine them.", + ...Array.from({ length: L1_CHUNKS }, (_, index) => `${index + 1}. \`${chunkCommand(nonce, index + 1)}\``), + "Each command ends with one line of the form `L1-...-END`.", + "Then save this issue's document with key `brief` (title \"Brief\"), in markdown, containing each of those END lines exactly,", + `each on its own line, and the line \`${doneMarker(nonce)}\` on its own line. Do not create files or other tasks.`, + "When the document is saved, mark this task done.", + ].join("\n"), + collect: { documents: ["brief"], files: () => [], commands: [] }, + grade(nonce: string, artifacts: OracleArtifacts): OracleResult { + const brief = artifacts.documents.brief ?? ""; + const lines = new Set(brief.split(/\r?\n/).map(line => line.trim().replace(/^[-*]\s+/, "").replace(/^`|`$/g, ""))); + const missing = Array.from({ length: L1_CHUNKS }, (_, index) => chunkEnd(nonce, index + 1)).filter(marker => !lines.has(marker)); + const checks = [ + { name: "issue_done", ok: artifacts.issueStatus === "done", detail: `status=${artifacts.issueStatus ?? "unknown"}` }, + { name: "brief_has_every_chunk_end", ok: !missing.length, detail: missing.length ? `missing ${missing.join(",")}` : `${L1_CHUNKS} chunks` }, + { name: "brief_has_done_marker", ok: lines.has(doneMarker(nonce)) }, + ...contentionChecks(artifacts.contention), + ]; + return { verdict: checks.every(check => check.ok) ? "pass" : "fail", checks }; + }, +}; + +export interface PressureStats { writers: number; updates: number; rowsTouched: number; errors: number; lastError: string | null; startedAt: number; stoppedAt: number | null } + +/** + * Harness-owned writers that touch the wave company's running run rows back + * to back. Each statement keeps every value (`updated_at = updated_at`) but + * still writes a new row version, which is exactly what invalidates a + * repeatable-read snapshot that later locks the row. + */ +export class RunRowPressure { + readonly stats: PressureStats; + #stopped = false; + #workers: Promise[] = []; + #db: Db; + + constructor(connectionString: string, private readonly companyId: string, writers: number) { + this.#db = createDb(connectionString); + this.stats = { writers, updates: 0, rowsTouched: 0, errors: 0, lastError: null, startedAt: Date.now(), stoppedAt: null }; + } + + start() { + for (let index = 0; index < this.stats.writers; index += 1) this.#workers.push(this.#loop()); + return this; + } + + async #loop() { + while (!this.#stopped) { + try { + const rows = await this.#db.$client.unsafe( + "update heartbeat_runs set updated_at = updated_at where company_id::text = $1 and status = 'running' returning id", [this.companyId]); + this.stats.updates += 1; + this.stats.rowsTouched += rows.length; + // Nothing running: idle briefly instead of spinning on an empty update. + if (!rows.length) await delay(50); + } catch (error) { + // The server (and its database) restart during faults; keep going once it is back. + this.stats.errors += 1; + this.stats.lastError = error instanceof Error ? error.message.slice(0, 200) : String(error); + await delay(500); + } + } + } + + async stop(): Promise { + this.#stopped = true; + await Promise.all(this.#workers); + this.stats.stoppedAt = Date.now(); + await this.#db.$client.end({ timeout: 5 }).catch(() => undefined); + return this.stats; + } +} diff --git a/tests/restart-chaos/grader.ts b/tests/restart-chaos/grader.ts index 4fc16669af6..14815988041 100644 --- a/tests/restart-chaos/grader.ts +++ b/tests/restart-chaos/grader.ts @@ -1,5 +1,6 @@ import { UI_RETRY_BUTTON_MISSING, UI_RETRY_NO_REQUEST, type FaultReceipt, type OracleResult, type RuleId, type RuleResult, type TaskGrade, type TaskRecord, type TaskSample } from "./types.js"; import { g1Decision } from "./workloads.js"; +import { enforcedUniversalFailures, gradeUniversal, type UniversalDbEvidence } from "./universal.js"; /** Pure R1–R6 grading over observer samples (plan §2). No I/O. */ @@ -245,7 +246,8 @@ export function gradeR6(record: TaskRecord): RuleResult { return leaks.length ? { verdict: "fail", detail: `leak: ${leaks.join("; ")}` } : { verdict: "pass", detail: "clean rest" }; } -export function gradeTask(record: TaskRecord, faults: readonly FaultReceipt[], budgetMs: number, oracle: OracleResult): TaskGrade { +export function gradeTask(record: TaskRecord, faults: readonly FaultReceipt[], budgetMs: number, oracle: OracleResult, + universalDb?: UniversalDbEvidence): TaskGrade { const crossing = crossingFaults(record, faults); const rules: Record = { R1: gradeR1(record, budgetMs, faults), @@ -256,9 +258,13 @@ export function gradeTask(record: TaskRecord, faults: readonly FaultReceipt[], b R6: gradeR6(record), }; const failed = (Object.entries(rules) as Array<[RuleId, RuleResult]>).find(([, rule]) => rule.verdict === "fail"); - const pass = !failed && oracle.verdict === "pass"; + // Universal checks run on every task whatever its workload; only enforced ones fail it (universal.ts). + const universal = gradeUniversal(record, universalDb ? { db: universalDb } : {}); + const [universalFailure] = enforcedUniversalFailures(universal); + const pass = !failed && oracle.verdict === "pass" && !universalFailure; let signature: string | null = null; if (failed) signature = `${failed[0]}:${failed[1].detail.split(/[\s:(]/)[0]}:${record.plan.workload}:${record.plan.fault}`; else if (oracle.verdict === "fail") signature = `T:${oracle.checks.find(check => !check.ok)?.name ?? "unknown"}:${record.plan.workload}:${record.plan.fault}`; - return { crossed: crossing.length > 0, crossingFaults: crossing, rules, T: oracle, pass, signature }; + else if (universalFailure) signature = `${universalFailure[0]}:${universalFailure[1].detail.split(/[\s:(]/)[0]}:${record.plan.workload}:${record.plan.fault}`; + return { crossed: crossing.length > 0, crossingFaults: crossing, rules, T: oracle, universal, pass, signature }; } diff --git a/tests/restart-chaos/invariants.ts b/tests/restart-chaos/invariants.ts index 92066badffc..3dec3d0e5fb 100644 --- a/tests/restart-chaos/invariants.ts +++ b/tests/restart-chaos/invariants.ts @@ -180,3 +180,29 @@ export async function checkPlanInvariants(input: { await db.$client.end({ timeout: 5 }).catch(() => undefined); } } + +/** + * U3 evidence from the wave database (universal.ts): every failed run's error + * text and every recovery action's cause, so a bare "Error" or an empty + * message is caught even where the board's run list hides the text. Read once + * at wave end, before the instance is removed; grouped by issue. + */ +export async function readUniversalDbEvidence(db: Db, companyId: string): Promise> { + const runs = await db.$client.unsafe( + `select id::text as id, coalesce(context_snapshot->>'issueId', '') as "issueId", left(error, 400) as error + from heartbeat_runs where company_id::text = $1 and status in ('failed', 'timed_out')`, [companyId]) as Row[]; + const actions = await db.$client.unsafe( + `select id::text as id, source_issue_id::text as "issueId", kind, left(cause, 400) as cause, left(next_action, 400) as "nextAction" + from issue_recovery_actions where company_id::text = $1`, [companyId]) as Row[]; + const byIssue = new Map; recoveryActions: Array<{ id: string; kind: string; cause: string | null; nextAction: string | null }> }>(); + const entry = (issueId: string) => { + let value = byIssue.get(issueId); + if (!value) byIssue.set(issueId, value = { runMessages: new Map(), recoveryActions: [] }); + return value; + }; + // A run is matched to its task by id in the grader, so runs without an issue id are kept under "". + for (const run of runs) entry(text(run.issueId)).runMessages.set(text(run.id), typeof run.error === "string" ? run.error : null); + for (const action of actions) entry(text(action.issueId)).recoveryActions.push({ id: text(action.id), kind: text(action.kind), + cause: typeof action.cause === "string" ? action.cause : null, nextAction: typeof action.nextAction === "string" ? action.nextAction : null }); + return byIssue; +} diff --git a/tests/restart-chaos/ledger.ts b/tests/restart-chaos/ledger.ts index e843d49e2f5..d527fdc558b 100644 --- a/tests/restart-chaos/ledger.ts +++ b/tests/restart-chaos/ledger.ts @@ -34,6 +34,9 @@ export interface LedgerLine { masks?: string[]; /** Crash-point suite: the named point this wave armed. */ crashPoint?: string; + /** Universal checks U1–U4 (universal.ts): verdict per rule, and the details of warnings that did not fail the task. */ + universal?: Record; + universalWarnings?: string[]; } export function ledgerLine(plan: WavePlan, record: TaskRecord, grade: TaskGrade, gitSha: string, evidence: string | null, @@ -52,6 +55,11 @@ export function ledgerLine(plan: WavePlan, record: TaskRecord, grade: TaskGrade, recordedAt: new Date().toISOString(), ...(plan.options.masks?.length ? { masks: [...plan.options.masks] } : {}), ...(plan.options.crashPoint ? { crashPoint: plan.options.crashPoint } : {}), + ...(grade.universal ? { + universal: Object.fromEntries(Object.entries(grade.universal).map(([id, rule]) => [id, rule.verdict])), + universalWarnings: Object.entries(grade.universal).filter(([, rule]) => !rule.enforced && rule.verdict === "fail") + .map(([id, rule]) => `${id}: ${rule.detail}`), + } : {}), }; } @@ -87,6 +95,13 @@ export function renderSummary(lines: readonly LedgerLine[], latestWave?: string) out.push(`| ${line.taskId} ${line.issue ?? ""} | ${line.adapter} ${line.workload} ${line.fault}${line.trigger ? ` ^${line.trigger}${line.triggerMissed?.length ? " (missed)" : ""}` : ""}${line.masks?.length ? ` (${line.masks.map(mask => `${mask}-masked`).join(", ")})` : ""} | ${line.crossed ? "yes" : "no"} | ${line.rules.R1} | ${line.rules.R2} | ${line.rules.R3} | ${line.rules.R4} | ${line.rules.R5} | ${line.rules.R6} | ${line.T} | ${line.pass ? "PASS" : `FAIL ${line.signature}`} |`); } out.push(""); + // Universal checks that are still warnings (universal.ts) never fail a task; list them so they are seen. + const warnings = lines.filter(line => line.waveId === latestWave && line.universalWarnings?.length); + if (warnings.length) { + out.push("Universal check warnings (not failing):", ""); + for (const line of warnings) for (const warning of line.universalWarnings!) out.push(`- ${line.taskId}: ${warning}`); + out.push(""); + } } out.push("## Cells", "", "| Cell | Pass | Total | Crossed |", "| --- | --- | --- | --- |"); for (const [key, cell] of [...cells].sort()) out.push(`| ${key} | ${cell.pass} | ${cell.total} | ${cell.crossed} |`); diff --git a/tests/restart-chaos/main.ts b/tests/restart-chaos/main.ts index c9e5056c934..7d5da9dd9e1 100644 --- a/tests/restart-chaos/main.ts +++ b/tests/restart-chaos/main.ts @@ -25,7 +25,9 @@ import { DaytonaWaveSweeper, leaksForTask, unattributedLeaks, type DaytonaSweepR import { knownSecretValues, redactKnownSecrets, writeEvidenceBundle } from "./evidence.js"; import { gradeTask } from "./grader.js"; import { releaseInstance } from "../runner-e2e/instance-retention.js"; -import { checkPlanInvariants, summarizeViolations, type InvariantCheck } from "./invariants.js"; +import { checkPlanInvariants, readUniversalDbEvidence, summarizeViolations, type InvariantCheck } from "./invariants.js"; +import { collectContentionEvidence, RunRowPressure } from "./contention.js"; +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"; @@ -198,6 +200,22 @@ async function main(): Promise { const pointResults: Array }> = []; let currentPoint: (PointResult & { baseline: Set }) | null = null; let databaseUrl: string | null = null; + // The wave's own isolated database (never another instance's). + const waveDatabaseUrl = () => { + if (!databaseUrl) { + const config = JSON.parse(readFileSync(supervisor.paths.configPath, "utf8")) as { database?: { connectionString?: string } }; + databaseUrl = config.database?.connectionString ?? null; + if (!databaseUrl) throw new Error("The instance config has no database connection string"); + } + return databaseUrl; + }; + let pressure: RunRowPressure | null = null; + const stopPressure = async () => { + if (!pressure) return; + const stats = await pressure.stop(); + pressure = null; + timeline.add("db_pressure_stopped", { ...stats }); + }; const violationKey = (violation: { rule: string; kind: string; recordId: string }) => `${violation.rule}:${violation.kind}:${violation.recordId}`; if (crashPoint) { supervisor.crashPointHoldMs = plan.options.crashPointHoldMs ?? 0; @@ -285,6 +303,11 @@ async function main(): Promise { sweeper = new DaytonaWaveSweeper({ apiKey: process.env.DAYTONA_API_KEY!, companyId: owned.companyId, environmentId: owned.environmentId, api }); } }); spendTarget = { api, companyId: lab.companyId }; + // --db-pressure: contend with the wave's running run rows from outside the server (contention.ts). + if (plan.options.dbPressure) { + pressure = new RunRowPressure(waveDatabaseUrl(), lab.companyId, plan.options.dbPressure).start(); + timeline.add("db_pressure_started", { writers: plan.options.dbPressure }); + } // The instance root holds the secrets master key and the database: oracle code may not read it. const protectedRoots = [os.homedir(), supervisor.paths.temporaryRoot, resultsRoot]; board = serverEnvironment.SERVE_UI === "false" ? null : new BoardUi(supervisor.baseUrl); @@ -351,12 +374,7 @@ async function main(): Promise { }; const invariantsNow = async () => { - if (!databaseUrl) { - const config = JSON.parse(readFileSync(supervisor.paths.configPath, "utf8")) as { database?: { connectionString?: string } }; - databaseUrl = config.database?.connectionString ?? null; - if (!databaseUrl) throw new Error("Crash-point wave: the instance config has no database connection string"); - } - return checkPlanInvariants({ connectionString: databaseUrl, companyId: lab.companyId, + return checkPlanInvariants({ connectionString: waveDatabaseUrl(), companyId: lab.companyId, paperclipHome: supervisor.paths.home, instanceId: "default" }); }; const startCrashPoint = async (batch: TaskObserver[]) => { @@ -462,6 +480,7 @@ async function main(): Promise { await delay(1_000); } await Promise.all(faultWork); + await stopPressure(); checkWorktree("wave_end"); if (worktreeChanged) harnessError ??= new Error("The worktree changed during the wave (build, install or branch switch); later restarts may have run different code"); // A storm whose task pool ran dry early did not cover its storm; it is not a passing wave. @@ -488,6 +507,15 @@ async function main(): Promise { } const attributed = new Set(); const secrets = knownSecretValues(); + // U3 and L1 read the wave database once, before the instance is removed. + const evidenceDb = createDb(waveDatabaseUrl()); + const universalByIssue = await readUniversalDbEvidence(evidenceDb, lab.companyId).catch(error => { + timeline.add("universal_db_evidence_failed", { error: String(error) }); return null; }); + const allRunMessages = new Map([...(universalByIssue?.values() ?? [])].flatMap(entry => [...entry.runMessages])); + const costSummary = launched.some(observer => observer.plan.workload === "L1") + ? await api.get>(`/api/companies/${lab.companyId}/costs/summary?period=all`) + .then(summary => typeof summary.spendCents === "number" + ? { spendCents: summary.spendCents, pendingRunCount: Number(summary.pendingRunCount ?? 0) } : null).catch(() => null) : null; for (const observer of launched) { observer.record.sweep = await observer.sweep(); if (sandboxSweep) { @@ -499,11 +527,16 @@ async function main(): Promise { let artifacts: OracleArtifacts | null = null; if (observer.record.issueId) artifacts = await observer.collectArtifacts().catch(() => null); if (artifacts && providerFaults && observer.plan.providerFault) artifacts.providerFaultEvents = await providerFaults.eventsFor(observer.plan.nonce); + if (artifacts && observer.plan.workload === "L1") { + const runIds = [...new Set(observer.record.samples.flatMap(sample => (sample.runs ?? []).map(run => run.id)))]; + artifacts.contention = await collectContentionEvidence(evidenceDb, lab.companyId, runIds, costSummary); + } const graded = artifacts ? observer.definition.grade(observer.plan.nonce, artifacts, observer.workloadAgents()) : { verdict: "fail" as const, checks: [{ name: "artifacts_collected", ok: false, detail: observer.record.launchError ?? "collection failed" }] }; // A provider-fault task whose fault never happened tested nothing. const oracle = withProviderFaultCheck(graded, observer.plan, artifacts?.providerFaultEvents ?? []); - const grade = gradeTask(observer.record, faults, observer.definition.budgetMs, oracle); + const grade = gradeTask(observer.record, faults, observer.definition.budgetMs, oracle, universalByIssue ? { + runMessages: allRunMessages, recoveryActions: universalByIssue.get(observer.record.issueId ?? "")?.recoveryActions ?? [] } : undefined); let evidence: string | null = null; if (!grade.pass) { anyFailure = true; @@ -514,8 +547,10 @@ async function main(): Promise { await writeFile(path.join(waveDir, `${observer.plan.taskId}.grade.json`), `${JSON.stringify({ record: observer.record, grade }, null, 2)}\n`, { mode: 0o600 }); lines.push(ledgerLine(plan, observer.record, grade, gitSha, evidence, faults)); timeline.add("task_graded", { taskId: observer.plan.taskId, pass: grade.pass, crossed: grade.crossed, signature: grade.signature, - rules: Object.fromEntries(Object.entries(grade.rules).map(([id, rule]) => [id, rule.verdict])), T: grade.T.verdict }); + rules: Object.fromEntries(Object.entries(grade.rules).map(([id, rule]) => [id, rule.verdict])), T: grade.T.verdict, + universal: Object.fromEntries(Object.entries(grade.universal ?? {}).map(([id, rule]) => [id, `${rule.verdict}${rule.enforced ? "" : " (warning)"}`])) }); } + await evidenceDb.$client.end({ timeout: 5 }).catch(() => undefined); await writeFile(path.join(waveDir, "faults.json"), `${JSON.stringify(faults, null, 2)}\n`, { mode: 0o600 }); const speed = await (async () => { const childTasks = await childSpeedTasksInTurn(launched); @@ -562,6 +597,7 @@ async function main(): Promise { harnessError = error; timeline.add("harness_error", { error: error instanceof Error ? error.stack ?? error.message : String(error) }); } finally { + await stopPressure().catch(error => timeline.add("db_pressure_stop_failed", { error: String(error) })); // 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); diff --git a/tests/restart-chaos/types.ts b/tests/restart-chaos/types.ts index 787bef559e5..e16036657f4 100644 --- a/tests/restart-chaos/types.ts +++ b/tests/restart-chaos/types.ts @@ -12,7 +12,8 @@ export type FaultKind = (typeof FAULT_KINDS)[number]; export type TaskFaultKind = FaultKind | "K13" | ProviderFaultKind; /** G1: an approval-gated action across a restart. H1: a budget hard stop crossed mid-run with a restart (its own wave). */ -export const WORKLOAD_IDS = ["A1", "B1", "B5", "C1", "M1", "G1", "H1", "R1", "S1", "Q1"] as const; +/** 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; export type WorkloadId = (typeof WORKLOAD_IDS)[number]; export const ADAPTER_IDS = ["codex", "claude", "opencode"] as const; @@ -90,6 +91,8 @@ export interface WaveOptions { providerFaultMessage?: ProviderMessage; /** PV1 delays, cycled across tasks (`--provider-fault-delays`). */ providerFaultDelaysMs?: number[]; + /** `--db-pressure `: harness-owned writers touching the wave's running run rows back to back (contention.ts). */ + dbPressure?: number; } export interface StormFault { @@ -256,6 +259,8 @@ export interface OracleArtifacts { question?: QuestionArtifacts; /** PV1–PV3: the shim's log lines for this task. */ providerFaultEvents?: ProviderFaultEvent[]; + /** L1: accounting and run-log evidence read from the wave database at rest. */ + contention?: import("./contention.js").ContentionEvidence; } /** What the board did for a governance task (G1, H1), and what the observer saw. */ @@ -349,6 +354,8 @@ export interface TaskGrade { crossingFaults: string[]; rules: Record; T: OracleResult; + /** Universal checks U1–U4 (universal.ts); absent on grades written before them. */ + universal?: import("./universal.js").UniversalGrade; pass: boolean; signature: string | null; } diff --git a/tests/restart-chaos/universal-audit.ts b/tests/restart-chaos/universal-audit.ts new file mode 100644 index 00000000000..4f9520c5adb --- /dev/null +++ b/tests/restart-chaos/universal-audit.ts @@ -0,0 +1,104 @@ +/** + * Run the universal checks U1–U4 over saved wave evidence, to measure false + * positives before a rule is enforced. + * + * node cli/node_modules/tsx/dist/cli.mjs tests/restart-chaos/universal-audit.ts [--json] [...] + * + * Each results dir holds `w-*` wave directories with `.grade.json`. A + * task counts as known-good when it passed R1–R6 and T; a wave is known-good + * when every graded task in it passed. Every rule is graded as a warning here, + * whatever its enforcement, so all flags are shown. + */ +import { readdirSync, readFileSync } from "node:fs"; +import path from "node:path"; +import type { TaskGrade, TaskRecord } from "./types.js"; +import { DEFAULT_UNIVERSAL_BOUNDS, gradeUniversal, type UniversalBounds, type UniversalId } from "./universal.js"; + +export interface AuditFlag { wave: string; task: string; workload: string; adapter: string; taskPassed: boolean; waveGood: boolean; rule: UniversalId; detail: string } +export interface AuditReport { + waves: number; tasks: number; knownGoodTasks: number; knownGoodWaves: number; + byRule: Record; + flags: AuditFlag[]; + /** Per metric, over tasks that passed R1–R6+T: p50, p95, p99 and max. */ + passingMetrics: Record; +} + +function percentiles(values: number[]) { + const sorted = [...values].sort((a, b) => a - b); + const at = (q: number) => sorted[Math.min(sorted.length - 1, Math.floor(q * sorted.length))] ?? 0; + return { n: sorted.length, p50: at(0.5), p95: at(0.95), p99: at(0.99), max: sorted[sorted.length - 1] ?? 0 }; +} + +const RULES: UniversalId[] = ["U1", "U2", "U3", "U4"]; + +export function auditWaves(roots: readonly string[], bounds: UniversalBounds = DEFAULT_UNIVERSAL_BOUNDS): AuditReport { + const report: AuditReport = { waves: 0, tasks: 0, knownGoodTasks: 0, knownGoodWaves: 0, + byRule: Object.fromEntries(RULES.map(rule => [rule, { flagged: 0, onKnownGoodTasks: 0, onKnownGoodWaves: 0 }])) as AuditReport["byRule"], flags: [], + passingMetrics: {} }; + const metrics = new Map(); + for (const root of roots) { + let waves: string[] = []; + try { waves = readdirSync(root).filter(name => name.startsWith("w-")).sort(); } catch { continue; } + for (const wave of waves) { + const dir = path.join(root, wave); + const graded: Array<{ task: string; record: TaskRecord; grade: TaskGrade }> = []; + for (const file of readdirSync(dir).filter(name => /^t\d+\.grade\.json$/.test(name)).sort()) { + try { + const parsed = JSON.parse(readFileSync(path.join(dir, file), "utf8")) as { record: TaskRecord; grade: TaskGrade }; + graded.push({ task: file.replace(".grade.json", ""), ...parsed }); + } catch { /* a truncated grade file is not evidence */ } + } + if (!graded.length) continue; + report.waves += 1; + // Known-good is R1–R6 and T only: `grade.pass` on newer grades also includes enforced universal checks. + const known = (grade: TaskGrade) => grade.T.verdict === "pass" && Object.values(grade.rules).every(rule => rule.verdict !== "fail"); + const waveGood = graded.every(entry => known(entry.grade)); + if (waveGood) report.knownGoodWaves += 1; + for (const { task, record, grade } of graded) { + report.tasks += 1; + const passed = known(grade); + if (passed) report.knownGoodTasks += 1; + const universal = gradeUniversal(record, { bounds, enforced: new Set() }); + // A grade saved with universal checks graded U3 with the wave database (run messages, recovery + // actions) that the saved record lacks: keep that verdict instead of a samples-only recomputation. + if (grade.universal?.U3) universal.U3 = { ...grade.universal.U3, enforced: false }; + for (const rule of RULES) { + if (passed) for (const [name, value] of Object.entries(universal[rule].metrics ?? {})) metrics.set(name, [...(metrics.get(name) ?? []), value]); + if (universal[rule].verdict !== "fail") continue; + report.byRule[rule].flagged += 1; + if (passed) report.byRule[rule].onKnownGoodTasks += 1; + if (waveGood) report.byRule[rule].onKnownGoodWaves += 1; + report.flags.push({ wave: `${path.basename(path.resolve(root, "../../.."))}:${wave}`, task, workload: record.plan.workload, + adapter: record.plan.adapter, taskPassed: passed, waveGood, rule, detail: universal[rule].detail }); + } + } + } + } + report.passingMetrics = Object.fromEntries([...metrics].map(([name, values]) => [name, percentiles(values)])); + return report; +} + +export function renderAudit(report: AuditReport): string { + const lines = [ + `Universal check audit: ${report.waves} waves (${report.knownGoodWaves} all-pass), ${report.tasks} tasks (${report.knownGoodTasks} passed R1–R6+T)`, + "", + "| Rule | Flagged tasks | On passing tasks | On all-pass waves |", + "| --- | --- | --- | --- |", + ...RULES.map(rule => `| ${rule} | ${report.byRule[rule].flagged} | ${report.byRule[rule].onKnownGoodTasks} | ${report.byRule[rule].onKnownGoodWaves} |`), + "", + "Passing tasks: | metric | n | p50 | p95 | p99 | max |", + ...Object.entries(report.passingMetrics).map(([name, value]) => `| ${name} | ${value.n} | ${value.p50} | ${value.p95} | ${value.p99} | ${value.max} |`), + "", + ...report.flags.map(flag => `${flag.rule} ${flag.wave}/${flag.task} ${flag.adapter}/${flag.workload} task=${flag.taskPassed ? "pass" : "fail"}: ${flag.detail}`), + ]; + return lines.join("\n"); +} + +if (import.meta.url === `file://${process.argv[1]}` || process.argv[1]?.endsWith("universal-audit.ts")) { + const args = process.argv.slice(2); + const json = args.includes("--json"); + const roots = args.filter(arg => arg !== "--json").map(arg => path.resolve(arg)); + if (!roots.length) { console.error("usage: universal-audit.ts [--json] [...]"); process.exit(2); } + const report = auditWaves(roots); + console.info(json ? JSON.stringify(report, null, 2) : renderAudit(report)); +} diff --git a/tests/restart-chaos/universal.test.ts b/tests/restart-chaos/universal.test.ts new file mode 100644 index 00000000000..ecac50768a3 --- /dev/null +++ b/tests/restart-chaos/universal.test.ts @@ -0,0 +1,163 @@ +import { describe, expect, it } from "vitest"; +import { gradeTask } from "./grader.js"; +import type { OracleArtifacts, RunSample, TaskRecord, TaskSample } from "./types.js"; +import { bareCauseMessage, DEFAULT_UNIVERSAL_BOUNDS as BOUNDS, gradeU1, gradeU2, gradeU3, gradeU4, gradeUniversal, universalBounds } from "./universal.js"; +import { workloadDefinitions } from "./workloads.js"; + +const T0 = Date.parse("2026-11-09T12:00:00Z"); +const iso = (offsetMs: number) => new Date(T0 + offsetMs).toISOString(); +const SEC = 1_000; + +function run(id: string, status: string, created: number, finished: number | null = null, extra: Partial = {}): RunSample { + return { id, status, createdAt: iso(created), startedAt: status === "queued" ? null : iso(created + 500), + finishedAt: finished === null ? null : iso(finished), errorCode: null, retryOfRunId: null, wakeCommentIds: [], livenessState: null, + lastEvidenceAt: status === "running" ? iso(created + 500) : null, ...extra }; +} +function sample(at: number, issueStatus: string, runs: RunSample[], extra: Partial = {}): TaskSample { + return { at: T0 + at, reachable: true, serverReady: true, issueStatus, interruption: null, activeRecoveryAction: false, + executionBlocker: false, scheduledRetry: false, runs, ...extra }; +} +/** A poll every 3 s from `from` to `to` (ms), built by `make`. */ +function polls(from: number, to: number, make: (at: number) => TaskSample): TaskSample[] { + const out: TaskSample[] = []; + for (let at = from; at <= to; at += 3 * SEC) out.push(make(at)); + return out; +} +function record(samples: TaskSample[], extra: Partial = {}): TaskRecord { + return { + plan: { taskId: "t01", index: 0, workload: "A1", adapter: "codex", fault: "K1", fireAfterMs: 20_000, holdMs: 0, followUpAfterMs: null, nonce: "abcd1234" }, + issueId: "issue-1", identifier: "CHA-1", createdAt: T0, launchError: null, samples, retry: null, restAt: samples.at(-1)?.at ?? null, + finalStatus: "done", deadlineAt: T0 + 3_600_000, timedOut: false, followUp: null, faultIds: [], + sweep: { activeRuns: [], runningWorkspaceOperations: [], activeRecoveryAction: false, error: null }, + preFaultFiles: null, lostFiles: null, ...extra, + }; +} +const retryable = { state: "failed", canRetry: true, retryWait: false, workspaceFailure: null, observedRunId: "r1", userAction: null }; + +describe("U1 liveness", () => { + it("passes a task whose run keeps working, then finishes", () => { + const samples = [...polls(0, 120 * SEC, at => sample(at, "in_progress", [run("r1", "running", 0, null, { lastEvidenceAt: iso(at) })])), + sample(123 * SEC, "done", [run("r1", "succeeded", 0, 121 * SEC)])]; + expect(gradeU1(record(samples), BOUNDS).verdict).toBe("pass"); + }); + + it("fails a run that stays queued past the queue bound with nothing else for the user (the #20361 shape)", () => { + const samples = polls(0, 9 * 60 * SEC, at => sample(at, "todo", [run("r1", "queued", 0)])); + const result = gradeU1(record(samples, { restAt: null, finalStatus: null, timedOut: true }), BOUNDS); + expect(result.verdict).toBe("fail"); + expect(result.detail).toMatch(/^no_owner_no_action .* in todo with r1 queued/); + }); + + it("treats a visible Retry or a waiting question as actionable", () => { + const failedRun = run("r1", "failed", 0, 5 * SEC, { errorCode: "process_lost" }); + const withRetry = polls(0, 10 * 60 * SEC, at => sample(at, "in_progress", [failedRun], { interruption: retryable })); + expect(gradeU1(record(withRetry), BOUNDS).verdict).toBe("pass"); + const question = { ...retryable, state: "waiting_for_user", canRetry: false }; + expect(gradeU1(record(polls(0, 10 * 60 * SEC, at => sample(at, "in_progress", [failedRun], { interruption: question }))), BOUNDS).verdict).toBe("pass"); + }); + + it("does not count outage time toward the bound", () => { + const quiet = (at: number) => sample(at, "in_progress", [run("r1", "failed", 0, 5 * SEC, { errorCode: "process_lost" })]); + const samples = [...polls(10 * SEC, 100 * SEC, quiet), { at: T0 + 101 * SEC, reachable: false, serverReady: false }, + { at: T0 + 400 * SEC, reachable: false, serverReady: false }, ...polls(401 * SEC, 480 * SEC, quiet)]; + expect(gradeU1(record(samples, { restAt: null }), BOUNDS).verdict).toBe("pass"); + }); + + it("does not let a retry row keep a stuck dispatched retry run alive", () => { + const samples = polls(0, 9 * 60 * SEC, at => sample(at, "todo", [run("r1", "queued", 0)], { scheduledRetry: true })); + expect(gradeU1(record(samples, { restAt: null }), BOUNDS).verdict).toBe("fail"); + const pending = polls(0, 9 * 60 * SEC, at => sample(at, "todo", [run("r0", "failed", 0, SEC, { errorCode: "process_lost" })], { scheduledRetry: true })); + expect(gradeU1(record(pending, { restAt: null }), BOUNDS).verdict).toBe("pass"); + }); + + it("fails a silent running run once it stops counting as an owner", () => { + const samples = polls(0, 7 * 60 * SEC, at => sample(at, "in_progress", [run("r1", "running", 0)])); + expect(gradeU1(record(samples, { restAt: null }), BOUNDS).verdict).toBe("fail"); + }); +}); + +describe("U2 retry loops", () => { + const failed = (id: string, at: number, code = "adapter_failed") => run(id, "failed", at, at + 5 * SEC, { errorCode: code }); + it("fails the same failure repeating without the task changing state", () => { + const runs: RunSample[] = []; + const samples = [0, 1, 2, 3].map(index => { runs.push(failed(`r${index}`, index * 20 * SEC)); return sample(index * 20 * SEC + 6 * SEC, "in_progress", [...runs]); }); + expect(gradeU2(record(samples), BOUNDS)).toMatchObject({ verdict: "fail", detail: expect.stringMatching(/same_failure_repeated 4x failed:adapter_failed/) }); + }); + it("resets the streak when a run succeeds between failures", () => { + const succeeded = run("ok", "succeeded", 7 * SEC, 9 * SEC); + const samples = [sample(60 * SEC, "in_progress", [failed("r0", 0), succeeded, failed("r1", 10 * SEC), failed("r2", 20 * SEC), failed("r3", 30 * SEC)])]; + expect(gradeU2(record(samples), BOUNDS)).toMatchObject({ verdict: "pass", metrics: { sameFailureRuns: 3 } }); + }); + it("resets the streak when the issue changed state between failures", () => { + const all = [failed("r0", 0), failed("r1", 20 * SEC), failed("r2", 40 * SEC), failed("r3", 60 * SEC)]; + const samples = [sample(6 * SEC, "in_progress", all.slice(0, 1)), sample(26 * SEC, "in_progress", all.slice(0, 2)), + sample(30 * SEC, "blocked", all.slice(0, 2)), sample(46 * SEC, "in_progress", all.slice(0, 3)), sample(66 * SEC, "in_progress", all)]; + expect(gradeU2(record(samples), BOUNDS)).toMatchObject({ verdict: "pass", metrics: { sameFailureRuns: 2 } }); + }); + it("passes when the failures differ or stay within the bound", () => { + const samples = [sample(10 * SEC, "in_progress", [failed("r1", 0), failed("r2", 6 * SEC, "process_lost"), failed("r3", 12 * SEC)])]; + expect(gradeU2(record(samples), BOUNDS).verdict).toBe("pass"); + }); +}); + +describe("U3 diagnosable failures", () => { + it("flags a generic code, and a bare cause in the stored message", () => { + const samples = [sample(10 * SEC, "done", [run("r1", "failed", 0, 5 * SEC, { errorCode: "adapter_failed" })])]; + expect(gradeU3(record(samples)).verdict).toBe("fail"); + const replay = [sample(10 * SEC, "done", [run("r2", "failed", 0, 5 * SEC, { errorCode: "native_event_replay_conflict" })])]; + expect(gradeU3(record(replay)).verdict).toBe("pass"); + const stalled = "native_event_replay_conflict: runner_event_commit_stalled: runner events kept failing to commit without progress " + + "(last cause: Error at source seq 12); automatic recovery is stopped."; + expect(gradeU3(record(replay), { runMessages: new Map([["r2", stalled]]), recoveryActions: [] }).verdict).toBe("fail"); + const coded = stalled.replace("last cause: Error", "last cause: 40001"); + expect(gradeU3(record(replay), { runMessages: new Map([["r2", coded]]), recoveryActions: [] }).verdict).toBe("pass"); + }); + it("flags a recovery action with an empty or bare cause", () => { + const db = { runMessages: new Map(), recoveryActions: [{ id: "a1", kind: "stalled_issue", cause: "Error", nextAction: "Retry the task" }] }; + expect(gradeU3(record([sample(0, "done", [])]), db).verdict).toBe("fail"); + }); + it("recognizes bare causes", () => { + expect(bareCauseMessage("Error")).toBe(true); + expect(bareCauseMessage("")).toBe(true); + expect(bareCauseMessage("failed (cause: unknown_error)")).toBe(true); + expect(bareCauseMessage("Error at source seq 7")).toBe(true); + expect(bareCauseMessage("Process exited with code 1: ENOENT spawn codex")).toBe(false); + }); +}); + +describe("U4 stale notices", () => { + it("fails a Retry notice still shown after a later run succeeded", () => { + const runs = [run("r1", "failed", 0, 5 * SEC, { errorCode: "process_lost" }), run("r2", "succeeded", 10 * SEC, 30 * SEC)]; + const samples = polls(31 * SEC, 31 * SEC + 2 * 60 * SEC, at => sample(at, "done", runs, { interruption: retryable })); + expect(gradeU4(record(samples), BOUNDS)).toMatchObject({ verdict: "fail", detail: expect.stringMatching(/^stale_notice Retry offered after r2 succeeded/) }); + }); + it("passes when the notice clears", () => { + const runs = [run("r1", "failed", 0, 5 * SEC, { errorCode: "process_lost" }), run("r2", "succeeded", 10 * SEC, 30 * SEC)]; + const samples = [sample(31 * SEC, "done", runs, { interruption: retryable }), sample(34 * SEC, "done", runs)]; + expect(gradeU4(record(samples), BOUNDS).verdict).toBe("pass"); + }); +}); + +describe("universal checks in the task grade", () => { + const artifacts: OracleArtifacts = { issueStatus: "done", documents: { brief: "CHAOS-abcd1234\n- one\n- two\n- three\n" }, + agentComments: [], files: {}, commands: {}, workspacePath: null }; + it("fails a task on an enforced rule and keeps a warning rule from failing it", () => { + const runs = [run("r1", "failed", 0, 5 * SEC, { errorCode: "adapter_failed" }), run("r2", "succeeded", 10 * SEC, 30 * SEC)]; + const stale = polls(31 * SEC, 31 * SEC + 2 * 60 * SEC, at => sample(at, "done", runs, { interruption: retryable })); + const grade = gradeTask(record(stale), [], 600_000, workloadDefinitions.A1.grade("abcd1234", artifacts)); + expect(grade.universal?.U4).toMatchObject({ verdict: "fail", enforced: true }); + expect(grade.universal?.U3).toMatchObject({ verdict: "fail", enforced: false }); + expect(grade.pass).toBe(false); + expect(grade.signature).toMatch(/^U4:stale_notice:A1:K1$/); + + const clean = [sample(31 * SEC, "done", runs)]; + const warned = gradeTask(record(clean), [], 600_000, workloadDefinitions.A1.grade("abcd1234", artifacts)); + expect(warned.universal?.U3.verdict).toBe("fail"); + expect(warned.pass).toBe(true); + }); + it("reads bounds from the environment", () => { + expect(universalBounds({ RESTART_CHAOS_U1_ORPHAN_SECONDS: "60", RESTART_CHAOS_U2_SAME_FAILURE_RUNS: "5" })).toMatchObject({ orphanMs: 60_000, sameFailureRuns: 5 }); + expect(() => universalBounds({ RESTART_CHAOS_U1_ORPHAN_SECONDS: "-1" })).toThrow(); + expect(gradeUniversal(record([sample(0, "done", [])])).U1.verdict).toBe("pass"); + }); +}); diff --git a/tests/restart-chaos/universal.ts b/tests/restart-chaos/universal.ts new file mode 100644 index 00000000000..435a9327b0f --- /dev/null +++ b/tests/restart-chaos/universal.ts @@ -0,0 +1,287 @@ +/** + * Universal checks U1–U4: scenario-independent rules graded on every task of + * every wave, whatever its workload. Pure functions over observer samples (no + * I/O), so they also run over saved evidence (`universal-audit.ts`). + * + * U1 liveness every non-terminal task has a live owner or a visible, actionable item + * U2 retry loop the same failure does not repeat without the task changing state + * U3 diagnosable every failed run names a specific cause + * U4 stale notices no failure or Retry notice outlives later successful work + * + * Each rule is either `enforced` (a violation fails the task) or a warning + * (reported in the grade and ledger, never fails the task). The split comes + * from the false-positive audit over saved waves; see README "Universal checks". + */ +import type { RunSample, TaskRecord, TaskSample } from "./types.js"; + +export type UniversalId = "U1" | "U2" | "U3" | "U4"; +export interface UniversalResult { verdict: "pass" | "fail" | "n/a"; detail: string; enforced: boolean; + /** Measurements behind the verdict (milliseconds or counts), for bound tuning. */ + metrics?: Record } +export type UniversalGrade = Record; + +export interface UniversalBounds { + /** U1: longest ready-server time a non-terminal task may spend with no live owner and no actionable item. */ + orphanMs: number; + /** U1: longest ready-server time a run may stay queued without starting. */ + queuedMs: number; + /** U1: a `running` run with no event, activity or start evidence for this long is not a live owner. */ + silentRunMs: number; + /** U2: most consecutive failed runs with one failure code while the issue status stays the same. */ + sameFailureRuns: number; + /** U4: longest a failure or Retry notice may remain after a later run succeeded. */ + staleNoticeMs: number; +} + +/** + * Defaults from the 2026-11-09 audit of 74 saved waves (1,592 tasks; README + * "Universal checks"). Override with RESTART_CHAOS_U1_ORPHAN_SECONDS, + * RESTART_CHAOS_U1_QUEUED_SECONDS, RESTART_CHAOS_U2_SAME_FAILURE_RUNS and + * RESTART_CHAOS_U4_STALE_SECONDS. + */ +export const DEFAULT_UNIVERSAL_BOUNDS: UniversalBounds = { + orphanMs: 3 * 60_000, + queuedMs: 5 * 60_000, + silentRunMs: 3 * 60_000, + sameFailureRuns: 3, + staleNoticeMs: 60_000, +}; + +/** Rules that fail a task. The rest are warnings until the audit shows no false positive on known-good waves. */ +export const ENFORCED_UNIVERSAL: ReadonlySet = new Set(["U1", "U2", "U4"]); + +export function universalBounds(env: NodeJS.ProcessEnv = process.env): UniversalBounds { + const seconds = (name: string, fallback: number) => { + const raw = env[name]; + if (raw === undefined || raw === "") return fallback; + const value = Number(raw); + if (!Number.isFinite(value) || value <= 0) throw new Error(`${name} must be a positive number`); + return value; + }; + return { + orphanMs: seconds("RESTART_CHAOS_U1_ORPHAN_SECONDS", DEFAULT_UNIVERSAL_BOUNDS.orphanMs / 1000) * 1000, + queuedMs: seconds("RESTART_CHAOS_U1_QUEUED_SECONDS", DEFAULT_UNIVERSAL_BOUNDS.queuedMs / 1000) * 1000, + silentRunMs: DEFAULT_UNIVERSAL_BOUNDS.silentRunMs, + sameFailureRuns: Math.round(seconds("RESTART_CHAOS_U2_SAME_FAILURE_RUNS", DEFAULT_UNIVERSAL_BOUNDS.sameFailureRuns)), + staleNoticeMs: seconds("RESTART_CHAOS_U4_STALE_SECONDS", DEFAULT_UNIVERSAL_BOUNDS.staleNoticeMs / 1000) * 1000, + }; +} + +const TERMINAL_ISSUE = new Set(["done", "cancelled"]); +const ms = (iso: string | null | undefined) => (iso ? Date.parse(iso) : NaN); +const seconds = (value: number) => `${Math.round(value / 1000)}s`; +const ready = (sample: TaskSample) => sample.reachable && sample.serverReady; + +/** + * Codes that name no cause a person can act on. `adapter_failed` is the + * catch-all an adapter returns when it did not classify its failure; the board + * shows it as "adapter failed". + */ +export const GENERIC_FAILURE_CODES: ReadonlySet = new Set(["", "error", "unknown", "unknown_error", "failed", "adapter_failed", "internal_error"]); +const FAILED_RUN = new Set(["failed", "timed_out"]); + +/** A message whose only cause is a bare exception class or "unknown" (the #20385 "last cause: Error" class). */ +export function bareCauseMessage(message: string | null | undefined): boolean { + const text = (message ?? "").trim(); + if (!text) return true; + if (/^(error|unknown|unknown_error|internal error)\.?$/i.test(text)) return true; + return /\b(?:last cause|cause|reason)\s*[:=]\s*\(?\s*(?:Error|unknown_error|unknown|undefined|null)\s*\)?(?:\s|[;,.)]|$)/.test(text) + || /^Error at source seq \d+$/.test(text) || /:\s*Error at source seq \d+/.test(text); +} + +/** Latest view of each run across the samples. */ +function finalRuns(record: TaskRecord): Map { + const runs = new Map(); + for (const sample of record.samples) { + if (!sample.reachable) continue; + for (const run of sample.runs ?? []) { + const prior = runs.get(run.id); + const finished = !["queued", "running", "scheduled_retry"].includes(run.status); + runs.set(run.id, { ...run, firstSeenAt: prior?.firstSeenAt ?? sample.at, + finishedSeenAt: prior?.finishedSeenAt ?? (finished ? sample.at : null), + statusWhenFinished: prior?.statusWhenFinished ?? (finished ? sample.issueStatus ?? null : null) }); + } + } + return runs; +} + +/** U1: what keeps this task moving at this sample, or null when nothing does. */ +export function liveOwner(sample: TaskSample, queuedFor: ReadonlyMap, bounds: UniversalBounds): string | null { + for (const run of sample.runs ?? []) { + if (run.status === "running") { + if (run.toolInFlight || run.processAlive) return `run ${run.id} running (tool or process alive)`; + const evidence = Math.max(ms(run.lastEvidenceAt) || 0, ms(run.startedAt) || 0); + if (!evidence || sample.at - evidence <= bounds.silentRunMs) return `run ${run.id} running`; + } + if (run.status === "queued" || run.status === "scheduled_retry") { + if ((queuedFor.get(run.id) ?? 0) <= bounds.queuedMs) return `run ${run.id} ${run.status}`; + } + } + // A retry row stays on the issue after its run is dispatched; once a run exists the run's own bounds above decide. + const dispatched = (sample.runs ?? []).some(run => ["queued", "running", "scheduled_retry"].includes(run.status)); + if (!dispatched && (sample.scheduledRetry || sample.interruption?.retryWait || sample.interruption?.state === "retry_scheduled")) return "retry timer"; + if (sample.activeRecoveryAction && sample.recoveryOwner !== "board") return "system recovery action"; + return null; +} + +/** U1: an item a person can see and act on (Retry, a question, an approval, a board-owned recovery, a named blocker). */ +export function actionableItem(sample: TaskSample): string | null { + const interruption = sample.interruption; + if (interruption?.canRetry) return "Retry"; + if (interruption?.userAction) return `user action ${interruption.userAction}`; + if (interruption && (interruption.state === "waiting_for_user" || interruption.state === "failed")) return `interruption ${interruption.state}`; + if (sample.activeRecoveryAction && sample.recoveryOwner === "board") return "board recovery action"; + if (sample.executionBlocker) return "execution blocker"; + // Blocked and in-review issues wait on another issue, an approval or a person by definition. + if (sample.issueStatus === "blocked" || sample.issueStatus === "in_review") return `status ${sample.issueStatus}`; + return null; +} + +export function gradeU1(record: TaskRecord, bounds: UniversalBounds): Omit { + if (!record.samples.some(ready)) return { verdict: "n/a", detail: "no ready sample" }; + // Ready-server time each run has spent queued (outages do not count). + const queuedFor = new Map(); + let orphanMs = 0, worstOrphan = 0, orphanFrom: number | null = null, lastAt: number | null = null; + let worstQueued = 0, worstQueuedRun = ""; + for (const sample of record.samples) { + // Outages pause the clock: a down server is the fault's time, not the task's. + if (!ready(sample)) { lastAt = null; continue; } + for (const run of sample.runs ?? []) { + if (run.status === "queued" || run.status === "scheduled_retry") { + const waited = (queuedFor.get(run.id) ?? 0) + (lastAt !== null && queuedFor.has(run.id) ? sample.at - lastAt : 0); + queuedFor.set(run.id, waited); + if (waited > worstQueued) { worstQueued = waited; worstQueuedRun = run.id; } + } + } + if (TERMINAL_ISSUE.has(sample.issueStatus ?? "")) { orphanMs = 0; orphanFrom = null; lastAt = sample.at; continue; } + const owned = liveOwner(sample, queuedFor, bounds) ?? actionableItem(sample); + if (owned) { orphanMs = 0; orphanFrom = null; } + else { + if (orphanFrom === null) orphanFrom = sample.at; + else if (lastAt !== null) orphanMs += sample.at - lastAt; + worstOrphan = Math.max(worstOrphan, orphanMs); + if (orphanMs > bounds.orphanMs) { + const queued = (sample.runs ?? []).filter(run => run.status === "queued" || run.status === "scheduled_retry"); + return { verdict: "fail", metrics: { ownerlessMs: orphanMs, queuedMs: worstQueued }, detail: `no_owner_no_action ${seconds(orphanMs)} in ${sample.issueStatus ?? "unknown"}` + + (queued.length ? ` with ${queued.map(run => `${run.id} queued ${seconds(queuedFor.get(run.id) ?? 0)}`).join(",")}` : "") + + ` (interruption ${sample.interruption?.state ?? "none"})` }; + } + } + lastAt = sample.at; + } + return { verdict: "pass", metrics: { ownerlessMs: worstOrphan, queuedMs: worstQueued }, detail: `longest ownerless ${seconds(worstOrphan)}; longest queued ${seconds(worstQueued)}${worstQueuedRun ? ` (${worstQueuedRun})` : ""}` }; +} + +/** The issue status seen between two failures (both finish polls included) was not one value throughout. */ +function issueStatusChanged(record: TaskRecord, before: { finishedSeenAt: number | null; statusWhenFinished: string | null }, + after: { finishedSeenAt: number | null; statusWhenFinished: string | null }): boolean { + if (before.statusWhenFinished !== after.statusWhenFinished) return true; + const [from, to] = [before.finishedSeenAt ?? 0, after.finishedSeenAt ?? Number.POSITIVE_INFINITY]; + return record.samples.some(sample => sample.reachable && sample.at >= from && sample.at <= to && sample.issueStatus !== before.statusWhenFinished); +} + +export function gradeU2(record: TaskRecord, bounds: UniversalBounds): Omit { + // Every finished run in creation order: a succeeded run between two failures is progress and ends the streak. + const finished = [...finalRuns(record).values()].filter(run => run.finishedSeenAt !== null) + .sort((a, b) => (ms(a.createdAt) || a.firstSeenAt) - (ms(b.createdAt) || b.firstSeenAt)); + const isFailure = (run: RunSample) => FAILED_RUN.has(run.status) || run.status === "interrupted"; + if (!finished.some(isFailure)) return { verdict: "n/a", detail: "no failed run" }; + let streak = 0, worst = 0, worstCode = ""; + let previous: (typeof finished)[number] | null = null; + for (const current of finished) { + if (!isFailure(current)) { + // A cancellation (for example a superseded continuation) is neither progress nor a repeat. + if (current.status !== "cancelled") { streak = 0; previous = null; } + continue; + } + const same = previous !== null && `${previous.status}:${previous.errorCode}` === `${current.status}:${current.errorCode}` + && !issueStatusChanged(record, previous, current); + streak = same ? streak + 1 : 1; + previous = current; + if (streak > worst) { worst = streak; worstCode = `${current.status}:${current.errorCode ?? "none"}`; } + } + return worst > bounds.sameFailureRuns + ? { verdict: "fail", metrics: { sameFailureRuns: worst }, detail: `same_failure_repeated ${worst}x ${worstCode} without a task state change` } + : { verdict: "pass", metrics: { sameFailureRuns: worst }, detail: `most repeats of one failure ${worst} (${worstCode})` }; +} + +/** What the wave database adds for U3: each run's error text and the task's recovery actions. */ +export interface UniversalDbEvidence { + runMessages: ReadonlyMap; + recoveryActions: ReadonlyArray<{ id: string; kind: string; cause: string | null; nextAction: string | null }>; +} + +/** U3: failure codes from the samples; with database evidence, also run error text and recovery-action causes. */ +export function gradeU3(record: TaskRecord, db?: UniversalDbEvidence): Omit { + const failed = [...finalRuns(record).values()].filter(run => FAILED_RUN.has(run.status)); + const actions = db?.recoveryActions ?? []; + if (!failed.length && !actions.length) return { verdict: "n/a", detail: "no failed run or recovery action" }; + const messages = db?.runMessages; + const vague: string[] = []; + for (const action of actions) { + if (bareCauseMessage(action.cause) || bareCauseMessage(action.nextAction)) { + vague.push(`recovery ${action.id}(${action.kind}) cause=${JSON.stringify((action.cause ?? "").slice(0, 80))} next=${JSON.stringify((action.nextAction ?? "").slice(0, 80))}`); + } + } + for (const run of failed) { + const code = run.errorCode ?? ""; + if (GENERIC_FAILURE_CODES.has(code.trim().toLowerCase())) vague.push(`${run.id}:code=${code || "none"}`); + else if (messages?.has(run.id) && bareCauseMessage(messages.get(run.id))) { + vague.push(`${run.id}:${code} message=${JSON.stringify((messages.get(run.id) ?? "").slice(0, 120))}`); + } + } + return vague.length + ? { verdict: "fail", detail: `undiagnosable_failure ${vague.slice(0, 4).join("; ")}${vague.length > 4 ? ` (+${vague.length - 4})` : ""}` } + : { verdict: "pass", detail: `${failed.length} failed run(s) name a specific cause${messages ? " and message" : ""}; ${actions.length} recovery action(s)` }; +} + +/** A failure or Retry notice the board shows for this sample. */ +function failureNotice(sample: TaskSample): string | null { + const interruption = sample.interruption; + if (!interruption) return null; + if (interruption.canRetry) return "Retry offered"; + if (interruption.state === "failed") return "failed notice"; + return null; +} + +export function gradeU4(record: TaskRecord, bounds: UniversalBounds): Omit { + let staleFrom: number | null = null, worst = 0, what = ""; + const latestFinish = new Map(); + for (const sample of record.samples) { + if (!ready(sample)) continue; + for (const run of sample.runs ?? []) latestFinish.set(run.id, run); + const runs = [...latestFinish.values()].sort((a, b) => (ms(a.createdAt) || 0) - (ms(b.createdAt) || 0)); + const latest = runs[runs.length - 1]; + const active = runs.some(run => ["queued", "running", "scheduled_retry"].includes(run.status)); + const succeededLast = latest?.status === "succeeded" && !active; + const notice = succeededLast ? failureNotice(sample) : null; + if (!notice) { staleFrom = null; continue; } + staleFrom ??= sample.at; + if (sample.at - staleFrom > worst) { worst = sample.at - staleFrom; what = `${notice} after ${latest!.id} succeeded (status ${sample.issueStatus ?? "unknown"})`; } + } + // A notice still shown at rest is stale however briefly it was observed. + const last = [...record.samples].reverse().find(ready); + const atRest = record.restAt !== null && staleFrom !== null && last !== undefined && failureNotice(last) !== null; + if (worst > bounds.staleNoticeMs || atRest) return { verdict: "fail", metrics: { staleNoticeMs: worst }, detail: `stale_notice ${what} for ${seconds(worst)}${atRest ? ", still shown at rest" : ""}` }; + return { verdict: "pass", metrics: { staleNoticeMs: worst }, detail: worst ? `notice cleared ${seconds(worst)} after success` : "no notice after success" }; +} + +export function gradeUniversal(record: TaskRecord, options: { bounds?: UniversalBounds; db?: UniversalDbEvidence; + enforced?: ReadonlySet } = {}): UniversalGrade { + const bounds = options.bounds ?? universalBounds(); + const enforced = options.enforced ?? ENFORCED_UNIVERSAL; + const results = { + U1: gradeU1(record, bounds), + U2: gradeU2(record, bounds), + U3: gradeU3(record, options.db), + U4: gradeU4(record, bounds), + }; + return Object.fromEntries(Object.entries(results).map(([id, result]) => [id, { ...result, enforced: enforced.has(id as UniversalId) }])) as UniversalGrade; +} + +/** Violations that fail the task (enforced rules only). */ +export const enforcedUniversalFailures = (grade: UniversalGrade) => + (Object.entries(grade) as Array<[UniversalId, UniversalResult]>).filter(([, result]) => result.enforced && result.verdict === "fail"); +/** Violations reported as warnings. */ +export const universalWarnings = (grade: UniversalGrade) => + (Object.entries(grade) as Array<[UniversalId, UniversalResult]>).filter(([, result]) => !result.enforced && result.verdict === "fail"); diff --git a/tests/restart-chaos/workloads.ts b/tests/restart-chaos/workloads.ts index fd8b60a7605..6cc01a57211 100644 --- a/tests/restart-chaos/workloads.ts +++ b/tests/restart-chaos/workloads.ts @@ -1,4 +1,5 @@ import { questionHandoffWorkload } from "./question-handoff.js"; +import { l1Workload } from "./contention.js"; import type { OracleArtifacts, OracleResult, WorkloadId } from "./types.js"; /** Deterministic bench output: sum of squares 1..240. */ @@ -168,6 +169,7 @@ function budgetChecks(artifacts: OracleArtifacts) { export const workloadDefinitions: Record = { Q1: questionHandoffWorkload, + L1: l1Workload, G1: { id: "G1", label: "Approval-gated action across a restart",