diff --git a/server/src/__tests__/execution-preparation-phases.test.ts b/server/src/__tests__/execution-preparation-phases.test.ts new file mode 100644 index 00000000000..3888f48b8a6 --- /dev/null +++ b/server/src/__tests__/execution-preparation-phases.test.ts @@ -0,0 +1,63 @@ +import { randomUUID } from "node:crypto"; +import { setTimeout as delay } from "node:timers/promises"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { createRunPreparationTiming, drainRunPreparationPhases, measureRunPhase, withRunExecutionTiming } from "../services/execution-preparation-timing.js"; + +describe("run preparation phases for the run log", () => { + afterEach(() => { vi.unstubAllEnvs(); }); + it("records slow named phases per run, marks failures, and forgets them once drained", async () => { + const runId = randomUUID(); + const time = createRunPreparationTiming(runId); + await time("workspace_provision", () => delay(70)); + await time("workspace_binding", async () => undefined); + await expect(time("runner_binary", async () => { await delay(60); throw new Error("upload failed"); })).rejects.toThrow("upload failed"); + await Promise.resolve(); + const phases = drainRunPreparationPhases(runId); + expect(phases.map(phase => [phase.name, phase.outcome])).toEqual([ + ["heartbeat.prepare.workspace_provision", "ok"], + ["heartbeat.prepare.runner_binary", "failed"], + ]); + expect(phases[0]!.endedAtMs - phases[0]!.startedAtMs).toBeGreaterThanOrEqual(50); + expect(drainRunPreparationPhases(runId)).toEqual([]); + }); + + it("records nested run phases inside a preparation phase without a tracer", async () => { + const runId = randomUUID(); + await createRunPreparationTiming(runId)("runtime_selection", async () => { + await measureRunPhase("runtime_selection.provider_discovery", () => delay(60)); + await measureRunPhase("runtime_selection.skill_assets", async () => undefined); + }); + expect(drainRunPreparationPhases(runId).map(phase => phase.name)).toEqual([ + "runtime_selection.provider_discovery", "heartbeat.prepare.runtime_selection", + ]); + }); + + it("never records an execution attempt, so a resumed run's prepare holds only its own preparation", async () => { + const runId = randomUUID(); + await createRunPreparationTiming(runId)("workspace", () => delay(60)); + expect(drainRunPreparationPhases(runId)).toHaveLength(1); + await withRunExecutionTiming(runId, async () => { + await delay(60); + await measureRunPhase("workspace_sync.transfer", () => delay(60)); + }); + expect(drainRunPreparationPhases(runId)).toEqual([]); + }); + + it("records nested preparation steps when an OTLP endpoint is set too", async () => { + vi.stubEnv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://127.0.0.1:9"); + const runId = randomUUID(); + await createRunPreparationTiming(runId)("runtime_selection", async () => { + await measureRunPhase("runtime_selection.workspace_sync", () => delay(60)); + }); + expect(drainRunPreparationPhases(runId).map(phase => phase.name)).toEqual([ + "runtime_selection.workspace_sync", "heartbeat.prepare.runtime_selection", + ]); + }); + + it("keeps runs apart", async () => { + const [first, second] = [randomUUID(), randomUUID()]; + await createRunPreparationTiming(first)("instructions", () => delay(60)); + expect(drainRunPreparationPhases(second)).toEqual([]); + expect(drainRunPreparationPhases(first)).toHaveLength(1); + }); +}); diff --git a/server/src/services/execution-preparation-timing.ts b/server/src/services/execution-preparation-timing.ts index af5a9bebcf4..89a9790fe45 100644 --- a/server/src/services/execution-preparation-timing.ts +++ b/server/src/services/execution-preparation-timing.ts @@ -73,7 +73,7 @@ export type ExecutionPhase = | "workspace.checkout.cache_lookup" | "workspace.checkout.cache_publish" | "runtime_selection.review" | "runtime_selection.contract_read" | "runtime_selection.contract_ensure" | "runtime_selection.resume" | "runtime_selection.recovery_profile" | "runtime_selection.profile" | "runtime_selection.interactions" - | "runtime_selection.provider_discovery" | "runtime_selection.bootstrap" | "runtime_selection.run_write" | "runtime_selection.finalization_write" + | "runtime_selection.provider_discovery" | "runtime_selection.codex_cli_version" | "runtime_selection.runtime_context" | "runtime_selection.bootstrap" | "runtime_selection.run_write" | "runtime_selection.finalization_write" | "runtime_selection.transaction" | "runtime_selection.transaction.entry" | "runtime_selection.transaction.commit" | "runtime_selection.transaction.rollback" | "runtime_selection.row_lock" | "runtime_selection.workspace_sync" | "runtime_selection.skill_assets" | "runtime_context.instructions" | "runtime_context.skills" | "runtime_context.mcp" @@ -147,7 +147,55 @@ export function isRunPhaseActive(phase: ExecutionPhase): boolean { return endpointEnabled() && owned?.phase === phase && !owned.ended; } +/** + * Named preparation phases per run for the run log, so a slow step before + * dispatch is visible without an OTLP endpoint. Bounded: phases under 50 ms are + * left out, at most 200 per run, and at most 256 runs are held until drained. + */ +export type RecordedPreparationPhase = { name: string; startedAtMs: number; endedAtMs: number; outcome: "ok" | "failed" }; +const recordedPreparationPhases = new Map(); +const MIN_RECORDED_PHASE_MS = 50; +const MAX_RECORDED_PHASES_PER_RUN = 200; +const MAX_RECORDED_RUNS = 256; + +function isPreparationPhaseName(name: string): boolean { + return name.startsWith("heartbeat.prepare."); +} + +function recordPreparationPhase(runId: string, phase: RecordedPreparationPhase) { + if (phase.endedAtMs - phase.startedAtMs < MIN_RECORDED_PHASE_MS) return; + let phases = recordedPreparationPhases.get(runId); + if (!phases) { + if (recordedPreparationPhases.size >= MAX_RECORDED_RUNS) { + const oldest = recordedPreparationPhases.keys().next().value; + if (oldest !== undefined) recordedPreparationPhases.delete(oldest); + } + phases = []; + recordedPreparationPhases.set(runId, phases); + } + if (phases.length < MAX_RECORDED_PHASES_PER_RUN) phases.push(phase); +} + +/** Returns and forgets the recorded preparation phases of a run. */ +export function drainRunPreparationPhases(runId: string): RecordedPreparationPhase[] { + const phases = recordedPreparationPhases.get(runId) ?? []; + recordedPreparationPhases.delete(runId); + return phases; +} + function measure(runId: string, name: string, work: () => T | Promise, independent = false, counts: { fileCount?: number; bytes?: number } = {}, activeParent?: Context): Promise { + const startedAtMs = Date.now(); + const operation = measureTraced(runId, name, work, independent, counts, activeParent); + // Only pre-dispatch preparation reaches the run log here; execution attempts + // and runtime timings belong to the turn, not to the next attempt's prepare. + if (isPreparationPhaseName(name)) operation.then( + () => recordPreparationPhase(runId, { name, startedAtMs, endedAtMs: Date.now(), outcome: "ok" }), + () => recordPreparationPhase(runId, { name, startedAtMs, endedAtMs: Date.now(), outcome: "failed" }), + ); + return operation; +} + +function measureTraced(runId: string, name: string, work: () => T | Promise, independent = false, counts: { fileCount?: number; bytes?: number } = {}, activeParent?: Context): Promise { let operation: Promise | undefined; const run = (span?: { setStatus(status: { code: number }): void; end(): void }, parent?: Context) => { if (operation) return operation; @@ -170,7 +218,15 @@ function measure(runId: string, name: string, work: () => T | Promise, ind })); return operation; }; - if (!endpointEnabled()) return withDatabaseOperation(name, async () => work()); + if (!endpointEnabled()) { + // No tracer: still scope the run so nested measureRunPhase steps reach the run log. + const previous = ownership.getStore(); + const owned: RunScope = { runId, phase: name, parent: undefined, ended: false, + details: !independent && previous?.runId === runId ? previous.details : new Map() }; + return withDatabaseOperation(name, () => ownership.run(owned, async () => { + try { return await work(); } finally { owned.ended = true; } + })); + } try { const parent = independent ? undefined : activeParent ?? getRunExecutionTraceContext(runId); const trigger = parent ? undefined : trace.getActiveSpan()?.spanContext(); @@ -211,13 +267,25 @@ export function withRunResponsePresentationTiming(runId: string, work: () => export function measureRunPhase(phase: ExecutionPhase, work: () => T | Promise, counts?: { fileCount?: number; bytes?: number }): Promise { return withDatabaseOperation(phase, async () => { const owned = ownership.getStore(); - if (!endpointEnabled() || !owned || owned.ended) return work(); + if (!owned || owned.ended) return work(); const count = owned.details.get(phase) ?? 0; if (count >= 16) return work(); owned.details.set(phase, count + 1); - // A startup step can be more specific than the owning run phase. Keep that - // causal parent while the child publishes its own context to plugin workers. - return measure(owned.runId, phase, work, false, counts, getActiveStepContext()?.parentContext as Context | undefined); + // A step inside a preparation phase reaches the run log with or without a tracer. + const recordedRunId = isPreparationPhaseName(owned.phase) ? owned.runId : null; + const startedAtMs = Date.now(); + let outcome: RecordedPreparationPhase["outcome"] = "failed"; + try { + // A startup step can be more specific than the owning run phase. Keep that + // causal parent while the child publishes its own context to plugin workers. + const value = endpointEnabled() + ? await measure(owned.runId, phase, work, false, counts, getActiveStepContext()?.parentContext as Context | undefined) + : await work(); + outcome = "ok"; + return value; + } finally { + if (recordedRunId) recordPreparationPhase(recordedRunId, { name: phase, startedAtMs, endedAtMs: Date.now(), outcome }); + } }); } diff --git a/server/src/services/heartbeat/runtime-execution.ts b/server/src/services/heartbeat/runtime-execution.ts index 88223c355cc..86f529253cd 100644 --- a/server/src/services/heartbeat/runtime-execution.ts +++ b/server/src/services/heartbeat/runtime-execution.ts @@ -14,7 +14,7 @@ import { environmentService } from "../environments.js"; import { writeExecutionControl } from "../execution-control-notifications.js"; import { inspectPluginWorkspaceForFinalization } from "../native-runtime/native-plugin-workspace-finalization.js"; import { AsyncLocalStorage } from "node:async_hooks"; -import { measureRunPhase } from "../execution-preparation-timing.js"; +import { drainRunPreparationPhases, measureRunPhase } from "../execution-preparation-timing.js"; import { parseNativeSessionGoalControl } from "./scheduling.js"; import { isHeartbeatRunTerminalStatus } from "./run-lifecycle.js"; import { @@ -601,6 +601,12 @@ export async function executeHeartbeatRuntime(db: Db, input: HeartbeatRuntimeExe }); if (wakeIngressSpan) nativeRunnerPreparationSpans.unshift(wakeIngressSpan); + // Named preparation phases (workspace, configuration, runner staging) + // that ran before dispatch, as run-log spans under task.prepare. + nativeRunnerPreparationSpans.push(...drainRunPreparationPhases(run.id).map(phase => ({ + name: phase.name, startedAtMs: phase.startedAtMs, endedAtMs: phase.endedAtMs, + ...(phase.outcome === "failed" ? { outcome: "failed" as const } : {}), + }))); nativeRunnerPreparationSpans.push( ...buildNativeHeartbeatPreparationSpans({ runCreatedAtMs, diff --git a/server/src/services/heartbeat/runtime-selection.ts b/server/src/services/heartbeat/runtime-selection.ts index da5dc8269b9..9d0e54afe62 100644 --- a/server/src/services/heartbeat/runtime-selection.ts +++ b/server/src/services/heartbeat/runtime-selection.ts @@ -733,14 +733,14 @@ export async function selectHeartbeatRuntime(db: Db, input: HeartbeatRuntimeSele ? await museRunnerBroker(db).snapshot(agent.companyId, agent.id, String(parseObject(agent.adapterConfig).museBindingId ?? "")) : undefined; const dotBinding = ["openai_dot_mcp", "grok_bot_mcp"].includes(nativeRuntimeResolution.profile.backend) ? await dotRunnerBroker(db).snapshot(agent.companyId, agent.id, String(parseObject(agent.adapterConfig)[nativeRuntimeResolution.profile.backend === "grok_bot_mcp" ? "externalBindingId" : "dotBindingId"] ?? "")) : undefined; - const nativeRuntimeContext = await buildNativeRuntimeContext({ + const nativeRuntimeContext = await measureRunPhase("runtime_selection.runtime_context", () => buildNativeRuntimeContext({ db, agent, runId: run.id, runtimeConfig, runtimeSkillEntries, instructionWorkingCopy: instructionCopy ? { rootPath: instructionCopy.executionRoot, entryPath: instructionCopy.entryFile, ...(isAgentDirectoryCopy(instructionCopy) ? { kind: "agent_files" as const } : {}) } : undefined, - }); + })); getNativeFreshSessionHandoff = nativeReviewRequest ? undefined : getFreshSessionHandoff; const nativeProviderConfig = nativeRuntimeResolution.profile.backend === "cursor_cloud_api" || nativeRuntimeResolution.profile.backend === "codex_app_server" @@ -761,12 +761,12 @@ export async function selectHeartbeatRuntime(db: Db, input: HeartbeatRuntimeSele adapterConfig: nativeProviderConfig, managedProfile, agentCoreProfile, dotBinding, museBinding, }); const codexCliVersion = agent.adapterType === "paperclip_runner" && requestedNativeProvider.provider === "codex" - ? await readRemoteCodexModelCliVersion({ + ? await measureRunPhase("runtime_selection.codex_cli_version", () => readRemoteCodexModelCliVersion({ model: requestedNativeProvider.model, target: executionTarget, remoteCodexPath: runtimeEnv.PAPERCLIP_RUNNER_REMOTE_CODEX_PATH, remoteCodexNpmSpec: runtimeEnv.PAPERCLIP_RUNNER_REMOTE_CODEX_NPM_SPEC, - }) : null; + })) : null; const nativeProvider = codexCliVersion ? resolvePaperclipRunnerNativeProviderInput({ companyId: agent.companyId,