From e3029c632b26b68595adc1a704a387833b3cefcb Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Tue, 13 Oct 2026 10:16:10 -0500 Subject: [PATCH] feat: trace controller preparation before agent dispatch (#16178) Trace eleven controller preparation operations with the existing hashed run correlation and preserve errors without exporting exception text. Verify asynchronous child-span parenting with a real OpenTelemetry SDK and context manager. Full exact-head CI and E2E passed after one documented fixture-startup retry; Greptile Apex 5/5. Co-Authored-By: Paperclip --- doc/observability.md | 18 +++++++ server/package.json | 2 + server/src/services/heartbeat.ts | 42 ++++++++------- .../preparation-timing-context.test.ts | 53 +++++++++++++++++++ .../heartbeat/preparation-timing.test.ts | 49 +++++++++++++++++ .../services/heartbeat/preparation-timing.ts | 25 +++++++++ 6 files changed, 169 insertions(+), 20 deletions(-) create mode 100644 server/src/services/heartbeat/preparation-timing-context.test.ts create mode 100644 server/src/services/heartbeat/preparation-timing.test.ts create mode 100644 server/src/services/heartbeat/preparation-timing.ts diff --git a/doc/observability.md b/doc/observability.md index 4a331900469..626e84dced1 100644 --- a/doc/observability.md +++ b/doc/observability.md @@ -161,6 +161,24 @@ This document also holds three local instrumentation contracts: native runner traces, sandbox startup traces, and sandbox duplex transport instrumentation. Those sections follow below. +## Controller preparation phases + +With an operator OTLP endpoint, `paperclip.heartbeat-preparation` emits active +`heartbeat.prepare.` spans around environment selection, execution +configuration, session preparation, workspace preparation, GitHub environment +and launcher preparation, run output initialization, workspace validation, +runtime services, instruction files, and runtime selection. Database and HTTP +instrumentation can attach child spans during each awaited operation. + +These spans exist before provider dispatch, including setup failures. Each has +only the existing `paperclip.task.run.run_id` attribute: the first 12 hexadecimal +characters of the run UUID's SHA-256 hash. Failed operations set error status +without recording exception messages, paths, inputs, or returned content. +Search by that hash without a root-name filter to include these preparation +traces; the provider lifecycle trace is created later and can have a different +trace ID. Without an endpoint, tracing remains a no-op. These spans add no +first-party Telemetry events or database run-log writes. + ## Native Runner Trace Spans Paperclip Runner task runs emit a single foldable OpenTelemetry trace. This is diff --git a/server/package.json b/server/package.json index 82088f5548b..0ce4ac72f0f 100644 --- a/server/package.json +++ b/server/package.json @@ -125,6 +125,8 @@ "acpx" ], "devDependencies": { + "@opentelemetry/context-async-hooks": "2.10.0", + "@opentelemetry/sdk-trace": "2.10.0", "@paperclipai/paperclip-runner": "workspace:*", "@types/compression": "^1.8.1", "@types/express": "^5.0.0", diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 4a6d6a32611..fa3b28245a8 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -1,3 +1,4 @@ +import { createHeartbeatPreparationTiming } from "./heartbeat/preparation-timing.js"; import { createHeartbeatInstructionCopyRelease, createHeartbeatInstructionWorkingCopy } from "./heartbeat/instruction-working-copy.js"; import { prepareHeartbeatRuntimeServiceIntents, startHeartbeatRuntimeServices, recordHeartbeatAdapterRuntimeServices } from "./heartbeat/runtime-services.js"; import { prepareHeartbeatExecutionConfiguration } from "./heartbeat/execution-configuration.js"; @@ -1918,6 +1919,7 @@ export function heartbeatService( if (run.runtimeMode === "legacy" && run.controllerBootId && run.controllerBootId !== legacyControllerBootId) return; activeRunExecutions.add(run.id); + const timePreparation = createHeartbeatPreparationTiming(run.id); const executionControl = createAdapterExecutionControl(); // This coarse scope also covers host finalization after the adapter returns. // Nested scopes refine the label without making any termination claim. @@ -2613,7 +2615,7 @@ export function heartbeatService( materializeNativeChatWorktree, explicitTaskDirectory, useIsolatedTaskDirectory, - } = await selectHeartbeatEnvironment(db, { + } = await timePreparation("environment_selection", () => selectHeartbeatEnvironment(db, { run, agent, task: { issueRef, issueContext, issueId, context, executionProjectId, workspaceProjectContext, boundSourceWorkspace, hasSelectedWorkspaceSource: Boolean(selectedWorkspaceSource), taskWorkspaceIntent: taskWorkspaceIntentRow?.intent ?? null, responsibleUserId }, @@ -2623,9 +2625,9 @@ export function heartbeatService( }, config: { config, sourceWorkspaceConfig, issueAssigneeOverrides }, services: { executionWorkspacesSvc, environmentsSvc, instanceSettings, findSharedWorkspaceHolder }, - }); + })); let nativeChatExpectedCwd = selectedNativeChatExpectedCwd; - const executionConfiguration = await prepareHeartbeatExecutionConfiguration(db, { + const executionConfiguration = await timePreparation("execution_configuration", () => prepareHeartbeatExecutionConfiguration(db, { run, agent, task: { issueId, context, responsibleUserId, acceptedPlanContinuationWake, acceptedPlanWakeRoutingDecision, taskProjectContext, routineEnvContext }, config: { sourceWorkspaceConfig, aiBinding }, @@ -2635,7 +2637,7 @@ export function heartbeatService( onFailureSecretsResolved: (readSecrets) => { readFailureReportSecrets = readSecrets; }, onManagedAiRuntimeAcquired: (runtime) => { managedAiRuntime = runtime; }, onIdentityRedactorPrepared: (redactor) => { identityRedactor = redactor; }, - }); + })); if (!executionConfiguration.prepared) return; const { mergedConfig, configSnapshot, runScopedMentionedSkillKeys, githubSelection, useHostGitHub, @@ -2649,7 +2651,7 @@ export function heartbeatService( ...identityRedactor.values, ...collectRunFailureSecretValues(runtimeConfig.env, secretKeys), ]; - const preparedSession = await prepareHeartbeatSession(db, { + const preparedSession = await timePreparation("session", () => prepareHeartbeatSession(db, { run, agent, task: { context, issueRef, issueContext, taskProjectContext, workspaceProjectContext, wakeCommentId, acceptedPlanContinuationWake, acceptedPlanWakeRoutingDecision }, session: { taskSession, taskSessionDecodedParams, explicitResumeSessionParams, explicitResumeSessionDisplayId, sessionCodec }, @@ -2663,7 +2665,7 @@ export function heartbeatService( }, environment: { environmentResolution, selectedEnvironmentId, selectedEnvironmentForConfig }, getLatestAgentConfigRevision, - }); + })); const { sessionConfigMetadata, configuredModel, resetTaskSession, sessionResetReason, taskSessionForRun, getFreshSessionHandoff, previousSessionParams } = preparedSession; const { resolvedWorkspace, @@ -2685,7 +2687,7 @@ export function heartbeatService( workspaceRealization, executionTarget, workspaceRestoreSource, - } = await prepareHeartbeatWorkspace(db, { + } = await timePreparation("workspace", () => prepareHeartbeatWorkspace(db, { run, agent, controllerLease, nativeRunnerPreparationSpans, task: { issueRef, @@ -2737,7 +2739,7 @@ export function heartbeatService( resolveReusedGitWorkspaceAnchor, appendRunEvent, }, - }); + })); const preparedInstructions = instructionWorkingCopy = createHeartbeatInstructionWorkingCopy({ db, instructionCopies, readRun: () => run!, agent, context, session: { taskSession, taskSessionForRun, sessionConfigMetadata }, @@ -2901,7 +2903,7 @@ export function heartbeatService( } else { delete context.paperclipScratch; } - const gitExecutionEnv = await prepareGitHubExecutionEnvironment({ + const gitExecutionEnv = await timePreparation("github_environment", () => prepareGitHubExecutionEnvironment({ target: executionTarget, cwd: executionWorkspace.cwd, env: Object.fromEntries( @@ -2915,12 +2917,12 @@ export function heartbeatService( networkAccess: trustPreset.kind === "standard" && process.env.PAPERCLIP_RUNNER_NETWORK_ACCESS !== "disabled", - }); + })); runtimeConfig = { ...runtimeConfig, env: gitExecutionEnv }; for (const key of MANAGED_GITHUB_TOKEN_KEYS) secretKeys.add(key); context.githubAuthenticationMode = useHostGitHub ? "host" : "managed"; if (!useHostGitHub) { - const githubLaunchers = await prepareHeartbeatGitHubLaunchers({ + const githubLaunchers = await timePreparation("github_launchers", () => prepareHeartbeatGitHubLaunchers({ native: agent.adapterType === "paperclip_runner", githubConfigured: githubSelection.configured, agentId: agent.id, @@ -2936,7 +2938,7 @@ export function heartbeatService( responsibleUserId: responsibleUserId ?? "", scope: "github_credentials", })?.token ?? "", - }); + })); githubLauncherLocation = githubLaunchers.cleanupLocation; runtimeConfig = { ...runtimeConfig, env: githubLaunchers.env }; secretKeys.add("PAPERCLIP_GITHUB_BROKER_TOKEN"); @@ -3167,7 +3169,7 @@ export function heartbeatService( }); const { currentUserRedactionOptions, appendIdentityRedactedLog, onLog, onAdapterMeta, onAdapterEvent } = - await runOutput.begin(currentRun); + await timePreparation("run_output", () => runOutput.begin(currentRun)); if (runScopedMentionedSkillKeys.length > 0) { await onLog( "stdout", @@ -3178,7 +3180,7 @@ export function heartbeatService( const logEntry = formatRuntimeWorkspaceWarningLog(warning); await onLog(logEntry.stream, logEntry.chunk); } - await assertGitSensitiveAdapterWorkspaceValid({ + await timePreparation("workspace_validation", () => assertGitSensitiveAdapterWorkspaceValid({ adapterType: agent.adapterType, agentId: agent.id, issue: issueRef @@ -3195,17 +3197,17 @@ export function heartbeatService( executionTarget, environmentDriver: selectedEnvironment.driver, leaseMetadata: activeEnvironmentLease.lease.metadata, - }); + })); const adapterEnv = Object.fromEntries( Object.entries(parseObject(runtimeConfig.env)).filter( (entry): entry is [string, string] => typeof entry[0] === "string" && typeof entry[1] === "string", ), ); - const runtimeServices = await startHeartbeatRuntimeServices(db, { + const runtimeServices = await timePreparation("runtime_services", () => startHeartbeatRuntimeServices(db, { run, agent, issueId, issueRef, executionWorkspace, context, issuesSvc, onLog, persistedExecutionWorkspace, hostExecutionWorkspaceConfig, skipHostProvision, adapterEnv, workspaceOperationRecorder, - }); + })); const adapter = getServerAdapter(agent.adapterType); const durableGoalControlRun = readNonEmptyString(context.goalControlRequestId) !== null || @@ -3251,8 +3253,8 @@ export function heartbeatService( const hasInstructionFilesystem = nativeRuntimeResolution.kind !== "native" ? adapter.supportsInstructionsBundle === true : !["claude_managed_agents_api", "aws_agentcore_harness_api", "openai_dot_mcp"].includes(nativeRuntimeResolution.profile.backend); - runtimeConfig = await preparedInstructions.prepare({ runtimeConfig, nativeRuntimeResolution, hasInstructionFilesystem }); - const runtimeSelection = await selectHeartbeatRuntime(db, { + runtimeConfig = await timePreparation("instructions", () => preparedInstructions.prepare({ runtimeConfig, nativeRuntimeResolution, hasInstructionFilesystem })); + const runtimeSelection = await timePreparation("runtime_selection", () => selectHeartbeatRuntime(db, { run, agent, nativeRuntimeResolution, @@ -3319,7 +3321,7 @@ export function heartbeatService( nativeOwnershipHeld = true; }, stopControllerLease: () => controllerLease.stop(), - }); + })); if (!runtimeSelection.selected) return; const { nativeExecution, nativeRunnerInstanceId, getNativeFreshSessionHandoff } = runtimeSelection; nativeWorkspaceSync = runtimeSelection.nativeWorkspaceSync; diff --git a/server/src/services/heartbeat/preparation-timing-context.test.ts b/server/src/services/heartbeat/preparation-timing-context.test.ts new file mode 100644 index 00000000000..8ae9daf918c --- /dev/null +++ b/server/src/services/heartbeat/preparation-timing-context.test.ts @@ -0,0 +1,53 @@ +import { createHash } from "node:crypto"; +import { setImmediate } from "node:timers/promises"; +import { context, trace } from "@opentelemetry/api"; +import { AsyncLocalStorageContextManager } from "@opentelemetry/context-async-hooks"; +import { TracerProvider, InMemorySpanExporter, SimpleSpanProcessor } from "@opentelemetry/sdk-trace"; +import { expect, it } from "vitest"; +import { createHeartbeatPreparationTiming } from "./preparation-timing.js"; + +it("parents child spans to their own preparation phase across overlapping awaited runs", async () => { + const manager = new AsyncLocalStorageContextManager().enable(); + const exporter = new InMemorySpanExporter(); + const provider = new TracerProvider({ spanProcessors: [new SimpleSpanProcessor({ exporter })] }); + expect(context.setGlobalContextManager(manager)).toBe(true); + expect(trace.setGlobalTracerProvider(provider)).toBe(true); + try { + const gates = [0, 1].map(() => { + let resolve!: () => void; + const promise = new Promise(release => { resolve = release; }); + return { promise, resolve }; + }); + const runs = ["first-run", "second-run"]; + const pending = runs.map((run, index) => createHeartbeatPreparationTiming(run)("workspace", async () => { + await gates[index].promise; + await setImmediate(); + const child = trace.getTracer("preparation-context-test").startSpan(`child.${run}`); + await setImmediate(); + child.end(); + })); + expect(trace.getActiveSpan()).toBeUndefined(); + gates[1].resolve(); + gates[0].resolve(); + await Promise.all(pending); + await provider.forceFlush(); + const spans = exporter.getFinishedSpans(); + expect(spans).toHaveLength(4); + const phases = runs.map(run => spans.find(span => span.name === "heartbeat.prepare.workspace" + && span.attributes["paperclip.task.run.run_id"] === createHash("sha256").update(run).digest("hex").slice(0, 12))!); + expect(phases.every(Boolean)).toBe(true); + expect(phases[0].spanContext().traceId).not.toBe(phases[1].spanContext().traceId); + for (const [index, run] of runs.entries()) { + const child = spans.find(span => span.name === `child.${run}`)!; + expect(child.parentSpanContext?.spanId).toBe(phases[index].spanContext().spanId); + expect(child.spanContext().traceId).toBe(phases[index].spanContext().traceId); + expect(phases[index].parentSpanContext).toBeUndefined(); + } + expect(trace.getActiveSpan()).toBeUndefined(); + } finally { + trace.disable(); + context.disable(); + manager.disable(); + await provider.shutdown(); + } +}); diff --git a/server/src/services/heartbeat/preparation-timing.test.ts b/server/src/services/heartbeat/preparation-timing.test.ts new file mode 100644 index 00000000000..d6b1005f527 --- /dev/null +++ b/server/src/services/heartbeat/preparation-timing.test.ts @@ -0,0 +1,49 @@ +import { createHash } from "node:crypto"; +import { beforeEach, describe, expect, it, vi } from "vitest"; +import { createHeartbeatPreparationTiming } from "./preparation-timing.js"; + +const handles = vi.hoisted(() => ({ spans: [] as Array<{ setStatus: ReturnType; end: ReturnType }> })); +const startActiveSpan = vi.hoisted(() => vi.fn((_name, _options, callback) => { + const span = { setStatus: vi.fn(), end: vi.fn() }; + handles.spans.push(span); + return callback(span); +})); +vi.mock("@opentelemetry/api", () => ({ + SpanStatusCode: { ERROR: 2 }, + trace: { getTracer: () => ({ startActiveSpan }) }, +})); + +describe("heartbeat preparation timing", () => { + beforeEach(() => { vi.clearAllMocks(); handles.spans.length = 0; }); + + it("correlates preparation with the existing run hash without recording content", async () => { + const runId = "private-run-id"; + const time = createHeartbeatPreparationTiming(runId); + const result = { secret: "private result" }; + await expect(time("instructions", async () => result)).resolves.toBe(result); + expect(startActiveSpan).toHaveBeenCalledExactlyOnceWith("heartbeat.prepare.instructions", { + attributes: { "paperclip.task.run.run_id": createHash("sha256").update(runId).digest("hex").slice(0, 12) }, + }, expect.any(Function)); + expect(handles.spans[0].end).toHaveBeenCalledOnce(); + expect(handles.spans[0].setStatus).not.toHaveBeenCalled(); + }); + + it("ends a failed setup phase and preserves its exact error without recording its text", async () => { + const error = new Error("private filesystem or credential detail"); + await expect(createHeartbeatPreparationTiming("run")("workspace", async () => { throw error; })).rejects.toBe(error); + expect(handles.spans[0].setStatus).toHaveBeenCalledExactlyOnceWith({ code: 2 }); + expect(handles.spans[0].end).toHaveBeenCalledOnce(); + }); + + it("keeps overlapping operations open until their own work settles", async () => { + const time = createHeartbeatPreparationTiming("run"); + let release!: () => void; + const waiting = time("workspace", () => new Promise(resolve => { release = resolve; })); + await time("runtime_services", async () => undefined); + expect(handles.spans[0].end).not.toHaveBeenCalled(); + expect(handles.spans[1].end).toHaveBeenCalledOnce(); + release(); + await waiting; + expect(handles.spans[0].end).toHaveBeenCalledOnce(); + }); +}); diff --git a/server/src/services/heartbeat/preparation-timing.ts b/server/src/services/heartbeat/preparation-timing.ts new file mode 100644 index 00000000000..2d089fe4375 --- /dev/null +++ b/server/src/services/heartbeat/preparation-timing.ts @@ -0,0 +1,25 @@ +import { createHash } from "node:crypto"; +import { SpanStatusCode, trace } from "@opentelemetry/api"; + +type PreparationPhase = "environment_selection" | "execution_configuration" | "session" + | "workspace" | "github_environment" | "github_launchers" | "run_output" + | "workspace_validation" | "runtime_services" | "instructions" | "runtime_selection"; + +export function createHeartbeatPreparationTiming(runId: string) { + const runIdHash = createHash("sha256").update(runId).digest("hex").slice(0, 12); + return (phase: PreparationPhase, work: () => Promise): Promise => + trace.getTracer("paperclip.heartbeat-preparation").startActiveSpan( + `heartbeat.prepare.${phase}`, + { attributes: { "paperclip.task.run.run_id": runIdHash } }, + async (span) => { + try { + return await work(); + } catch (error) { + span.setStatus({ code: SpanStatusCode.ERROR }); + throw error; + } finally { + span.end(); + } + }, + ); +}