mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-11 05:11:49 +01:00
test(restart-chaos): add database contention workload and universal checks U1-U4 (#20421)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - The restart chaos harness (`tests/restart-chaos/`) runs real agent tasks on an isolated control plane and grades them. > - Two production failures on 2026-11-09 passed the harness. Its tasks made little output and little accounting, so database write contention never happened (#20385). Its grading was per workload, so a task stuck for a reason no scenario targeted was not caught (#20361). > - The harness needs a contention workload and checks that apply to every task. > - This pull request adds the L1 workload, a `--db-pressure` knob, and the universal checks U1–U4. > - The benefit is that the #20385 failure class and unexplained stuck or misleading tasks now fail a wave. ## Linked Issues or Issue Description Refs #20385, #20383, #20361. **What happened?** The harness ran real embedded PostgreSQL, but no task streamed enough run-log output while usage receipts committed. The #20385 receipt race (SQLSTATE 40001, retained outbox, `native_event_replay_conflict`) could not occur. Also, no rule checked every task for an owner or an actionable item, for retry loops, for diagnosable failures, or for stale notices. **Expected behavior** A chaos wave can create run-row contention between run-log writes and accounting writes and check that accounting stays exact. Every task of every wave gets scenario-independent liveness and notice checks. **Steps to reproduce** Run an L1 wave with `--db-pressure 4` against the parent of the #20385 merge (red), then against master (green). Commands are in Verification. ## What Changed - `contention.ts` (new): workload L1 (five 3,000-line shell outputs as five tool calls, then a brief), its accounting oracle, and `RunRowPressure`, the `--db-pressure <writers>` knob. The writers run `update heartbeat_runs set updated_at = updated_at` on the wave company's running runs, in the wave's own isolated database only. - L1 oracle: no `native_event_replay_conflict` or `runner_event_commit_stalled`; each producer event journaled once; a terminal receipt for each succeeded native run; finished runs accounted; per-run cost equals its cost events; the board's spend equals the cost-event sum with no runs pending accounting; and proof that the task produced contention. - `universal.ts` (new): U1 liveness, U2 no silent retry loop, U3 diagnosable failures, U4 no stale notices. `grader.ts` grades them on every task. U1, U2 and U4 fail a task. U3 is a warning. - `invariants.ts`: `readUniversalDbEvidence` reads failed-run error text and recovery-action causes from the wave database at wave end, for U3. - `universal-audit.ts` (new): runs U1–U4 over saved wave evidence to measure false positives and bound headroom. - `contention-db.test.ts` (new): on real PostgreSQL, under the pressure writers, the pre-#20385 repeatable-read receipt write exhausts its retries with 40001 and the current writer commits. - Small registry edits: `types.ts` (L1 id, `dbPressure`, optional `contention` and `universal` fields), `workloads.ts` (L1 entry), `config.ts` (`--db-pressure`, L1 not a default), `main.ts` (start and stop the writers, read database evidence before teardown), `ledger.ts` (universal verdicts and warnings in the ledger and `SUMMARY.md`). - README: "Database contention" and "Universal checks" sections, with the audit table and bound reasons. ## Verification - `contention-db.test.ts` (real embedded PostgreSQL): with `--db-pressure` writers (4), the pre-#20385 repeatable-read receipt write exhausted its retries with 40001 in 8 of 8 attempts; the current READ COMMITTED `persistUsageReceipt` committed 8 of 8. This is the harness-level red/green for #20385. - Live L1 waves (`--env local --adapters codex,claude --workloads L1 --faults K1 --concurrency 4 --tasks 4 --seed 20385 --db-pressure 4`), red on the #20385 parent `bf818280f2` with this harness overlaid and green on this branch: queued behind the shared one-wave-at-a-time lock and not yet run. The first red attempt failed at startup (the fresh checkout lacked a built plugin SDK; no tasks ran). Live results are not claimed here. - `pnpm exec vitest run --config tests/restart-chaos/vitest.config.ts`: all harness unit tests pass, including the new `universal.test.ts`, `contention.test.ts` and `contention-db.test.ts`. - `node scripts/typecheck-test-projects.mjs tests/restart-chaos`: passes. - Universal-check audit over 74 saved waves (1,592 tasks): U1 7 flags (one C1 stall and six R1 runs queued 8 min behind a retry row, the #20361 shape), U2 1, U3 22, U4 0. None of U1, U2 or U4 flags a passing task or an all-pass wave. U3 flags 3 passing tasks (all `adapter_failed`), so it ships as a warning. ## Risks - U1, U2 and U4 can now fail a task that passed R1–R6 and T. The audit found no such case in 70 saved waves. The bounds are environment-configurable. - `--db-pressure` is opt-in and writes no values. It writes only to the wave's own temporary database. - Other agents edit the same harness. This change keeps registry edits small; `grader.ts` and `invariants.ts` changes are additive. - No product code changes. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, used through Claude Code with tool use (shell, file editing, GitHub CLI). ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude <noreply@anthropic.com>
This commit is contained in:
DottaandClaude authored and GitHub committed
1 parent
0c7a628fd1
commit
5ab98b0063
14 files changed
+1099
-18
No files matched your search
@@ -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 <writers>` (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 <n>` 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 <writers>`** (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<n>:...`). 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 <results dir> [...]
|
||||
```
|
||||
|
||||
| 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/<wave>/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. |
|
||||
|
||||
@@ -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<WorkloadId>(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<WorkloadId>(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) {
|
||||
|
||||
@@ -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<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
||||
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<T>(companyId: string, work: () => Promise<T>) {
|
||||
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);
|
||||
});
|
||||
@@ -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> = {}): 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> = {}): ContentionEvidence => ({ runs,
|
||||
companyCostEventCents: 12.5, summary: { spendCents: 12.5, pendingRunCount: 0 }, error: null, ...extra });
|
||||
const failing = (result: ReturnType<typeof contentionChecks>) => 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/);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,219 @@
|
||||
/**
|
||||
* Database contention: L1 (high-output, high-accounting) and the harness-side
|
||||
* run-row pressure knob (`--db-pressure <writers>`).
|
||||
*
|
||||
* 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<ContentionEvidence> {
|
||||
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<Record<string, unknown>>;
|
||||
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<void>[] = [];
|
||||
#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<PressureStats> {
|
||||
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;
|
||||
}
|
||||
}
|
||||
@@ -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<RuleId, RuleResult> = {
|
||||
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 };
|
||||
}
|
||||
@@ -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<Map<string, import("./universal.js").UniversalDbEvidence>> {
|
||||
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<string, { runMessages: Map<string, string | null>; 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;
|
||||
}
|
||||
@@ -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<string, string>;
|
||||
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} |`);
|
||||
|
||||
@@ -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<number> {
|
||||
const pointResults: Array<PointResult & { baseline?: Set<string> }> = [];
|
||||
let currentPoint: (PointResult & { baseline: Set<string> }) | 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<number> {
|
||||
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<number> {
|
||||
};
|
||||
|
||||
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<number> {
|
||||
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<number> {
|
||||
}
|
||||
const attributed = new Set<string>();
|
||||
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<Record<string, unknown>>(`/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<number> {
|
||||
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<number> {
|
||||
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<number> {
|
||||
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);
|
||||
|
||||
@@ -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 <writers>`: 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<RuleId, RuleResult>;
|
||||
T: OracleResult;
|
||||
/** Universal checks U1–U4 (universal.ts); absent on grades written before them. */
|
||||
universal?: import("./universal.js").UniversalGrade;
|
||||
pass: boolean;
|
||||
signature: string | null;
|
||||
}
|
||||
@@ -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] <results dir> [...]
|
||||
*
|
||||
* Each results dir holds `w-*` wave directories with `<task>.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<UniversalId, { flagged: number; onKnownGoodTasks: number; onKnownGoodWaves: number }>;
|
||||
flags: AuditFlag[];
|
||||
/** Per metric, over tasks that passed R1–R6+T: p50, p95, p99 and max. */
|
||||
passingMetrics: Record<string, { n: number; p50: number; p95: number; p99: number; max: number }>;
|
||||
}
|
||||
|
||||
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<string, number[]>();
|
||||
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] <results dir> [...]"); process.exit(2); }
|
||||
const report = auditWaves(roots);
|
||||
console.info(json ? JSON.stringify(report, null, 2) : renderAudit(report));
|
||||
}
|
||||
@@ -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> = {}): 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> = {}): 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> = {}): 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");
|
||||
});
|
||||
});
|
||||
@@ -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<string, number> }
|
||||
export type UniversalGrade = Record<UniversalId, UniversalResult>;
|
||||
|
||||
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<UniversalId> = new Set<UniversalId>(["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<string> = 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<string, RunSample & { firstSeenAt: number; finishedSeenAt: number | null; statusWhenFinished: string | null }> {
|
||||
const runs = new Map<string, RunSample & { firstSeenAt: number; finishedSeenAt: number | null; statusWhenFinished: string | null }>();
|
||||
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<string, number>, 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<UniversalResult, "enforced"> {
|
||||
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<string, number>();
|
||||
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<UniversalResult, "enforced"> {
|
||||
// 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<string, string | null>;
|
||||
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<UniversalResult, "enforced"> {
|
||||
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<UniversalResult, "enforced"> {
|
||||
let staleFrom: number | null = null, worst = 0, what = "";
|
||||
const latestFinish = new Map<string, RunSample>();
|
||||
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<UniversalId> } = {}): 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");
|
||||
@@ -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<WorkloadId, WorkloadDefinition> = {
|
||||
Q1: questionHandoffWorkload,
|
||||
L1: l1Workload,
|
||||
G1: {
|
||||
id: "G1",
|
||||
label: "Approval-gated action across a restart",
|
||||
|
||||
Reference in new issue
Block a user