feat(sandbox): stream duplex run logs and bind the duplex telemetry recorder

Stream sandbox run logs on the duplex transport with the same gate and the
same log line as the file transport. The duplex path starts no file bridge
worker, so the host creates the log directory before the tail starts. The
step is best effort and never tears down a working duplex transport.

Bind the fixed duplex telemetry surface to real host sinks. The host builds
one recorder per run: the spans go to the OTel tracer, the guarded counters
go to the tool-runtime metric store, and the transport event goes to the
run-event path. Each sink runs guarded and fire-and-forget, so a telemetry
failure never breaks the request path. The server attaches the recorder to
the sandbox target next to the runner; a run with no recorder keeps the safe
no-op default. Update the telemetry README to name the counter store file.

Co-authored-by: Paperclip <noreply@paperclip.ing>

server/src/__tests__/duplex-telemetry-recorder.test.ts

@@ -0,0 +1,138 @@
+import { describe, expect, it } from "vitest";
+
+import {
+  DUPLEX_COUNTER_CHANNEL_OPEN_TOTAL,
+  DUPLEX_COUNTER_FALLBACK_TOTAL,
+  DUPLEX_COUNTER_LOSS_TOTAL,
+  DUPLEX_SPAN_CHANNEL_OPEN,
+  DUPLEX_SPAN_REQUEST,
+  DUPLEX_TRANSPORT_EVENT,
+} from "@paperclipai/adapter-utils/duplex-telemetry";
+import {
+  createHostDuplexTelemetryRecorder,
+  foldDuplexCounterMetric,
+  type DuplexTelemetrySpan,
+  type DuplexTelemetryTracer,
+} from "../services/duplex-telemetry-recorder.js";
+
+// A recording tracer that captures each span's name, attributes, and end time.
+function createRecordingTracer(): {
+  tracer: DuplexTelemetryTracer;
+  spans: Array<{ name: string; startTime?: number; attributes: Record<string, string | number | boolean>; endTime?: number }>;
+} {
+  const spans: Array<{ name: string; startTime?: number; attributes: Record<string, string | number | boolean>; endTime?: number }> = [];
+  const tracer: DuplexTelemetryTracer = {
+    startSpan(name, options) {
+      const record = { name, startTime: options?.startTime, attributes: {} as Record<string, string | number | boolean>, 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<string, unknown> }> = [];
+    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" } },
+    ]);
+  });
+});

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 });
+    },
+  };
+}

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<AdapterExecutionTarget | null> {
   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,

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<EnvironmentRealizationResult> {
     const {
       environment,
@@ -515,6 +522,7 @@ export function environmentRunOrchestrator(
         leaseMetadata: (lease.metadata as Record<string, unknown> | null) ?? null,
         lease,
         environmentRuntime,
+        duplexTelemetryRecorder: input.duplexTelemetryRecorder ?? null,
       });
       const realizationMode = workspaceRealization.mode === "in_place" ? "in_place" : "copy";
       const authoritativeRoot =

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,