diff --git a/doc/plans/2026-11-07-durable-ownership-and-reconciliation.md b/doc/plans/2026-11-07-durable-ownership-and-reconciliation.md index f20302066b8..46b73a402a2 100644 --- a/doc/plans/2026-11-07-durable-ownership-and-reconciliation.md +++ b/doc/plans/2026-11-07-durable-ownership-and-reconciliation.md @@ -99,6 +99,12 @@ Restart recovery latency, from `ready` to the recovered run's next provider even | codex | 6.5 / 8.0 | | opencode | 25.5 / 38.5 | +Since 2026-11-09 this full latency is reported, not judged: the provider's saved-session load (`provider.session.load`) grows with session size and with how many sessions resume at once (Codex at concurrency 6: 3.0–6.3 s of a 5.5–8.5 s recovery). The judged row is Paperclip's own share, `restart_recovery_control_plane`: from `ready` to the start of the provider's session load, or to the next provider event when nothing reloads. It covers claim, prepare, runtime stage and runner launch, which measured 1.2–2.0 s per run with six Codex runs resuming at once (wave `w-20261109T045320-s9807`). + +| Adapter | Paperclip share budget (s, p50 / p95) | +|---|---| +| claude, codex, opencode | 2.5 / 4.0 | + These budgets measure the product default `lifecycleMode: per_turn`. Phase 1's TTFT work (runner and provider-session reuse, baseline finding 1) should lower them. Lower budgets go in the same table. ### Phase 1 report-only data: local chaos waves (2026-11-07) diff --git a/doc/run-log-events.md b/doc/run-log-events.md index e6f63526d73..fa2a6c92db9 100644 --- a/doc/run-log-events.md +++ b/doc/run-log-events.md @@ -64,6 +64,11 @@ instance run log and add no first-party Telemetry or OpenTelemetry export. ## Provider-pack preparation timing +Every `run.performance.span` payload also carries `wallStartedAtMs`, the span's own +start time in epoch milliseconds (since 2026-11-09). Span rows can be written late +through deferred diagnostics, so readers take a span's start from this field (or +from the trace start plus `startOffsetMs`), never from the row's write time. + `run.performance.span` also records `task.time_to_first_token` once per native dispatch at the first newly committed deliverable text: assistant or reasoning `item.delta`/`item.completed`, a redaction tail released at completion, or an @@ -97,8 +102,15 @@ budgets come from the Phase 0 local baseline and are listed in the first text delta, tool-call start or reasoning item; named `ttft` and judged on `task.time_to_first_token` before 2026-11-08), `agent_boot` (end of `heartbeat.queue` to the end of `runner.turn.submit` or `runner.session.resume`) -or `restart_recovery` (startup recovery `ready` to the recovered run's first -provider event). The payload names `adapter` (`codex`, `claude` or `opencode`), +or `restart_recovery_control_plane` (startup recovery `ready` to the start of the +provider's saved-session load, `provider.session.load`, or to the recovered run's +first provider event when nothing reloads). That is Paperclip's own share of a +recovery: claim, prepare, runtime stage and runner launch. The full recovery +(`ready` to the first provider event) and the provider's session load grow with +session size and with how many sessions resume at once, so they are reported in +the payload as `restartTotalMs` and `providerLoadMs` and are not judged; events +written before 2026-11-09 judged the full latency as `restart_recovery`. +The payload names `adapter` (`codex`, `claude` or `opencode`), `environment` (`local`), `startKind` (`cold`, `warm` or `unknown`; an unknown start uses the looser budget), `measuredMs`, `budgetP50Ms`, `budgetP95Ms`, `percentile` (`p95`), `baseline`, and, for restart recovery, `disposition`. @@ -106,7 +118,7 @@ For `time_to_first_activity`, `measuredMs` counts from run creation: when the in run to exist (for example a comment posted while the previous turn was still running), that wait is excluded from the judgment and reported as `inputWaitMs`. Each metric is checked at most once per dispatch. A restart-recovery dispatch is -judged only on `restart_recovery`, because its first activity and boot span the outage. Time to first text (`task.time_to_first_token`) is reported without a budget. Remote runs and other adapters have +judged only on `restart_recovery_control_plane`, because its first activity and boot span the outage. Time to first text (`task.time_to_first_token`) is reported without a budget. Remote runs and other adapters have no budget yet. The event is written in detached diagnostic work after the measurement, so it adds no wait before `turn.start`. It is a run-log row only and never goes to Telemetry. diff --git a/server/src/services/native-runtime/native-run-trace.ts b/server/src/services/native-runtime/native-run-trace.ts index 137ca901612..9224fd9dfee 100644 --- a/server/src/services/native-runtime/native-run-trace.ts +++ b/server/src/services/native-runtime/native-run-trace.ts @@ -262,11 +262,16 @@ function safeAttributes( * seams inherit that context. The persisted run-event copy retains the same * semantic parent name for developer installs without an OTLP collector. */ +/** The runner span for the provider loading its saved session on resume. */ +const PROVIDER_SESSION_LOAD_SPAN = "provider.session.load"; + export function createNativeRunTrace(input: { runId: string; startedAtMs?: number; onEvent?: NativeRunTraceSink; traceContext?: StartupTraceContextHandle; + /** Called once when the trace finishes. */ + onFinish?: () => void; /** Speed-budget subject (invariant 8). Absent or remote means no check. */ budget?: { adapter: SpeedBudgetAdapter | null; @@ -279,6 +284,8 @@ export function createNativeRunTrace(input: { const tracing = input.traceContext ?? getStartupTraceContext("paperclip.native-runner"); const traceStartedAtMs = finiteMs(input.startedAtMs ?? Date.now()); + /** Start times of the provider's saved-session loads (restart recovery split). */ + const providerLoadStartsMs: number[] = []; const runIdHash = hashedId(input.runId); let rootSpan: SpanHandle; try { @@ -429,6 +436,9 @@ export function createNativeRunTrace(input: { durationMs, outcome, ...attributes, + // The span's own start time. The row can be written late (deferred + // diagnostics), so readers must not infer the start from its write time. + wallStartedAtMs: startedAtMs, }, }); } catch { @@ -447,6 +457,11 @@ export function createNativeRunTrace(input: { } = {}, ): NativeRunSpanScope => { const startedAtMs = finiteMs(options.startedAtMs ?? Date.now()); + if (name === PROVIDER_SESSION_LOAD_SPAN) { + providerLoadStartsMs.push(startedAtMs); + // The measurement can arrive before its provider span is recorded; judge now. + if (pendingRecovery) judgeRestartRecovery(); + } const attributes = safeAttributes(options.attributes); const parent = parentFor(options.parentName); let span: SpanHandle; @@ -635,8 +650,13 @@ export function createNativeRunTrace(input: { const finish = async (outcome: NativeRunSpanOutcome): Promise => { if (ended) return; + try { input.onFinish?.(); } catch { /* Diagnostics cannot change runner control flow. */ } // Settled diagnostics can defer more (a budget breach after a deferred - // measurement), so drain until none remain. + // measurement), so drain until none remain. Judge restart recovery after + // the first drain, so a deferred provider span is counted, then drain its + // breach report. + while (pendingDiagnostics.size > 0) await Promise.all([...pendingDiagnostics]); + judgeRestartRecovery(); while (pendingDiagnostics.size > 0) await Promise.all([...pendingDiagnostics]); if (ended) return; ended = true; @@ -666,8 +686,28 @@ export function createNativeRunTrace(input: { }; /** Restart recovery latency: control plane ready to the recovered run's - * next provider event, checked once per run. */ + * next provider event, checked once per run. Only Paperclip's share is + * judged: ready to the start of the provider's session load (or to the + * event when nothing reloads). It is judged as soon as both are known: at + * the check when the load span is already recorded or nothing reloads (a + * reattached runner, whose dispositions launch nothing), when the load span is recorded, and otherwise when + * the run finishes. A run that detaches for another restart therefore keeps + * its warning unless the detach lands between the event and a deferred span. */ let recoveryChecked = false; + let pendingRecovery: RestartRecoveryMeasurement | null = null; + const judgeRestartRecovery = (): void => { + const measurement = pendingRecovery; + pendingRecovery = null; + if (!input.budget || !measurement || measurement.readyAtMs === null) return; + const readyAtMs = measurement.readyAtMs; + const loads = providerLoadStartsMs.filter(at => at >= readyAtMs && at <= measurement.committedAtMs); + const loadStartedAtMs = loads.length ? Math.min(...loads) : null; + const totalMs = measurement.committedAtMs - readyAtMs; + reportBudgetBreach(evaluateSpeedBudget({ adapter: input.budget.adapter, environment: input.budget.environment, + metric: "restart_recovery_control_plane", start: null, measuredMs: (loadStartedAtMs ?? measurement.committedAtMs) - readyAtMs, + disposition: measurement.disposition, restartTotalMs: totalMs, + providerLoadMs: loadStartedAtMs === null ? 0 : measurement.committedAtMs - loadStartedAtMs })); + }; const checkRestartRecovery = (measurement: RestartRecoveryMeasurement): void => { if (!input.budget || recoveryChecked || measurement.readyAtMs === null) return; // An event emitted before ready (buffered by a surviving runner and @@ -676,15 +716,20 @@ export function createNativeRunTrace(input: { if (measurement.emittedAtMs !== null && measurement.emittedAtMs < measurement.readyAtMs) return; if (measurement.committedAtMs < measurement.readyAtMs) return; recoveryChecked = true; - reportBudgetBreach(evaluateSpeedBudget({ adapter: input.budget.adapter, environment: input.budget.environment, - metric: "restart_recovery", start: null, measuredMs: measurement.committedAtMs - measurement.readyAtMs, - disposition: measurement.disposition })); + pendingRecovery = measurement; + const readyAtMs = measurement.readyAtMs; + // Only a launched runner reloads a provider session (native-restart-recovery). + if (!["resume_dead_runner", "bootstrap_incomplete"].includes(measurement.disposition) + || providerLoadStartsMs.some(at => at >= readyAtMs && at <= measurement.committedAtMs)) judgeRestartRecovery(); }; return { traceStartedAtMs, setBudgetStart, checkRestartRecovery, + /** Judge a pending restart-recovery check now (the controller is detaching + * this run for another restart and will not finish the trace). */ + flushRestartRecovery: () => judgeRestartRecovery(), start, annotate, activate, diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 3e46964c3f9..8512171085d 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -380,6 +380,9 @@ export class NativeControllerDetachedForRestartError extends Error { const activeNativeSessions = new Map(); // Shutdown can race provider startup before onSession publishes its handle. const nativeRunsDetachingForRestart = new Map(); +/** Flushes a run's pending restart-recovery budget check when the controller + * detaches it for another restart (its trace never finishes). */ +const nativeRunRecoveryFlushers = new Map void>(); // A restart must not cut authority rotation between archiving the previous // runner state and authenticating its replacement. Keep the startup owner @@ -465,6 +468,7 @@ export async function detachNativeSessionsForRestart( continue; } nativeRunsDetachingForRestart.set(runId, Symbol("native controller detach")); + try { nativeRunRecoveryFlushers.get(runId)?.(); } catch { /* Diagnostics cannot change detach. */ } let active = activeNativeSessions.get(runId); if (!active) { // Only a startup registered before this fence may complete read-only @@ -9592,6 +9596,24 @@ async function executePaperclipNativeSessionWithinScope( prepareAfterClaim?: () => Promise[0]>, cleanupOnly = false, ownDiagnosticLease?: (lease: NativeDiagnosticLease) => void, +): Promise { + // Drop this attempt's restart-recovery flush however it ends (a failure that + // never finishes its trace included), without touching a newer attempt's. + const runId = input.execution.binding.runId; + const priorFlush = nativeRunRecoveryFlushers.get(runId); + try { + return await executePaperclipNativeSessionAttempt(input, prepareAfterClaim, cleanupOnly, ownDiagnosticLease); + } finally { + const current = nativeRunRecoveryFlushers.get(runId); + if (current !== undefined && current !== priorFlush) nativeRunRecoveryFlushers.delete(runId); + } +} + +async function executePaperclipNativeSessionAttempt( + input: Parameters[0], + prepareAfterClaim?: () => Promise[0]>, + cleanupOnly = false, + ownDiagnosticLease?: (lease: NativeDiagnosticLease) => void, ): Promise { if (input.execution.provider.kind === "acpx" && input.execution.provider.agent === "antigravity") { const candidate = resolveAntigravityCandidateAdmission({ @@ -9649,16 +9671,25 @@ async function executePaperclipNativeSessionWithinScope( preparationSpans, Date.now(), ); + const traceRunId = input.execution.binding.runId; + const flushRecovery = () => { + if (nativeRunRecoveryFlushers.get(traceRunId) === flushRecovery) nativeRunRecoveryFlushers.delete(traceRunId); + trace.flushRestartRecovery(); + }; const trace = createNativeRunTrace({ - runId: input.execution.binding.runId, + runId: traceRunId, startedAtMs: preparationStarts.runStartedAtMs, onEvent: input.onEvent, + onFinish: () => { + if (nativeRunRecoveryFlushers.get(traceRunId) === flushRecovery) nativeRunRecoveryFlushers.delete(traceRunId); + }, budget: { adapter: speedBudgetAdapter(input.execution.provider), environment: !input.runnerExecutionTarget || input.runnerExecutionTarget.kind === "local" ? "local" : "remote", restartRecovery: Boolean(input.restartRecovery), }, }); + nativeRunRecoveryFlushers.set(traceRunId, flushRecovery); await emitStartupReuse(input.onEvent, "environment", environmentStartupReuse(input.runnerExecutionTarget)); if (!input.runnerExecutionTarget || input.runnerExecutionTarget.kind === "local") { await emitStartupReuse(input.onEvent, "workspace", { state: "not_applicable", reason: "not_applicable" }); diff --git a/server/src/services/native-runtime/native-speed-budgets.test.ts b/server/src/services/native-runtime/native-speed-budgets.test.ts index 1b54906f324..523d377fd73 100644 --- a/server/src/services/native-runtime/native-speed-budgets.test.ts +++ b/server/src/services/native-runtime/native-speed-budgets.test.ts @@ -134,12 +134,12 @@ describe("speed budget checks in the native run trace", () => { it("checks restart recovery latency once, from ready to the first provider event", async () => { const { trace, breaches } = traced({ adapter: "opencode", environment: "local" }); - const late = LOCAL_SPEED_BUDGETS.opencode.restart_recovery.p95Ms + 1_000; + const late = LOCAL_SPEED_BUDGETS.opencode.restart_recovery_control_plane.p95Ms + 1_000; trace.checkRestartRecovery({ readyAtMs: 10_000, emittedAtMs: 10_000 + late, committedAtMs: 10_000 + late, disposition: "resume_dead_runner" }); trace.checkRestartRecovery({ readyAtMs: 10_000, emittedAtMs: 10_000 + late * 2, committedAtMs: 10_000 + late * 2, disposition: "resume_dead_runner" }); await trace.finish("ok"); - expect(breaches()).toEqual([expect.objectContaining({ metric: "restart_recovery", measuredMs: late, + expect(breaches()).toEqual([expect.objectContaining({ metric: "restart_recovery_control_plane", measuredMs: late, disposition: "resume_dead_runner", startKind: "unknown" })]); }); @@ -148,22 +148,79 @@ describe("speed budget checks in the native run trace", () => { await trace.record({ name: "heartbeat.queue", startedAtMs: 0, endedAtMs: 1 }); await trace.record({ name: "runner.session.resume", startedAtMs: 1, endedAtMs: 120_000 }); await trace.record({ name: "task.time_to_first_agent_activity", startedAtMs: 0, endedAtMs: 120_000 }); - const at = LOCAL_SPEED_BUDGETS.codex.restart_recovery.p95Ms + 1; + const at = LOCAL_SPEED_BUDGETS.codex.restart_recovery_control_plane.p95Ms + 1; trace.checkRestartRecovery({ readyAtMs: 0, emittedAtMs: at, committedAtMs: at, disposition: "resume_dead_runner" }); await trace.finish("ok"); - expect(breaches().map(breach => breach.metric)).toEqual(["restart_recovery"]); + expect(breaches().map(breach => breach.metric)).toEqual(["restart_recovery_control_plane"]); + }); + + it("judges only Paperclip's share of a recovery, not the provider's session load", async () => { + const { trace, breaches } = traced({ adapter: "codex", environment: "local", restartRecovery: true }); + const total = LOCAL_SPEED_BUDGETS.codex.restart_recovery.p95Ms + 3_000; + // Ready → claim, prepare and launch in 1.5 s; the provider reloads its long session. + await trace.record({ name: "provider.session.load", startedAtMs: 1_500, endedAtMs: total - 500 }); + trace.checkRestartRecovery({ readyAtMs: 0, emittedAtMs: total, committedAtMs: total, disposition: "resume_dead_runner" }); + await trace.finish("ok"); + expect(breaches()).toEqual([]); + }); + + it("reports a slow control-plane share with the full latency and the provider's share", async () => { + const { trace, breaches } = traced({ adapter: "codex", environment: "local", restartRecovery: true }); + const control = LOCAL_SPEED_BUDGETS.codex.restart_recovery_control_plane.p95Ms + 2_000; + trace.checkRestartRecovery({ readyAtMs: 0, emittedAtMs: control + 3_000, committedAtMs: control + 3_000, disposition: "resume_dead_runner" }); + // The provider span is committed after the provider event that ended the recovery. + await trace.record({ name: "provider.session.load", startedAtMs: control, endedAtMs: control + 2_800 }); + await trace.finish("ok"); + expect(breaches()).toEqual([expect.objectContaining({ metric: "restart_recovery_control_plane", measuredMs: control, + restartTotalMs: control + 3_000, providerLoadMs: 3_000 })]); + }); + + it("flushes a pending recovery check when the controller detaches the run", async () => { + const { trace, breaches } = traced({ adapter: "codex", environment: "local", restartRecovery: true }); + const late = LOCAL_SPEED_BUDGETS.codex.restart_recovery_control_plane.p95Ms + 1_000; + // A launched runner whose load span has not been written yet. + trace.checkRestartRecovery({ readyAtMs: 0, emittedAtMs: late, committedAtMs: late, disposition: "resume_dead_runner" }); + await new Promise(resolve => setTimeout(resolve, 0)); + expect(breaches()).toEqual([]); + trace.flushRestartRecovery(); + await new Promise(resolve => setTimeout(resolve, 0)); + expect(breaches()).toEqual([expect.objectContaining({ metric: "restart_recovery_control_plane", measuredMs: late })]); + }); + + it("logs each span's own start time in its run-log row", async () => { + const events: AdapterRuntimeEvent[] = []; + const trace = createNativeRunTrace({ runId: "run-1", startedAtMs: 1_000, onEvent: async event => { events.push(event); } }); + await trace.record({ name: "provider.session.load", startedAtMs: 4_000, endedAtMs: 9_000 }); + await trace.finish("ok"); + expect(events.find(event => (event.payload as Record)?.span === "provider.session.load")?.payload) + .toMatchObject({ startOffsetMs: 3_000, durationMs: 5_000, wallStartedAtMs: 4_000 }); + }); + + it("judges a recovery before the run finishes, so a later detach for restart keeps the warning", async () => { + const control = LOCAL_SPEED_BUDGETS.codex.restart_recovery_control_plane.p95Ms + 1_000; + // The load span arrives after the provider event: judged when it is recorded. + const resumed = traced({ adapter: "codex", environment: "local", restartRecovery: true }); + resumed.trace.checkRestartRecovery({ readyAtMs: 0, emittedAtMs: control + 2_000, committedAtMs: control + 2_000, disposition: "resume_dead_runner" }); + await resumed.trace.record({ name: "provider.session.load", startedAtMs: control, endedAtMs: control + 1_900 }); + // A reattached runner reloads nothing: judged at the check. + const reattached = traced({ adapter: "codex", environment: "local", restartRecovery: true }); + reattached.trace.checkRestartRecovery({ readyAtMs: 0, emittedAtMs: control, committedAtMs: control, disposition: "reattach_existing_runner" }); + // Neither trace finishes (the controller detached for another restart). + await new Promise(resolve => setTimeout(resolve, 0)); + expect(resumed.breaches()).toEqual([expect.objectContaining({ metric: "restart_recovery_control_plane", measuredMs: control })]); + expect(reattached.breaches()).toEqual([expect.objectContaining({ metric: "restart_recovery_control_plane", measuredMs: control, providerLoadMs: 0 })]); }); it("does not spend the restart-recovery check on an event emitted before ready", async () => { const { trace, breaches } = traced({ adapter: "codex", environment: "local", restartRecovery: true }); - const late = LOCAL_SPEED_BUDGETS.codex.restart_recovery.p95Ms + 1_000; + const late = LOCAL_SPEED_BUDGETS.codex.restart_recovery_control_plane.p95Ms + 1_000; // Buffered during the outage, committed after ready. trace.checkRestartRecovery(restartRecoveryMeasurementFor({ event: { eventType: "item.delta", emittedAt: new Date(9_000).toISOString() }, startup: { phase: "ready", updatedAt: new Date(10_000).toISOString() }, committedAtMs: 10_500, disposition: "reattach_existing_runner" })!); trace.checkRestartRecovery(restartRecoveryMeasurementFor({ event: { eventType: "item.delta", emittedAt: new Date(10_000 + late - 200).toISOString() }, startup: { phase: "ready", updatedAt: new Date(10_000).toISOString() }, committedAtMs: 10_000 + late, disposition: "reattach_existing_runner" })!); await trace.finish("ok"); - expect(breaches()).toEqual([expect.objectContaining({ metric: "restart_recovery", measuredMs: late })]); + expect(breaches()).toEqual([expect.objectContaining({ metric: "restart_recovery_control_plane", measuredMs: late })]); }); it("measures a resumed attempt's boot from its own claim, not the run's first queue span", async () => { @@ -189,11 +246,11 @@ describe("speed budget checks in the native run trace", () => { it("measures to the commit time, so delivery delays after emission count", async () => { const { trace, breaches } = traced({ adapter: "codex", environment: "local", restartRecovery: true }); - const budget = LOCAL_SPEED_BUDGETS.codex.restart_recovery.p95Ms; + const budget = LOCAL_SPEED_BUDGETS.codex.restart_recovery_control_plane.p95Ms; // Emitted 1 s after ready, committed 20 s after ready. trace.checkRestartRecovery({ readyAtMs: 0, emittedAtMs: 1_000, committedAtMs: budget + 12_000, disposition: "resume_dead_runner" }); await trace.finish("ok"); - expect(breaches()).toEqual([expect.objectContaining({ metric: "restart_recovery", measuredMs: budget + 12_000 })]); + expect(breaches()).toEqual([expect.objectContaining({ metric: "restart_recovery_control_plane", measuredMs: budget + 12_000 })]); }); it("writes nothing for remote runs or without a budget subject", async () => { diff --git a/server/src/services/native-runtime/native-speed-budgets.ts b/server/src/services/native-runtime/native-speed-budgets.ts index cc6302b88bb..93234dc64c8 100644 --- a/server/src/services/native-runtime/native-speed-budgets.ts +++ b/server/src/services/native-runtime/native-speed-budgets.ts @@ -22,7 +22,16 @@ export type SpeedBudgetAdapter = "claude" | "codex" | "opencode"; * a reported metric without a budget: tool-first adapters such as Claude often * call tools before they write text. The budget numbers are the TTFT baseline's. */ -export type SpeedBudgetMetric = "time_to_first_activity" | "agent_boot" | "restart_recovery"; +/** + * `restart_recovery_control_plane`: control plane ready to the moment the + * recovered run's provider starts loading its saved session + * (`provider.session.load`), or to the run's next provider event when nothing + * reloads (a reattached runner). That is Paperclip's own share of a recovery: + * claim, prepare, runtime stage and runner launch. The provider's session load + * grows with session size and with how many sessions resume at once, so the + * full `restart_recovery` latency is reported without a budget. + */ +export type SpeedBudgetMetric = "time_to_first_activity" | "agent_boot" | "restart_recovery" | "restart_recovery_control_plane"; export type SpeedBudgetStart = "cold" | "warm"; export type SpeedBudget = { p50Ms: number; p95Ms: number }; @@ -43,22 +52,29 @@ const s = (seconds: number) => Math.round(seconds * 1000); export const LOCAL_SPEED_BUDGETS: Readonly; agent_boot: Record; + /** Full recovery (ready → next provider event). Reported, not judged. */ restart_recovery: SpeedBudget; + /** Paperclip's share of a recovery; judged. Measured 1.2–2.0 s per run with six + * Codex runs resuming at once (wave w-20261109T045320-s9807, local). */ + restart_recovery_control_plane: SpeedBudget; }>> = { claude: { time_to_first_activity: { cold: { p50Ms: s(16.5), p95Ms: s(30) }, warm: { p50Ms: s(9.5), p95Ms: s(28.5) } }, agent_boot: { cold: { p50Ms: s(9), p95Ms: s(20.5) }, warm: { p50Ms: s(3.5), p95Ms: s(16.5) } }, restart_recovery: { p50Ms: s(15), p95Ms: s(22.5) }, + restart_recovery_control_plane: { p50Ms: s(2.5), p95Ms: s(4) }, }, codex: { time_to_first_activity: { cold: { p50Ms: s(6), p95Ms: s(17) }, warm: { p50Ms: s(10), p95Ms: s(28.5) } }, agent_boot: { cold: { p50Ms: s(2.5), p95Ms: s(7) }, warm: { p50Ms: s(5), p95Ms: s(16.5) } }, restart_recovery: { p50Ms: s(6.5), p95Ms: s(8) }, + restart_recovery_control_plane: { p50Ms: s(2.5), p95Ms: s(4) }, }, opencode: { time_to_first_activity: { cold: { p50Ms: s(17.5), p95Ms: s(49) }, warm: { p50Ms: s(26.5), p95Ms: s(88) } }, agent_boot: { cold: { p50Ms: s(12), p95Ms: s(37) }, warm: { p50Ms: s(22), p95Ms: s(65.5) } }, restart_recovery: { p50Ms: s(25.5), p95Ms: s(38.5) }, + restart_recovery_control_plane: { p50Ms: s(2.5), p95Ms: s(4) }, }, }; @@ -77,6 +93,7 @@ export function speedBudgetFor(adapter: SpeedBudgetAdapter, metric: SpeedBudgetM start: SpeedBudgetStart | null): SpeedBudget { const budgets = LOCAL_SPEED_BUDGETS[adapter]; if (metric === "restart_recovery") return budgets.restart_recovery; + if (metric === "restart_recovery_control_plane") return budgets.restart_recovery_control_plane; const byStart = budgets[metric]; if (start) return byStart[start]; return { p50Ms: Math.max(byStart.cold.p50Ms, byStart.warm.p50Ms), p95Ms: Math.max(byStart.cold.p95Ms, byStart.warm.p95Ms) }; @@ -98,6 +115,10 @@ export type SpeedBudgetBreach = { /** First activity only: time the input waited for its run to exist (for example a * comment posted while the previous turn was running). Not judged. */ inputWaitMs?: number; + /** Control-plane recovery only: the full ready → next provider event latency + * and the provider's share of it (session load onwards). Not judged. */ + restartTotalMs?: number; + providerLoadMs?: number; }; /** A breach when one measured run exceeds its p95 budget, else null. */ @@ -109,6 +130,8 @@ export function evaluateSpeedBudget(input: { measuredMs: number; disposition?: string; inputWaitMs?: number; + restartTotalMs?: number; + providerLoadMs?: number; }): SpeedBudgetBreach | null { if (!input.adapter || input.environment !== "local" || !Number.isFinite(input.measuredMs) || input.measuredMs < 0) return null; const budget = speedBudgetFor(input.adapter, input.metric, input.start); @@ -118,6 +141,8 @@ export function evaluateSpeedBudget(input: { startKind: input.start ?? "unknown", measuredMs: Math.round(input.measuredMs), budgetP50Ms: budget.p50Ms, budgetP95Ms: budget.p95Ms, percentile: "p95", baseline: SPEED_BUDGET_BASELINE, ...(input.disposition ? { disposition: input.disposition } : {}), + ...(input.restartTotalMs !== undefined ? { restartTotalMs: Math.round(input.restartTotalMs) } : {}), + ...(input.providerLoadMs !== undefined ? { providerLoadMs: Math.round(input.providerLoadMs) } : {}), ...(input.inputWaitMs ? { inputWaitMs: Math.round(input.inputWaitMs) } : {}), }; } diff --git a/tests/restart-chaos/ttft-baseline-stats.test.ts b/tests/restart-chaos/ttft-baseline-stats.test.ts index 39a5490fc9b..52b90bc79cf 100644 --- a/tests/restart-chaos/ttft-baseline-stats.test.ts +++ b/tests/restart-chaos/ttft-baseline-stats.test.ts @@ -119,12 +119,44 @@ 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", readyAt, latencyMs: 3_250 }, - { runId: "r2", adapter: "codex", kind: "K1", disposition: "none", readyAt, latencyMs: null }, + { runId: "r1", adapter: "codex", kind: "K1", disposition: "resume_dead_runner", readyAt, latencyMs: 3_250, controlPlaneMs: 3_250, providerLoadMs: 0 }, + { runId: "r2", adapter: "codex", kind: "K1", disposition: "none", readyAt, latencyMs: null, controlPlaneMs: null, providerLoadMs: null }, ]); }); }); +describe("restart recovery split", () => { + // The trace starts at the epoch; the queue span is written promptly and anchors it. + const queue = { eventType: "run.performance.span", createdAt: at(100), payload: { span: "heartbeat.queue", startOffsetMs: 0, durationMs: 100 } }; + const load = (writtenAt: number) => ({ eventType: "run.performance.span", createdAt: at(writtenAt), + payload: { span: "provider.session.load", startOffsetMs: 11_500, durationMs: 4_500 } }); + + it("splits a recovery at the start of the provider's session load", () => { + const readyAt = Date.parse(at(10_000)); + const recovered = run("r1", "restart", [queue, load(16_000), { eventType: "session.resumed", createdAt: at(16_400) }]); + expect(recoverySamples([recovered], [{ kind: "K2", readyAt, runIds: ["r1"] }])).toEqual([ + { runId: "r1", adapter: "codex", kind: "K2", disposition: "none", readyAt, latencyMs: 6_400, controlPlaneMs: 1_500, providerLoadMs: 4_900 }, + ]); + }); + + it("uses the start the server logged even when every span row was written late", () => { + const readyAt = Date.parse(at(10_000)); + // All rows deferred by 8 s, so no write time can anchor the trace. + const late = (span: string, offset: number, duration: number) => ({ eventType: "run.performance.span", createdAt: at(offset + duration + 8_000), + payload: { span, startOffsetMs: offset, durationMs: duration, wallStartedAtMs: Date.parse(at(offset)) } }); + const recovered = run("r1", "restart", [late("heartbeat.queue", 0, 100), { eventType: "session.resumed", createdAt: at(16_400) }, + late("provider.session.load", 11_500, 4_500)]); + expect(recoverySamples([recovered], [{ kind: "K2", readyAt, runIds: ["r1"] }])[0]).toMatchObject({ controlPlaneMs: 1_500, providerLoadMs: 4_900 }); + }); + + it("takes the load start from the span's offset, not its (possibly deferred) write time", () => { + const readyAt = Date.parse(at(10_000)); + // The load span's run-log row is written 5 s late, after the provider event. + const recovered = run("r1", "restart", [queue, { eventType: "session.resumed", createdAt: at(16_400) }, load(21_000)]); + expect(recoverySamples([recovered], [{ kind: "K2", readyAt, runIds: ["r1"] }])[0]).toMatchObject({ controlPlaneMs: 1_500, providerLoadMs: 4_900 }); + }); +}); + describe("restart sample guard", () => { const tool = (eventType: string, executionId: string, extra: Record = {}): BaselineEvent => ({ eventType, createdAt: at(0), payload: { prpEvent: { payload: { executionId, ...extra } } }, diff --git a/tests/restart-chaos/ttft-baseline-stats.ts b/tests/restart-chaos/ttft-baseline-stats.ts index 9b1cdfd690f..ce407120494 100644 --- a/tests/restart-chaos/ttft-baseline-stats.ts +++ b/tests/restart-chaos/ttft-baseline-stats.ts @@ -132,9 +132,42 @@ 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; readyAt: number; latencyMs: number | null } +export interface RecoverySample { runId: string; adapter: string; kind: string; disposition: string; readyAt: number; latencyMs: number | null; + /** Paperclip's share: ready to the start of the provider's session load (`provider.session.load`), or the whole + * latency when nothing reloads. The judged row. */ + controlPlaneMs: number | null; + /** The provider's share: session load start to the next provider event. Reported. */ + providerLoadMs: number | null } /** Control plane ready → the recovered run's next provider event, keyed by runner disposition. */ +/** The run's trace start: every span records its offset from it. A span event is + * written when the span ends, possibly through deferred work, so each one gives + * an upper bound (write time minus offset and duration); the earliest is used. */ +export function spanTraceStartMs(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 candidate = Date.parse(event.createdAt) - (Number(event.payload.startOffsetMs) + Number(event.payload.durationMs)); + if (Number.isFinite(candidate) && (earliest === null || candidate < earliest)) earliest = candidate; + } + return earliest; +} + +/** Stamps each provider session load with its wall-clock start (`wallStartMs`). + * The server writes the span's own start as `wallStartedAtMs`; runs logged before + * that field existed fall back to the trace start plus the span's offset, never to + * its write time. Call on the raw run, before outage removal rewrites offsets. */ +export function stampProviderLoadStarts(run: BaselineRun): BaselineRun { + const anchor = spanTraceStartMs(run); + return { ...run, events: run.events.map(event => { + if (event.eventType !== "run.performance.span" || event.payload?.span !== "provider.session.load" || event.payload.wallStartMs !== undefined) return event; + const logged = Number(event.payload.wallStartedAtMs); + const offset = Number(event.payload.startOffsetMs); + const start = Number.isFinite(logged) ? logged : anchor !== null && Number.isFinite(offset) ? anchor + offset : null; + return start === null ? event : { ...event, payload: { ...event.payload, wallStartMs: start } }; + }) }; +} + export function recoverySamples(runs: readonly BaselineRun[], faults: readonly BaselineFault[]): RecoverySample[] { const byId = new Map(runs.map(run => [run.runId, run])); const samples: RecoverySample[] = []; @@ -146,9 +179,19 @@ 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); + const nextAt = next ? Date.parse(next.createdAt) : null; + const stamped = stampProviderLoadStarts(run); + const loadStarts = nextAt === null ? [] : stamped.events.flatMap(event => { + if (event.eventType !== "run.performance.span" || event.payload?.span !== "provider.session.load") return []; + const start = Number(event.payload.wallStartMs); + return Number.isFinite(start) && start >= fault.readyAt && start <= nextAt ? [start] : []; + }); + const loadStart = loadStarts.length ? Math.min(...loadStarts) : null; 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 }); + latencyMs: nextAt === null ? null : nextAt - fault.readyAt, + controlPlaneMs: nextAt === null ? null : (loadStart ?? nextAt) - fault.readyAt, + providerLoadMs: nextAt === null ? null : loadStart === null ? 0 : nextAt - loadStart }); } } return samples; diff --git a/tests/restart-chaos/ttft-baseline.ts b/tests/restart-chaos/ttft-baseline.ts index 2f1fe6d633d..893f18616c4 100644 --- a/tests/restart-chaos/ttft-baseline.ts +++ b/tests/restart-chaos/ttft-baseline.ts @@ -204,7 +204,7 @@ function report(runs: BaselineRun[], faults: BaselineFault[]): string { lines.push("", "| Pre-turn step | n | p50 | p95 | Adapters |", "|---|---|---|---|---|"); for (const step of rankPreTurnSteps(runs).slice(0, 25)) lines.push(`| \`${step.step}\` | ${step.n} | ${ms(step.p50)} | ${ms(step.p95)} | ${step.adapters.join(", ")} |`); const samples = recoverySamples(runs, faults); - lines.push("", "| Adapter | Fault | Disposition | n | Recovery p50 | Recovery p95 | No provider event |", "|---|---|---|---|---|---|---|"); + lines.push("", "| Adapter | Fault | Disposition | n | Recovery p50 | Recovery p95 | Paperclip share p50 / p95 | No provider event |", "|---|---|---|---|---|---|---|---|"); const groups = new Map(); for (const sample of samples) { const key = `${sample.adapter}|${sample.kind}|${sample.disposition}`; @@ -213,7 +213,8 @@ function report(runs: BaselineRun[], faults: BaselineFault[]): string { for (const [key, group] of [...groups.entries()].sort()) { const [adapter, kind, disposition] = key.split("|"); const d = distribution(group.flatMap(sample => sample.latencyMs === null ? [] : [sample.latencyMs])); - lines.push(`| ${adapter} | ${kind} | ${disposition} | ${d.n} | ${ms(d.p50)} | ${ms(d.p95)} | ${group.length - d.n} |`); + const control = distribution(group.flatMap(sample => sample.controlPlaneMs === null ? [] : [sample.controlPlaneMs])); + lines.push(`| ${adapter} | ${kind} | ${disposition} | ${d.n} | ${ms(d.p50)} | ${ms(d.p95)} | ${ms(control.p50)} / ${ms(control.p95)} | ${group.length - d.n} |`); } return `${lines.join("\n")}\n`; } diff --git a/tests/restart-chaos/wave-speed.test.ts b/tests/restart-chaos/wave-speed.test.ts index 293ba5d1756..c376cd83ee2 100644 --- a/tests/restart-chaos/wave-speed.test.ts +++ b/tests/restart-chaos/wave-speed.test.ts @@ -45,7 +45,23 @@ describe("wave speed grade", () => { 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"] }); + // Nothing reloaded, so the whole latency is Paperclip's share and is judged. + expect(speed.rows.find(row => row.metric === "restart_recovery_control_plane")).toMatchObject({ adapter: "claude", breaches: ["p50", "p95"] }); + expect(speed.rows.find(row => row.metric === "restart_recovery")).toMatchObject({ budget: null, breaches: [], measured: { p50: 30_000 } }); + }); + + it("judges Paperclip's share of a recovery and only reports the provider's session load", () => { + const loadEnd = 10_000 + 1_500 + 9_000; + const recovered: BaselineRun = { runId: "r", adapter: "codex", environment: "local", phase: "cold", events: [ + { eventType: "run.performance.span", createdAt: new Date(100).toISOString(), payload: { span: "heartbeat.queue", startOffsetMs: 0, durationMs: 100 } }, + { eventType: "run.performance.span", createdAt: new Date(loadEnd).toISOString(), + payload: { span: "provider.session.load", startOffsetMs: 11_500, durationMs: 9_000 } }, + { eventType: "session.resumed", createdAt: new Date(loadEnd + 300).toISOString() }] }; + const speed = gradeWaveSpeed("local", [recovered], [{ kind: "K2", readyAt: 10_000, runIds: ["r"] }], 0); + expect(speed.rows.find(row => row.metric === "restart_recovery_control_plane")).toMatchObject({ measured: { p50: 1_500 }, breaches: [] }); + expect(speed.rows.find(row => row.metric === "restart_provider_load")).toMatchObject({ measured: { p50: 9_300 }, budget: null, breaches: [] }); + expect(speed.rows.find(row => row.metric === "restart_recovery")).toMatchObject({ measured: { p50: 10_800 }, budget: null }); + expect(speed.breaches).toBe(0); }); it("removes a later outage from restart recovery", () => { @@ -54,7 +70,7 @@ describe("wave speed grade", () => { 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: [] }); + expect(speed.rows.find(row => row.metric === "restart_recovery_control_plane")).toMatchObject({ measured: { p50: 2_000 }, breaches: [] }); }); it("has no budget for remote environments or unbudgeted adapters", () => { diff --git a/tests/restart-chaos/wave-speed.ts b/tests/restart-chaos/wave-speed.ts index bbadc7d7128..c57973fa00b 100644 --- a/tests/restart-chaos/wave-speed.ts +++ b/tests/restart-chaos/wave-speed.ts @@ -1,7 +1,7 @@ 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, + baselineRows, distribution, recoverySamples, runMetrics, runSpans, runWarmth, stampProviderLoadStarts, type BaselineEvent, type BaselineFault, type BaselineRun, type Distribution, } from "./ttft-baseline-stats.js"; import type { AdapterId, EnvironmentKind, FaultReceipt, RunSample } from "./types.js"; @@ -11,7 +11,8 @@ 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" | "workspace_provision" | "restart_recovery"; start: "cold" | "warm" | "restart"; + adapter: string; metric: "time_to_first_activity" | "ttft" | "agent_boot" | "workspace_provision" | "restart_recovery" + | "restart_recovery_control_plane" | "restart_provider_load"; start: "cold" | "warm" | "restart"; measured: Distribution; budget: SpeedBudget | null; breaches: Array<"p50" | "p95">; } @@ -33,9 +34,11 @@ 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" || metric === "workspace_provision") return null; + // Reported without a budget: the full recovery and the provider's session load + // grow with session size and simultaneous resumes; Paperclip's share is judged. + if (metric === "ttft" || metric === "workspace_provision" || metric === "restart_recovery" || metric === "restart_provider_load") return null; const budgets = LOCAL_SPEED_BUDGETS[key] as unknown as Record | SpeedBudget>; - if (metric === "restart_recovery") return budgets.restart_recovery as SpeedBudget; + if (metric === "restart_recovery_control_plane") return budgets.restart_recovery_control_plane 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"]; @@ -180,12 +183,24 @@ export function gradeWaveSpeed(environment: EnvironmentKind, runs: readonly Base } } 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)) }); + const outageFree = (readyAt: number, ms: number | null) => ms === null ? null : Math.max(0, clock(readyAt + ms) - clock(readyAt)); + const samples = recoverySamples(runs, faults).map(sample => sample.latencyMs === null ? sample : { + ...sample, latencyMs: outageFree(sample.readyAt, sample.latencyMs), + controlPlaneMs: outageFree(sample.readyAt, sample.controlPlaneMs), + providerLoadMs: sample.providerLoadMs === null || sample.controlPlaneMs === null ? sample.providerLoadMs + : Math.max(0, clock(sample.readyAt + sample.controlPlaneMs + sample.providerLoadMs) - clock(sample.readyAt + sample.controlPlaneMs)), + }); 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) }); + for (const [metric, pick] of [["restart_recovery", (sample: typeof samples[number]) => sample.latencyMs], + ["restart_recovery_control_plane", (sample: typeof samples[number]) => sample.controlPlaneMs], + ["restart_provider_load", (sample: typeof samples[number]) => sample.providerLoadMs]] as const) { + const measured = distribution(samples.flatMap(sample => { + const value = sample.adapter === adapter ? pick(sample) : null; + return value === null ? [] : [value]; + })); + const budget = budgetFor(environment, adapter, metric, "restart"); + rows.push({ adapter, metric, 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); @@ -248,7 +263,8 @@ export async function collectWaveSpeed(api: ChaosApi, environment: EnvironmentKi } 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); + // Stamp provider load starts from raw offsets before outage removal rewrites them. + const trimmed = withoutOutage(stampProviderLoadStarts(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 });