Add shared native turn attribution boundaries (#18213)

## Thinking Path

> - Paperclip manages work performed by AI agents.
> - Native execution crosses the plugin worker, Runner, sidecar, and
durable event boundaries.
> - Existing outer spans leave repeated preparation and admission work
unattributed.
> - Optimizations need a common baseline with accurate ownership and
event meanings.
> - This change extends the existing opt-in tracing contract with
bounded timing receipts.
> - It preserves execution policy and keeps provider internals opaque
where the API provides no timing proof.

## Linked Issues or Issue Description

**What existing behavior does this improve?**
Attribution of native task preparation, retained/cold Runner handoff,
and completion-to-text work.

**Current behavior**
Worker spans use a flat provider contract. Workspace calls lack a
consistent dispatch envelope. Submitted-to-started timing is named as a
provider queue even though it includes other work. The transport does
not expose a shared breakdown of its waits.

**Proposed improvement**
Add invocation-scoped nesting, bounded numeric remote receipts,
current-run transport timing, durable receipt intervals, and distinct
completion/text milestones. Reuse the existing exporter. No latency
optimization is included.

Related public work: #17842 consolidates existing startup tracing tests;
it does not add this runtime coverage.

## What Changed

- Extend the optional SDK tracer with nested async scopes and completed
timing receipts. Keep the host responsible for parent identity,
capability checks, attributes, clocks, depth, and count limits.
- Add dispatch and bounded host-read timings for environment and
workspace providers.
- Resolve transport parents on each call so retained sessions do not
retain an earlier run's trace.
- Split cold/resume/attach/prepare/registration/command/started-event
waits. Add opt-in numeric ACPX Runner/sidecar status and prompt-start
receipts.
- Distinguish first completion request, validated tool receipt, final
result commit, first nonempty text, and event persistence. Rename the
submitted-to-started interval to match its meaning.
- Link completed earlier remote lifecycles to their host-owned observer
rather than placing them under a later polling request. Preserve
sampling, timestamp limits and budgets.
- Document the versioned coverage map, causal clock limits, and baseline
acceptance requirements.

## Verification

- Current head: GitHub CI is fully green. Standard Greptile and Apex are
both 5/5, with no unresolved review threads. Full-suite validation was
performed in CI; follow-up local validation was limited to affected
packages and tests.

- Review follow-up: early rejection, missing provider capability and
startup-parent regressions were reproduced and fixed. The affected
worker/SDK/timing set passes 128 tests; 28 timing/context tests, SDK
typecheck and direct server TypeScript checking pass. The server
typecheck wrapper reached an unrelated Runner rebuild and stopped
because Cargo was absent from PATH; full Runner/image validation remains
in CI.

- CI failure accounting: the earlier Turbo cache/force conflict and
Linux OpenCode shell assertion were reproduced and diagnosed. A direct
proc probe ruled out permissions; a portable shell comparison identified
Bash skipping BASH_ENV with stripped SHLVL and socket stdin. Upstream
#18230 now supplies the CI and ignored-stdin fixture fixes; this PR
adopts them without duplicate CI changes. The strict Linux OOM assertion
passed before this source-equivalent merge.
- Retry follow-up: the three insertion/replay/failure timing cases and
server typecheck pass. The existing database regression now asserts a
single marker across identical result replay; the current-head CI
integration gate passes.
- Upstream CI alignment: all three real-Turbo 2.11.10 integration cases
pass. The repository policy gate rejected committing a generated
lockfile; the final PR contains no lockfile edit and uses the maintained
inline stale-lockfile recovery.
- Linked-lifecycle follow-up: 44 focused host/export tests pass (three
optional SDK-dependent cases skip); SDK and server typechecks pass. The
API contract test verifies an explicit root context, host-owned observer
link and unsampled suppression.
- SDK typecheck and focused worker/invocation tests: 30 passed,
including clock changes, bounded legacy IPC, and manual-span
creation-scope parentage.
- Focused reviewed host/native/transaction-boundary checks: 44 passed.
The 86-test worker-manager suite passes after updating its exact
assertions for the host-owned timing budget found by CI. Earlier
preparation/first-text checks also passed.
- Affected server and Runner TypeScript checks pass after building local
workspace declarations.
- Merge follow-up: the profile/candidate startup subset passes 94 tests,
the four Grok source-identity checks pass, and Runner TypeScript
checking passes. A subsequent execution-factory merge passes 21 focused
timing/trace tests and server TypeScript checking.
- Pinned Rust build and the ACPX Rust subset pass. The hostile receipt
projection test passes. Source profile compatibility checks: 127 passed,
one existing skip. Grok identity checks: four passed.
- Broad transport/ACPX suite: 410 passed after building all fake
provider binaries. Earlier attempts failed on missing
Runner/fake-provider outputs; building the required fixtures resolved
those failures.
- The first CI run correctly rejected stale source attestations. After
integrating current master, Grok advances 13→14 and Copilot 84→85;
master’s historical fixtures and Pi 35 stay byte-identical. Provider
dependencies, models and permission defaults are unchanged. Identity
generation/checks now pass.
- The added real-Postgres result lock regression could not initialize
its local embedded cluster. The insertion/replay/failure unit
regressions pass. The current-head CI integration gate passes.
- Comprehensive repository checks and image builds are delegated to CI,
as requested for this instrumentation pass. No deployment, model run,
collector-ingestion proof, or latency improvement is claimed.

## Risks

The new SDK methods are optional. Legacy span calls and endpoint-off
behavior remain supported. The host accepts only closed timing names and
bounded attributes. Remote durations are anchored at local receipt;
transport/scheduling uncertainty remains explicit. Trace trees have
count/depth limits. The first rollout must inspect actual collector
exports before treating the baseline as accepted. Fresh live
qualification for the new source identities remains a rollout
requirement. Revert these commits to remove the new measurements; no
schema migration is required.

## Model Used

OpenAI Codex, GPT-6 family, with reasoning and local code/tool
execution. The exact runtime model ID and context-window size are not
exposed in this session.

## Checklist

- [x] I have included a thinking path that traces from project context
to this change
- [x] I have specified the model used (with version and capability
details)
- [x] I have checked ROADMAP.md and confirmed this PR does not duplicate
planned core work
- [x] I have searched GitHub for duplicate or related PRs and linked
them above
- [x] I have described the issue in-PR following the enhancement
template
- [x] I have not referenced internal/instance-local Paperclip issues or
links
- [x] My branch name describes the change and contains no internal issue
details
- [x] I have run tests locally and they pass
- [x] I have added or updated tests where applicable
- [x] I have updated relevant documentation to reflect my changes
- [x] I have considered and documented any risks above
- [x] All Paperclip CI gates are green
- [x] Standard Greptile and Apex are both 5/5, with no unresolved
findings
- [x] I will address all Greptile and reviewer comments before
requesting merge

---------

Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
Nicky LeachandPaperclip authored and GitHub committed 2026-10-29 17:45:56 -07:00
1 parent 2aa3ea47e3
commit 1d571de5f9
40 files changed
+939 -127

No files matched your search

@@ -386,6 +386,24 @@ describe.skipIf(!otelSdk)("recordProviderPluginSpan native duration", () => {
}
});
it("exports an earlier lifecycle as a separate trace linked to its observer",async()=>{
const {api,sdk}=otelSdk!;
const exporter=new sdk.InMemorySpanExporter();
const provider=new sdk.BasicTracerProvider({spanProcessors:[new sdk.SimpleSpanProcessor(exporter)]});
api.trace.setGlobalTracerProvider(provider);
try {
const {recordProviderPluginSpan}=await import("../instrumentation.js");
const parent={traceId:"0af7651916cd43dd8448eb211c80319c",spanId:"b7ad6b7169203331",traceFlags:1};
const end=Date.now()-100;
recordProviderPluginSpan({name:"plugin.remote.lifecycle",parent,linked:true,attributes:{},startTimeMs:end-500,endTimeMs:end});
const span=exporter.getFinishedSpans()[0]!;
expect(span.parentSpanContext).toBeUndefined();
expect(span.spanContext().traceId).not.toBe(parent.traceId);
expect(span.links[0].context).toMatchObject(parent);
expect(Math.round(hrTimeToMs(span.duration as [number,number]))).toBe(500);
} finally {api.trace.disable();}
});
it("opens and ends the span now when the timestamp pair is absent", async () => {
const { api, sdk } = otelSdk!;
const exporter = new sdk.InMemorySpanExporter();
@@ -413,3 +431,20 @@ describe.skipIf(!otelSdk)("recordProviderPluginSpan native duration", () => {
}
});
});
it("gives a linked lifecycle an explicit root context and a host-owned observer link",async()=>{
const root={},active={};const end=vi.fn();
const startSpan=vi.fn(()=>({end,setStatus:vi.fn(),spanContext:()=>({traceId:"a".repeat(32),spanId:"b".repeat(16),traceFlags:1})}));
const setSpanContext=vi.fn(()=>active);
vi.doMock("node:module",()=>({createRequire:()=>()=>({ROOT_CONTEXT:root,context:{active:()=>active},trace:{getTracer:()=>({startSpan}),setSpanContext}})}));
try {
const {recordProviderPluginSpan}=await importFreshInstrumentation();
const parent={traceId:"0af7651916cd43dd8448eb211c80319c",spanId:"b7ad6b7169203331",traceFlags:1};
recordProviderPluginSpan({name:"plugin.remote.lifecycle",parent,linked:true,attributes:{},startTimeMs:100,endTimeMs:500});
expect(startSpan).toHaveBeenCalledWith("plugin.remote.lifecycle",{attributes:{},startTime:100,links:[{context:{...parent,isRemote:true}}]},root);
expect(setSpanContext).not.toHaveBeenCalled();expect(end).toHaveBeenCalledWith(500);
startSpan.mockClear();
recordProviderPluginSpan({name:"plugin.remote.lifecycle",parent:{...parent,traceFlags:0},linked:true,attributes:{},startTimeMs:100,endTimeMs:500});
expect(startSpan).not.toHaveBeenCalled();
} finally {vi.doUnmock("node:module");vi.resetModules();}
});
@@ -6,6 +6,7 @@ import {
buildHostServices,
clampProviderSpanAttributes,
parseTraceparent,
recordWorkerProviderSpan,
} from "../services/plugin-host-services.js";
// Capture every span the host trust boundary hands to the real tracer.
@@ -329,3 +330,52 @@ describe("clampProviderSpanAttributes", () => {
expect(clampProviderSpanAttributes({ [A.transferDirection]: 1 })).toEqual({});
});
});
describe("nested worker timing receipts",()=>{
it("uses host-generated parents, clamps content and bounds child clocks",()=>{
mockRecordSpan.mockReset();
const childParent={traceId:"0af7651916cd43dd8448eb211c80319c",spanId:"1111111111111111",traceFlags:1};
mockRecordSpan.mockReturnValue(childParent);
const now=Date.now();
recordWorkerProviderSpan({name:"plugin.worker",startTimeMs:now-100,endTimeMs:now,children:[
{name:"plugin.remote.operation",startTimeMs:now-90,endTimeMs:now-10,attributes:{"plugin.operation":"script.settings",secret:"private"}},
{name:"plugin.runtime.verify",startTimeMs:now-200,endTimeMs:now},
]},{traceparent:VALID_TRACEPARENT} as WorkerHostCallContext);
expect(mockRecordSpan).toHaveBeenCalledTimes(2);
expect(mockRecordSpan.mock.calls[1][0]).toMatchObject({parent:childParent,attributes:{"plugin.operation":"script.settings"}});
expect(JSON.stringify(mockRecordSpan.mock.calls)).not.toContain("private");
});
});
it("allows a workspace provider timing receipt without environment-driver authority",async()=>{
mockRecordSpan.mockReset();
await handlersFor(["workspaces.manage"])["span.record"]({name:"plugin.worker"},{traceparent:VALID_TRACEPARENT} as WorkerHostCallContext);
expect(mockRecordSpan).toHaveBeenCalledTimes(1);
});
it("retains short children of a multi-hour worker invocation and shares its aggregate budget",()=>{
mockRecordSpan.mockReset();mockRecordSpan.mockReturnValue({traceId:"0af7651916cd43dd8448eb211c80319c",spanId:"1111111111111111",traceFlags:1});
const now=Date.now(),context={traceparent:VALID_TRACEPARENT,providerSpanBudget:{remaining:3}};
recordWorkerProviderSpan({name:"plugin.worker",startTimeMs:now-7200000,endTimeMs:now,children:[
{name:"pack",startTimeMs:now-7199990,endTimeMs:now-7199980}
]},context);
recordWorkerProviderSpan({name:"pack"},context);recordWorkerProviderSpan({name:"pack"},context);
expect(mockRecordSpan).toHaveBeenCalledTimes(3);
expect(mockRecordSpan.mock.calls[1][0]).toMatchObject({name:"sandbox.daytona.pack",startTimeMs:now-7199990});
});
it("links bounded earlier remote lifecycles to their host-owned observer",()=>{
mockRecordSpan.mockReset();mockRecordSpan.mockReturnValue({traceId:"0af7651916cd43dd8448eb211c80319c",spanId:"1111111111111111",traceFlags:1});
const now=Date.now();
recordWorkerProviderSpan({name:"plugin.worker",startTimeMs:now-10,endTimeMs:now,children:[
{name:"plugin.remote.lifecycle",relationship:"linked",startTimeMs:now-1000,endTimeMs:now-100,children:[
{name:"plugin.remote.operation",startTimeMs:now-900,endTimeMs:now-200}
]},
{name:"plugin.remote.operation",relationship:"linked",startTimeMs:now-1000,endTimeMs:now-100},
{name:"plugin.remote.lifecycle",relationship:"linked",startTimeMs:now-700000,endTimeMs:now-100},
]},{traceparent:VALID_TRACEPARENT});
expect(mockRecordSpan).toHaveBeenCalledTimes(3);
expect(mockRecordSpan.mock.calls[1][0]).toMatchObject({name:"plugin.remote.lifecycle",linked:true,startTimeMs:now-1000});
expect(mockRecordSpan.mock.calls[2][0].linked).toBeUndefined();
});
@@ -27,6 +27,7 @@ vi.mock("../middleware/logger.js", () => {
});
import { logger } from "../middleware/logger.js";
import { withRunExecutionTiming } from "../services/execution-preparation-timing.js";
import {
appendStderrExcerpt,
createDuplexRouteSlotController,
@@ -236,7 +237,8 @@ describe("plugin-worker-manager stderr failure context", () => {
}
});
it("does not emit an unhandled rejection when a plugin responds with terminated before callers attach handlers", async () => {
it.each(["", "http://collector.test"])("does not emit an unhandled rejection before callers attach handlers (endpoint %s)", async endpoint => {
vi.stubEnv("OTEL_EXPORTER_OTLP_ENDPOINT", endpoint);
const unhandledRejection = vi.fn();
process.on("unhandledRejection", unhandledRejection);
@@ -255,28 +257,35 @@ describe("plugin-worker-manager stderr failure context", () => {
try {
await handle.start();
const pendingCall = handle.call(
"environmentExecute" as keyof HostToWorkerMethods,
{
driverKey: "e2b",
companyId: "company-1",
environmentId: "environment-1",
config: {},
lease: { providerLeaseId: "lease-1" },
command: "echo",
} as HostToWorkerMethods[keyof HostToWorkerMethods][0],
);
await withRunExecutionTiming("early-worker-error", async () => {
let settled!: () => void;
const workerSettled = new Promise<void>(resolve => { settled = resolve; });
const pendingCall = handle.call(
"environmentExecute" as keyof HostToWorkerMethods,
{
driverKey: "e2b",
companyId: "company-1",
environmentId: "environment-1",
config: {},
lease: { providerLeaseId: "lease-1" },
command: "echo",
} as HostToWorkerMethods[keyof HostToWorkerMethods][0],
undefined, undefined, settled,
);
await new Promise((resolve) => setImmediate(resolve));
await workerSettled;
await new Promise((resolve) => setImmediate(resolve));
await expect(pendingCall).rejects.toBeInstanceOf(JsonRpcCallError);
await expect(pendingCall).rejects.toMatchObject({
message: expect.stringContaining("terminated"),
await expect(pendingCall).rejects.toBeInstanceOf(JsonRpcCallError);
await expect(pendingCall).rejects.toMatchObject({
message: expect.stringContaining("terminated"),
});
expect(unhandledRejection).not.toHaveBeenCalled();
});
expect(unhandledRejection).not.toHaveBeenCalled();
} finally {
process.off("unhandledRejection", unhandledRejection);
await handle.stop().catch(() => undefined);
vi.unstubAllEnvs();
}
});
@@ -352,7 +361,7 @@ describe("plugin-worker-manager stderr failure context", () => {
});
expect(companiesGet).toHaveBeenCalledWith(
{ companyId: "company-a" },
{ invocationScope: { companyId: "company-a" } },
{ invocationScope: { companyId: "company-a" }, traceparent: undefined, providerSpanBudget: { remaining: 256 } },
);
} finally {
await handle.stop().catch(() => undefined);
@@ -389,7 +398,7 @@ describe("plugin-worker-manager stderr failure context", () => {
expect(companiesGet).toHaveBeenCalledWith(
{ companyId: "company-1" },
{ invocationScope: { companyId: "company-1" } },
{ invocationScope: { companyId: "company-1" }, traceparent: undefined, providerSpanBudget: { remaining: 256 } },
);
} finally {
await handle.stop().catch(() => undefined);
+12 -6
View File
@@ -235,6 +235,7 @@ export function traceparentFromContextToken(contextToken: unknown): string | und
*/
export function recordProviderPluginSpan(input: {
name: string;
linked?: boolean;
parent: ParsedTraceparent;
attributes: Record<string, string | number | boolean>;
status?: { code: number; message?: string };
@@ -244,7 +245,7 @@ export function recordProviderPluginSpan(input: {
startTimeMs?: number;
/** The optional span end time as an epoch-millisecond number. */
endTimeMs?: number;
}): void {
}): ParsedTraceparent | undefined {
try {
const require = createRequire(import.meta.url);
const api = require("@opentelemetry/api") as {
@@ -253,6 +254,7 @@ export function recordProviderPluginSpan(input: {
setSpanContext(context: unknown, spanContext: unknown): unknown;
};
context?: { active(): unknown };
ROOT_CONTEXT?: unknown;
};
const trace = api.trace;
const context = api.context;
@@ -263,22 +265,26 @@ export function recordProviderPluginSpan(input: {
traceFlags: input.parent.traceFlags,
isRemote: true,
};
const parentContext = trace.setSpanContext(context.active(), remoteSpanContext);
if (input.linked && (!api.ROOT_CONTEXT || (input.parent.traceFlags & 1) === 0)) return;
const parentContext = input.linked ? api.ROOT_CONTEXT : trace.setSpanContext(context.active(), remoteSpanContext);
const tracer = trace.getTracer("paperclip.startup");
// Pass the true start time as the OpenTelemetry `startTime` option, so the
// span opens at its real wall-clock start. An epoch-millisecond number is a
// valid OpenTelemetry `TimeInput`.
const startSpanOptions =
input.startTimeMs !== undefined
? { attributes: input.attributes, startTime: input.startTimeMs }
: { attributes: input.attributes };
const startSpanOptions = {
attributes: input.attributes,
...(input.startTimeMs !== undefined ? { startTime: input.startTimeMs } : {}),
...(input.linked ? { links: [{ context: remoteSpanContext }] } : {}),
};
const span = tracer.startSpan(input.name, startSpanOptions, parentContext);
const childParent = (span as typeof span & { spanContext?(): ParsedTraceparent }).spanContext?.();
if (input.status) span.setStatus(input.status);
// Pass the true end time to `span.end`, so the span ends at its real
// wall-clock end and shows its true native width. When the end time is
// absent, the span ends now, so its native width is near zero.
if (input.endTimeMs !== undefined) span.end(input.endTimeMs);
else span.end();
return childParent;
} catch {
// Observability must not change control flow.
}
@@ -1,3 +1,4 @@
import { getActiveStepContext, runWithRuntimeParent } from "@paperclipai/adapter-utils/acpx-engine/startup-timing";
import { afterEach, beforeEach, expect, it, vi } from "vitest";
import { createRunPreparationTiming, measureRunPhase, recordRepositoryObjectCacheDetail, withRunExecutionTiming, withRunWorkspaceFinalizationTiming } from "./execution-preparation-timing.js";
import { ROOT_CONTEXT, trace, SpanStatusCode, type Context } from "@opentelemetry/api";
@@ -141,3 +142,23 @@ it("does not let cache attribute failures replace the operation result", async (
await expect(withRunExecutionTiming("run", () => measureRunPhase("workspace.checkout.cache_publish", work))).resolves.toBe(42);
expect(work).toHaveBeenCalledOnce();
});
it("keeps plugin dispatch beneath its active startup step and gives the worker the dispatch parent", async () => {
const parents = new Map<string, Context>();
const spans = new Map<string, { end: ReturnType<typeof vi.fn>; setStatus: ReturnType<typeof vi.fn> }>();
handles.startActiveSpan.mockImplementation((name, _options, parent, callback) => {
parents.set(name, parent);
const span = { end: vi.fn(), setStatus: vi.fn() };
spans.set(name, span);
return callback(span);
});
const step = trace.wrapSpanContext({ traceId: "1".repeat(32), spanId: "2".repeat(16), traceFlags: 1 });
const stepContext = trace.setSpan(ROOT_CONTEXT, step);
await withRunExecutionTiming("startup", () => runWithRuntimeParent(stepContext, async () => {
await measureRunPhase("execution.plugin_dispatch", () => {
expect(trace.getSpan(getActiveStepContext()!.parentContext as Context)).toBe(spans.get("execution.plugin_dispatch"));
});
expect(getActiveStepContext()!.parentContext).toBe(stepContext);
}));
expect(parents.get("execution.plugin_dispatch")).toBe(stepContext);
});
@@ -1,5 +1,5 @@
import { AsyncLocalStorage } from "node:async_hooks";
import { EXECUTION_PREPARATION_SPAN_ATTRS, runWithRuntimeParent } from "@paperclipai/adapter-utils/acpx-engine/startup-timing";
import { EXECUTION_PREPARATION_SPAN_ATTRS, getActiveStepContext, runWithRuntimeParent } from "@paperclipai/adapter-utils/acpx-engine/startup-timing";
import { createHash } from "node:crypto";
import { isSpanContextValid, ROOT_CONTEXT, SpanStatusCode, trace, type Context } from "@opentelemetry/api";
@@ -102,7 +102,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.plugin_prepare" | "execution.plugin_project_auth" | "execution.plugin_submit" | "execution.plugin_connection"
| "execution.plugin_dispatch" | "execution.plugin_prepare" | "execution.plugin_project_auth" | "execution.plugin_submit" | "execution.plugin_connection"
| "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"
@@ -143,7 +143,7 @@ export function isRunPhaseActive(phase: ExecutionPhase): boolean {
return endpointEnabled() && owned?.phase === phase && !owned.ended;
}
function measure<T>(runId: string, name: string, work: () => T | Promise<T>, independent = false, counts: { fileCount?: number; bytes?: number } = {}): Promise<T> {
function measure<T>(runId: string, name: string, work: () => T | Promise<T>, independent = false, counts: { fileCount?: number; bytes?: number } = {}, activeParent?: Context): Promise<T> {
let operation: Promise<T> | undefined;
const run = (span?: { setStatus(status: { code: number }): void; end(): void }, parent?: Context) => {
if (operation) return operation;
@@ -168,7 +168,7 @@ function measure<T>(runId: string, name: string, work: () => T | Promise<T>, ind
};
if (!endpointEnabled()) return (async () => work())();
try {
const parent = independent ? undefined : getRunExecutionTraceContext(runId);
const parent = independent ? undefined : activeParent ?? getRunExecutionTraceContext(runId);
const trigger = parent ? undefined : trace.getActiveSpan()?.spanContext();
const traced = trace.getTracer(name.startsWith("heartbeat.prepare.") ? "paperclip.heartbeat-preparation"
: name.startsWith("native.controller.") ? "paperclip.native-controller" : "paperclip.execution").startActiveSpan(name, {
@@ -202,7 +202,9 @@ export async function measureRunPhase<T>(phase: ExecutionPhase, work: () => T |
const count = owned.details.get(phase) ?? 0;
if (count >= 16) return work();
owned.details.set(phase, count + 1);
return measure(owned.runId, phase, work, false, counts);
// 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);
}
export function beginRunPhase(phase: ExecutionPhase): (failed?: boolean) => void {
@@ -439,7 +439,7 @@ describe("native runner performance trace", () => {
).toBe("agent.turn");
await trace.measure("provider.dynamic", async () => undefined);
await trace.record({
name: "provider.turn.queue",
name: "runner.turn.submitted_to_started",
parentName: "agent.turn",
startedAtMs: 1_085,
endedAtMs: 1_090,
@@ -469,7 +469,7 @@ describe("native runner performance trace", () => {
expect(parentOf("stage.asset.ca_bundle")).toBe("stage.sync");
expect(parentOf("agent.turn")).toBe("native.session.execute");
expect(parentOf("provider.dynamic")).toBe("agent.turn");
expect(parentOf("provider.turn.queue")).toBe("agent.turn");
expect(parentOf("runner.turn.submitted_to_started")).toBe("agent.turn");
expect(spans.filter((span) => span.name === "task.provider_session")).toHaveLength(1);
expect(spans.some((span) => span.name === "task.provider_session.measured")).toBe(false);
@@ -150,6 +150,7 @@ const NOOP_SPAN: SpanHandle = {
const SAFE_NATIVE_RUN_SPAN_ATTRIBUTE_KEYS = new Set([
"adapter",
"bytesTransferred",
"inputBytes", "toolSchemaBytes", "toolCount", "contextReused", "tokenDataAvailable",
"connectionOwner",
"driver",
"duration_ms",
@@ -1,3 +1,4 @@
import { measureRunnerOperation, recordRunnerTimingReceipt } from "./runner-operation-timing.js";
import { notifyExternalAgentSettlement } from "../../modules/external-agents/index.js";
import { notifyNativeResultCommit } from "../native-result-commit-notifications.js";
import { releaseRetainedNativeSessionCleanup } from "../retained-native-session-cleanup-notifications.js";
@@ -8642,13 +8643,29 @@ async function executePaperclipNativeSessionWithinScope(
privateKeyPem: input.runnerEnvironment?.PAPERCLIP_AGENT_PRIVATE_KEY,
assertControllerActive,
assertRunAdmissionActive,
onResultCommitted: async () => {
const at=Date.now();
await trace.record({name:"completion.result_committed",parentName:"native.session.execute",startedAtMs:at,endedAtMs:at});
},
onEventPersistence: async (event, startedAtMs, endedAtMs) => {
if (["turn.submitted", "turn.started", "turn.accepted", "run.result.accepted"].includes(event.eventType)) {
await trace.record({name: "runner.event.persist", parentName: "native.session.execute", startedAtMs, endedAtMs,
attributes:{eventKind:event.eventType}});
}
},
onCommittedEvent: async (event) => {
if (event.eventType === "run.result.proposed") {
const at = Date.now();
await trace.record({name:"completion.tool_accepted",parentName:"native.session.execute",startedAtMs:at,endedAtMs:at});
}
if (event.eventType === "harness.diagnostic") {
await recordNativeProviderPerformanceSpans(trace, parseProviderPerformanceSpans(event.payload, Date.now()));
}
const textKind = !firstTokenRecorded ? observeDeliverableText(event) : null;
if (textKind) {
firstTokenRecorded = true;
const textAt = Date.now();
await trace.record({name:"response.first_nonempty_text",parentName:"native.session.execute",startedAtMs:textAt,endedAtMs:textAt});
await trace.record({
name: "task.time_to_first_token",
parentName: "task.provider_session",
@@ -8737,7 +8754,7 @@ async function executePaperclipNativeSessionWithinScope(
if (event.eventType === "turn.started" && turnStartedAtMs === null) {
turnStartedAtMs = milestoneAtMs;
await trace.record({
name: "provider.turn.queue",
name: "runner.turn.submitted_to_started",
parentName: "agent.turn",
startedAtMs: turnSubmittedAtMs ?? milestoneAtMs,
endedAtMs: milestoneAtMs,
@@ -12273,6 +12290,8 @@ async function createRunnerdBackendWithinSessionClaim(
return nativeCompletionFeedback(input.db, current.runId, result);
},
codexTransportFactory: recovery => createRunnerdCodexTransport({ provider: provider.kind, stateDirectory: root,
measureOperation: process.env.OTEL_EXPORTER_OTLP_ENDPOINT?.trim() ? measureRunnerOperation : undefined,
recordTimingReceipt: process.env.OTEL_EXPORTER_OTLP_ENDPOINT?.trim() ? recordRunnerTimingReceipt : undefined,
...owner.driver, runnerArtifact: input.controllerPersistence?.runnerArtifact ?? description.runner,
controlPlaneRegistration: input.controllerPersistence ? async (authority, identity, signal) => {
await nativeRunnerCheckpointRegistration({
@@ -13755,6 +13774,8 @@ async function createRunnerdBackendWithinSessionClaim(
),
codexTransportFactory: (recoveryContext) => {
const bundle = createRunnerdCodexTransport({
measureOperation: process.env.OTEL_EXPORTER_OTLP_ENDPOINT?.trim() ? measureRunnerOperation : undefined,
recordTimingReceipt: process.env.OTEL_EXPORTER_OTLP_ENDPOINT?.trim() ? recordRunnerTimingReceipt : undefined,
onSpawn: input.onSpawn,
...(input.execution.provider.kind === "codex" ? { disableCodexWebSearch: "webSearch" in input.execution.provider && input.execution.provider.webSearch === "disabled" } : {}),
homeMode: preconfiguredHome ? "preconfigured" : "isolated",
@@ -72,3 +72,11 @@ describe("native tool timing", () => {
expect(spans.filter((span) => span.span === "tool.request.input_stream").map((span) => span.durationMs)).toEqual([10, 31]);
});
});
it("marks the first completion tool request once per turn without arguments",async()=>{
const {spans,timing}=fixture();
const completion=(id:string)=>({...event("started",id,1000),payload:{executionId:id,name:"paperclip_finish",arguments:"private"}});
await timing.observe(completion("a"));await timing.observe(completion("b"));
expect(spans.filter(span=>span.span==="completion.request.first")).toHaveLength(1);
expect(JSON.stringify(spans)).not.toContain("private");
});
@@ -9,11 +9,13 @@ function operation(value: unknown): string {
/** Provider stream IDs and MCP request IDs are separate namespaces. Never join by name or order. */
export function createNativeToolTrace(trace: NativeRunTrace, now = Date.now) {
let firstCompletion = false;
const announced = new Map<string, { startedAtMs: number; lastInputAtMs?: number; operation: string }>();
return {
observe(event: { eventType: string; emittedAt: string; payload: unknown }): void | Promise<void> {
if (["turn.started", "turn.completed", "turn.failed", "turn.cancelled"].includes(event.eventType)) {
announced.clear();
firstCompletion = false;
return;
}
if (!event.eventType.startsWith("tool.execution.")) return;
@@ -25,6 +27,10 @@ export function createNativeToolTrace(trace: NativeRunTrace, now = Date.now) {
if (announced.has(id)) return;
if (announced.size >= 512) announced.delete(announced.keys().next().value!);
announced.set(id, { startedAtMs: time, operation: operation(payload?.name) });
if (!firstCompletion && typeof payload?.name === "string" && /^(?:mcp__paperclip__)?paperclip_(?:finish|block)$/.test(payload.name)) {
firstCompletion = true;
return trace.record({name:"completion.request.first",parentName:"agent.turn",startedAtMs:time,endedAtMs:time});
}
return;
}
const pending = announced.get(id);
@@ -1120,6 +1120,7 @@ describe("PaperclipControlPlanePort conformance", () => {
credentialBindings: [],
runtimeContext: nativeRuntimeContextFixture(),
});
let resultCommitMarkers = 0;
const port = new PaperclipControlPlanePort(db, {
companyId: identity.companyId,
issueId: taskIssueId,
@@ -1130,7 +1131,7 @@ describe("PaperclipControlPlanePort conformance", () => {
completionContractSha256: "phase6-task-contract",
sourceInstanceId: taskRunnerId,
controlPlaneSourceInstanceId: "phase6-control-plane",
});
}, {onResultCommitted:async()=>{resultCommitMarkers++;}});
const completed = await executeNativeSession({
input: execution,
backend,
@@ -1210,6 +1211,7 @@ describe("PaperclipControlPlanePort conformance", () => {
callerResultId: `phase6-scripted-runner:${taskRunId}:result`,
callerDedupeKey: `${taskRunId}:phase6-task-contract`,
})).rejects.toThrow("structured_result_replay_conflict");
expect(resultCommitMarkers).toBe(1);
await Promise.all([
finalizeNativeRun({ db, runId: taskRunId, workspaceFinalizeStatus: "succeeded" }),
finalizeNativeRun({ db, runId: taskRunId, workspaceFinalizeStatus: "succeeded" }),
@@ -1345,10 +1347,14 @@ describe("PaperclipControlPlanePort conformance", () => {
nativeSessionId: identity.sessionId, runnerInstanceId: conformanceRunnerId,
completionContractId: contractId, completionContractSha256: contractSha,
});
let resultCommitted = false;
const port = kind === "port" ? new PaperclipControlPlanePort(db, {
...identity, runId, completionContractId: contractId, completionContractSha256: contractSha,
sourceInstanceId: conformanceRunnerId, controlPlaneSourceInstanceId: "result-lock-order",
}) : new NativeRunCoordinatorStore(db, {
}, {onResultCommitted:async()=>{
const rows=await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId,runId));
expect(rows).toHaveLength(1);resultCommitted=true;
}}) : new NativeRunCoordinatorStore(db, {
...identity, runId, completionContractId: contractId, completionContractSha256: contractSha,
normalizedSessionId: identity.sessionId, runnerSourceInstanceId: conformanceRunnerId,
completionContractRevision: "standalone-v1", completionContractCriterionIds: ["objective"],
@@ -1371,11 +1377,13 @@ describe("PaperclipControlPlanePort conformance", () => {
await new Promise(resolve => setTimeout(resolve, 10));
}
expect(waiting).toBe(true);
expect(resultCommitted).toBe(false);
await tx.execute(sql`set local lock_timeout = '1s'`);
await tx.update(heartbeatRuns).set({ updatedAt: new Date() }).where(eq(heartbeatRuns.id, runId));
});
} finally {
expect(await completion).toBeNull();
expect(resultCommitted).toBe(kind === "port");
}
});
@@ -86,6 +86,8 @@ export class PaperclipControlPlanePort implements ControlPlanePort {
#sessionId: string | null = null;
readonly #assertControllerActive?: () => void;
readonly #assertRunAdmissionActive?: () => void;
readonly #onResultCommitted?: () => Promise<void>;
readonly #onEventPersistence?: (event: PrpEvent, startedAtMs: number, endedAtMs: number) => Promise<void>;
readonly #onCommittedEvent?: (event: PrpEvent) => Promise<void>;
readonly #onDuplicateEvent?: (event: PrpEvent) => Promise<void>;
@@ -99,6 +101,8 @@ export class PaperclipControlPlanePort implements ControlPlanePort {
assertControllerActive?: () => void;
/** Read-only admission may finish an existing startup solely for restart detachment. */
assertRunAdmissionActive?: () => void;
onResultCommitted?: () => Promise<void>;
onEventPersistence?: (event: PrpEvent, startedAtMs: number, endedAtMs: number) => Promise<void>;
onCommittedEvent?: (event: PrpEvent) => Promise<void>;
onDuplicateEvent?: (event: PrpEvent) => Promise<void>;
} = {},
@@ -109,6 +113,8 @@ export class PaperclipControlPlanePort implements ControlPlanePort {
this.#identityRedactor = createAgentIdentityRedactor(options.privateKeyPem);
this.#binding = structuredClone(binding);
this.#onCommittedEvent = options.onCommittedEvent;
this.#onEventPersistence = options.onEventPersistence;
this.#onResultCommitted = options.onResultCommitted;
this.#onDuplicateEvent = options.onDuplicateEvent;
}
@@ -217,6 +223,7 @@ export class PaperclipControlPlanePort implements ControlPlanePort {
}
async appendEvent(value: NativeRunEvent | PrpEvent, options?: { signal: AbortSignal }) {
const persistenceStartedAt = Date.now(), persistenceClock = performance.now();
this.#assertActive(options);
if (!isPrpEvent(value)) throw new Error("native_legacy_event_not_supported");
const validated = validatePrpEvent(value);
@@ -287,6 +294,7 @@ export class PaperclipControlPlanePort implements ControlPlanePort {
return receipt;
});
this.#assertActive(options);
try { await this.#onEventPersistence?.(event, persistenceStartedAt, persistenceStartedAt + performance.now() - persistenceClock); } catch {}
if (persisted.disposition === "committed") {
try {
publishLiveEvent({
@@ -377,7 +385,7 @@ export class PaperclipControlPlanePort implements ControlPlanePort {
canonicalSha256,
});
await this.#db.transaction(async (tx) => {
const insertedResult = await this.#db.transaction(async (tx) => {
// Result insertion checks the task foreign key. Acquire that parent lock
// before the run, matching task mutations that subsequently update a run.
// Otherwise concurrent chat/status writes can deadlock after generation.
@@ -401,7 +409,7 @@ export class PaperclipControlPlanePort implements ControlPlanePort {
if (existing) {
this.#assertActive(options);
if (existing.canonicalSha256 !== canonicalSha256) throw new Error("structured_result_replay_conflict");
return;
return false;
}
const resultJson = {
result: validated.result,
@@ -457,6 +465,8 @@ export class PaperclipControlPlanePort implements ControlPlanePort {
updatedAt: new Date(),
}).where(eq(heartbeatRuns.id, this.#binding.runId));
this.#assertActive(options);
return true;
});
if (insertedResult) { try { await this.#onResultCommitted?.(); } catch {} }
}
}
@@ -0,0 +1,14 @@
import { expect,it,vi } from "vitest";
import { PaperclipControlPlanePort } from "./paperclip-control-plane-port.js";
import { CONTROL_PLANE_CONFORMANCE_OPEN as OPEN,CONTROL_PLANE_CONFORMANCE_RESULT as result,CONTROL_PLANE_CONFORMANCE_TERMINAL as terminal } from "../../vendor/paperclip-runner/testing.js";
it.each(["inserted","replayed","failed"])("marks only a new result after transaction success (%s)",async(outcome)=>{
let resolve!:(inserted:boolean)=>void,reject!:(error:unknown)=>void;
const transaction=vi.fn(()=>new Promise<boolean>((yes,no)=>{resolve=yes;reject=no;}));
const committed=vi.fn(async()=>{});
const port=new PaperclipControlPlanePort({transaction} as never,{...OPEN.identity,completionContractId:"fixture",completionContractSha256:"fixture",sourceInstanceId:"fixture",controlPlaneSourceInstanceId:"fixture"},{onResultCommitted:committed});
const pending=port.completeRun({result,terminal});
expect(transaction).toHaveBeenCalledOnce();expect(committed).not.toHaveBeenCalled();
if(outcome === "failed"){const error=Error("transaction rejected");reject(error);await expect(pending).rejects.toBe(error);expect(committed).not.toHaveBeenCalled();}
else{resolve(outcome === "inserted");await pending;expect(committed).toHaveBeenCalledTimes(outcome === "inserted" ? 1 : 0);}
});
@@ -0,0 +1,39 @@
import { afterEach, describe, expect, it, vi } from "vitest";
const mocks=vi.hoisted(()=>({parent:undefined as unknown, spans:[] as any[]}));
vi.mock("@opentelemetry/api",()=>({SpanStatusCode:{ERROR:2},trace:{
getTracer:()=>({startSpan:(name:string,options:unknown,parent:unknown)=>{const span={name,options,parent,end:vi.fn(),setStatus:vi.fn()};mocks.spans.push(span);return span;}}),
setSpan:(parent:unknown,span:unknown)=>({parent,span}),
}}));
vi.mock("@paperclipai/adapter-utils/acpx-engine/startup-timing",()=>({
getActiveStepContext:()=>mocks.parent?{parentContext:mocks.parent}:null,
runWithRuntimeParent:(_parent:unknown,work:()=>unknown)=>work(),
}));
import { measureRunnerOperation, recordRunnerTimingReceipt } from "./runner-operation-timing.js";
afterEach(()=>{vi.unstubAllEnvs();mocks.spans=[];mocks.parent=undefined;});
describe("Runner operation ownership",()=>{
it("does no tracer work when disabled and preserves the operation",async()=>{
vi.stubEnv("OTEL_EXPORTER_OTLP_ENDPOINT","");mocks.parent={run:"a"};let count=0;
expect(await measureRunnerOperation("runner.transport.resume",async()=>++count)).toBe(1);
expect(mocks.spans).toEqual([]);
});
it("reads each current run parent and closes concurrent failures",async()=>{
vi.stubEnv("OTEL_EXPORTER_OTLP_ENDPOINT","http://collector.test");
const a={run:"a"},b={run:"b"};mocks.parent=a;
let done!:()=>void;const pending=measureRunnerOperation("runner.transport.resume",()=>new Promise<void>(resolve=>done=resolve));
await Promise.resolve();mocks.parent=b;
const error=Error("private");await expect(measureRunnerOperation("runner.transport.attach",async()=>{throw error;})).rejects.toBe(error);
done();await pending;
expect(mocks.spans.map(span=>span.parent)).toEqual([a,b]);
expect(mocks.spans.every(span=>span.end.mock.calls.length===1)).toBe(true);
expect(mocks.spans[1].setStatus).toHaveBeenCalledWith({code:2});
});
it("validates numeric receipt bounds and rejects names or content",()=>{
vi.stubEnv("OTEL_EXPORTER_OTLP_ENDPOINT","http://collector.test");mocks.parent={run:"a"};
recordRunnerTimingReceipt({runnerMs:100,roundTripMs:120,sidecarMs:80,statusMs:30,promptMs:40,statusOffsetMs:0,promptOffsetMs:40,secret:"private"});
expect(mocks.spans.map(span=>span.name)).toEqual(["runner.remote.turn_start","runner.sidecar.turn_start","runner.sidecar.status_read","runner.sidecar.prompt_started"]);
expect(JSON.stringify(mocks.spans)).not.toContain("private");
recordRunnerTimingReceipt({runnerMs:10,roundTripMs:120,sidecarMs:80,statusMs:30,promptMs:40,statusOffsetMs:0,promptOffsetMs:40});
recordRunnerTimingReceipt({runnerMs:100,roundTripMs:120,sidecarMs:80,statusMs:NaN,promptMs:40,statusOffsetMs:0,promptOffsetMs:40});
expect(mocks.spans).toHaveLength(4);
});
});
@@ -0,0 +1,51 @@
import { trace, SpanStatusCode, type Context, type Span } from "@opentelemetry/api";
import { getActiveStepContext, runWithRuntimeParent } from "@paperclipai/adapter-utils/acpx-engine/startup-timing";
const phases = new Set(["runner.transport.cold_start", "runner.transport.resume", "runner.transport.attach", "runner.transport.prepare",
"runner.transport.turn_start", "runner.transport.started_event", "runner.transport.command_ack", "runner.transport.registration", "runner.transport.provider_identity"]);
const keys = new Set(["inputBytes", "toolSchemaBytes", "toolCount", "contextReused", "tokenDataAvailable"]);
// Resolve the parent on every call: a retained transport belongs to several successive runs.
export function measureRunnerOperation<T>(phase: string, work: () => Promise<T>, attributes?: Record<string, number | boolean>): Promise<T> {
const parent = getActiveStepContext()?.parentContext;
if (!process.env.OTEL_EXPORTER_OTLP_ENDPOINT?.trim() || !parent || !phases.has(phase)) return work();
let operation: Promise<T> | undefined;
const once = () => operation ??= Promise.resolve().then(work);
let span: Span | undefined;
try {
const bounded = Object.fromEntries(Object.entries(attributes ?? {}).filter(([key,value])=>keys.has(key) &&
(typeof value === "boolean" || Number.isSafeInteger(value) && value >= 0 && value <= 1e12)));
span = trace.getTracer("paperclip.native-runner").startSpan(phase,{attributes:bounded},parent as Context);
const currentSpan = span;
const child = trace.setSpan(parent as Context,currentSpan);
return runWithRuntimeParent(child,once).then(value=>{try {currentSpan.end();}catch{} return value;},error=>{
try{currentSpan.setStatus({code:SpanStatusCode.ERROR});currentSpan.end();}catch{} throw error;
});
} catch { try { span?.end(); } catch {} return once(); }
}
/** Numeric same-process durations, causally anchored at receipt. Remainders are unattributed. */
export function recordRunnerTimingReceipt(raw: unknown): void {
const parent = getActiveStepContext()?.parentContext;
if (!process.env.OTEL_EXPORTER_OTLP_ENDPOINT?.trim() || !parent || !raw || typeof raw !== "object") return;
const row = raw as Record<string, unknown>;
const values = ["runnerMs", "sidecarMs", "statusMs", "promptMs", "statusOffsetMs", "promptOffsetMs", "roundTripMs"].map(key=>row[key]);
if (!values.every(value=>typeof value === "number" && Number.isFinite(value) && value >= 0 && value <= 300000)) return;
const [runner,sidecar,status,prompt,statusOffset,promptOffset,roundTrip] = values as number[];
if (runner > roundTrip || sidecar > runner || statusOffset + status > promptOffset || promptOffset + prompt > sidecar) return;
const opened: Array<{span:Span; end:number}> = [];
const end = Date.now(), start=end-runner, sidecarStart=end-sidecar;
try {
const tracer = trace.getTracer("paperclip.native-runner");
const root = tracer.startSpan("runner.remote.turn_start",{startTime:start},parent as Context);
opened.push({span:root,end});
const rootContext=trace.setSpan(parent as Context,root);
const child=tracer.startSpan("runner.sidecar.turn_start",{startTime:sidecarStart},rootContext);
opened.push({span:child,end});
const childContext=trace.setSpan(rootContext,child);
const settings=tracer.startSpan("runner.sidecar.status_read",{startTime:sidecarStart+statusOffset},childContext);opened.push({span:settings,end:sidecarStart+statusOffset+status});
const acknowledgement=tracer.startSpan("runner.sidecar.prompt_started",{startTime:sidecarStart+promptOffset},childContext);opened.push({span:acknowledgement,end:sidecarStart+promptOffset+prompt});
} catch {} finally {
for (const item of opened.reverse()) { try { item.span.end(item.end); } catch {} }
}
}
+46 -13
View File
@@ -537,6 +537,10 @@ const KNOWN_PROVIDER_SPAN_NAMES: ReadonlySet<string> = new Set([
"postUploadCommand",
"session.open",
"session.close",
"plugin.worker", "plugin.host.identity.read", "plugin.host.config.read", "plugin.host.progress.read", "plugin.host.progress.write",
"plugin.workspace.prepare", "plugin.storage.queue", "plugin.storage.execute",
"plugin.runtime.queue", "plugin.runtime.prepare", "plugin.runtime.reconcile", "plugin.runtime.verify",
"plugin.remote.request", "plugin.remote.operation", "plugin.remote.lifecycle",
]);
/** Clamp the span name to a closed, namespaced set. A known name maps to
@@ -547,7 +551,7 @@ const KNOWN_PROVIDER_SPAN_NAMES: ReadonlySet<string> = new Set([
* `provider` family attribute on the span instead of this literal. */
function clampProviderSpanName(raw: unknown): string {
const name = typeof raw === "string" && KNOWN_PROVIDER_SPAN_NAMES.has(raw) ? raw : "other";
return `sandbox.daytona.${name}`;
return name.startsWith("plugin.") ? name : `sandbox.daytona.${name}`;
}
/** The closed allowlist of attribute keys a provider span may carry. The host
@@ -560,6 +564,7 @@ const PROVIDER_SPAN_ATTR_ALLOWLIST: ReadonlySet<string> = new Set<string>([
SPAN_ATTRS.transferWallMs,
SPAN_ATTRS.transferGuardCount,
SPAN_ATTRS.transferDirection,
"plugin.operation", "plugin.reused", "plugin.item_count", "plugin.bytes", "plugin.call_count",
]);
/** The subset of allowed keys that carry a finite number. */
@@ -567,6 +572,7 @@ const PROVIDER_SPAN_NUMERIC_ATTRS: ReadonlySet<string> = new Set<string>([
SPAN_ATTRS.packWallMs,
SPAN_ATTRS.transferWallMs,
SPAN_ATTRS.transferGuardCount,
"plugin.item_count", "plugin.bytes", "plugin.call_count",
]);
/** The closed value set for the `outcome` attribute. */
@@ -589,6 +595,11 @@ export function clampProviderSpanAttributes(
if (!raw) return clamped;
for (const [key, value] of Object.entries(raw)) {
if (!PROVIDER_SPAN_ATTR_ALLOWLIST.has(key)) continue;
if (key === "plugin.operation") {
if (typeof value === "string" && ["request", "queue.wait", "launch.wait", "mount.refresh", "guest.setup", "process.execute", "script.workspace", "script.git", "script.settings", "script.publication", "cleanup", "process.start", "hello", "checkpoint.load"].includes(value)) clamped[key] = value;
continue;
}
if (key === "plugin.reused") { if (typeof value === "boolean") clamped[key] = value; continue; }
if (key === SPAN_ATTRS.provider) {
clamped[key] = normalizeProviderFamily(typeof value === "string" ? value : undefined);
continue;
@@ -602,7 +613,7 @@ export function clampProviderSpanAttributes(
continue;
}
if (PROVIDER_SPAN_NUMERIC_ATTRS.has(key)) {
if (typeof value === "number" && Number.isFinite(value)) clamped[key] = value;
if (typeof value === "number" && Number.isFinite(value) && (!key.startsWith("plugin.") || value >= 0 && value <= 1e12)) clamped[key] = value;
continue;
}
}
@@ -664,12 +675,14 @@ const MAX_PROVIDER_SPAN_END_SKEW_MS = 60 * 1000; // 1 minute
function validateProviderSpanTimes(
startTimeMs: unknown,
endTimeMs: unknown,
maxDurationMs = MAX_PROVIDER_SPAN_DURATION_MS,
maxAgeMs = MAX_PROVIDER_SPAN_START_AGE_MS,
): { startTimeMs: number; endTimeMs: number } | undefined {
if (typeof startTimeMs !== "number" || !Number.isFinite(startTimeMs)) return undefined;
if (typeof endTimeMs !== "number" || !Number.isFinite(endTimeMs)) return undefined;
if (startTimeMs > endTimeMs) return undefined;
if (endTimeMs - startTimeMs > MAX_PROVIDER_SPAN_DURATION_MS) return undefined;
if (Date.now() - startTimeMs > MAX_PROVIDER_SPAN_START_AGE_MS) return undefined;
if (endTimeMs - startTimeMs > maxDurationMs) return undefined;
if (Date.now() - startTimeMs > maxAgeMs) return undefined;
if (endTimeMs - Date.now() > MAX_PROVIDER_SPAN_END_SKEW_MS) return undefined;
return { startTimeMs, endTimeMs };
}
@@ -691,20 +704,40 @@ export function recordWorkerProviderSpan(
status?: { code?: unknown; message?: unknown };
startTimeMs?: unknown;
endTimeMs?: unknown;
children?: unknown;
relationship?: unknown;
},
context: WorkerHostCallContext | undefined,
): void {
const parent = parseTraceparent(context?.traceparent);
if (!parent) return; // reject a missing or malformed traceparent
const times = validateProviderSpanTimes(params.startTimeMs, params.endTimeMs);
const status = clampSpanStatus(params.status);
recordProviderPluginSpan({
name: clampProviderSpanName(params.name),
parent,
attributes: clampProviderSpanAttributes(params.attributes),
...(status ? { status } : {}),
...(times ? { startTimeMs: times.startTimeMs, endTimeMs: times.endTimeMs } : {}),
});
const budget = context?.providerSpanBudget ?? {remaining:256};
const longInvocation = params.name === "plugin.worker";
const invocationLimitMs = 24 * 60 * 60 * 1000;
const visit = (params: { name: string; attributes?: Record<string, unknown>; status?: { code?: unknown }; startTimeMs?: unknown; endTimeMs?: unknown; children?: unknown; relationship?: unknown }, parent: ParsedTraceparent, depth: number, enclosing?: { startTimeMs: number; endTimeMs: number }) => {
if (depth > 12 || budget.remaining-- <= 0) return;
const times = validateProviderSpanTimes(params.startTimeMs, params.endTimeMs,
depth === 0 && longInvocation ? invocationLimitMs : MAX_PROVIDER_SPAN_DURATION_MS,
longInvocation ? invocationLimitMs : MAX_PROVIDER_SPAN_START_AGE_MS);
const linked = params.relationship === "linked" && params.name === "plugin.remote.lifecycle";
if (params.relationship !== undefined && !linked) return;
if (linked && (!times || Date.now() - times.startTimeMs > MAX_PROVIDER_SPAN_DURATION_MS)) return;
if (enclosing && (!times || (!linked && times.startTimeMs < enclosing.startTimeMs) || times.endTimeMs > enclosing.endTimeMs)) return;
const status = clampSpanStatus(params.status);
const childParent = recordProviderPluginSpan({
name: clampProviderSpanName(params.name),
parent,
...(linked ? { linked: true } : {}),
attributes: clampProviderSpanAttributes(params.attributes),
...(status ? { status } : {}),
...(times ? { startTimeMs: times.startTimeMs, endTimeMs: times.endTimeMs } : {}),
});
if (childParent && times && Array.isArray(params.children)) for (const child of params.children.slice(0, 256)) {
if (budget.remaining <= 0) break;
if (child && typeof child === "object" && typeof child.name === "string") visit(child, childParent, depth + 1, times);
}
};
try { visit(params, parent, 0); } catch {}
}
export function buildHostServices(
+22 -7
View File
@@ -66,6 +66,8 @@ import {
type LoginCommandKey,
} from "./login-command.js";
import { logger } from "../middleware/logger.js";
import { measureRunPhase } from "./execution-preparation-timing.js";
import { context as otelContext } from "@opentelemetry/api";
import { traceparentFromContextToken } from "../instrumentation.js";
// ---------------------------------------------------------------------------
@@ -539,6 +541,7 @@ interface PendingRequest {
}
interface ActiveInvocation {
providerSpanBudget: {remaining:number};
scope: PluginInvocationScope;
timer?: ReturnType<typeof setTimeout>;
// The host-minted W3C `traceparent` for the active startup span, or undefined
@@ -1198,20 +1201,18 @@ export function createPluginWorkerHandle(
}
function registerInvocation(scope: PluginInvocationScope, ttlMs?: number): PluginInvocationContext {
// Mint a W3C `traceparent` from the active startup span, so the worker's
// provider span can parent to it. The host keeps the value on its own record
// (below) and never trusts the worker to supply the parent. Outside a
// measured startup step there is no active span, so this is undefined.
// Keep the owning operation's parent host-side; workspace/runtime operations
// use the same per-invocation contract as measured provider startup.
const activeStep = getActiveStepContext();
const traceparent = activeStep
? traceparentFromContextToken(activeStep.parentContext)
: undefined;
: process.env.OTEL_EXPORTER_OTLP_ENDPOINT?.trim() ? traceparentFromContextToken(otelContext.active()) : undefined;
const invocation: PluginInvocationContext = {
id: randomUUID(),
scope,
...(traceparent ? { traceparent } : {}),
};
const entry: ActiveInvocation = { scope, traceparent };
const entry: ActiveInvocation = { scope, traceparent, providerSpanBudget:{remaining:256} };
if (ttlMs !== undefined) {
entry.timer = setTimeout(() => {
activeInvocations.delete(invocation.id);
@@ -2746,7 +2747,7 @@ export function createPluginWorkerHandle(
}
const entry = activeInvocations.get(invocationId);
if (!entry) return { invalidInvocationScope: true };
return { invocationScope: entry.scope, traceparent: entry.traceparent };
return { invocationScope: entry.scope, traceparent: entry.traceparent, providerSpanBudget:entry.providerSpanBudget };
}
/**
@@ -3377,6 +3378,20 @@ export function createPluginWorkerHandle(
timeoutMs?: number,
executeLogSink?: ExecuteLogSink,
onWorkerSettled?: () => void,
): Promise<HostToWorkerMethods[M][1]> {
const dispatched = measureRunPhase("execution.plugin_dispatch", () => callInternalRaw(method, params, timeoutMs, executeLogSink, onWorkerSettled));
// Tracing wraps the RPC in a new promise. Guard the promise handed to the
// caller as well, without swallowing its eventual rejection for that caller.
void dispatched.catch(() => undefined);
return dispatched;
}
function callInternalRaw<M extends HostToWorkerMethodName>(
method: M,
params: HostToWorkerMethods[M][0],
timeoutMs?: number,
executeLogSink?: ExecuteLogSink,
onWorkerSettled?: () => void,
): Promise<HostToWorkerMethods[M][1]> {
const rpcPromise = new Promise<HostToWorkerMethods[M][1]>((resolve, reject) => {
const idleControl = method === "prepareIdleSleep" || method === "releaseIdleSleep";