From ec9d9320f32b459b3b9f69a937073ef406a443ba Mon Sep 17 00:00:00 2001 From: Dotta Date: Thu, 15 Oct 2026 19:42:15 -0500 Subject: [PATCH] fix(observability): trace pre-dispatch controller waits Co-Authored-By: Paperclip --- doc/observability.md | 8 +++++ .../execution-preparation-timing.test.ts | 25 ++++++++++++-- .../services/execution-preparation-timing.ts | 1 + server/src/services/heartbeat.ts | 5 +-- .../heartbeat/runtime-execution.test.ts | 34 +++++++++++++++++++ .../services/heartbeat/runtime-execution.ts | 17 ++++++---- 6 files changed, 78 insertions(+), 12 deletions(-) diff --git a/doc/observability.md b/doc/observability.md index f50ba6c3141..6068bc13022 100644 --- a/doc/observability.md +++ b/doc/observability.md @@ -242,6 +242,14 @@ preparation, execution and settlement spans. They are controller work, not model turn time, and carry only the existing run hash. Nested phases overlap their containing backend or lifecycle intervals; do not add both durations. +Before provider dispatch, `execution.usage_startup` measures durable usage-recorder +initialization. `execution.mcp_handoff` measures native MCP server resolution or +legacy managed MCP configuration. `execution.dispatch_admission` measures the +final admission gate, including its nested `execution.budget_reservation` span. +The admission span ends after the provider promise is handed off, not after the +provider turn finishes. These spans use only the existing run hash. They close +gaps between preparation and execution; their nested intervals are not additive. + All preparation timing uses the endpoint gate, even when another library has registered a tracer provider. Tracer lookup, callback, status or ending failures cannot repeat work or replace its result or error. Automatic database children diff --git a/server/src/services/execution-preparation-timing.test.ts b/server/src/services/execution-preparation-timing.test.ts index f6467f5b54b..17bda2f4191 100644 --- a/server/src/services/execution-preparation-timing.test.ts +++ b/server/src/services/execution-preparation-timing.test.ts @@ -1,5 +1,5 @@ import { afterEach, beforeEach, expect, it, vi } from "vitest"; -import { createRunPreparationTiming } from "./execution-preparation-timing.js"; +import { createRunPreparationTiming, measureRunPhase, withRunExecutionTiming } from "./execution-preparation-timing.js"; const handles = vi.hoisted(() => ({ getTracer: vi.fn(), startActiveSpan: vi.fn() })); vi.mock("@opentelemetry/api", () => ({ @@ -37,6 +37,25 @@ it.each(["lookup", "start", "after_callback", "missing_callback", "rejected", "e it.each([undefined, "", " "])("does not consult tracing for endpoint %s", async endpoint => { vi.stubEnv("OTEL_EXPORTER_OTLP_ENDPOINT", endpoint); handles.getTracer.mockClear(); - await expect(createRunPreparationTiming("run", "native.controller")("backend_setup", () => 42)).resolves.toBe(42); - expect(handles.getTracer).not.toHaveBeenCalled(); + const warn = vi.spyOn(console, "warn"); + const error = vi.spyOn(console, "error"); + try { + await expect(createRunPreparationTiming("run", "native.controller")("backend_setup", () => 42)).resolves.toBe(42); + for (const phase of ["execution.usage_startup", "execution.mcp_handoff", "execution.dispatch_admission", "execution.budget_reservation"] as const) { + const result = {}; + const work = vi.fn(() => result); + await expect(withRunExecutionTiming("run", () => measureRunPhase(phase, work))).resolves.toBe(result); + expect(work).toHaveBeenCalledOnce(); + const original = new Error("original startup failure"); + const failed = vi.fn(() => { throw original; }); + await expect(withRunExecutionTiming("run", () => measureRunPhase(phase, failed))).rejects.toBe(original); + expect(failed).toHaveBeenCalledOnce(); + } + expect(handles.getTracer).not.toHaveBeenCalled(); + expect(warn).not.toHaveBeenCalled(); + expect(error).not.toHaveBeenCalled(); + } finally { + warn.mockRestore(); + error.mockRestore(); + } }); diff --git a/server/src/services/execution-preparation-timing.ts b/server/src/services/execution-preparation-timing.ts index 0df68c93e30..baa3858b33a 100644 --- a/server/src/services/execution-preparation-timing.ts +++ b/server/src/services/execution-preparation-timing.ts @@ -74,6 +74,7 @@ export type ExecutionPhase = | "execution.runtime" | "execution.workspace_ownership" | "execution.workspace_finalize" | "execution.workspace_restore" | "execution.workspace_commit" | "execution.workspace_ownership_wait" | "execution.realtime" | "execution.usage" | "execution.native_finalize" + | "execution.usage_startup" | "execution.mcp_handoff" | "execution.dispatch_admission" | "execution.budget_reservation" | "completion.run" | "completion.output" | "completion.terminal_write" | "completion.cost" | "completion.liveness" | "completion.summary" | "response.transaction" | "response.transaction.entry" | "response.transaction.commit" | "response.transaction.rollback" | "response.row_lock" | "response.write" | "response.resolve" | "response.authorize" | "response.add_comment" | "response.activity" | "response.publish"; diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 221ed1632f5..63bc20da0f8 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -2870,8 +2870,9 @@ export function heartbeatService( ) => { // Admission can wait for accounting locks. Finish that wait before the // final ownership gate, whose callback must enter the adapter directly. - const reservation = await reserveRunBudget(db, run.companyId, run.id, - readNonEmptyString(runLedgerScope.projectId), runLedgerScope, runOptions.nativeLeaseOwner); + const reservation = await measureRunPhase("execution.budget_reservation", () => + reserveRunBudget(db, run.companyId, run.id, + readNonEmptyString(runLedgerScope.projectId), runLedgerScope, runOptions.nativeLeaseOwner)); let entered = false; try { return await dispatchResolvedInteractionContinuationAfterAdmission((markDispatchStarted) => { diff --git a/server/src/services/heartbeat/runtime-execution.test.ts b/server/src/services/heartbeat/runtime-execution.test.ts index f124baeb711..f8d9ee53c02 100644 --- a/server/src/services/heartbeat/runtime-execution.test.ts +++ b/server/src/services/heartbeat/runtime-execution.test.ts @@ -246,6 +246,13 @@ describe.skipIf(!support.supported)("heartbeat runtime execution boundary", () = }]); await provider.forceFlush(); const spans = exporter.getFinishedSpans(); + const attempt = spans.find(span => span.name === "task.execution_attempt")!; + expect(attempt).toBeDefined(); + for (const name of ["execution.usage_startup", "execution.mcp_handoff", "execution.dispatch_admission"]) { + const startup = spans.filter(span => span.name === name); + expect(startup).toHaveLength(1); + expect(startup[0].parentSpanContext?.spanId).toBe(attempt.spanContext().spanId); + } const copyback = spans.filter(span => span.name === "execution.workspace_restore"); const commit = spans.filter(span => span.name === "execution.workspace_commit"); expect(copyback).toHaveLength(1); @@ -260,6 +267,33 @@ describe.skipIf(!support.supported)("heartbeat runtime execution boundary", () = } }); + it("ends admission timing while the admitted provider is still running", async () => { + const { input, run } = await fixture(true); + const endpoint = process.env.OTEL_EXPORTER_OTLP_ENDPOINT; + vi.stubEnv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://collector.test"); + const exporter = new InMemorySpanExporter(); + const provider = new TracerProvider({ spanProcessors: [new SimpleSpanProcessor({ exporter })] }); + expect(trace.setGlobalTracerProvider(provider)).toBe(true); + let release!: (result: Awaited>) => void; + const providerResult = new Promise>>(resolve => { release = resolve; }); + vi.mocked(executePaperclipNativeSession).mockReturnValueOnce(providerResult); + const running = withRunExecutionTiming(run.id, () => executeHeartbeatRuntime(db, input)); + try { + await vi.waitFor(() => { + expect(exporter.getFinishedSpans().filter(span => span.name === "execution.dispatch_admission")).toHaveLength(1); + }, { timeout: 5_000 }); + expect(executePaperclipNativeSession).toHaveBeenCalledOnce(); + expect(exporter.getFinishedSpans().find(span => span.name === "task.execution_attempt")).toBeUndefined(); + expect(await finalizations()).toHaveLength(0); + } finally { + release({ exitCode: 0, signal: null, timedOut: false }); + await running; + vi.stubEnv("OTEL_EXPORTER_OTLP_ENDPOINT", endpoint); + trace.disable(); + await provider.shutdown(); + } + }); + it.each(["codex", "grok_bot", "openai_dot"] as const)("retains user files only when %s can access the workspace after a dispatch failure", async provider => { const { input, run } = await fixture(true, provider); input.workspace.nativeWorkspaceSync = { diff --git a/server/src/services/heartbeat/runtime-execution.ts b/server/src/services/heartbeat/runtime-execution.ts index c2971935867..bdd459d45cc 100644 --- a/server/src/services/heartbeat/runtime-execution.ts +++ b/server/src/services/heartbeat/runtime-execution.ts @@ -231,6 +231,8 @@ export async function executeHeartbeatRuntime(db: Db, input: HeartbeatRuntimeExe const { nativeInstructionWorkingCopy, collectStoppedInstructions, nativeInstructionReservation, releaseInstructionCopy } = input.instructions; const { providerTraceCapture, nativeRunnerPreparationSpans, attestedQuestionResponseAtMs, attemptStartedAtMs, environmentAcquireStartedAtMs, environmentRealizeEndedAtMs } = input.trace; const { executionControl, executionPhaseContext, dispatchResolvedInteractionContinuationWithAtomicGate } = input.control; + const dispatchWithTiming: HeartbeatRuntimeExecutionInput["control"]["dispatchResolvedInteractionContinuationWithAtomicGate"] = dispatch => + measureRunPhase("execution.dispatch_admission", () => dispatchResolvedInteractionContinuationWithAtomicGate(dispatch)); const { identityRedactor, appendIdentityRedactedLog, onLog, onAdapterEvent, onAdapterMeta } = input.output; const { executionWorkspacesSvc, instanceSettings, envOrchestrator, upsertTaskSession, getRun, cancelRunInternal, enqueueWakeup, persistRunProcessMetadata, recordCurrentHeartbeatRunRuntimeProgress, dispatchPendingNativeStatusWakeups } = input.services; const localAgentJwtScope = @@ -481,7 +483,8 @@ export async function executeHeartbeatRuntime(db: Db, input: HeartbeatRuntimeExe adapterFinalizeOutcome = status; }; - const usageRecorder = await createRunUsageRecorder(db, { companyId: run.companyId, runId: run.id, adapterType: agent.adapterType }); + const usageRecorder = await measureRunPhase("execution.usage_startup", () => + createRunUsageRecorder(db, { companyId: run.companyId, runId: run.id, adapterType: agent.adapterType })); input.effects.onUsageCaptureReady(usageRecorder.persistFailure); let adapterResult: AdapterExecutionResult; const runGoalControlRequestId = readNonEmptyString( @@ -496,12 +499,12 @@ export async function executeHeartbeatRuntime(db: Db, input: HeartbeatRuntimeExe nativeExecution.runtimeContext.mcp.bindingId ? nativeExecution.runtimeContext.mcp.digest : null; - const nativeMcpServers = await buildPaperclipRuntimeMcpServers({ + const nativeMcpServers = await measureRunPhase("execution.mcp_handoff", () => buildPaperclipRuntimeMcpServers({ db, agent, runId: run.id, expectedAssignmentDigest: expectedNativeMcpDigest, - }); + })); if ("runtimeContext" in nativeExecution) { if (nativeMcpServers.length > 1) throw new Error( @@ -564,7 +567,7 @@ export async function executeHeartbeatRuntime(db: Db, input: HeartbeatRuntimeExe }), ); const guardedDispatch = - await dispatchResolvedInteractionContinuationWithAtomicGate( + await dispatchWithTiming( (markDispatchStarted) => { return executePaperclipNativeSession({ db, @@ -772,19 +775,19 @@ export async function executeHeartbeatRuntime(db: Db, input: HeartbeatRuntimeExe if (runtimeTools && runtimeToolDelivery === "invocation_context") { adapterContext.paperclipRuntimeTools = runtimeTools; } - const managedMcpConfig = await createManagedMcpRunConfig({ + const managedMcpConfig = await measureRunPhase("execution.mcp_handoff", () => createManagedMcpRunConfig({ db, agent, runId: run.id, config: runtimeConfig, projectId: issueRef?.projectId ?? null, issueId: issueRef?.id ?? null, - }); + })); if (managedMcpConfig) { adapterContext.paperclipManagedMcp = managedMcpConfig; } const guardedDispatch = - await dispatchResolvedInteractionContinuationWithAtomicGate( + await dispatchWithTiming( (markDispatchStarted) => { input.effects.onLegacyAdapterEntered(); return withAdapterExecutionPhase(executionPhaseContext, "adapter_execution", () => adapter.execute({