diff --git a/tests/restart-chaos/ttft-baseline-stats.ts b/tests/restart-chaos/ttft-baseline-stats.ts index 3e5bda1f661..9b1cdfd690f 100644 --- a/tests/restart-chaos/ttft-baseline-stats.ts +++ b/tests/restart-chaos/ttft-baseline-stats.ts @@ -18,6 +18,7 @@ export interface BaselineRun { /** Driver intent: a new task, a follow-up comment on the same task, or a run crossed by a restart. */ phase: "cold" | "warm" | "restart" | "midturn" | "verify"; events: BaselineEvent[]; + workspaceCreated?: boolean; } export interface BaselineFault { diff --git a/tests/restart-chaos/wave-speed.test.ts b/tests/restart-chaos/wave-speed.test.ts index e14a56440ec..293ba5d1756 100644 --- a/tests/restart-chaos/wave-speed.test.ts +++ b/tests/restart-chaos/wave-speed.test.ts @@ -2,7 +2,7 @@ import { describe, expect, it } from "vitest"; import type { BaselineRun } from "./ttft-baseline-stats.js"; import type { ChaosApi } from "./api.js"; import type { FaultReceipt } from "./types.js"; -import { breachesOf, budgetFor, collectWaveSpeed, faultBeforeFirstActivity, faultRuns, firstActivityMs, gradeWaveSpeed, outageWindows, speedMarkdown, traceStartMs, withoutOutage } from "./wave-speed.js"; +import { bootWithoutProvisionMs, breachesOf, budgetFor, collectWaveSpeed, faultBeforeFirstActivity, 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, extra: Record = {}) => @@ -71,9 +71,10 @@ describe("wave speed grade", () => { it("treats a run that a fault hit before its first activity as a restart, not a cold start", async () => { const hit = traced(1_000_000, [["heartbeat.queue", 0, 100], ["task.time_to_first_agent_activity", 0, 30_000]]); - expect(faultBeforeFirstActivity(hit, [1_010_000], 1_000_000)).toBe(true); - expect(faultBeforeFirstActivity(hit, [1_040_000, 990_000], 1_000_000)).toBe(false); - expect(faultBeforeFirstActivity(hit, [1_005_000], 1_008_000)).toBe(false); + expect(faultBeforeFirstActivity(hit, [{ startedAt: 1_010_000, readyAt: 1_012_000 }], 1_000_000)).toBe(true); + expect(faultBeforeFirstActivity(hit, [{ startedAt: 1_040_000, readyAt: 1_041_000 }, { startedAt: 980_000, readyAt: 990_000 }], 1_000_000)).toBe(false); + expect(faultBeforeFirstActivity(hit, [{ startedAt: 1_005_000, readyAt: 1_006_000 }], 1_008_000)).toBe(false); + expect(faultBeforeFirstActivity(hit, [{ startedAt: 996_000, readyAt: 1_025_000 }], 1_000_000)).toBe(true); const clean = traced(2_000_000, [["heartbeat.queue", 0, 100], ["task.time_to_first_agent_activity", 0, 4_000]]); const events: Record = { hit: [...hit.events, { eventType: "turn.started", createdAt: new Date(1_013_000).toISOString() }], clean: clean.events }; @@ -88,6 +89,54 @@ describe("wave speed grade", () => { expect(speed.rows.find(row => row.metric === "restart_recovery")).toMatchObject({ adapter: "claude", measured: { n: 1 } }); }); + it("reports first-time workspace provisioning on its own row and keeps it out of agent boot", () => { + const provisioned: BaselineRun = { runId: "p", adapter: "claude", environment: "local", phase: "cold", workspaceCreated: true, + events: [span("heartbeat.queue", 0, 100), span("heartbeat.prepare.workspace_provision", 200, 22_000), + span("runner.turn.submit", 22_500, 3_000), span("task.time_to_first_agent_activity", 0, 27_000)] }; + expect(bootWithoutProvisionMs(provisioned)).toBe(25_400 - 22_000); + expect(bootWithoutProvisionMs({ ...provisioned, workspaceCreated: false })).toBe(25_400); + const speed = gradeWaveSpeed("local", [provisioned], [], 0); + expect(speed.rows.find(row => row.metric === "agent_boot")).toMatchObject({ measured: { p50: 3_400 }, breaches: [] }); + expect(speed.rows.find(row => row.metric === "workspace_provision")).toMatchObject({ budget: null, measured: { n: 1, p50: 22_000 } }); + expect(gradeWaveSpeed("local", [run("q", "claude", 4_000)], [], 0).rows.some(row => row.metric === "workspace_provision")).toBe(false); + }); + + it("grades a retry successor as restart and reads provisioning creation from workspace operations", async () => { + const created = traced(3_000_000, [["heartbeat.queue", 0, 100], ["heartbeat.prepare.workspace_provision", 200, 20_000], + ["runner.turn.submit", 20_300, 2_000], ["task.time_to_first_agent_activity", 0, 23_000]]); + const successor = traced(4_000_000, [["heartbeat.queue", 0, 30_000], ["task.time_to_first_agent_activity", 0, 35_000]]); + const overload = traced(6_000_000, [["heartbeat.queue", 0, 100], ["task.time_to_first_agent_activity", 0, 5_000]]); + const interrupted = traced(3_500_000, [["heartbeat.queue", 0, 100]]); + const events: Record = { created: created.events, successor: successor.events, overload: overload.events, interrupted: interrupted.events }; + const idOf = (route: string) => route.split("/")[3]!; + const api = { get: async (route: string) => route.endsWith("/workspace-operations") + ? [{ phase: "workspace_provision", metadata: { created: idOf(route) === "created" } }] + : events[idOf(route)]!.map((event: object, seq: number) => ({ ...event, seq })) } as unknown as ChaosApi; + const sampleRun = (id: string, createdAt: number, finishedAt: number, retryOfRunId: string | null) => ({ id, status: "succeeded", retryOfRunId, + createdAt: new Date(createdAt).toISOString(), finishedAt: new Date(finishedAt).toISOString() }); + const fault = { kind: "K1", startedAt: 3_550_000, readyAt: 3_560_000 } as FaultReceipt; + const speed = await collectWaveSpeed(api, "local", [ + { issueId: "i1", adapter: "claude", samples: [{ runs: [sampleRun("created", 3_000_000, 3_060_000, null), sampleRun("overload", 6_000_000, 6_060_000, "provider-overloaded")] }] }, + { issueId: "i2", adapter: "claude", samples: [{ runs: [sampleRun("interrupted", 3_500_000, 3_555_000, null), sampleRun("successor", 4_000_000, 4_060_000, "interrupted")] }] }, + ] as never, [fault]); + expect(speed.rows.find(row => row.metric === "time_to_first_activity" && row.start === "cold")).toMatchObject({ measured: { n: 1, p50: 23_000 } }); + expect(speed.rows.find(row => row.metric === "time_to_first_activity" && row.start === "warm")).toMatchObject({ measured: { n: 1, p50: 5_000 } }); + expect(speed.recoveryWithoutProviderEvent).toBe(1); + expect(speed.rows.find(row => row.metric === "workspace_provision")).toMatchObject({ measured: { n: 1, p50: 20_000 } }); + expect(speed.rows.find(row => row.metric === "agent_boot" && row.start === "cold")).toMatchObject({ measured: { p50: 2_200 } }); + }); + + it("records an unread workspace-operations list as a speed collection error", async () => { + const run = traced(7_000_000, [["heartbeat.queue", 0, 100], ["task.time_to_first_agent_activity", 0, 4_000]]); + const api = { get: async (route: string) => { + if (route.endsWith("/workspace-operations")) throw new Error("timeout"); + return run.events.map((event, seq) => ({ ...event, seq })); + } } as unknown as ChaosApi; + const speed = await collectWaveSpeed(api, "local", [{ issueId: "i", adapter: "claude", samples: [{ runs: [{ id: "r", status: "succeeded", + retryOfRunId: null, createdAt: new Date(7_000_000).toISOString(), finishedAt: new Date(7_060_000).toISOString() }] }] }] as never, []); + expect(speed.errors.join(" ")).toMatch(/workspace operations unread: timeout/); + }, 15_000); + 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]]); @@ -114,9 +163,10 @@ describe("wave speed grade", () => { expect(runMetrics(withoutOutage(run, windows).run).bootMs).toBe(1_000); }); - it("attributes each fault to the runs it crossed", () => { + it("attributes each fault to the runs it crossed and the runs created during its outage", () => { 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"] }]); + { id: "during", createdAt: 1_500, finishedAt: null }, { id: "after", createdAt: 2_500, finishedAt: null }])) + .toEqual([{ kind: "K2", readyAt: 2_000, runIds: ["live", "during"] }]); }); }); diff --git a/tests/restart-chaos/wave-speed.ts b/tests/restart-chaos/wave-speed.ts index 2c0cb82c831..bbadc7d7128 100644 --- a/tests/restart-chaos/wave-speed.ts +++ b/tests/restart-chaos/wave-speed.ts @@ -11,7 +11,7 @@ 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"; + adapter: string; metric: "time_to_first_activity" | "ttft" | "agent_boot" | "workspace_provision" | "restart_recovery"; start: "cold" | "warm" | "restart"; measured: Distribution; budget: SpeedBudget | null; breaches: Array<"p50" | "p95">; } @@ -33,7 +33,7 @@ export function budgetFor(environment: EnvironmentKind, adapter: string, metric: start: SpeedRow["start"]): SpeedBudget | null { const key = BUDGET_ADAPTER[adapter as AdapterId]; if (environment !== "local" || !key) return null; - if (metric === "ttft") return null; + if (metric === "ttft" || metric === "workspace_provision") 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 @@ -51,7 +51,7 @@ export function breachesOf(measured: Distribution, budget: SpeedBudget | null): 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)) + runIds: runs.filter(run => run.createdAt <= fault.readyAt! && (run.finishedAt === null || run.finishedAt >= fault.startedAt)) .map(run => run.id) })); } @@ -100,14 +100,14 @@ export function withoutOutage(run: BaselineRun, windows: readonly OutageWindow[] return { run: { ...run, events }, adjusted, removedMs, anchored: true }; } -export function faultBeforeFirstActivity(run: BaselineRun, faultStarts: readonly number[], runCreatedAt: number): boolean { +export function faultBeforeFirstActivity(run: BaselineRun, windows: readonly OutageWindow[], runCreatedAt: number): boolean { const traceStart = traceStartMs(run); if (traceStart === null) return false; const spans = runSpans(run); const first = spans.find(span => span.name === "task.time_to_first_agent_activity") ?? spans.find(span => span.name === "task.time_to_first_token"); const end = first ? traceStart + first.startOffsetMs + first.durationMs : Number.POSITIVE_INFINITY; - return faultStarts.some(at => at >= runCreatedAt && at <= end); + return windows.some(window => window.startedAt <= end && window.readyAt >= runCreatedAt); } export function firstActivityMs(run: BaselineRun): number | null { @@ -115,6 +115,42 @@ export function firstActivityMs(run: BaselineRun): number | null { return metrics.firstActivityAfterRunCreatedMs ?? metrics.ttftAfterRunCreatedMs; } +export function provisionMs(run: BaselineRun): number | null { + if (!run.workspaceCreated) return null; + return runSpans(run).find(span => span.name === "heartbeat.prepare.workspace_provision")?.durationMs ?? null; +} + +async function workspaceCreated(api: ChaosApi, runId: string, attempts = 3): Promise { + let lastError: unknown; + for (let attempt = 1; attempt <= attempts; attempt += 1) { + try { + const rows = await api.get(`/api/heartbeat-runs/${runId}/workspace-operations`, 15_000); + return (rows ?? []).some(row => row.phase === "workspace_provision" && (row.metadata as Json | null)?.created === true); + } catch (error) { + lastError = error; + if (attempt < attempts) await new Promise(resolve => setTimeout(resolve, 1_000 * attempt)); + } + } + throw new Error(`workspace operations unread: ${lastError instanceof Error ? lastError.message : String(lastError)}`); +} + +export function bootWithoutProvisionMs(run: BaselineRun): number | null { + const boot = runMetrics(run).bootMs; + return boot === null ? null : Math.max(0, boot - (provisionMs(run) ?? 0)); +} + +function byGroup(runs: readonly BaselineRun[], value: (run: BaselineRun) => number | null): Map { + const groups = new Map(); + for (const run of runs) { + if (run.phase === "restart" || run.phase === "verify") continue; + const measured = value(run); + if (measured === null) continue; + const key = `${run.adapter}|${runWarmth(run)}`; + groups.set(key, [...(groups.get(key) ?? []), measured]); + } + return groups; +} + function firstActivityByGroup(runs: readonly BaselineRun[]): Map { const groups = new Map(); for (const run of runs) { @@ -132,10 +168,13 @@ export function gradeWaveSpeed(environment: EnvironmentKind, runs: readonly Base outage: { adjusted: number; removedMs: number; windows?: readonly OutageWindow[] } = { adjusted: 0, removedMs: 0 }): WaveSpeed { const rows: SpeedRow[] = []; const activity = firstActivityByGroup(runs); + const boots = byGroup(runs, bootWithoutProvisionMs), provisions = byGroup(runs, provisionMs); for (const row of baselineRows(runs)) { - const firstActivity = activity.get(`${row.adapter}|${row.warmth}`) ?? []; + const key = `${row.adapter}|${row.warmth}`; + const firstActivity = activity.get(key) ?? []; for (const [metric, measured] of [["time_to_first_activity", distribution(firstActivity)], ["ttft", row.ttftAfterRunCreated], - ["agent_boot", row.boot]] as const) { + ["agent_boot", distribution(boots.get(key) ?? [])], ["workspace_provision", distribution(provisions.get(key) ?? [])]] as const) { + if (metric === "workspace_provision" && !measured.n) continue; const budget = budgetFor(environment, row.adapter, metric, row.warmth); rows.push({ adapter: row.adapter, metric, start: row.warmth, measured, budget, breaches: breachesOf(measured, budget) }); } @@ -150,7 +189,7 @@ export function gradeWaveSpeed(environment: EnvironmentKind, runs: readonly Base } 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 activity = firstActivityMs(run), boot = bootWithoutProvisionMs(run); 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) @@ -190,16 +229,25 @@ export async function collectWaveSpeed(api: ChaosApi, environment: EnvironmentKi const errors: string[] = []; const windows = outageWindows(faults); const outage = { adjusted: 0, removedMs: 0, windows }; - for (const task of tasks) { + const perTask = tasks.map(task => { 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!)); + return { task, ordered: [...seen.values()].sort((a, b) => Date.parse(a.createdAt!) - Date.parse(b.createdAt!)) }; + }); + const faultHit = new Set(faultRuns(faults, perTask.flatMap(({ ordered }) => ordered.map(run => ({ id: run.id, + createdAt: Date.parse(run.createdAt!), finishedAt: run.finishedAt ? Date.parse(run.finishedAt) : null })))).flatMap(fault => fault.runIds)); + for (const { task, ordered } of perTask) { 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) }; - if (faultBeforeFirstActivity(raw, faults.filter(fault => fault.readyAt !== null).map(fault => fault.startedAt), - Date.parse(run.createdAt!))) raw.phase = "restart"; + try { + raw.workspaceCreated = await workspaceCreated(api, run.id); + } catch (error) { + errors.push(`${run.id}: ${error instanceof Error ? error.message : String(error)}`); + } + if (run.retryOfRunId && faultHit.has(run.retryOfRunId)) raw.phase = "restart"; + if (faultBeforeFirstActivity(raw, windows, Date.parse(run.createdAt!))) raw.phase = "restart"; const trimmed = withoutOutage(raw, windows); outage.adjusted += trimmed.adjusted; outage.removedMs += trimmed.removedMs; runs.push(trimmed.run);