diff --git a/packages/adapter-utils/src/acpx-engine/execute.ts b/packages/adapter-utils/src/acpx-engine/execute.ts index e04c375f6c3..becb090d85a 100644 --- a/packages/adapter-utils/src/acpx-engine/execute.ts +++ b/packages/adapter-utils/src/acpx-engine/execute.ts @@ -15,6 +15,7 @@ import type { import { adapterExecutionTargetSessionIdentity, describeAdapterExecutionTarget, + adapterExecutionTargetDuplexTelemetryRecorder, adapterExecutionTargetEnablesSandboxDuplexBridge, formatAdapterExecutionTimeoutErrorMessage, formatAdapterExecutionTimeoutStartLogLine, @@ -2040,6 +2041,7 @@ async function buildRuntime(input: { timeoutSec, hostApiToken: env.PAPERCLIP_API_KEY, enableSandboxDuplexBridge: adapterExecutionTargetEnablesSandboxDuplexBridge(remoteTarget), + duplexTelemetryRecorder: adapterExecutionTargetDuplexTelemetryRecorder(remoteTarget), onLog: input.ctx.onLog, getRuntimeParentContext: input.getRuntimeParentContext, runtimeSpan: input.runtimeSpan, diff --git a/packages/adapter-utils/src/execution-target-sandbox.test.ts b/packages/adapter-utils/src/execution-target-sandbox.test.ts index 619ed4656e8..4be42c59d70 100644 --- a/packages/adapter-utils/src/execution-target-sandbox.test.ts +++ b/packages/adapter-utils/src/execution-target-sandbox.test.ts @@ -15,6 +15,7 @@ import { import { DEFAULT_REMOTE_SANDBOX_ADAPTER_TIMEOUT_SEC, + adapterExecutionTargetDuplexTelemetryRecorder, adapterExecutionTargetEnablesSandboxDuplexBridge, adapterExecutionTargetSessionIdentity, adapterExecutionTargetToRemoteSpec, @@ -3041,6 +3042,170 @@ describe("sandbox adapter execution targets", () => { expect(control.stopCount).toBeGreaterThanOrEqual(1); }, 20000); + it("streams run logs on the duplex path under the same gate and log line as the file path", async () => { + const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-duplex-runlog-")); + cleanupDirs.push(rootDir); + const remoteCwd = path.join(rootDir, "workspace"); + await mkdir(remoteCwd, { recursive: true }); + const api = await startRecordingApiServer(); + const { runner } = makeDuplexSelectionRunner(); + const logs: Array<{ stream: "stdout" | "stderr"; chunk: string }> = []; + const target: AdapterSandboxExecutionTarget = { + kind: "remote", + transport: "sandbox", + providerKey: "daytona", + remoteCwd, + timeoutMs: 30_000, + runner, + streamRunLogs: true, + effectiveCapabilities: duplexCapabilities(true), + }; + + const bridge = await startAdapterExecutionTargetPaperclipBridge({ + runId: "run-duplex-log", + target, + runtimeRootDir: path.join(remoteCwd, ".paperclip-runtime", "codex"), + adapterKey: "codex", + hostApiToken: "real-run-jwt", + hostApiUrl: api.origin, + enableSandboxDuplexBridge: true, + onLog: async (stream, chunk) => { + logs.push({ stream, chunk }); + }, + }); + try { + // The duplex transport served, and it still streams run logs with the same + // gate and the same log line as the file path. + expect(bridge?.env.PAPERCLIP_API_BRIDGE_MODE).toBe("duplex_v1"); + expect(bridge?.runLogTail).toBeTruthy(); + expect(combinedStream(logs, "stdout")).toContain("Sandbox run log streaming enabled"); + const wrapped = bridge!.runLogTail!.create().wrapCommand("agent-cli", ["--message", "hello world"]); + expect(wrapped.args.join("\n")).toContain("tee -a"); + expect(wrapped.args.join("\n")).toContain("agent-cli"); + } finally { + await bridge?.stop(); + await api.close(); + } + }, 20000); + + it("returns no run-log tail on the duplex path when streaming is opted out", async () => { + const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-duplex-runlog-off-")); + cleanupDirs.push(rootDir); + const remoteCwd = path.join(rootDir, "workspace"); + await mkdir(remoteCwd, { recursive: true }); + const api = await startRecordingApiServer(); + const { runner } = makeDuplexSelectionRunner(); + const target: AdapterSandboxExecutionTarget = { + kind: "remote", + transport: "sandbox", + providerKey: "daytona", + remoteCwd, + timeoutMs: 30_000, + runner, + streamRunLogs: false, + effectiveCapabilities: duplexCapabilities(true), + }; + + const bridge = await startAdapterExecutionTargetPaperclipBridge({ + runId: "run-duplex-log-off", + target, + runtimeRootDir: path.join(remoteCwd, ".paperclip-runtime", "codex"), + adapterKey: "codex", + hostApiToken: "real-run-jwt", + hostApiUrl: api.origin, + enableSandboxDuplexBridge: true, + }); + try { + expect(bridge?.env.PAPERCLIP_API_BRIDGE_MODE).toBe("duplex_v1"); + expect(bridge?.runLogTail ?? null).toBeNull(); + } finally { + await bridge?.stop(); + await api.close(); + } + }, 20000); + + it("routes duplex channel-open and fallback records to a recorder attached on the server seam", async () => { + const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-duplex-recorder-")); + cleanupDirs.push(rootDir); + const remoteCwd = path.join(rootDir, "workspace"); + await mkdir(remoteCwd, { recursive: true }); + const api = await startRecordingApiServer(); + + const counters: DuplexTelemetryCounterRecord[] = []; + const recorder: DuplexTelemetryRecorder = { + recordSpan() {}, + incrementCounter(record) { + counters.push(record); + }, + emitEvent() {}, + }; + + // A channel open reaches the recorder on the duplex success path. The host + // attaches the recorder to the sandbox target on the same seam as the + // runner; the caller reads it with the accessor and passes it to the bridge. + const openTarget: AdapterSandboxExecutionTarget = { + kind: "remote", + transport: "sandbox", + providerKey: "daytona", + remoteCwd, + timeoutMs: 30_000, + runner: makeDuplexSelectionRunner().runner, + effectiveCapabilities: duplexCapabilities(true), + duplexTelemetryRecorder: recorder, + }; + const openBridge = await startAdapterExecutionTargetPaperclipBridge({ + runId: "run-duplex-open", + target: openTarget, + runtimeRootDir: path.join(remoteCwd, ".paperclip-runtime", "codex"), + adapterKey: "codex", + hostApiToken: "real-run-jwt", + hostApiUrl: api.origin, + enableSandboxDuplexBridge: true, + duplexTelemetryRecorder: adapterExecutionTargetDuplexTelemetryRecorder(openTarget), + }); + try { + expect(openBridge?.env.PAPERCLIP_API_BRIDGE_MODE).toBe("duplex_v1"); + const open = counters.find((record) => record.metric === DUPLEX_COUNTER_CHANNEL_OPEN_TOTAL); + expect(open?.dimensions.transport).toBe("duplex"); + expect(open?.dimensions.provider).toBe("daytona"); + } finally { + await openBridge?.stop(); + } + + // A fallback reaches the same recorder. The kill switch off records a + // gate_off fallback with the file transport. + counters.length = 0; + const fallbackTarget: AdapterSandboxExecutionTarget = { + kind: "remote", + transport: "sandbox", + providerKey: "daytona", + remoteCwd, + timeoutMs: 30_000, + runner: makeDuplexSelectionRunner().runner, + effectiveCapabilities: duplexCapabilities(true), + duplexTelemetryRecorder: recorder, + }; + const fallbackBridge = await startAdapterExecutionTargetPaperclipBridge({ + runId: "run-duplex-fallback", + target: fallbackTarget, + runtimeRootDir: path.join(remoteCwd, ".paperclip-runtime", "codex"), + adapterKey: "codex", + hostApiToken: "real-run-jwt", + hostApiUrl: api.origin, + enableSandboxDuplexBridge: false, + duplexTelemetryRecorder: adapterExecutionTargetDuplexTelemetryRecorder(fallbackTarget), + }); + try { + expect(fallbackBridge?.env.PAPERCLIP_API_BRIDGE_MODE).toBe("queue_v1"); + const fallback = counters.find((record) => record.metric === DUPLEX_COUNTER_FALLBACK_TOTAL); + expect(fallback?.dimensions.fallback_reason).toBe("gate_off"); + expect(fallback?.dimensions.transport).toBe("file"); + } finally { + await fallbackBridge?.stop(); + await api.close(); + } + }, 20000); + it.each([ { name: "the kill switch is off with the capability granted", enable: false, capability: true }, { name: "the capability is absent with the kill switch on", enable: true, capability: false }, diff --git a/packages/adapter-utils/src/execution-target.ts b/packages/adapter-utils/src/execution-target.ts index 4d93f11db01..f9a2749a1be 100644 --- a/packages/adapter-utils/src/execution-target.ts +++ b/packages/adapter-utils/src/execution-target.ts @@ -173,6 +173,13 @@ export interface AdapterSandboxExecutionTarget extends AdapterExecutionTargetWor * set to `false` to explicitly opt out back to batch-at-end delivery. */ streamRunLogs?: boolean | null; + /** + * The injected duplex telemetry recorder for this run. The host attaches it on + * the same seam as `runner`, so this live object stays on the host and never + * enters the sandbox environment. The bridge binds it to the fixed duplex + * observability surface. Absent means the safe no-op default. + */ + duplexTelemetryRecorder?: DuplexTelemetryRecorder | null; } export type AdapterExecutionTarget = @@ -365,6 +372,19 @@ export function adapterExecutionTargetEnablesSandboxDuplexBridge( ); } +/** + * Read the injected duplex telemetry recorder off a target. Only a sandbox + * target with a recorder attached returns it. Every other target returns null, + * so the bridge falls back to the safe no-op recorder. + */ +export function adapterExecutionTargetDuplexTelemetryRecorder( + target: AdapterExecutionTarget | null | undefined, +): DuplexTelemetryRecorder | null { + return target?.kind === "remote" && target.transport === "sandbox" + ? target.duplexTelemetryRecorder ?? null + : null; +} + export function adapterExecutionTargetRemoteCwd( target: AdapterExecutionTarget | null | undefined, localCwd: string, @@ -2340,6 +2360,37 @@ function deriveNestedDuplexBrokerBudgets(forwardTimeoutMs: number): DuplexBroker return { forwardTimeoutMs, responseBudgetMs, gatewayWaitMs }; } +/** + * Create the run-log directory on the sandbox before the tail starts. The file + * bridge worker creates this directory on the file path. The duplex path starts + * no worker, so the host creates the directory here. The tail then reads a real + * directory from its first tick. + * + * This step is best effort. The broker already serves the duplex transport when + * the host reaches it, and the tail wrap command runs its own `mkdir -p` as a + * backstop. So a create failure must not tear down a working duplex transport; + * the host swallows it and still builds the tail. The log line names no raw + * error, so no raw error rides a log line here. + */ +async function ensureSandboxRunLogDirectory(input: { + runner: CommandManagedRuntimeRunner; + remoteCwd: string; + logsDir: string; + shellCommand: "bash" | "sh"; + timeoutMs: number | null | undefined; +}): Promise { + try { + await input.runner.execute({ + command: input.shellCommand, + args: shellCommandArgs(`mkdir -p ${shellQuote(input.logsDir)}`), + cwd: input.remoteCwd, + timeoutMs: input.timeoutMs ?? undefined, + }); + } catch { + // Best effort: the tail wrap command creates the directory before it writes. + } +} + /** * The duplex readiness gate. The gate owns the single data listener and the * single exit listener of the channel while the host waits for a valid READY @@ -2877,13 +2928,34 @@ export async function startAdapterExecutionTargetPaperclipBridge(input: { "stdout", "[paperclip] Sandbox duplex transport ready; serving the host-assigned origin.\n", ); + // Stream run logs on the duplex path with the same gate and the same + // log line as the file path. The duplex path starts no file-bridge + // worker, so create the log directory before the tail starts. + let duplexRunLogTail: SandboxRunLogTailFactory | null = null; + if (target.transport === "sandbox" && target.streamRunLogs !== false) { + const duplexLogsDir = sandboxCallbackBridgeDirectories(queueDir).logsDir; + await ensureSandboxRunLogDirectory({ + runner, + remoteCwd: target.remoteCwd, + logsDir: duplexLogsDir, + shellCommand, + timeoutMs: bridgeTimeoutMs, + }); + duplexRunLogTail = createSandboxRunLogTailFactory({ + runner, + remoteCwd: target.remoteCwd, + logsDir: duplexLogsDir, + shellCommand, + }); + await onLog("stdout", "[paperclip] Sandbox run log streaming enabled for this run.\n"); + } return { env: { PAPERCLIP_API_URL: sandboxOrigin, PAPERCLIP_API_KEY: bridgeToken, PAPERCLIP_API_BRIDGE_MODE: SANDBOX_CALLBACK_BRIDGE_DUPLEX_MODE, }, - runLogTail: null, + runLogTail: duplexRunLogTail, stop: async () => { // Close the channel before lease release. The broker sends an orderly // close and releases the route, then stops the child, so no live diff --git a/packages/adapters/claude-local/src/server/execute.ts b/packages/adapters/claude-local/src/server/execute.ts index 54e715ce325..4eadb926b6e 100644 --- a/packages/adapters/claude-local/src/server/execute.ts +++ b/packages/adapters/claude-local/src/server/execute.ts @@ -15,6 +15,7 @@ import { ensureAdapterExecutionTargetCommandResolvable, ensureAdapterExecutionTargetRuntimeCommandInstalled, prepareAdapterExecutionTargetRuntime, + adapterExecutionTargetDuplexTelemetryRecorder, adapterExecutionTargetEnablesSandboxDuplexBridge, readAdapterExecutionTarget, resolveAdapterExecutionTargetTimeoutSec, @@ -683,6 +684,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise; endTime?: number }>; +} { + const spans: Array<{ name: string; startTime?: number; attributes: Record; endTime?: number }> = []; + const tracer: DuplexTelemetryTracer = { + startSpan(name, options) { + const record = { name, startTime: options?.startTime, attributes: {} as Record, endTime: undefined as number | undefined }; + spans.push(record); + const span: DuplexTelemetrySpan = { + setAttribute(key, value) { + record.attributes[key] = value; + }, + end(endTime) { + record.endTime = endTime; + }, + }; + return span; + }, + }; + return { tracer, spans }; +} + +describe("createHostDuplexTelemetryRecorder", () => { + it("records the channel-open span with only the closed dimension keys", () => { + const { tracer, spans } = createRecordingTracer(); + const recorder = createHostDuplexTelemetryRecorder({ + tracer, + incrementCounter: () => {}, + emitTransportEvent: () => {}, + now: () => 1_000, + }); + + recorder.recordSpan({ + name: DUPLEX_SPAN_CHANNEL_OPEN, + dimensions: { provider: "daytona", transport: "duplex", outcome: "ok" }, + }); + + expect(spans).toHaveLength(1); + expect(spans[0].name).toBe(DUPLEX_SPAN_CHANNEL_OPEN); + expect(spans[0].attributes).toEqual({ provider: "daytona", transport: "duplex", outcome: "ok" }); + // The channel-open span carries no latency, so it opens and ends at the same instant. + expect(spans[0].startTime).toBe(1_000); + expect(spans[0].endTime).toBe(1_000); + }); + + it("makes the request span duration equal the measured latency", () => { + const { tracer, spans } = createRecordingTracer(); + const recorder = createHostDuplexTelemetryRecorder({ + tracer, + incrementCounter: () => {}, + emitTransportEvent: () => {}, + now: () => 5_000, + }); + + recorder.recordSpan({ + name: DUPLEX_SPAN_REQUEST, + dimensions: { provider: "other", transport: "duplex", outcome: "error" }, + latencyMs: 250, + }); + + expect(spans[0].startTime).toBe(4_750); + expect(spans[0].endTime).toBe(5_000); + }); + + it("folds the discriminating dimension into the counter metric name", () => { + expect( + foldDuplexCounterMetric({ + metric: DUPLEX_COUNTER_CHANNEL_OPEN_TOTAL, + dimensions: { provider: "daytona", transport: "duplex", outcome: "ok" }, + }), + ).toBe(DUPLEX_COUNTER_CHANNEL_OPEN_TOTAL); + expect( + foldDuplexCounterMetric({ + metric: DUPLEX_COUNTER_FALLBACK_TOTAL, + dimensions: { provider: "daytona", transport: "file", outcome: "error", fallback_reason: "gate_off" }, + }), + ).toBe(`${DUPLEX_COUNTER_FALLBACK_TOTAL}.gate_off`); + expect( + foldDuplexCounterMetric({ + metric: DUPLEX_COUNTER_LOSS_TOTAL, + dimensions: { provider: "daytona", transport: "duplex", outcome: "error", loss_class: "post_dispatch" }, + }), + ).toBe(`${DUPLEX_COUNTER_LOSS_TOTAL}.post_dispatch`); + }); + + it("forwards the counter through the guarded sink with the folded metric", () => { + const metrics: string[] = []; + const recorder = createHostDuplexTelemetryRecorder({ + tracer: createRecordingTracer().tracer, + incrementCounter: (metric) => metrics.push(metric), + emitTransportEvent: () => {}, + }); + + recorder.incrementCounter({ + metric: DUPLEX_COUNTER_FALLBACK_TOTAL, + dimensions: { provider: "daytona", transport: "file", outcome: "error", fallback_reason: "ready_timeout" }, + }); + + expect(metrics).toEqual([`${DUPLEX_COUNTER_FALLBACK_TOTAL}.ready_timeout`]); + }); + + it("forwards the transport event with its name and dimensions", () => { + const events: Array<{ name: string; dimensions: Record }> = []; + const recorder = createHostDuplexTelemetryRecorder({ + tracer: createRecordingTracer().tracer, + incrementCounter: () => {}, + emitTransportEvent: (event) => events.push(event), + }); + + recorder.emitEvent({ + name: DUPLEX_TRANSPORT_EVENT, + dimensions: { provider: "daytona", transport: "duplex", outcome: "ok" }, + }); + + expect(events).toEqual([ + { name: DUPLEX_TRANSPORT_EVENT, dimensions: { provider: "daytona", transport: "duplex", outcome: "ok" } }, + ]); + }); +}); diff --git a/server/src/services/duplex-telemetry-recorder.ts b/server/src/services/duplex-telemetry-recorder.ts new file mode 100644 index 00000000000..8a4192dae66 --- /dev/null +++ b/server/src/services/duplex-telemetry-recorder.ts @@ -0,0 +1,113 @@ +import { + DUPLEX_DIMENSION_KEYS, + DUPLEX_SPAN_REQUEST, + type DuplexTelemetryCounterRecord, + type DuplexTelemetryDimensions, + type DuplexTelemetryEventRecord, + type DuplexTelemetryRecorder, + type DuplexTelemetrySpanRecord, +} from "@paperclipai/adapter-utils/duplex-telemetry"; + +/** + * The host binding for the fixed duplex telemetry surface. This module maps each + * recorder call to a real host sink: the span to the OTel tracer, the counter to + * the guarded counter store in `tool-runtime-metrics.ts`, and the transport event + * to the run-event path. The mapping is the single boundary, so the closed + * dimension keys and the counter folding stay in one place. + * + * The recorder never throws. The facade in `duplex-telemetry.ts` wraps each + * synchronous call in a swallow, but an async sink can still reject after the + * call returns. Each async sink here runs fire-and-forget with its own catch, so + * a sink failure never breaks the request path. + */ + +/** The minimal span the recorder opens. A real OTel span satisfies it; the + * no-op tracer's span satisfies it too. */ +export interface DuplexTelemetrySpan { + setAttribute(key: string, value: string | number | boolean): void; + end(endTime?: number): void; +} + +/** The minimal tracer surface the recorder calls. `getStartupTracer` returns a + * real or a no-op implementation that satisfies it. The optional second argument + * carries an explicit start time, so the request span duration equals the + * measured latency. */ +export interface DuplexTelemetryTracer { + startSpan(name: string, options?: { startTime?: number }): DuplexTelemetrySpan; +} + +export interface HostDuplexTelemetryRecorderInput { + /** The OTel tracer for the two duplex spans. */ + tracer: DuplexTelemetryTracer; + /** + * Increment one guarded host counter by its metric name. The caller binds it + * to `incrementToolRuntimeMetricCounter` with the company id, inside a swallow. + */ + incrementCounter(metric: string): void; + /** + * Emit one transport event to the run-event path. The caller binds it to the + * run-events bridge, inside a swallow. + */ + emitTransportEvent(event: { name: string; dimensions: DuplexTelemetryDimensions }): void; + /** The wall clock. The default is `Date.now`. Tests inject a fixed clock. */ + now?: () => number; +} + +/** + * Fold the discriminating dimension into the counter metric name. The guarded + * counter store keys on `(companyId, metric)` with no label column, so the + * fallback reason or the loss class rides the metric name to keep the analytic + * breakdown. Both dimension sets are closed and low-cardinality, so the metric + * cardinality stays bounded. A record with neither dimension uses the base name. + */ +export function foldDuplexCounterMetric(record: DuplexTelemetryCounterRecord): string { + if (record.dimensions.fallback_reason) { + return `${record.metric}.${record.dimensions.fallback_reason}`; + } + if (record.dimensions.loss_class) { + return `${record.metric}.${record.dimensions.loss_class}`; + } + return record.metric; +} + +/** + * Build the host duplex telemetry recorder. The recorder receives only + * already-mapped dimensions, so a raw provider key never reaches a sink. Every + * span attribute uses only the closed dimension keys. + */ +export function createHostDuplexTelemetryRecorder( + input: HostDuplexTelemetryRecorderInput, +): DuplexTelemetryRecorder { + const now = input.now ?? Date.now; + + const setDimensionAttributes = (span: DuplexTelemetrySpan, dimensions: DuplexTelemetryDimensions): void => { + for (const key of DUPLEX_DIMENSION_KEYS) { + const value = dimensions[key]; + if (typeof value === "string") { + span.setAttribute(key, value); + } + } + }; + + return { + recordSpan(record: DuplexTelemetrySpanRecord): void { + // The request span carries a latency, so start it in the past and end it + // now, so the span duration equals the measured latency. The channel-open + // span carries no latency, so it opens and ends at the same instant. + const end = now(); + const latencyMs = + record.name === DUPLEX_SPAN_REQUEST && typeof record.latencyMs === "number" && Number.isFinite(record.latencyMs) + ? Math.max(0, record.latencyMs) + : 0; + const span = input.tracer.startSpan(record.name, { startTime: end - latencyMs }); + setDimensionAttributes(span, record.dimensions); + span.end(end); + }, + incrementCounter(record: DuplexTelemetryCounterRecord): void { + input.incrementCounter(foldDuplexCounterMetric(record)); + }, + emitEvent(record: DuplexTelemetryEventRecord): void { + input.emitTransportEvent({ name: record.name, dimensions: record.dimensions }); + }, + }; +} diff --git a/server/src/services/environment-execution-target.ts b/server/src/services/environment-execution-target.ts index 672ae9b76e0..90fc7ac3c9b 100644 --- a/server/src/services/environment-execution-target.ts +++ b/server/src/services/environment-execution-target.ts @@ -5,6 +5,7 @@ import { adapterExecutionTargetToRemoteSpec, type AdapterExecutionTarget, } from "@paperclipai/adapter-utils/execution-target"; +import type { DuplexTelemetryRecorder } from "@paperclipai/adapter-utils/duplex-telemetry"; import { clampSpanLabel, getActiveStepContext, @@ -200,6 +201,11 @@ export async function resolveEnvironmentExecutionTarget(input: { // gated server tracer, which is a no-op when tracing is off. Tests inject a // recording tracer. tracer?: ExecTracer; + // The host duplex telemetry recorder. The seam stamps it onto the sandbox + // target next to the runner, so the live object stays on the host and never + // enters the sandbox environment. Absent keeps the safe no-op default in the + // bridge, so the surface stays inert until the host injects a real recorder. + duplexTelemetryRecorder?: DuplexTelemetryRecorder | null; }): Promise { if (input.environment.driver === "local") { return { @@ -313,6 +319,10 @@ export async function resolveEnvironmentExecutionTarget(input: { shellCommand, remoteCwd, enableSandboxDuplexBridge, + // Attach the host duplex telemetry recorder next to the runner. The bridge + // binds it to the fixed observability surface. Absent keeps the no-op + // default, so the surface stays inert on a run with no injected recorder. + duplexTelemetryRecorder: input.duplexTelemetryRecorder ?? null, ...(effectiveCapabilities ? { effectiveCapabilities: Object.freeze({ ...effectiveCapabilities }) } : {}), environmentId: input.environment.id ?? null, leaseId: input.leaseId ?? null, diff --git a/server/src/services/environment-run-orchestrator.ts b/server/src/services/environment-run-orchestrator.ts index c907d4c177e..3a76dcfec13 100644 --- a/server/src/services/environment-run-orchestrator.ts +++ b/server/src/services/environment-run-orchestrator.ts @@ -42,6 +42,7 @@ import { type AdapterRemoteExecutionSpec, type AdapterWorkspaceRealization, } from "@paperclipai/adapter-utils/execution-target"; +import type { DuplexTelemetryRecorder } from "@paperclipai/adapter-utils/duplex-telemetry"; import { buildWorkspaceRealizationRequest } from "./workspace-realization.js"; import { executionWorkspaceService } from "./execution-workspaces.js"; import { logActivity } from "./activity-log.js"; @@ -347,6 +348,12 @@ export function environmentRunOrchestrator( executionWorkspace: RealizedExecutionWorkspace; effectiveExecutionWorkspaceMode: string | null; persistedExecutionWorkspace: ExecutionWorkspace | null; + /** + * The host duplex telemetry recorder for this run. The orchestrator threads + * it to `resolveEnvironmentExecutionTarget`, which stamps it on the sandbox + * target. Absent keeps the safe no-op default in the bridge. + */ + duplexTelemetryRecorder?: DuplexTelemetryRecorder | null; }): Promise { const { environment, @@ -515,6 +522,7 @@ export function environmentRunOrchestrator( leaseMetadata: (lease.metadata as Record | null) ?? null, lease, environmentRuntime, + duplexTelemetryRecorder: input.duplexTelemetryRecorder ?? null, }); const realizationMode = workspaceRealization.mode === "in_place" ? "in_place" : "copy"; const authoritativeRoot = diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 6a74c765c3d..c3560ee9193 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -69,7 +69,9 @@ import { workspaceOperations, } from "@paperclipai/db"; import { conflict, HttpError, notFound } from "../errors.js"; -import { getStartupTraceContext } from "../instrumentation.js"; +import { getStartupTraceContext, getStartupTracer } from "../instrumentation.js"; +import { createHostDuplexTelemetryRecorder } from "./duplex-telemetry-recorder.js"; +import { incrementToolRuntimeMetricCounter } from "./tool-runtime-metrics.js"; import { logger } from "../middleware/logger.js"; import { createGitRemoteAuthProvider, @@ -15178,6 +15180,33 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) lease: acquiredEnvironment.lease, leaseContext: acquiredEnvironment.leaseContext, }; + // The host duplex telemetry recorder for this run. It binds the fixed duplex + // observability surface to real sinks: the spans to the OTel tracer, the + // guarded counters to the tool-runtime metric store, and the transport event + // to the run-event path. Each sink runs guarded and fire-and-forget, so a + // telemetry failure never breaks the run. The orchestrator stamps it on the + // sandbox target; a non-duplex run keeps the safe no-op default in the bridge. + const duplexTelemetryRecorder = createHostDuplexTelemetryRecorder({ + tracer: getStartupTracer(), + incrementCounter: (metric) => { + void incrementToolRuntimeMetricCounter(db, { + companyId: run.companyId, + metric, + }).catch(() => {}); + }, + emitTransportEvent: (event) => { + void (async () => { + const eventRun = run; + const seq = await nextRunEventSeq(eventRun.id); + await appendRunEvent(eventRun, seq, { + eventType: event.name, + stream: "system", + level: event.dimensions.outcome === "error" ? "warn" : "info", + payload: { ...event.dimensions }, + }); + })().catch(() => {}); + }, + }); const realizationResult = await envOrchestrator.realizeForRun({ environment: selectedEnvironment, lease: activeEnvironmentLease.lease, @@ -15188,6 +15217,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) executionWorkspace, effectiveExecutionWorkspaceMode, persistedExecutionWorkspace, + duplexTelemetryRecorder, }); activeEnvironmentLease = { ...activeEnvironmentLease,