diff --git a/tests/restart-chaos/config.ts b/tests/restart-chaos/config.ts index 69900e93a4e..1395a16a8fa 100644 --- a/tests/restart-chaos/config.ts +++ b/tests/restart-chaos/config.ts @@ -154,6 +154,9 @@ export function parseCommand(argv: readonly string[], now = Date.now()): ParsedC ? { crashPointHoldMs: integer(values["crash-point-hold-ms"], "crash-point-hold-ms", 0, 0, 600_000) } : {}), }; if (!crashPoint && values["crash-point-hold-ms"] !== undefined) throw new Error("--crash-point-hold-ms applies only to --crash-point"); + if (options.workloads.includes("M1") && options.adapters.length < 2) { + throw new Error("--workloads M1 delegates between two agents; pass at least two --adapters"); + } if (env !== "daytona" && values["daytona-cpu"] !== undefined) throw new Error("--daytona-cpu applies only to --env daytona"); if (env !== "daytona" && values["daytona-reuse-lease"]) throw new Error("--daytona-reuse-lease applies only to --env daytona"); if (options.daytonaCpu !== undefined) { diff --git a/tests/restart-chaos/grader.test.ts b/tests/restart-chaos/grader.test.ts index 416fc214eb0..f300f9567ed 100644 --- a/tests/restart-chaos/grader.test.ts +++ b/tests/restart-chaos/grader.test.ts @@ -68,17 +68,20 @@ describe("R1–R6 grading on fixture timelines", () => { expect(grade.signature).toBe("R1:rested_not_done:A1:K1"); }); - it("passes R3 when one Retry starts a run that completes the task", () => { + it("fails R3 when the task needed a Retry, even one that completed it, and names a second click", () => { const samples = [ sample(60_000, "blocked", [run("r1", "failed", 1_000, 30_000)], { interruption: { state: "failed", canRetry: true, retryWait: false, workspaceFailure: null, observedRunId: "r1", userAction: null } }), sample(80_000, "in_progress", [run("r1", "failed", 1_000, 30_000), run("r2", "running", 75_000, null, { retryOfRunId: "r1" })]), sample(140_000, "done", [run("r1", "failed", 1_000, 30_000), run("r2", "succeeded", 75_000, 130_000)]), ]; const retried = record(samples, { retry: { at: T0 + 74_000, observedRunId: "r1", httpStatus: 200, accepted: true, error: null }, restAt: T0 + 140_000 }); - expect(gradeR3(retried)).toMatchObject({ verdict: "pass" }); + expect(gradeR3(retried)).toEqual({ verdict: "fail", detail: "retry_needed (failed)" }); const grade = gradeTask(retried, [fault], workloadDefinitions.A1.budgetMs, oracle()); expect(grade.rules.R1.verdict).toBe("n/a"); - expect(grade.pass).toBe(true); + expect(grade.pass).toBe(false); + const twice = record(samples, { retry: { at: T0 + 74_000, observedRunId: "r1", httpStatus: 200, accepted: true, error: null }, + priorRetries: [{ at: T0 + 50_000, observedRunId: "r0", httpStatus: 200, accepted: true, error: null }], restAt: T0 + 140_000 }); + expect(gradeR3(twice).detail).toBe("retry_needed_2_clicks (failed)"); }); it("fails R3 for a dead Retry (accepted but no new run)", () => { @@ -189,6 +192,29 @@ describe("R1–R6 grading on fixture timelines", () => { }); describe("T-oracles", () => { + it("grades M1 delegation: two done subtasks, part 1 delegated, and a summary of both", () => { + const agents = { self: { id: "self", name: "Chaos codex" }, other: { id: "other", name: "Chaos claude" } }; + const child = (n: number, assigneeAgentId: string, status = "done") => + ({ id: `c${n}`, status, assigneeAgentId, documents: { part: `PART${n}-abcd1234\nIdempotency keys.` } }); + const artifacts: OracleArtifacts = { issueStatus: "done", agentComments: [], files: {}, commands: {}, workspacePath: null, + documents: { summary: "PART1-abcd1234\nPART2-abcd1234" }, children: [child(1, "other"), child(2, "self")] }; + expect(workloadDefinitions.M1.grade("abcd1234", artifacts, agents).verdict).toBe("pass"); + expect(workloadDefinitions.M1.description("abcd1234", agents)).toContain("assigned to Chaos claude"); + const undelegated = { ...artifacts, children: [child(1, "self"), child(2, "self")] }; + expect(workloadDefinitions.M1.grade("abcd1234", undelegated, agents).checks.find(check => check.name === "part1_delegated")?.ok).toBe(false); + const pending = { ...artifacts, children: [child(1, "other", "in_progress"), child(2, "self")] }; + expect(workloadDefinitions.M1.grade("abcd1234", pending, agents).verdict).toBe("fail"); + expect(workloadDefinitions.M1.grade("abcd1234", { ...artifacts, children: [child(1, "other")] }, agents).verdict).toBe("fail"); + const both = { ...artifacts, children: [{ ...child(1, "other"), documents: { part: "PART1-abcd1234\nPART2-abcd1234" } }, + { ...child(2, "self"), documents: { part: "nothing" } }] }; + expect(workloadDefinitions.M1.grade("abcd1234", both, agents).checks.find(check => check.name === "parts_in_distinct_subtasks")?.ok).toBe(false); + const wrongPart2 = { ...artifacts, children: [child(1, "other"), child(2, "other")] }; + expect(workloadDefinitions.M1.grade("abcd1234", wrongPart2, agents).checks.find(check => check.name === "part2_kept")?.ok).toBe(false); + const single = { self: agents.self, other: agents.self }; + expect(workloadDefinitions.M1.grade("abcd1234", { ...artifacts, children: [child(1, "self"), child(2, "self")] }, single) + .checks.find(check => check.name === "two_agents")?.ok).toBe(false); + }); + it("grades A1 and C1 markers", () => { expect(workloadDefinitions.A1.grade("abcd1234", passingBrief).verdict).toBe("pass"); expect(workloadDefinitions.C1.grade("abcd1234", passingBrief).checks.find(check => check.name.includes("followup"))?.ok).toBe(false); diff --git a/tests/restart-chaos/grader.ts b/tests/restart-chaos/grader.ts index aaead120c04..03b565b187e 100644 --- a/tests/restart-chaos/grader.ts +++ b/tests/restart-chaos/grader.ts @@ -100,7 +100,18 @@ export function gradeR3(record: TaskRecord): RuleResult { const newRun = record.samples.some(sample => (sample.runs ?? []).some(run => (ms(run.createdAt) || 0) >= retryAt - 1_000)); if (!newRun) return { verdict: "fail", detail: "retry_no_new_run" }; if (record.finalStatus !== "done") return { verdict: "fail", detail: `retry_did_not_complete (status ${record.finalStatus ?? "unknown"}${record.timedOut ? ", timed out" : ""})` }; - return { verdict: "pass", detail: "one Retry started a run that completed the task" }; + const clicks = 1 + (record.priorRetries?.length ?? 0); + const failed = failedRunCodes(record).join(",") || "unknown"; + if (clicks > 1) return { verdict: "fail", detail: `retry_needed_${clicks}_clicks (${failed})` }; + return { verdict: "fail", detail: `retry_needed (${failed})` }; +} + +function failedRunCodes(record: TaskRecord): string[] { + const codes = new Map(); + for (const sample of record.samples) { + for (const run of sample.runs ?? []) if (run.status === "failed") codes.set(run.id, run.errorCode ?? "failed"); + } + return [...codes.values()]; } /** Run cancellations decided before adapter dispatch: the run never executed. */ diff --git a/tests/restart-chaos/main.ts b/tests/restart-chaos/main.ts index 73e06d45538..f8c765c4dee 100644 --- a/tests/restart-chaos/main.ts +++ b/tests/restart-chaos/main.ts @@ -21,6 +21,7 @@ import { knownSecretValues, redactKnownSecrets, writeEvidenceBundle } from "./ev import { gradeTask } from "./grader.js"; import { checkPlanInvariants, summarizeViolations, type InvariantCheck } from "./invariants.js"; import { appendLedger, ledgerLine, writeSummary, type LedgerLine } from "./ledger.js"; +import { collectWaveSpeed, speedMarkdown } from "./wave-speed.js"; import { TaskObserver } from "./observer.js"; import { StormRun, TASK_FAULT_LOG, observedFaultSchedule, planWave, replayRecordFor, replayWave, type TaskFaultLog } from "./scheduler.js"; import { chaosLifecycleOverrides, seedChaosLab } from "./seed.js"; @@ -179,6 +180,7 @@ async function main(): Promise { const faultWork: Promise[] = []; const lines: LedgerLine[] = []; let anyFailure = false; + let speedSection: string | null = null; let harnessError: unknown; process.on("uncaughtException", error => { if (isSocketTypeOfServiceRace(error)) { timeline.add("harness_socket_race_ignored", { error: String(error) }); return; } @@ -208,7 +210,8 @@ async function main(): Promise { const api = new ChaosApi(supervisor.baseUrl); const lab = await seedChaosLab({ api, env: plan.options.env, daytonaImage: plan.options.daytonaImage, daytonaCpu: plan.options.daytonaCpu, daytonaReuseLease: plan.options.daytonaReuseLease, - adapters: [...new Set(plan.tasks.map(task => task.adapter))], + adapters: [...new Set([...plan.tasks.map(task => task.adapter), + ...(plan.tasks.some(task => task.workload === "M1") ? plan.options.adapters : [])])], concurrency: plan.options.concurrency, budgetCents: plan.options.budgetCents, masks: plan.options.masks, fixtureRoot: path.join(supervisor.paths.temporaryRoot, "fixtures"), waveId: plan.waveId, timeline, adapterConfigOverrides: chaosLifecycleOverrides(plan.options.lifecycle), @@ -414,7 +417,7 @@ async function main(): Promise { } let artifacts: OracleArtifacts | null = null; if (observer.record.issueId) artifacts = await observer.collectArtifacts().catch(() => null); - const oracle = artifacts ? observer.definition.grade(observer.plan.nonce, artifacts) + const oracle = 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" }] }; const grade = gradeTask(observer.record, faults, observer.definition.budgetMs, oracle); let evidence: string | null = null; @@ -430,6 +433,19 @@ async function main(): Promise { rules: Object.fromEntries(Object.entries(grade.rules).map(([id, rule]) => [id, rule.verdict])), T: grade.T.verdict }); } await writeFile(path.join(waveDir, "faults.json"), `${JSON.stringify(faults, null, 2)}\n`, { mode: 0o600 }); + const speed = await (async () => { + const childTasks = (await Promise.all(launched.map(observer => observer.childSpeedTasks()))).flat(); + return collectWaveSpeed(api, plan.options.env, [...launched.flatMap(observer => observer.record.issueId + ? [{ issueId: observer.record.issueId, adapter: observer.plan.adapter, samples: observer.record.samples }] : []), ...childTasks], faults); + })().catch(error => { timeline.add("wave_speed_failed", { error: String(error) }); anyFailure = true; return null; }); + if (speed) { + speedSection = speedMarkdown(speed); + await writeFile(path.join(waveDir, "speed.json"), `${JSON.stringify(speed, null, 2)}\n`, { mode: 0o600 }); + timeline.add("wave_speed", { breaches: speed.breaches, runBudgetExceeded: speed.runBudgetExceeded, singleRunBreaches: speed.singleRunBreaches, + rows: speed.rows.map(row => ({ adapter: row.adapter, metric: row.metric, start: row.start, n: row.measured.n, + p50: row.measured.p50, p95: row.measured.p95, breaches: row.breaches })) }); + if (speed.breaches > 0 || speed.errors.length > 0) anyFailure = true; + } if (crashPoint) { if (currentPoint) { await finishCrashPoint(currentPoint); currentPoint = null; } const byTask = new Map(lines.map(line => [line.taskId, line])); @@ -500,7 +516,7 @@ async function main(): Promise { if (lines.length) { await appendLedger(resultsRoot, lines); const summary = await writeSummary(resultsRoot, plan.waveId); - await writeFile(path.join(waveDir, "SUMMARY.md"), summary, { mode: 0o600 }); + await writeFile(path.join(waveDir, "SUMMARY.md"), speedSection ? `${summary}\n${speedSection}` : summary, { mode: 0o600 }); console.info(`\n${summary}`); } if (keep || anyFailure || harnessError) timeline.add("instance_kept", { temporaryRoot: supervisor.paths.temporaryRoot }); diff --git a/tests/restart-chaos/observer.ts b/tests/restart-chaos/observer.ts index 672dec5f088..7d84f17e39d 100644 --- a/tests/restart-chaos/observer.ts +++ b/tests/restart-chaos/observer.ts @@ -9,12 +9,12 @@ import { ACTIVE_RUN_STATUSES, isStalledSample, STALL_LIMIT_MS } from "./grader.j import type { ChaosLab } from "./seed.js"; import type { Timeline } from "./timeline.js"; import type { - InterruptionSample, OracleArtifacts, RestSweep, RetryAttempt, RunSample, ScheduledFault, TaskPlan, TaskRecord, TaskSample, + AdapterId, InterruptionSample, OracleArtifacts, RestSweep, RetryAttempt, RunSample, ScheduledFault, TaskPlan, TaskRecord, TaskSample, } from "./types.js"; import { restartProcessStartMatches } from "../runner-e2e/process-tree-owner.js"; import { runOracleCommand } from "./oracle-sandbox.js"; import { taskFaultSchedule } from "./scheduler.js"; -import { workloadDefinitions } from "./workloads.js"; +import { workloadDefinitions, type WorkloadAgents } from "./workloads.js"; const IGNORED_DIRECTORIES = new Set([".git", ".bench", "__pycache__", "node_modules"]); const MAX_SNAPSHOT_FILES = 2_000; @@ -47,6 +47,7 @@ const RESULT_TRIGGER_EVENTS = new Set(["run.result.proposed", "run.result.accept /** Rest must hold for this long before the observer acts on it. */ export const REST_CONFIRM_MS = 12_000; +export const CHILD_WAKE_GRACE_MS = 60_000; /** Extra time granted after a Retry click before R3 is graded as timed out. */ export const RETRY_BUDGET_MS = 15 * 60_000; @@ -139,6 +140,7 @@ export class TaskObserver { readonly record: TaskRecord; issue: Json | null = null; private restCandidateSince: number | null = null; + private childrenSettledAt: number | null = null; private stallSince: number | null = null; private readonly faultSchedule: ScheduledFault[]; private readonly pendingFaults: ScheduledFault[]; @@ -200,7 +202,7 @@ export class TaskObserver { try { const issue = await this.api.post(`/api/companies/${this.lab.companyId}/issues`, { title: definition.title(this.plan.nonce), - description: definition.description(this.plan.nonce), + description: definition.description(this.plan.nonce, this.workloadAgents()), status: "todo", priority: "medium", ...(assign ? { assigneeAgentId: this.lab.agents[this.plan.adapter].id } : {}), @@ -457,6 +459,15 @@ export class TaskObserver { // C1: a follow-up posted after rest must be served before the task counts as resting. if (this.definition.followUp && !this.record.followUp) return; const faultSinceRetry = this.record.retry !== null && (this.record.lastFaultAt ?? 0) > this.record.retry.at; + if (status !== "done" && this.definition.collect.childDocuments && this.record.issueId) { + const children = await this.collectChildren(this.record.issueId).catch(() => null); + const pending = !children || children.some(child => !["done", "cancelled"].includes(child.status ?? "")); + this.childrenSettledAt = pending ? null : (this.childrenSettledAt ?? now); + if (pending || now - this.childrenSettledAt! < CHILD_WAKE_GRACE_MS) { + this.restCandidateSince = null; + return; + } + } if (status !== "done" && (!this.record.retry || faultSinceRetry) && sample.interruption?.canRetry) { if (this.record.retry) (this.record.priorRetries ??= []).push(this.record.retry); this.record.retry = await this.retry(sample, now); @@ -501,6 +512,44 @@ export class TaskObserver { return null; } + workloadAgents(): WorkloadAgents { + const self = this.lab.agents[this.plan.adapter]; + const other = Object.values(this.lab.agents).find(agent => agent.id !== self.id) ?? self; + return { self, other }; + } + + async childSpeedTasks(): Promise }>> { + if (!this.definition.collect.childDocuments || !this.record.issueId) return []; + const rows = await this.api.get(`/api/companies/${this.lab.companyId}/issues?view=index&parentIds=${encodeURIComponent(this.record.issueId)}&limit=20`); + const tasks: Array<{ issueId: string; adapter: AdapterId; samples: Array<{ runs?: RunSample[] }> }> = []; + for (const row of (rows ?? []).filter(child => child.parentId === this.record.issueId)) { + const adapter = (Object.entries(this.lab.agents) as Array<[AdapterId, { id: string }]>) + .find(([, agent]) => agent.id === row.assigneeAgentId)?.[0]; + if (!adapter) continue; + const runs = await this.api.get(`/api/issues/${row.id}/runs`, 15_000); + tasks.push({ issueId: String(row.id), adapter, samples: [{ runs: (runs ?? []).map(toRunSample) }] }); + } + return tasks; + } + + async collectChildren(issueId: string): Promise> { + const keys = this.definition.collect.childDocuments; + if (!keys) return []; + const rows = await this.api.get(`/api/companies/${this.lab.companyId}/issues?view=index&parentIds=${encodeURIComponent(issueId)}&limit=20`) + .catch(() => []); + const children: NonNullable = []; + for (const row of (rows ?? []).filter(child => child.parentId === issueId)) { + const documents: Record = {}; + for (const key of keys) { + const document = await this.api.get(`/api/issues/${row.id}/documents/${key}`).catch(() => null); + documents[key] = typeof document?.body === "string" ? document.body : null; + } + children.push({ id: String(row.id), status: typeof row.status === "string" ? row.status : null, + assigneeAgentId: typeof row.assigneeAgentId === "string" ? row.assigneeAgentId : null, documents }); + } + return children; + } + async collectArtifacts(): Promise { const definition = this.definition; const issueId = this.record.issueId!; @@ -538,6 +587,7 @@ export class TaskObserver { agentComments: (comments ?? []).filter(comment => comment.authorAgentId).map(comment => String(comment.body ?? "")), files, commands, workspacePath, ...(this.plan.workload === "B5" ? { benchProgressObserved: this.record.benchProgressSeen?.length ?? 0 } : {}), + ...(definition.collect.childDocuments ? { children: await this.collectChildren(issueId) } : {}), }; } diff --git a/tests/restart-chaos/ttft-baseline-stats.test.ts b/tests/restart-chaos/ttft-baseline-stats.test.ts index 2a1b7173b96..f82fd5159b5 100644 --- a/tests/restart-chaos/ttft-baseline-stats.test.ts +++ b/tests/restart-chaos/ttft-baseline-stats.test.ts @@ -92,8 +92,8 @@ describe("ttft baseline statistics", () => { ]); const silent = run("r2", "restart", []); expect(recoverySamples([recovered, silent], [{ kind: "K1", readyAt, runIds: ["r1", "r2", "missing"] }])).toEqual([ - { runId: "r1", adapter: "codex", kind: "K1", disposition: "resume_dead_runner", latencyMs: 3_250 }, - { runId: "r2", adapter: "codex", kind: "K1", disposition: "none", latencyMs: null }, + { runId: "r1", adapter: "codex", kind: "K1", disposition: "resume_dead_runner", readyAt, latencyMs: 3_250 }, + { runId: "r2", adapter: "codex", kind: "K1", disposition: "none", readyAt, latencyMs: null }, ]); }); }); diff --git a/tests/restart-chaos/ttft-baseline-stats.ts b/tests/restart-chaos/ttft-baseline-stats.ts index 9f90c0ecad6..5043f886feb 100644 --- a/tests/restart-chaos/ttft-baseline-stats.ts +++ b/tests/restart-chaos/ttft-baseline-stats.ts @@ -117,7 +117,7 @@ export function runMetrics(run: BaselineRun): RunMetrics { const PROVIDER_EVENT_TYPES = new Set(["turn.submitted", "turn.started", "turn.accepted", "item.started", "item.delta", "item.completed", "tool.execution.started", "tool.execution.progressed", "tool.execution.completed", "session.resumed"]); -export interface RecoverySample { runId: string; adapter: string; kind: string; disposition: string; latencyMs: number | null } +export interface RecoverySample { runId: string; adapter: string; kind: string; disposition: string; readyAt: number; latencyMs: number | null } /** Control plane ready → the recovered run's next provider event, keyed by runner disposition. */ export function recoverySamples(runs: readonly BaselineRun[], faults: readonly BaselineFault[]): RecoverySample[] { @@ -131,7 +131,7 @@ export function recoverySamples(runs: readonly BaselineRun[], faults: readonly B const transition = after.find(event => event.eventType === "native.recovery.transition" && Date.parse(event.createdAt) >= fault.readyAt - 60_000); const next = run.events.find(event => PROVIDER_EVENT_TYPES.has(event.eventType) && Date.parse(event.createdAt) >= fault.readyAt); - samples.push({ runId, adapter: run.adapter, kind: fault.kind, + samples.push({ runId, adapter: run.adapter, kind: fault.kind, readyAt: fault.readyAt, disposition: String(transition?.payload?.runnerDisposition ?? "none"), latencyMs: next ? Date.parse(next.createdAt) - fault.readyAt : null }); } diff --git a/tests/restart-chaos/types.ts b/tests/restart-chaos/types.ts index 179acc04ee6..9e9776270f8 100644 --- a/tests/restart-chaos/types.ts +++ b/tests/restart-chaos/types.ts @@ -5,7 +5,7 @@ export type FaultKind = (typeof FAULT_KINDS)[number]; /** A task's planned fault. "K13" (restart storm) tasks have none of their own; the wave's storm crosses them. */ export type TaskFaultKind = FaultKind | "K13"; -export const WORKLOAD_IDS = ["A1", "B1", "B5", "C1"] as const; +export const WORKLOAD_IDS = ["A1", "B1", "B5", "C1", "M1"] as const; export type WorkloadId = (typeof WORKLOAD_IDS)[number]; export const ADAPTER_IDS = ["codex", "claude", "opencode"] as const; @@ -209,6 +209,7 @@ export interface OracleArtifacts { workspacePath: string | null; /** B5: distinct `.bench/progress` values the harness itself saw while the task was live. */ benchProgressObserved?: number; + children?: Array<{ id: string; status: string | null; assigneeAgentId: string | null; documents: Record }>; } export interface OracleResult { verdict: Exclude; checks: Array<{ name: string; ok: boolean; detail?: string }> } diff --git a/tests/restart-chaos/wave-speed.test.ts b/tests/restart-chaos/wave-speed.test.ts new file mode 100644 index 00000000000..b283fc11f68 --- /dev/null +++ b/tests/restart-chaos/wave-speed.test.ts @@ -0,0 +1,101 @@ +import { describe, expect, it } from "vitest"; +import type { BaselineRun } from "./ttft-baseline-stats.js"; +import type { FaultReceipt } from "./types.js"; +import { breachesOf, budgetFor, faultRuns, firstActivityMs, gradeWaveSpeed, outageWindows, speedMarkdown, traceStartMs, withoutOutage } from "./wave-speed.js"; +import { runMetrics } from "./ttft-baseline-stats.js"; + +const span = (name: string, startOffsetMs: number, durationMs: number) => + ({ eventType: "run.performance.span", createdAt: "2026-11-08T00:00:00.000Z", payload: { span: name, startOffsetMs, durationMs } }); + +const run = (runId: string, adapter: string, ttftMs: number, extra: BaselineRun["events"] = []): BaselineRun => ({ + runId, adapter, environment: "local", phase: "cold", + events: [span("task.time_to_first_token", 0, ttftMs), span("heartbeat.queue", 0, 100), span("runner.turn.submit", 100, 900), ...extra], +}); + +describe("wave speed grade", () => { + it("judges aggregate TTFT and boot against the local budgets and counts single-run breaches", () => { + const fast = gradeWaveSpeed("local", [run("a", "codex", 4_000), run("b", "codex", 5_000)], [], 0); + expect(fast.breaches).toBe(0); + expect(fast.rows.find(row => row.metric === "time_to_first_activity")).toMatchObject({ adapter: "codex", start: "cold", breaches: [] }); + + const slow = gradeWaveSpeed("local", [run("a", "codex", 40_000), run("b", "codex", 40_000, + [{ eventType: "run.performance.budget_exceeded", createdAt: "2026-11-08T00:00:00.000Z", payload: {} }])], [], 1); + expect(slow.rows.find(row => row.metric === "time_to_first_activity")?.breaches).toEqual(["p50", "p95"]); + expect(slow.rows.find(row => row.metric === "ttft")).toMatchObject({ budget: null, breaches: [] }); + expect(slow).toMatchObject({ breaches: 4, singleRunBreaches: 2, runBudgetExceeded: 1 }); + const heldButFast = gradeWaveSpeed("local", [run("c", "codex", 4_000, + [{ eventType: "run.performance.budget_exceeded", createdAt: "2026-11-08T00:00:00.000Z", payload: {} }])], [], 1); + expect(heldButFast).toMatchObject({ breaches: 0, singleRunBreaches: 0, runBudgetExceeded: 1 }); + expect(speedMarkdown(slow)).toContain("| codex | time_to_first_activity | cold | 2 |"); + }); + + it("judges time to first agent activity when the run logs it, net of the wait behind a live turn", () => { + const withActivity = run("a", "codex", 40_000, [span("task.time_to_first_agent_activity", 0, 9_000), + span("comment.to_run_created", 0, 4_000)]); + expect(firstActivityMs(withActivity)).toBe(5_000); + expect(firstActivityMs(run("b", "codex", 7_000))).toBe(7_000); + const speed = gradeWaveSpeed("local", [withActivity], [], 0); + expect(speed.rows.find(row => row.metric === "time_to_first_activity")).toMatchObject({ breaches: [], measured: { p50: 5_000 } }); + expect(speed.rows.find(row => row.metric === "ttft")).toMatchObject({ budget: null, measured: { p50: 36_000 } }); + }); + + it("measures restart recovery from ready to the run's next provider event", () => { + const recovered: BaselineRun = { runId: "r", adapter: "claude", environment: "local", phase: "cold", + events: [{ eventType: "turn.started", createdAt: new Date(10_000 + 30_000).toISOString() }] }; + const speed = gradeWaveSpeed("local", [recovered], [{ kind: "K1", readyAt: 10_000, runIds: ["r"] }], 0); + expect(speed.rows.find(row => row.metric === "restart_recovery")).toMatchObject({ adapter: "claude", breaches: ["p50", "p95"] }); + }); + + it("removes a later outage from restart recovery", () => { + const recovered: BaselineRun = { runId: "r", adapter: "claude", environment: "local", phase: "cold", + events: [{ eventType: "turn.started", createdAt: new Date(10_000 + 302_000).toISOString() }] }; + const windows = outageWindows([{ startedAt: 12_000, readyAt: 312_000 } as FaultReceipt]); + const speed = gradeWaveSpeed("local", [recovered], [{ kind: "K1", readyAt: 10_000, runIds: ["r"] }], 0, [], + { adjusted: 0, removedMs: 0, windows }); + expect(speed.rows.find(row => row.metric === "restart_recovery")).toMatchObject({ measured: { p50: 2_000 }, breaches: [] }); + }); + + it("has no budget for remote environments or unbudgeted adapters", () => { + expect(budgetFor("daytona", "codex", "ttft", "cold")).toBeNull(); + expect(budgetFor("local", "unknown", "ttft", "cold")).toBeNull(); + expect(breachesOf({ n: 1, p50: 99_000, p95: 99_000, max: 99_000 }, null)).toEqual([]); + }); + + const traced = (trace: number, spans: Array<[string, number, number]>): BaselineRun => ({ + runId: "t", adapter: "claude", environment: "local", phase: "cold", + events: spans.map(([name, offset, duration]) => ({ eventType: "run.performance.span", + createdAt: new Date(trace + offset + duration).toISOString(), payload: { span: name, startOffsetMs: offset, durationMs: duration } })), + }); + + it("finds the trace start from when its spans were written, never later than it", () => { + expect(traceStartMs(traced(1_000_000, [["a", 0, 5_000], ["b", 2_000, 1_000]]))).toBe(1_000_000); + const late = traced(1_000_000, [["a", 0, 5_000]]); + late.events.push({ ...late.events[0]!, createdAt: new Date(1_090_000).toISOString() }); + expect(traceStartMs(late)).toBe(1_000_000); + }); + + it("removes outage time from a held TTFT span", () => { + const trace = 1_000_000; + const windows = outageWindows([{ startedAt: trace + 10_000, readyAt: trace + 260_000 } as FaultReceipt, + { startedAt: 0, readyAt: null } as unknown as FaultReceipt]); + const trimmed = withoutOutage(traced(trace, [["task.time_to_first_token", 0, 300_000]]), windows); + expect(trimmed).toMatchObject({ adjusted: 1, removedMs: 250_000, anchored: true }); + const speed = gradeWaveSpeed("local", [trimmed.run], [], 0, [], trimmed); + expect(speed.rows.find(row => row.metric === "time_to_first_activity")?.measured.p50).toBe(50_000); + expect(speedMarkdown(speed)).toContain("Outage time removed from 1 spans: 250.0 s"); + }); + + it("never moves a hold inside the queue into agent boot", () => { + const trace = 1_000_000; + const windows = [{ startedAt: trace + 10_000, readyAt: trace + 160_000 }]; + const run = traced(trace, [["heartbeat.queue", 0, 200_000], ["runner.turn.submit", 200_000, 1_000]]); + expect(runMetrics(run).bootMs).toBe(1_000); + expect(runMetrics(withoutOutage(run, windows).run).bootMs).toBe(1_000); + }); + + it("attributes each fault to the runs it crossed", () => { + const fault = { kind: "K2", startedAt: 1_000, readyAt: 2_000 } as FaultReceipt; + expect(faultRuns([fault], [{ id: "before", createdAt: 0, finishedAt: 500 }, { id: "live", createdAt: 0, finishedAt: null }, + { id: "after", createdAt: 1_500, finishedAt: null }])).toEqual([{ kind: "K2", readyAt: 2_000, runIds: ["live"] }]); + }); +}); diff --git a/tests/restart-chaos/wave-speed.ts b/tests/restart-chaos/wave-speed.ts new file mode 100644 index 00000000000..61afe2e1001 --- /dev/null +++ b/tests/restart-chaos/wave-speed.ts @@ -0,0 +1,220 @@ +import { LOCAL_SPEED_BUDGETS, type SpeedBudget, type SpeedBudgetAdapter } from "../../server/src/services/native-runtime/native-speed-budgets.js"; +import type { ChaosApi } from "./api.js"; +import { + baselineRows, distribution, recoverySamples, runMetrics, runSpans, runWarmth, + type BaselineEvent, type BaselineFault, type BaselineRun, type Distribution, +} from "./ttft-baseline-stats.js"; +import type { AdapterId, EnvironmentKind, FaultReceipt, RunSample } from "./types.js"; + +type Json = Record; + +export interface WaveSpeedTask { issueId: string | null; adapter: AdapterId; samples: Array<{ runs?: RunSample[] }> } + +export interface SpeedRow { + adapter: string; metric: "time_to_first_activity" | "ttft" | "agent_boot" | "restart_recovery"; start: "cold" | "warm" | "restart"; + measured: Distribution; budget: SpeedBudget | null; breaches: Array<"p50" | "p95">; +} + +export interface WaveSpeed { + environment: EnvironmentKind; + rows: SpeedRow[]; + outageAdjustedSpans: number; + outageRemovedMs: number; + recoveryWithoutProviderEvent: number; + runBudgetExceeded: number; + singleRunBreaches: number; + breaches: number; + errors: string[]; +} + +const BUDGET_ADAPTER: Partial> = { codex: "codex", claude: "claude", opencode: "opencode" }; + +export function budgetFor(environment: EnvironmentKind, adapter: string, metric: SpeedRow["metric"], + start: SpeedRow["start"]): SpeedBudget | null { + const key = BUDGET_ADAPTER[adapter as AdapterId]; + if (environment !== "local" || !key) return null; + if (metric === "ttft") return null; + const budgets = LOCAL_SPEED_BUDGETS[key] as unknown as Record | SpeedBudget>; + if (metric === "restart_recovery") return budgets.restart_recovery as SpeedBudget; + const byStart = (metric === "time_to_first_activity" ? budgets.time_to_first_activity ?? budgets.ttft : budgets[metric]) as + Record<"cold" | "warm", SpeedBudget>; + return byStart[start === "warm" ? "warm" : "cold"]; +} + +export function breachesOf(measured: Distribution, budget: SpeedBudget | null): Array<"p50" | "p95"> { + if (!budget) return []; + const out: Array<"p50" | "p95"> = []; + if (measured.p50 !== null && measured.p50 > budget.p50Ms) out.push("p50"); + if (measured.p95 !== null && measured.p95 > budget.p95Ms) out.push("p95"); + return out; +} + +export function faultRuns(faults: readonly FaultReceipt[], runs: ReadonlyArray<{ id: string; createdAt: number; finishedAt: number | null }>): BaselineFault[] { + return faults.filter(fault => fault.readyAt !== null).map(fault => ({ kind: fault.kind, readyAt: fault.readyAt!, + runIds: runs.filter(run => run.createdAt <= fault.startedAt && (run.finishedAt === null || run.finishedAt >= fault.startedAt)) + .map(run => run.id) })); +} + +export interface OutageWindow { startedAt: number; readyAt: number } + +export function outageWindows(faults: readonly FaultReceipt[]): OutageWindow[] { + return faults.filter(fault => fault.readyAt !== null).map(fault => ({ startedAt: fault.startedAt, readyAt: fault.readyAt! })); +} + +export function traceStartMs(run: BaselineRun): number | null { + let earliest: number | null = null; + for (const event of run.events) { + if (event.eventType !== "run.performance.span" || !event.payload) continue; + const written = Date.parse(event.createdAt); + const end = Number(event.payload.startOffsetMs) + Number(event.payload.durationMs); + if (!Number.isFinite(written) || !Number.isFinite(end)) continue; + const candidate = written - end; + if (earliest === null || candidate < earliest) earliest = candidate; + } + return earliest; +} + +export function outageFreeClock(windows: readonly OutageWindow[]): (at: number) => number { + return at => at - windows.reduce((total, window) => + total + Math.max(0, Math.min(at, window.readyAt) - window.startedAt), 0); +} + +export function withoutOutage(run: BaselineRun, windows: readonly OutageWindow[]): + { run: BaselineRun; adjusted: number; removedMs: number; anchored: boolean } { + const traceStart = traceStartMs(run); + if (traceStart === null) return { run, adjusted: 0, removedMs: 0, anchored: false }; + const clock = outageFreeClock(windows); + const origin = clock(traceStart); + let adjusted = 0, removedMs = 0; + const events = run.events.map(event => { + const payload = event.payload; + if (event.eventType !== "run.performance.span" || !payload) return event; + const offset = Number(payload.startOffsetMs), duration = Number(payload.durationMs); + if (!Number.isFinite(offset) || !Number.isFinite(duration)) return event; + const start = traceStart + offset, end = start + duration; + const startOffsetMs = clock(start) - origin, durationMs = clock(end) - clock(start); + if (startOffsetMs === offset && durationMs === duration) return event; + if (durationMs !== duration) { adjusted += 1; removedMs += duration - durationMs; } + return { ...event, payload: { ...payload, startOffsetMs, durationMs } }; + }); + return { run: { ...run, events }, adjusted, removedMs, anchored: true }; +} + +export function firstActivityMs(run: BaselineRun): number | null { + const metrics = runMetrics(run); + const span = runSpans(run).find(candidate => candidate.name === "task.time_to_first_agent_activity"); + if (!span) return metrics.ttftAfterRunCreatedMs; + return Math.max(0, span.durationMs - (metrics.inputWaitMs ?? 0)); +} + +function firstActivityByGroup(runs: readonly BaselineRun[]): Map { + const groups = new Map(); + for (const run of runs) { + if (run.phase === "restart" || run.phase === "verify") continue; + const value = firstActivityMs(run); + if (value === null) continue; + const key = `${run.adapter}|${runWarmth(run)}`; + groups.set(key, [...(groups.get(key) ?? []), value]); + } + return groups; +} + +export function gradeWaveSpeed(environment: EnvironmentKind, runs: readonly BaselineRun[], faults: readonly BaselineFault[], + runBudgetExceeded: number, errors: string[] = [], + outage: { adjusted: number; removedMs: number; windows?: readonly OutageWindow[] } = { adjusted: 0, removedMs: 0 }): WaveSpeed { + const rows: SpeedRow[] = []; + const activity = firstActivityByGroup(runs); + for (const row of baselineRows(runs)) { + const firstActivity = activity.get(`${row.adapter}|${row.warmth}`) ?? []; + for (const [metric, measured] of [["time_to_first_activity", distribution(firstActivity)], ["ttft", row.ttftAfterRunCreated], + ["agent_boot", row.boot]] as const) { + const budget = budgetFor(environment, row.adapter, metric, row.warmth); + rows.push({ adapter: row.adapter, metric, start: row.warmth, measured, budget, breaches: breachesOf(measured, budget) }); + } + } + const clock = outageFreeClock(outage.windows ?? []); + const samples = recoverySamples(runs, faults).map(sample => sample.latencyMs === null ? sample + : { ...sample, latencyMs: Math.max(0, clock(sample.readyAt + sample.latencyMs) - clock(sample.readyAt)) }); + for (const adapter of [...new Set(samples.map(sample => sample.adapter))].sort()) { + const measured = distribution(samples.flatMap(sample => sample.adapter === adapter && sample.latencyMs !== null ? [sample.latencyMs] : [])); + const budget = budgetFor(environment, adapter, "restart_recovery", "restart"); + rows.push({ adapter, metric: "restart_recovery", start: "restart", measured, budget, breaches: breachesOf(measured, budget) }); + } + const singleRunBreaches = runs.filter(run => run.phase !== "restart" && run.phase !== "verify").filter(run => { + const start = runWarmth(run); + const activity = firstActivityMs(run), boot = runMetrics(run).bootMs; + const activityBudget = budgetFor(environment, run.adapter, "time_to_first_activity", start); + const bootBudget = budgetFor(environment, run.adapter, "agent_boot", start); + return (activity !== null && activityBudget !== null && activity > activityBudget.p95Ms) + || (boot !== null && bootBudget !== null && boot > bootBudget.p95Ms); + }).length; + const breaches = rows.reduce((total, row) => total + row.breaches.length, 0) + singleRunBreaches; + return { environment, rows, recoveryWithoutProviderEvent: samples.filter(sample => sample.latencyMs === null).length, + runBudgetExceeded, singleRunBreaches, breaches, errors, outageAdjustedSpans: outage.adjusted, outageRemovedMs: outage.removedMs }; +} + +async function runEventsWithRetry(api: ChaosApi, runId: string, attempts = 3): Promise { + let lastError: unknown; + for (let attempt = 1; attempt <= attempts; attempt += 1) { + try { return await runEvents(api, runId); } catch (error) { + lastError = error; + if (attempt < attempts) await new Promise(resolve => setTimeout(resolve, 1_000 * attempt)); + } + } + throw lastError; +} + +async function runEvents(api: ChaosApi, runId: string): Promise { + const events: BaselineEvent[] = []; + let afterSeq = 0; + for (;;) { + const page = await api.get(`/api/heartbeat-runs/${runId}/events?afterSeq=${afterSeq}&limit=500`, 15_000); + for (const row of page) events.push({ eventType: String(row.eventType), createdAt: String(row.createdAt), payload: (row.payload ?? null) as Json | null }); + if (page.length < 500) return events; + afterSeq = Number(page[page.length - 1]!.seq); + } +} + +export async function collectWaveSpeed(api: ChaosApi, environment: EnvironmentKind, tasks: readonly WaveSpeedTask[], + faults: readonly FaultReceipt[]): Promise { + const runs: BaselineRun[] = []; + const timing: Array<{ id: string; createdAt: number; finishedAt: number | null }> = []; + const errors: string[] = []; + const windows = outageWindows(faults); + const outage = { adjusted: 0, removedMs: 0, windows }; + for (const task of tasks) { + const seen = new Map(); + for (const sample of task.samples) for (const run of sample.runs ?? []) if (run.createdAt) seen.set(run.id, run); + const ordered = [...seen.values()].sort((a, b) => Date.parse(a.createdAt!) - Date.parse(b.createdAt!)); + for (const [index, run] of ordered.entries()) { + try { + const raw: BaselineRun = { runId: run.id, adapter: task.adapter, environment, phase: index === 0 ? "cold" : "warm", + events: await runEventsWithRetry(api, run.id) }; + const trimmed = withoutOutage(raw, windows); + outage.adjusted += trimmed.adjusted; outage.removedMs += trimmed.removedMs; + runs.push(trimmed.run); + timing.push({ id: run.id, createdAt: Date.parse(run.createdAt!), finishedAt: run.finishedAt ? Date.parse(run.finishedAt) : null }); + } catch (error) { + errors.push(`${run.id}: ${error instanceof Error ? error.message : String(error)}`); + } + } + } + const runBudgetExceeded = runs.reduce((total, run) => + total + run.events.filter(event => event.eventType === "run.performance.budget_exceeded").length, 0); + return gradeWaveSpeed(environment, runs, faultRuns(faults, timing), runBudgetExceeded, errors, outage); +} + +export function speedMarkdown(speed: WaveSpeed): string { + const s = (value: number | null) => value === null ? "—" : `${(value / 1000).toFixed(1)} s`; + const lines = [`## Speed (${speed.environment})`, "", + `Breaches: ${speed.breaches} (aggregate p50/p95 rows plus ${speed.singleRunBreaches} runs over the p95 budget after outage removal; the server logged ${speed.runBudgetExceeded} budget_exceeded events before removal).`, + `Restart samples with no provider event after ready: ${speed.recoveryWithoutProviderEvent}.`, + `Outage time removed from ${speed.outageAdjustedSpans} spans: ${s(speed.outageRemovedMs)} (fault start to server ready; budgets judge the rest).`, "", + "| Adapter | Metric | Start | n | p50 | p95 | Budget p50 / p95 | Breach |", "|---|---|---|---|---|---|---|---|"]; + for (const row of speed.rows) { + lines.push(`| ${row.adapter} | ${row.metric} | ${row.start} | ${row.measured.n} | ${s(row.measured.p50)} | ${s(row.measured.p95)} | ` + + `${row.budget ? `${s(row.budget.p50Ms)} / ${s(row.budget.p95Ms)}` : "no budget"} | ${row.breaches.join(", ") || "—"} |`); + } + if (speed.errors.length) lines.push("", `Speed grade incomplete: ${speed.errors.length} run logs not read; the wave fails.`); + return `${lines.join("\n")}\n`; +} diff --git a/tests/restart-chaos/workloads.ts b/tests/restart-chaos/workloads.ts index a9f85a26fb2..e694a0e2547 100644 --- a/tests/restart-chaos/workloads.ts +++ b/tests/restart-chaos/workloads.ts @@ -14,13 +14,18 @@ export interface WorkloadDefinition { /** Time budget for the task excluding outage time (R1). */ budgetMs: number; title(nonce: string): string; - description(nonce: string): string; + description(nonce: string, agents?: WorkloadAgents): string; followUp?(nonce: string): string; /** Files and commands the observer must collect for the T-oracle. */ - collect: { documents: string[]; files(nonce: string): string[]; commands: Array<{ name: string; argv: string[] }> }; - grade(nonce: string, artifacts: OracleArtifacts): OracleResult; + collect: { documents: string[]; files(nonce: string): string[]; commands: Array<{ name: string; argv: string[] }>; + childDocuments?: string[] }; + grade(nonce: string, artifacts: OracleArtifacts, agents?: WorkloadAgents): OracleResult; } +export interface WorkloadAgents { self: { id: string; name: string }; other: { id: string; name: string } } + +const partMarker = (nonce: string, part: number) => `PART${part}-${nonce}`; + const briefMarker = (nonce: string) => `CHAOS-${nonce}`; const followUpMarker = (nonce: string) => `FOLLOWUP-${nonce}`; const notesFile = (nonce: string) => `NOTES-${nonce}.md`; @@ -66,7 +71,52 @@ const briefDescription = (nonce: string) => [ "Do not create files or other tasks. When the document is saved, mark this task done.", ].join("\n"); +const delegateDescription = (nonce: string, agents?: WorkloadAgents) => { + const self = agents?.self.name ?? "yourself"; + const other = agents?.other.name ?? "yourself"; + return [ + "Coordinate this task by delegating it to two subtasks, then summarize.", + `1. Create subtask "Part 1 ${nonce}" with this task as its parent, assigned to ${other}. Its description:`, + ` "Save this issue's document with key \`part\` (title \"Part\"), in markdown, containing the line \`${partMarker(nonce, 1)}\` exactly on its own line and one sentence on why retries need idempotency keys. Then mark this subtask done."`, + `2. Create subtask "Part 2 ${nonce}" with this task as its parent, assigned to ${self}. Same description, with the line \`${partMarker(nonce, 2)}\` instead.`, + "Do not do the subtasks' work yourself. Wait until both subtasks are done.", + `3. Then save this issue's document with key \`summary\` (title "Summary") containing both lines \`${partMarker(nonce, 1)}\` and \`${partMarker(nonce, 2)}\`, each on its own line.`, + "Then mark this task done.", + ].join("\n"); +}; + export const workloadDefinitions: Record = { + M1: { + id: "M1", + label: "Delegate to two subtasks", + usesRepository: false, + fireWindowMs: [20_000, 120_000], + budgetMs: 20 * 60_000, + title: nonce => `Chaos M1 delegation ${nonce}`, + description: delegateDescription, + collect: { documents: ["summary"], files: () => [], commands: [], childDocuments: ["part"] }, + grade(nonce, artifacts, agents) { + const children = artifacts.children ?? []; + const holders = (n: number) => children.filter(child => (child.documents.part ?? "").includes(partMarker(nonce, n))); + const part1 = holders(1), part2 = holders(2); + const distinct = part1.length === 1 && part2.length === 1 && part1[0]!.id !== part2[0]!.id; + const summary = artifacts.documents.summary ?? ""; + return result([ + statusCheck(artifacts), + { name: "two_agents", ok: Boolean(agents) && agents!.self.id !== agents!.other.id, + detail: agents ? `self=${agents.self.id} other=${agents.other.id}` : "no agents" }, + { name: "two_subtasks", ok: children.length === 2, detail: `children=${children.length}` }, + { name: "subtasks_done", ok: children.length > 0 && children.every(child => child.status === "done"), + detail: children.map(child => child.status).join(",") }, + { name: "parts_in_distinct_subtasks", ok: distinct, detail: `part1=${part1.length} part2=${part2.length}` }, + { name: "part1_delegated", ok: distinct && Boolean(agents) && part1[0]!.assigneeAgentId === agents!.other.id, + detail: `assignee=${part1[0]?.assigneeAgentId ?? "none"}` }, + { name: "part2_kept", ok: distinct && Boolean(agents) && part2[0]!.assigneeAgentId === agents!.self.id, + detail: `assignee=${part2[0]?.assigneeAgentId ?? "none"}` }, + { name: "summary_has_both_parts", ok: summary.includes(partMarker(nonce, 1)) && summary.includes(partMarker(nonce, 2)) }, + ]); + }, + }, A1: { id: "A1", label: "Brief to issue document",