mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-31 06:04:11 +01:00
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 <noreply@paperclip.ing>
This commit is contained in:
1 parent
0b0e53029f
commit
e3029c632b
6 files changed
+169
-20
No files matched your search
@@ -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.<phase>` 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
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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<void>(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();
|
||||
}
|
||||
});
|
||||
@@ -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<typeof vi.fn>; end: ReturnType<typeof vi.fn> }> }));
|
||||
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<void>(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();
|
||||
});
|
||||
});
|
||||
@@ -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 <T>(phase: PreparationPhase, work: () => Promise<T>): Promise<T> =>
|
||||
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();
|
||||
}
|
||||
},
|
||||
);
|
||||
}
|
||||
Reference in new issue
Block a user