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 f8ecd1089d8..9aaf87c9c03 100644 --- a/doc/plans/2026-11-07-durable-ownership-and-reconciliation.md +++ b/doc/plans/2026-11-07-durable-ownership-and-reconciliation.md @@ -69,6 +69,61 @@ Each metric is reported by adapter (Codex, Claude/ACPX, OpenCode) and environmen - RC-27: a 5-minute orphan reaper. 4. **Phase 0 measures, then sets budgets.** It records a baseline for every metric above. Phase 1 sets p50/p95 budgets per adapter and environment. From then on, every phase's exit criteria include "no regression against the baseline". The on-demand crash-point suite and the chaos soak also report TTFT and restart recovery latency. +### Speed budgets (Phase 1) + +Phase 1 sets these budgets from the local baseline in `2026-11-07-ttft-baseline.md`. Each budget is the baseline value times 1.2, rounded up to 0.5 s. The 20% headroom covers the small samples and the shared host of the baseline. Daytona and other remote environments get budgets after their own baseline. + +A single run is checked against its p95 budget. When a run exceeds it, the run log gets one `run.performance.budget_exceeded` row (`doc/run-log-events.md`). The p50 budgets apply to aggregates: chaos waves, the baseline driver, and staging. Every later phase's exit criteria include "no regression against these budgets". The table lives in code at `server/src/services/native-runtime/native-speed-budgets.ts`. + +Time to first token (s, p50 / p95): + +| Adapter | Cold | Warm | +|---|---|---| +| claude | 16.5 / 30.0 | 9.5 / 28.5 | +| codex | 6.0 / 17.0 | 10.0 / 28.5 | +| opencode | 17.5 / 49.0 | 26.5 / 88.0 | + +Agent boot (s, p50 / p95): + +| Adapter | Cold | Warm | +|---|---|---| +| claude | 9.0 / 20.5 | 3.5 / 16.5 | +| codex | 2.5 / 7.0 | 5.0 / 16.5 | +| opencode | 12.0 / 37.0 | 22.0 / 65.5 | + +Restart recovery latency, from `ready` to the recovered run's next provider event (s, p50 / p95). The source is the baseline's run 4, which has 2 samples per adapter and disposition, so the largest disposition row is used: + +| Adapter | Budget | +|---|---| +| claude | 15.0 / 22.5 | +| codex | 6.5 / 8.0 | +| opencode | 25.5 / 38.5 | + +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) + +Three local waves (`tests/restart-chaos`, `--env local`, one adapter each, 6 tasks, faults K1 and K2, seed 62) ran on the Phase 1 stack at `9fe186de36`. All 18 tasks passed R1–R6 and crossed a restart. The ownership report ran 13 passes per wave and saw at most 19–23 in-progress records at once. + +| Kind | Owner verdicts (record-passes, three waves) | +|---|---| +| `environment_run_lease_active` | alive 196 | +| `issue_execution_lock` | alive 80 | +| `issue_checkout_lock` | alive 80 | +| `wake_request_claimed` | alive 80 | +| `native_restart_recovery_claim` | alive 64 (re-stamped by the new boot) | +| `native_executor_claim` | alive 16 | +| `run_accounting_debt` | unprovable 4 (no owner stamp; transient) | + +Invariant violations: **0** for every rule and kind (I1, I2, I3, I4 and I7). No owner was ever proven gone, so I2 never fired. + +What the data does and does not show: + +- The existing per-subsystem paths settle dead-owner records within about 1 s of `ready`, before the first report pass. A pass every 30 s, triggered by work signals, does not catch those short windows. The crash-point suite (Phase 5) is the tool for them; the report measures records that stay stuck. +- These waves exercise only the run-lifecycle kinds. Remote, workspace and accounting kinds need Daytona waves, K4 and K6 faults, and longer storms. The overnight campaign's failures came mostly from those. +- Track 1 now stamps the PID space on `native_run_finalizations` (#19868) and `heartbeat_runs` (#19871), and the registry reads those stamps. Rows written before them stay `unprovable`. Phase 2 needs the same stamp on every kind it moves before that kind can settle by proof. +- Restart-crossing dispatches exceeded the TTFT and boot budgets because those metrics spanned the outage. The budget check now judges such dispatches only on restart-recovery latency. + ## Relationship to the zero-downtime upgrade work Separate work moves the control plane toward zero-downtime upgrades on an ephemeral filesystem. PostgreSQL holds execution authority and durable protocol state. Object storage holds workspace archives, provider checkpoints, logs, traces and skill assets. Replacement servers reuse stable encryption and signing secrets. Most of the foundations are on master; #19830 is still open. Entry points: diff --git a/doc/run-log-events.md b/doc/run-log-events.md index eb7d5afb676..8ae2618d9cf 100644 --- a/doc/run-log-events.md +++ b/doc/run-log-events.md @@ -87,6 +87,25 @@ not qualify. No deliverable text means no timing sample. A recovered dispatch ca record newly committed output from the original start; it does not reproduce earlier output timing. The event contains no text or provider identifiers. +## Speed budget breaches + +`run.performance.budget_exceeded` (schema `paperclip.run-performance-budget.v1`, +level `warn`) records one native run that exceeded its p95 speed budget. The +budgets come from the Phase 0 local baseline and are listed in +`doc/plans/2026-11-07-durable-ownership-and-reconciliation.md` ("Speed budgets"). +`metric` is `ttft` (`task.time_to_first_token`), `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`), +`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`. +Each metric is checked at most once per dispatch. A restart-recovery dispatch is +judged only on `restart_recovery`, because its TTFT and boot span the outage. 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. + The existing `run.performance.span` event also records `provider_pack.select`, `provider_pack.discover`, `provider_pack.verify`, `provider_pack.verify_preinstalled`, `provider_pack.link_preinstalled`, `provider_pack.clear_stale`, `provider_pack.upload`, diff --git a/server/src/services/native-runtime/native-run-trace.ts b/server/src/services/native-runtime/native-run-trace.ts index a2967612164..0685506fcec 100644 --- a/server/src/services/native-runtime/native-run-trace.ts +++ b/server/src/services/native-runtime/native-run-trace.ts @@ -9,6 +9,13 @@ import { getStartupTraceContext, type StartupTraceContextHandle, } from "../../instrumentation.js"; +import { + evaluateSpeedBudget, + SPEED_BUDGET_EXCEEDED_EVENT_TYPE, + type SpeedBudgetAdapter, + type SpeedBudgetBreach, + type SpeedBudgetStart, +} from "./native-speed-budgets.js"; export const NATIVE_RUN_SPAN_EVENT_TYPE = "run.performance.span"; export const NATIVE_RUN_TRACE_SCHEMA_VERSION = 3; @@ -260,6 +267,14 @@ export function createNativeRunTrace(input: { startedAtMs?: number; onEvent?: NativeRunTraceSink; traceContext?: StartupTraceContextHandle; + /** Speed-budget subject (invariant 8). Absent or remote means no check. */ + budget?: { + adapter: SpeedBudgetAdapter | null; + environment: "local" | "remote"; + /** A restart-recovery dispatch: its TTFT and boot span the outage, so only + * `restart_recovery` is judged. */ + restartRecovery?: boolean; + }; }) { const tracing = input.traceContext ?? getStartupTraceContext("paperclip.native-runner"); @@ -340,6 +355,45 @@ export function createNativeRunTrace(input: { } }; + // Speed budgets (invariant 8). Breaches are written detached, after the + // measurement, so the check never adds a wait to the turn path. + let budgetStart: SpeedBudgetStart | null = null; + let queueEndedAtMs: number | null = null; + let attemptStartedAtMs: number | null = null; + let bootChecked = false; + const reportBudgetBreach = (breach: SpeedBudgetBreach | null): void => { + if (!breach || !input.onEvent) return; + defer(async () => { + await input.onEvent?.({ + eventType: SPEED_BUDGET_EXCEEDED_EVENT_TYPE, + stream: "system", + level: "warn", + message: `speed budget exceeded: ${breach.metric} ${breach.measuredMs}ms > p95 ${breach.budgetP95Ms}ms (${breach.adapter}, ${breach.startKind})`, + payload: breach, + }); + }); + }; + const checkSpanBudget = (name: string, startedAtMs: number, endedAtMs: number): void => { + if (!input.budget || input.budget.restartRecovery) return; + const subject = { adapter: input.budget.adapter, environment: input.budget.environment }; + if (name === "heartbeat.queue") queueEndedAtMs = endedAtMs; + // A resumed attempt keeps the run's original queue span; its own claim is + // the start of this attempt's preparation. + if (name === "heartbeat.prepare_before_environment") attemptStartedAtMs = startedAtMs; + if (name === "task.time_to_first_token") { + reportBudgetBreach(evaluateSpeedBudget({ ...subject, metric: "ttft", start: budgetStart, + measuredMs: endedAtMs - startedAtMs })); + } + // Agent boot: run claimed (end of heartbeat.queue) to the provider session + // accepting turn.start (end of runner.turn.submit or runner.session.resume). + if ((name === "runner.turn.submit" || name === "runner.session.resume") && !bootChecked && queueEndedAtMs !== null) { + bootChecked = true; + const claimedAtMs = Math.max(queueEndedAtMs, attemptStartedAtMs ?? queueEndedAtMs); + reportBudgetBreach(evaluateSpeedBudget({ ...subject, metric: "agent_boot", start: budgetStart, + measuredMs: endedAtMs - claimedAtMs })); + } + }; + const emitCompletedSpanEvent = async ( span: NativeRunHistoricalSpan, ): Promise => { @@ -479,6 +533,7 @@ export function createNativeRunTrace(input: { outcome, attributes: state.attributes, }); + checkSpanBudget(state.name, state.startedAtMs, endedAtMs); }; const activate = (scope: NativeRunSpanScope): void => { @@ -570,7 +625,9 @@ export function createNativeRunTrace(input: { const finish = async (outcome: NativeRunSpanOutcome): Promise => { if (ended) return; - await Promise.all([...pendingDiagnostics]); + // Settled diagnostics can defer more (a budget breach after a deferred + // measurement), so drain until none remain. + while (pendingDiagnostics.size > 0) await Promise.all([...pendingDiagnostics]); if (ended) return; ended = true; const endedAtMs = Date.now(); @@ -593,8 +650,31 @@ export function createNativeRunTrace(input: { }); }; + /** Labels the run cold (fresh provider session) or warm (reused). */ + const setBudgetStart = (kind: SpeedBudgetStart): void => { + budgetStart ??= kind; + }; + + /** Restart recovery latency: control plane ready to the recovered run's + * next provider event, checked once per run. */ + let recoveryChecked = false; + const checkRestartRecovery = (measurement: RestartRecoveryMeasurement): void => { + if (!input.budget || recoveryChecked || measurement.readyAtMs === null) return; + // An event emitted before ready (buffered by a surviving runner and + // committed late) is not the recovered run's next event; it must not use + // the one check. + 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 })); + }; + return { traceStartedAtMs, + setBudgetStart, + checkRestartRecovery, start, annotate, activate, @@ -609,6 +689,16 @@ export function createNativeRunTrace(input: { export type NativeRunTrace = ReturnType; +/** Restart recovery latency: control plane ready to the recovered run's + * next provider event. The event is filtered on when it was emitted and + * measured to when it was committed (the baseline's run-log `created_at`). */ +export type RestartRecoveryMeasurement = { + readyAtMs: number | null; + emittedAtMs: number | null; + committedAtMs: number; + disposition: string; +}; + /** Group already-normalized provider measurements without exporting provider attributes. */ export async function recordNativeProviderPerformanceSpans(trace: NativeRunTrace, spans: NativeRunHistoricalSpan[]): Promise { if (!spans.length) return; diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 9d0c3b7f7a2..db064de7157 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -13,6 +13,8 @@ import { readProviderFailure } from "@paperclipai/shared"; import { nativeCancellationHasUnsettledOwner, notifyNativeCancellation, waitForNativeCancellation } from "./native-cancellation-delivery.js"; import { readClaimedNativeStopRecovery } from "./native-cancellation-request.js"; import { publishNativeHarnessBackup } from "./native-harness-backup-publication.js"; +import { restartRecoveryMeasurementFor, speedBudgetAdapter } from "./native-speed-budgets.js"; +import { getStartupRecoveryState } from "../../startup-recovery-state.js"; import { assertNativeHarnessBackupOwner, durableNativeHarnessStorage, NATIVE_HARNESS_DOWNLOAD_IDLE_TIMEOUT_MS, nativeHarnessBackupOwner, persistNativeHarnessBackup, restoreDurableNativeHarnessBackup } from "./native-harness-backup-storage.js"; import { restoreNativeProviderArtifacts, retainNativeProviderArtifacts, retainExistingNativeProviderArtifacts, @@ -9497,6 +9499,11 @@ async function executePaperclipNativeSessionWithinScope( runId: input.execution.binding.runId, startedAtMs: preparationStarts.runStartedAtMs, onEvent: input.onEvent, + budget: { + adapter: speedBudgetAdapter(input.execution.provider), + environment: !input.runnerExecutionTarget || input.runnerExecutionTarget.kind === "local" ? "local" : "remote", + restartRecovery: Boolean(input.restartRecovery), + }, }); await emitStartupReuse(input.onEvent, "environment", environmentStartupReuse(input.runnerExecutionTarget)); if (!input.runnerExecutionTarget || input.runnerExecutionTarget.kind === "local") { @@ -10262,6 +10269,13 @@ async function executePaperclipNativeSessionWithinScope( const milestoneAtMs = Number.isFinite(eventAtMs) ? eventAtMs : Date.now(); + // Restart recovery latency (invariant 8): control plane ready to this + // recovered run's first committed provider event. Checked once; run log only. + if (input.restartRecovery) { + const measurement = restartRecoveryMeasurementFor({ event, startup: getStartupRecoveryState(), + committedAtMs: Date.now(), disposition: input.restartRecovery.kind }); + if (measurement) trace.checkRestartRecovery(measurement); + } if ( event.eventType === "session.started" && sessionStartedAtMs === null @@ -10269,6 +10283,7 @@ async function executePaperclipNativeSessionWithinScope( sessionStartedAtMs = milestoneAtMs; sessionStartupMode = "bootstrap"; await emitStartupReuse(input.onEvent, "provider_session", { state: "fresh", reason: "created" }); + trace.setBudgetStart("cold"); await trace.record({ name: "runner.session.bootstrap", parentName: "runner.session.startup", @@ -10278,6 +10293,7 @@ async function executePaperclipNativeSessionWithinScope( } if (event.eventType === "session.resumed") { await emitStartupReuse(input.onEvent, "provider_session", { state: "reused", reason: "verified_reuse" }); + trace.setBudgetStart("warm"); } if ( event.eventType === "turn.submitted" && @@ -10643,6 +10659,7 @@ async function executePaperclipNativeSessionWithinScope( await emitStartupReuse(input.onEvent, "runner", { state: "reused", reason: "verified_reuse" }); executionContext?.assertOpen(); await emitStartupReuse(input.onEvent, "provider_session", { state: "reused", reason: "verified_reuse" }); + trace.setBudgetStart("warm"); }; try { if (executionContext) await executionContext.wait(emitReuse); else await emitReuse(); @@ -10840,6 +10857,7 @@ async function executePaperclipNativeSessionWithinScope( existingWarmSession = recovered.session; await emitStartupReuse(input.onEvent, "runner", { state: "reused", reason: "verified_reuse" }); await emitStartupReuse(input.onEvent, "provider_session", { state: "reused", reason: "verified_reuse" }); + trace.setBudgetStart("warm"); persistedWarmSession = null; } else { await recovered.session.close({ reason: "recovered warm session requires current process configuration" }); diff --git a/server/src/services/native-runtime/native-speed-budgets.test.ts b/server/src/services/native-runtime/native-speed-budgets.test.ts new file mode 100644 index 00000000000..02a6a6b4116 --- /dev/null +++ b/server/src/services/native-runtime/native-speed-budgets.test.ts @@ -0,0 +1,164 @@ +import { describe, expect, it } from "vitest"; +import type { AdapterRuntimeEvent } from "../../adapters/index.js"; +import { createNativeRunTrace } from "./native-run-trace.js"; +import { + evaluateSpeedBudget, + LOCAL_SPEED_BUDGETS, + RESTART_RECOVERY_PROVIDER_EVENT_TYPES, + restartRecoveryMeasurementFor, + SPEED_BUDGET_EXCEEDED_EVENT_TYPE, + speedBudgetAdapter, + speedBudgetFor, +} from "./native-speed-budgets.js"; + +describe("speed budgets", () => { + it("defines p50 below p95 for every adapter, metric and start", () => { + for (const budgets of Object.values(LOCAL_SPEED_BUDGETS)) { + for (const budget of [budgets.ttft.cold, budgets.ttft.warm, budgets.agent_boot.cold, budgets.agent_boot.warm, + budgets.restart_recovery]) { + expect(budget.p50Ms).toBeGreaterThan(0); + expect(budget.p50Ms).toBeLessThanOrEqual(budget.p95Ms); + } + } + }); + + it("maps native providers to budgeted adapters", () => { + expect(speedBudgetAdapter({ kind: "codex" })).toBe("codex"); + expect(speedBudgetAdapter({ kind: "opencode" })).toBe("opencode"); + expect(speedBudgetAdapter({ kind: "acpx", agent: "claude" })).toBe("claude"); + expect(speedBudgetAdapter({ kind: "acpx", agent: "cursor" })).toBeNull(); + expect(speedBudgetAdapter({ kind: "claude_managed" })).toBeNull(); + }); + + it("judges one run against p95 and uses the looser budget when the start kind is unknown", () => { + const codexCold = LOCAL_SPEED_BUDGETS.codex.ttft.cold.p95Ms; + expect(evaluateSpeedBudget({ adapter: "codex", environment: "local", metric: "ttft", start: "cold", measuredMs: codexCold })) + .toBeNull(); + expect(evaluateSpeedBudget({ adapter: "codex", environment: "local", metric: "ttft", start: "cold", measuredMs: codexCold + 1 })) + .toMatchObject({ metric: "ttft", adapter: "codex", startKind: "cold", budgetP95Ms: codexCold, percentile: "p95" }); + const loose = speedBudgetFor("codex", "ttft", null); + expect(loose.p95Ms).toBe(Math.max(codexCold, LOCAL_SPEED_BUDGETS.codex.ttft.warm.p95Ms)); + expect(evaluateSpeedBudget({ adapter: "codex", environment: "local", metric: "ttft", start: null, measuredMs: codexCold + 1 })) + .toBeNull(); + }); + + it("does not judge remote runs or adapters without a baseline", () => { + expect(evaluateSpeedBudget({ adapter: "codex", environment: "remote", metric: "ttft", start: "cold", measuredMs: 1e9 })).toBeNull(); + expect(evaluateSpeedBudget({ adapter: null, environment: "local", metric: "ttft", start: "cold", measuredMs: 1e9 })).toBeNull(); + }); +}); + +describe("speed budget checks in the native run trace", () => { + function traced(budget?: Parameters[0]["budget"]) { + const events: AdapterRuntimeEvent[] = []; + const trace = createNativeRunTrace({ runId: "run-1", startedAtMs: 0, budget, + onEvent: async event => { events.push(event); } }); + const breaches = () => events.filter(event => event.eventType === SPEED_BUDGET_EXCEEDED_EVENT_TYPE) + .map(event => event.payload as Record); + return { trace, breaches }; + } + + it("writes one run-log breach for a slow agent boot and none for a fast TTFT", async () => { + const { trace, breaches } = traced({ adapter: "codex", environment: "local" }); + trace.setBudgetStart("cold"); + await trace.record({ name: "heartbeat.queue", startedAtMs: 0, endedAtMs: 100 }); + const boot = LOCAL_SPEED_BUDGETS.codex.agent_boot.cold.p95Ms + 500; + await trace.record({ name: "runner.turn.submit", startedAtMs: 100, endedAtMs: 100 + boot }); + // A second submit in the same run is not a second boot. + await trace.record({ name: "runner.turn.submit", startedAtMs: 100, endedAtMs: 100 + boot * 2 }); + await trace.record({ name: "task.time_to_first_token", startedAtMs: 0, endedAtMs: 1_000 }); + await trace.finish("ok"); + + expect(breaches()).toEqual([expect.objectContaining({ + schema: "paperclip.run-performance-budget.v1", metric: "agent_boot", adapter: "codex", startKind: "cold", + measuredMs: boot, budgetP95Ms: LOCAL_SPEED_BUDGETS.codex.agent_boot.cold.p95Ms, + })]); + }); + + it("writes a breach for a slow TTFT recorded by deferred diagnostics before the trace closes", async () => { + const { trace, breaches } = traced({ adapter: "claude", environment: "local" }); + trace.setBudgetStart("warm"); + const ttft = LOCAL_SPEED_BUDGETS.claude.ttft.warm.p95Ms + 1; + trace.defer(async () => { + await trace.record({ name: "task.time_to_first_token", startedAtMs: 0, endedAtMs: ttft }); + }); + await trace.finish("ok"); + + expect(breaches()).toEqual([expect.objectContaining({ metric: "ttft", adapter: "claude", startKind: "warm", measuredMs: ttft })]); + }); + + 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; + 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, + disposition: "resume_dead_runner", startKind: "unknown" })]); + }); + + it("judges a restart-recovery dispatch only on recovery latency, not on TTFT or boot spanning the outage", async () => { + const { trace, breaches } = traced({ adapter: "codex", environment: "local", restartRecovery: true }); + 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_token", startedAtMs: 0, endedAtMs: 120_000 }); + const at = LOCAL_SPEED_BUDGETS.codex.restart_recovery.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"]); + }); + + 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; + // 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 })]); + }); + + it("measures a resumed attempt's boot from its own claim, not the run's first queue span", async () => { + const { trace, breaches } = traced({ adapter: "codex", environment: "local" }); + trace.setBudgetStart("warm"); + await trace.record({ name: "heartbeat.queue", startedAtMs: 0, endedAtMs: 100 }); + // The attempt resumed ten minutes later and booted in one second. + await trace.record({ name: "heartbeat.prepare_before_environment", startedAtMs: 600_000, endedAtMs: 600_200 }); + await trace.record({ name: "runner.session.resume", startedAtMs: 600_200, endedAtMs: 601_000 }); + await trace.finish("ok"); + expect(breaches()).toEqual([]); + }); + + it("counts only the baseline's provider events as a recovered run's next event", () => { + const startup = { phase: "ready", updatedAt: new Date(10_000).toISOString() }; + const measure = (eventType: string) => restartRecoveryMeasurementFor({ event: { eventType, emittedAt: new Date(11_000).toISOString() }, + startup, committedAtMs: 12_000, disposition: "resume_dead_runner" }); + expect(measure("item.delta")).toEqual({ readyAtMs: 10_000, emittedAtMs: 11_000, committedAtMs: 12_000, disposition: "resume_dead_runner" }); + expect(measure("harness.diagnostic")).toBeNull(); + expect(measure("run.performance.span")).toBeNull(); + expect(RESTART_RECOVERY_PROVIDER_EVENT_TYPES.has("session.resumed")).toBe(true); + }); + + 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; + // 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 })]); + }); + + it("writes nothing for remote runs or without a budget subject", async () => { + for (const budget of [{ adapter: "codex" as const, environment: "remote" as const }, undefined]) { + const { trace, breaches } = traced(budget); + await trace.record({ name: "heartbeat.queue", startedAtMs: 0, endedAtMs: 1 }); + await trace.record({ name: "runner.turn.submit", startedAtMs: 1, endedAtMs: 10_000_000 }); + await trace.record({ name: "task.time_to_first_token", startedAtMs: 0, endedAtMs: 10_000_000 }); + await trace.finish("ok"); + expect(breaches()).toEqual([]); + } + }); +}); diff --git a/server/src/services/native-runtime/native-speed-budgets.ts b/server/src/services/native-runtime/native-speed-budgets.ts new file mode 100644 index 00000000000..1d746946dd8 --- /dev/null +++ b/server/src/services/native-runtime/native-speed-budgets.ts @@ -0,0 +1,129 @@ +/** + * Speed budgets for native runs (plan invariant 8, Phase 1). + * + * Source: the Phase 0 local baseline (doc/plans/2026-11-07-ttft-baseline.md, + * merged in #19846: TTFT and boot from runs 1 and 3, restart recovery from + * run 4; Apple M5 Max, shared host). Each + * budget is the baseline value times 1.2, rounded up to 0.5 s. The table is + * recorded in doc/plans/2026-11-07-durable-ownership-and-reconciliation.md. + * + * A single run is checked against the p95 budget; p50 budgets apply to + * aggregates (chaos waves, the baseline driver). Remote environments have no + * budget until a Daytona baseline exists. Breaches go to the run log only, + * never to Telemetry (AGENTS.md §5.7). + */ + +export type SpeedBudgetAdapter = "claude" | "codex" | "opencode"; +export type SpeedBudgetMetric = "ttft" | "agent_boot" | "restart_recovery"; +export type SpeedBudgetStart = "cold" | "warm"; + +export type SpeedBudget = { p50Ms: number; p95Ms: number }; + +export const SPEED_BUDGET_BASELINE = "2026-11-07-ttft-baseline (local)"; +export const SPEED_BUDGET_EXCEEDED_EVENT_TYPE = "run.performance.budget_exceeded"; + +/** The provider events the baseline counts as a recovered run's next event + * (tests/restart-chaos/ttft-baseline-stats.ts). Diagnostics do not count. */ +export const RESTART_RECOVERY_PROVIDER_EVENT_TYPES: ReadonlySet = 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", +]); + +const s = (seconds: number) => Math.round(seconds * 1000); + +/** TTFT and agent boot by adapter and start kind; restart recovery by adapter. */ +export const LOCAL_SPEED_BUDGETS: Readonly; + agent_boot: Record; + restart_recovery: SpeedBudget; +}>> = { + claude: { + ttft: { 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) }, + }, + codex: { + ttft: { 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) }, + }, + opencode: { + ttft: { 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) }, + }, +}; + +/** Budgeted adapter of a native execution provider, or null (no budget). */ +export function speedBudgetAdapter(provider: { kind: string; agent?: unknown } | null | undefined): SpeedBudgetAdapter | null { + if (!provider) return null; + if (provider.kind === "codex") return "codex"; + if (provider.kind === "opencode") return "opencode"; + if (provider.kind === "acpx" && provider.agent === "claude") return "claude"; + return null; +} + +/** The budget for one measurement. An unknown start kind uses the looser of + * the cold and warm budgets, so an unlabelled run is never over-reported. */ +export function speedBudgetFor(adapter: SpeedBudgetAdapter, metric: SpeedBudgetMetric, + start: SpeedBudgetStart | null): SpeedBudget { + const budgets = LOCAL_SPEED_BUDGETS[adapter]; + if (metric === "restart_recovery") return budgets.restart_recovery; + 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) }; +} + +export type SpeedBudgetBreach = { + schema: "paperclip.run-performance-budget.v1"; + metric: SpeedBudgetMetric; + adapter: SpeedBudgetAdapter; + environment: "local"; + startKind: SpeedBudgetStart | "unknown"; + measuredMs: number; + budgetP50Ms: number; + budgetP95Ms: number; + /** Single runs are judged against p95. */ + percentile: "p95"; + baseline: string; + disposition?: string; +}; + +/** A breach when one measured run exceeds its p95 budget, else null. */ +export function evaluateSpeedBudget(input: { + adapter: SpeedBudgetAdapter | null; + environment: "local" | "remote"; + metric: SpeedBudgetMetric; + start: SpeedBudgetStart | null; + measuredMs: number; + disposition?: string; +}): 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); + if (input.measuredMs <= budget.p95Ms) return null; + return { + schema: "paperclip.run-performance-budget.v1", metric: input.metric, adapter: input.adapter, environment: "local", + 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 } : {}), + }; +} + +/** The restart-recovery measurement for one committed runner event, or null + * when the event is not a provider event (the baseline's set). */ +export function restartRecoveryMeasurementFor(input: { + event: { eventType: string; emittedAt?: string | null }; + startup: { phase: string; updatedAt: string }; + committedAtMs: number; + disposition: string; +}): { readyAtMs: number | null; emittedAtMs: number | null; committedAtMs: number; disposition: string } | null { + if (!RESTART_RECOVERY_PROVIDER_EVENT_TYPES.has(input.event.eventType)) return null; + const emitted = Date.parse(input.event.emittedAt ?? ""); + const ready = Date.parse(input.startup.updatedAt); + return { + readyAtMs: input.startup.phase === "ready" && Number.isFinite(ready) ? ready : null, + emittedAtMs: Number.isFinite(emitted) ? emitted : null, + committedAtMs: input.committedAtMs, + disposition: input.disposition, + }; +}