From c881bc460cac5fb9a7aed53b000771bc6df43a5d Mon Sep 17 00:00:00 2001 From: Priya Raman Date: Thu, 20 Aug 2026 04:57:22 +0000 Subject: [PATCH] feat(sandbox): add fixed duplex observability with a redacted telemetry surface Add the closed telemetry contract for the sandbox duplex transport: the two fixed spans, the four guarded counters, and the one transport event. A new provider-scoped facade maps each record to the fixed names and dimension keys. It maps the raw provider key through a one-value allowlist before every sink, so a plugin key never reaches a span, a counter, or the event; a non-approved key maps to the constant other. Wire the facade into the transport selection path and the broker. The selection path records the channel-open span, the fallback counter, and the transport event, and it maps each readiness failure to a fixed fallback reason. The broker records one request span with latency per delivered request and one loss record per terminal loss, classified as pre_dispatch or post_dispatch. Keep the per-request debug output off the duplex path: suppress the route and query proxy logs and pass no free-form logger to the broker, so no route, query, body, token, or raw provider error rides a log line, a span, an event, or a counter. Every recorder call sits inside an error swallow, so a telemetry failure never breaks the request path. The default recorder is a no-op, so the surface stays inert until the host binds a real recorder. Co-authored-by: Paperclip --- .../adapter-utils/src/duplex-bridge-broker.ts | 39 +- .../adapter-utils/src/duplex-telemetry.ts | 290 ++++++++++ .../src/execution-target-sandbox.test.ts | 498 ++++++++++++++++++ .../adapter-utils/src/execution-target.ts | 90 +++- 4 files changed, 905 insertions(+), 12 deletions(-) create mode 100644 packages/adapter-utils/src/duplex-telemetry.ts diff --git a/packages/adapter-utils/src/duplex-bridge-broker.ts b/packages/adapter-utils/src/duplex-bridge-broker.ts index fa615e3bae1..95aad013efa 100644 --- a/packages/adapter-utils/src/duplex-bridge-broker.ts +++ b/packages/adapter-utils/src/duplex-bridge-broker.ts @@ -38,6 +38,7 @@ import { type DuplexResponseFrame, type DuplexResponseOutcome, } from "./duplex-frame-codec.js"; +import type { DuplexOutcomeValue, DuplexTelemetry } from "./duplex-telemetry.js"; /** The lifecycle states of the broker. The broker moves through them in order. */ export type DuplexBrokerState = "opening" | "open" | "lost" | "closing" | "closed"; @@ -132,6 +133,13 @@ export interface DuplexBrokerOptions { onStateChange?: (state: DuplexBrokerState) => void; /** The sink for a diagnostic message. The broker never writes diagnostics to the channel. */ logger?: (message: string) => void; + /** + * The fixed observability facade. The broker records one request span per + * delivered request and one loss record per terminal loss. The facade maps each + * record to the fixed names and dimensions, so no route, query, body, token, or + * raw error rides a span or a counter. The default records nothing. + */ + telemetry?: DuplexTelemetry; } /** The broker handle the factory returns. */ @@ -176,6 +184,8 @@ interface PendingRequest { responded: boolean; forwardTimer: ReturnType; responseTimer: ReturnType; + /** The point the broker started to dispatch the request. It sets the span latency. */ + dispatchStartMs: number; } /** @@ -235,12 +245,18 @@ export function createDuplexBridgeBroker(options: DuplexBrokerOptions): DuplexBr // Loss is terminal. Record it one time and stop every activity. if (stopped) return; stopped = true; + // Classify the loss relative to the first dispatch. A loss after the broker + // dispatched a request is `post_dispatch`; a loss before any dispatch is + // `pre_dispatch`. The class rides the fixed loss counter, never the raw + // message. + const lossClass = seenRequestIds.size > 0 ? "post_dispatch" : "pre_dispatch"; clearHeartbeat(); clearPending(); lossRecord = { reason, message, atMs: now() }; setState("lost"); options.logger?.(`Duplex broker lost the channel (${reason}): ${message}`); options.onLoss?.(lossRecord); + options.telemetry?.recordLoss(lossClass); }; const writeFrame = (frame: DuplexFrame): boolean => { @@ -257,6 +273,7 @@ export function createDuplexBridgeBroker(options: DuplexBrokerOptions): DuplexBr id: string, result: DuplexBrokerForwardResult, outcome: DuplexResponseOutcome, + telemetryOutcome: DuplexOutcomeValue, ): void => { const entry = pending.get(id); if (!entry || entry.responded) return; @@ -267,6 +284,12 @@ export function createDuplexBridgeBroker(options: DuplexBrokerOptions): DuplexBr // Do not write on a lost or closed channel. The gateway answers its own // outstanding request on loss, so a late write would go to a dead channel. if (state !== "open") return; + // Record the request span for the delivered request. The span carries the + // latency and the outcome only; no route, query, body, or token rides it. + options.telemetry?.recordRequest({ + latencyMs: now() - entry.dispatchStartMs, + outcome: telemetryOutcome, + }); const frame: DuplexResponseFrame = { version: DUPLEX_FRAME_VERSION, type: "response", @@ -317,11 +340,18 @@ export function createDuplexBridgeBroker(options: DuplexBrokerOptions): DuplexBr }), }, "indeterminate", + "error", ); }, budgets.responseBudgetMs); forwardTimer.unref?.(); responseTimer.unref?.(); - pending.set(frame.id, { controller, responded: false, forwardTimer, responseTimer }); + pending.set(frame.id, { + controller, + responded: false, + forwardTimer, + responseTimer, + dispatchStartMs: record.dispatchStartMs, + }); forwardRequest(frame, { signal: controller.signal }).then( (result) => { @@ -332,7 +362,10 @@ export function createDuplexBridgeBroker(options: DuplexBrokerOptions): DuplexBr result.headers?.["x-paperclip-bridge-outcome"] === "indeterminate" ? "indeterminate" : "completed"; - respond(frame.id, result, outcome); + // The host delivered a real response, so the request span outcome is `ok`. + // A host application status (200, a 4xx, a 5xx) is still a delivered + // response; only a broker-synthesized failure below is `error`. + respond(frame.id, result, outcome, "ok"); }, (error) => { if (controller.signal.aborted) { @@ -354,6 +387,7 @@ export function createDuplexBridgeBroker(options: DuplexBrokerOptions): DuplexBr }), }, "indeterminate", + "error", ); return; } @@ -368,6 +402,7 @@ export function createDuplexBridgeBroker(options: DuplexBrokerOptions): DuplexBr body: JSON.stringify({ error: errorMessage(error) }), }, "completed", + "error", ); }, ); diff --git a/packages/adapter-utils/src/duplex-telemetry.ts b/packages/adapter-utils/src/duplex-telemetry.ts new file mode 100644 index 00000000000..156e8e36892 --- /dev/null +++ b/packages/adapter-utils/src/duplex-telemetry.ts @@ -0,0 +1,290 @@ +/** + * The fixed observability surface for the sandbox duplex transport. + * + * This module owns the one closed contract for every duplex telemetry sink: the + * span names, the counter names, the one event name, the dimension keys, and the + * enum values. Each name and value is a literal constant here, so the surface + * never drifts and a test asserts the exact set. + * + * The module also owns the provider allowlist. The `provider` dimension carries + * one approved public value, `daytona`. Any other plugin key maps to the constant + * `other`. The map runs inside this module before every sink, so a raw plugin key + * never reaches a span attribute, a counter label, or an event field. + * + * The module stays free of `@opentelemetry/api` and of the database. The host + * injects a {@link DuplexTelemetryRecorder}; the default is a no-op recorder, so + * the whole surface stays inert until the host binds a real recorder. Every + * recorder call sits inside an error swallow, so a telemetry failure never breaks + * the request path. + */ + +/** The span for one duplex channel-open attempt. */ +export const DUPLEX_SPAN_CHANNEL_OPEN = "sandbox.duplex.channel_open"; +/** The span for one duplex request. It carries the request latency. */ +export const DUPLEX_SPAN_REQUEST = "sandbox.duplex.request"; + +/** The one duplex transport event. The host emits it at each transport boundary. */ +export const DUPLEX_TRANSPORT_EVENT = "sandbox.duplex.transport"; + +/** The guarded counter for one successful channel open. */ +export const DUPLEX_COUNTER_CHANNEL_OPEN_TOTAL = "sandbox_duplex_channel_open_total"; +/** The guarded counter for one fallback to the file bridge. */ +export const DUPLEX_COUNTER_FALLBACK_TOTAL = "sandbox_duplex_fallback_total"; +/** The guarded counter for one terminal channel loss. */ +export const DUPLEX_COUNTER_LOSS_TOTAL = "sandbox_duplex_loss_total"; +/** The guarded counter for one leaked provider session on teardown. */ +export const DUPLEX_COUNTER_SESSION_LEAK_TOTAL = "sandbox_duplex_session_leak_total"; + +/** + * The closed dimension-key set. Every span attribute, counter label, and event + * field uses only these keys. A test asserts the exact set, so a new key never + * reaches a sink by accident. + */ +export const DUPLEX_DIMENSION_KEYS = [ + "provider", + "transport", + "outcome", + "fallback_reason", + "loss_class", +] as const; + +/** One dimension key from the closed set. */ +export type DuplexDimensionKey = (typeof DUPLEX_DIMENSION_KEYS)[number]; + +/** The transport a record is about. */ +export type DuplexTransportValue = "duplex" | "file"; + +/** The outcome of a record. */ +export type DuplexOutcomeValue = "ok" | "error"; + +/** The reason the host selected the file bridge instead of the duplex transport. */ +export type DuplexFallbackReason = + | "gate_off" + | "capability_absent" + | "open_failed" + | "ready_invalid" + | "ready_nonce_mismatch" + | "ready_timeout" + | "contaminated"; + +/** The class of a terminal loss, relative to the first request dispatch. */ +export type DuplexLossClass = "pre_dispatch" | "post_dispatch"; + +/** The one approved public provider value. */ +export const DUPLEX_APPROVED_PROVIDER = "daytona"; +/** The constant for any provider key outside the allowlist. */ +export const DUPLEX_PROVIDER_OTHER = "other"; + +/** The `provider` dimension value after the allowlist map. */ +export type DuplexProviderValue = typeof DUPLEX_APPROVED_PROVIDER | typeof DUPLEX_PROVIDER_OTHER; + +/** The host-owned closed provider allowlist. It holds one approved public value. */ +const APPROVED_PROVIDERS: ReadonlySet = new Set([DUPLEX_APPROVED_PROVIDER]); + +/** + * Map a raw provider key to the closed `provider` dimension value. Return the key + * when the allowlist holds it. Return `other` for any other value, so a raw + * plugin key never reaches a sink. A missing key also maps to `other`. + */ +export function normalizeDuplexProvider(key: string | null | undefined): DuplexProviderValue { + return typeof key === "string" && APPROVED_PROVIDERS.has(key) + ? (key as DuplexProviderValue) + : DUPLEX_PROVIDER_OTHER; +} + +/** + * The dimension bag a sink reads. Only the closed keys appear. The optional keys + * are present only when the record defines them, so a span or a counter never + * carries an empty dimension. + */ +export interface DuplexTelemetryDimensions { + provider: DuplexProviderValue; + transport: DuplexTransportValue; + outcome?: DuplexOutcomeValue; + fallback_reason?: DuplexFallbackReason; + loss_class?: DuplexLossClass; +} + +/** One span record the host records. The request span carries a latency. */ +export interface DuplexTelemetrySpanRecord { + name: string; + dimensions: DuplexTelemetryDimensions; + /** The request latency in milliseconds. Only the request span sets it. */ + latencyMs?: number; +} + +/** One counter increment the host records. */ +export interface DuplexTelemetryCounterRecord { + metric: string; + dimensions: DuplexTelemetryDimensions; +} + +/** One event the host emits. */ +export interface DuplexTelemetryEventRecord { + name: string; + dimensions: DuplexTelemetryDimensions; +} + +/** + * The injected low-level recorder. The host binds it to the existing telemetry + * pipeline: the span to the OTel tracer, the counter to the guarded counter + * store, and the event to the run-events bridge. The default is a no-op recorder. + * The recorder receives only already-mapped dimensions, so the raw provider key + * never reaches it. + */ +export interface DuplexTelemetryRecorder { + recordSpan(record: DuplexTelemetrySpanRecord): void; + incrementCounter(record: DuplexTelemetryCounterRecord): void; + emitEvent(record: DuplexTelemetryEventRecord): void; +} + +/** A no-op recorder. Every method does nothing, so the surface stays inert. */ +export const NOOP_DUPLEX_TELEMETRY_RECORDER: DuplexTelemetryRecorder = { + recordSpan() {}, + incrementCounter() {}, + emitEvent() {}, +}; + +/** One channel-open attempt. The caller reports exactly one terminal. */ +export interface DuplexChannelOpenAttempt { + /** + * The channel opened and readiness passed. Record the channel-open span with + * the `ok` outcome, increment the channel-open counter, and emit the transport + * event for the duplex transport. + */ + ready(): void; + /** + * The channel did not open or readiness failed. Record the channel-open span + * with the `error` outcome, increment the fallback counter, and emit the + * transport event for the file bridge. + */ + fallback(reason: DuplexFallbackReason): void; +} + +/** + * The bound telemetry facade the call sites use. It carries the normalized + * provider, so a call site never passes a raw key. It maps each semantic event to + * the fixed names and dimensions, then calls the recorder inside an error swallow. + */ +export interface DuplexTelemetry { + /** Begin a channel-open attempt. The caller reports `ready` or `fallback`. */ + startChannelOpen(): DuplexChannelOpenAttempt; + /** + * Record a fallback to the file bridge with no channel-open attempt. Use it for + * `gate_off` and `capability_absent`, where the host opens no channel. + */ + recordFallback(reason: DuplexFallbackReason): void; + /** Record one duplex request span with its latency and outcome. */ + recordRequest(record: { latencyMs: number; outcome: DuplexOutcomeValue }): void; + /** Record one terminal channel loss. */ + recordLoss(lossClass: DuplexLossClass): void; + /** Record one leaked provider session on teardown. */ + recordSessionLeak(): void; +} + +/** The options for {@link createDuplexTelemetry}. */ +export interface DuplexTelemetryOptions { + /** The injected recorder. The default is the no-op recorder. */ + recorder?: DuplexTelemetryRecorder | null; + /** The raw provider key. The facade maps it through the allowlist one time. */ + providerKey?: string | null; +} + +/** + * Build the bound telemetry facade. The facade normalizes the provider key one + * time, then reuses the mapped value for every record. Every recorder call sits + * inside a `try/catch`, so a throwing recorder never breaks the request path. A + * missing recorder yields a facade whose methods do nothing. + */ +export function createDuplexTelemetry(options: DuplexTelemetryOptions = {}): DuplexTelemetry { + const recorder = options.recorder ?? NOOP_DUPLEX_TELEMETRY_RECORDER; + const provider = normalizeDuplexProvider(options.providerKey); + + const safeSpan = (record: DuplexTelemetrySpanRecord): void => { + try { + recorder.recordSpan(record); + } catch { + // Observability must not break the request path. + } + }; + const safeCounter = (record: DuplexTelemetryCounterRecord): void => { + try { + recorder.incrementCounter(record); + } catch { + // Observability must not break the request path. + } + }; + const safeEvent = (record: DuplexTelemetryEventRecord): void => { + try { + recorder.emitEvent(record); + } catch { + // Observability must not break the request path. + } + }; + + const recordFallback = (reason: DuplexFallbackReason): void => { + const dimensions: DuplexTelemetryDimensions = { + provider, + transport: "file", + outcome: "error", + fallback_reason: reason, + }; + safeCounter({ metric: DUPLEX_COUNTER_FALLBACK_TOTAL, dimensions }); + safeEvent({ name: DUPLEX_TRANSPORT_EVENT, dimensions }); + }; + + return { + startChannelOpen(): DuplexChannelOpenAttempt { + let settled = false; + return { + ready(): void { + if (settled) return; + settled = true; + const dimensions: DuplexTelemetryDimensions = { + provider, + transport: "duplex", + outcome: "ok", + }; + safeSpan({ name: DUPLEX_SPAN_CHANNEL_OPEN, dimensions }); + safeCounter({ metric: DUPLEX_COUNTER_CHANNEL_OPEN_TOTAL, dimensions }); + safeEvent({ name: DUPLEX_TRANSPORT_EVENT, dimensions }); + }, + fallback(reason: DuplexFallbackReason): void { + if (settled) return; + settled = true; + // The channel-open span records the failed attempt on the duplex + // transport; the counter and the event record the file-bridge fallback. + safeSpan({ + name: DUPLEX_SPAN_CHANNEL_OPEN, + dimensions: { provider, transport: "duplex", outcome: "error" }, + }); + recordFallback(reason); + }, + }; + }, + recordFallback, + recordRequest(record: { latencyMs: number; outcome: DuplexOutcomeValue }): void { + safeSpan({ + name: DUPLEX_SPAN_REQUEST, + dimensions: { provider, transport: "duplex", outcome: record.outcome }, + latencyMs: record.latencyMs, + }); + }, + recordLoss(lossClass: DuplexLossClass): void { + const dimensions: DuplexTelemetryDimensions = { + provider, + transport: "duplex", + outcome: "error", + loss_class: lossClass, + }; + safeCounter({ metric: DUPLEX_COUNTER_LOSS_TOTAL, dimensions }); + safeEvent({ name: DUPLEX_TRANSPORT_EVENT, dimensions }); + }, + recordSessionLeak(): void { + safeCounter({ + metric: DUPLEX_COUNTER_SESSION_LEAK_TOTAL, + dimensions: { provider, transport: "duplex", outcome: "error" }, + }); + }, + }; +} diff --git a/packages/adapter-utils/src/execution-target-sandbox.test.ts b/packages/adapter-utils/src/execution-target-sandbox.test.ts index 52d768d61be..e3e887afd26 100644 --- a/packages/adapter-utils/src/execution-target-sandbox.test.ts +++ b/packages/adapter-utils/src/execution-target-sandbox.test.ts @@ -59,6 +59,20 @@ import { type DuplexBrokerRequestRecord, type DuplexBrokerState, } from "./duplex-bridge-broker.js"; +import { + DUPLEX_COUNTER_CHANNEL_OPEN_TOTAL, + DUPLEX_COUNTER_FALLBACK_TOTAL, + DUPLEX_COUNTER_LOSS_TOTAL, + DUPLEX_DIMENSION_KEYS, + DUPLEX_SPAN_CHANNEL_OPEN, + DUPLEX_SPAN_REQUEST, + DUPLEX_TRANSPORT_EVENT, + type DuplexTelemetryCounterRecord, + type DuplexTelemetryDimensions, + type DuplexTelemetryEventRecord, + type DuplexTelemetryRecorder, + type DuplexTelemetrySpanRecord, +} from "./duplex-telemetry.js"; const execFileAsync = promisify(execFile); @@ -3227,6 +3241,490 @@ describe("sandbox adapter execution targets", () => { await api.close(); } }, 20000); + + // One recording telemetry recorder. It captures every span, counter, and event + // the fixed duplex surface produces, so a test asserts the exact names, + // dimensions, and values. An optional `failEvery` flag makes every method throw, + // so a test proves a telemetry failure never breaks the request path. + function createRecordingDuplexRecorder(options: { failEvery?: boolean } = {}): { + recorder: DuplexTelemetryRecorder; + spans: DuplexTelemetrySpanRecord[]; + counters: DuplexTelemetryCounterRecord[]; + events: DuplexTelemetryEventRecord[]; + } { + const spans: DuplexTelemetrySpanRecord[] = []; + const counters: DuplexTelemetryCounterRecord[] = []; + const events: DuplexTelemetryEventRecord[] = []; + const recorder: DuplexTelemetryRecorder = { + recordSpan(record) { + if (options.failEvery) throw new Error("telemetry sink down"); + spans.push(record); + }, + incrementCounter(record) { + if (options.failEvery) throw new Error("telemetry sink down"); + counters.push(record); + }, + emitEvent(record) { + if (options.failEvery) throw new Error("telemetry sink down"); + events.push(record); + }, + }; + return { recorder, spans, counters, events }; + } + + // Every dimension key a record carries must be one of the fixed keys. The set is + // closed, so a new key never reaches a sink by accident. + function assertOnlyFixedDimensionKeys(dimensions: DuplexTelemetryDimensions | undefined): void { + expect(dimensions).toBeDefined(); + for (const key of Object.keys(dimensions ?? {})) { + expect(DUPLEX_DIMENSION_KEYS).toContain(key as (typeof DUPLEX_DIMENSION_KEYS)[number]); + } + } + + it("records a duplex request span with latency and the fixed dimension keys", async () => { + const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-duplex-obs-")); + cleanupDirs.push(rootDir); + const remoteCwd = path.join(rootDir, "workspace"); + await mkdir(remoteCwd, { recursive: true }); + const api = await startRecordingApiServer(); + const { runner, control } = makeDuplexSelectionRunner(); + const { recorder, spans, counters, events } = createRecordingDuplexRecorder(); + const target: AdapterSandboxExecutionTarget = { + kind: "remote", + transport: "sandbox", + providerKey: "daytona", + remoteCwd, + timeoutMs: 30_000, + runner, + effectiveCapabilities: duplexCapabilities(true), + }; + + const bridge = await startAdapterExecutionTargetPaperclipBridge({ + runId: "run-obs", + target, + runtimeRootDir: path.join(remoteCwd, ".paperclip-runtime", "codex"), + adapterKey: "codex", + hostApiToken: "real-run-jwt", + hostApiUrl: api.origin, + enableSandboxDuplexBridge: true, + duplexTelemetryRecorder: recorder, + }); + try { + expect(bridge?.env.PAPERCLIP_API_BRIDGE_MODE).toBe("duplex_v1"); + control.emitData( + encodeDuplexFrame({ + version: DUPLEX_FRAME_VERSION, + type: "request", + id: "req-obs", + method: "GET", + path: "/api/agents/me", + query: "", + headers: { authorization: "Bearer bridge-token" }, + body: "", + }), + ); + await waitForCondition( + () => control.written.length >= 1, + "the broker to write a duplex response frame", + 4000, + ); + + // The channel-open surface: the span, the counter, and the transport event. + const openSpan = spans.find((span) => span.name === DUPLEX_SPAN_CHANNEL_OPEN); + expect(openSpan).toBeDefined(); + expect(openSpan?.dimensions).toMatchObject({ provider: "daytona", transport: "duplex", outcome: "ok" }); + expect(counters.some((c) => c.metric === DUPLEX_COUNTER_CHANNEL_OPEN_TOTAL)).toBe(true); + expect( + events.some( + (e) => + e.name === DUPLEX_TRANSPORT_EVENT && + e.dimensions.transport === "duplex" && + e.dimensions.outcome === "ok", + ), + ).toBe(true); + + // The request span carries a numeric latency and only the fixed keys. + const requestSpan = spans.find((span) => span.name === DUPLEX_SPAN_REQUEST); + expect(requestSpan).toBeDefined(); + expect(typeof requestSpan?.latencyMs).toBe("number"); + expect(requestSpan?.latencyMs).toBeGreaterThanOrEqual(0); + expect(requestSpan?.dimensions).toMatchObject({ provider: "daytona", transport: "duplex", outcome: "ok" }); + assertOnlyFixedDimensionKeys(requestSpan?.dimensions); + } finally { + await bridge?.stop(); + await api.close(); + } + }, 20000); + + it("increments the fallback counter with an approved reason when the capability is absent", async () => { + const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-duplex-fb-")); + cleanupDirs.push(rootDir); + const remoteCwd = path.join(rootDir, "workspace"); + await mkdir(remoteCwd, { recursive: true }); + const api = await startRecordingApiServer(); + const { runner } = makeDuplexSelectionRunner(); + const { recorder, counters, events } = createRecordingDuplexRecorder(); + const target: AdapterSandboxExecutionTarget = { + kind: "remote", + transport: "sandbox", + providerKey: "daytona", + remoteCwd, + timeoutMs: 30_000, + runner, + effectiveCapabilities: duplexCapabilities(false), + }; + + const bridge = await startAdapterExecutionTargetPaperclipBridge({ + runId: "run-fb", + target, + runtimeRootDir: path.join(remoteCwd, ".paperclip-runtime", "codex"), + adapterKey: "codex", + hostApiToken: "real-run-jwt", + hostApiUrl: api.origin, + enableSandboxDuplexBridge: true, + duplexTelemetryRecorder: recorder, + }); + try { + expect(bridge?.env.PAPERCLIP_API_BRIDGE_MODE).toBe("queue_v1"); + const fallback = counters.find((c) => c.metric === DUPLEX_COUNTER_FALLBACK_TOTAL); + expect(fallback).toBeDefined(); + const approvedReasons = [ + "gate_off", + "capability_absent", + "open_failed", + "ready_invalid", + "ready_nonce_mismatch", + "ready_timeout", + "contaminated", + ]; + expect(approvedReasons).toContain(fallback?.dimensions.fallback_reason); + expect(fallback?.dimensions).toMatchObject({ transport: "file", outcome: "error" }); + assertOnlyFixedDimensionKeys(fallback?.dimensions); + // The transport event mirrors the fallback. + expect( + events.some( + (e) => e.name === DUPLEX_TRANSPORT_EVENT && e.dimensions.transport === "file", + ), + ).toBe(true); + } finally { + await bridge?.stop(); + await api.close(); + } + }, 20000); + + it.each([ + { name: "before any dispatch", dispatchFirst: false, expectedClass: "pre_dispatch" }, + { name: "after a dispatch", dispatchFirst: true, expectedClass: "post_dispatch" }, + ])("increments the loss counter with the loss class $name", async ({ dispatchFirst, expectedClass }) => { + const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-duplex-loss-")); + cleanupDirs.push(rootDir); + const remoteCwd = path.join(rootDir, "workspace"); + await mkdir(remoteCwd, { recursive: true }); + const api = await startRecordingApiServer(); + const { runner, control } = makeDuplexSelectionRunner(); + const { recorder, counters } = createRecordingDuplexRecorder(); + const target: AdapterSandboxExecutionTarget = { + kind: "remote", + transport: "sandbox", + providerKey: "daytona", + remoteCwd, + timeoutMs: 30_000, + runner, + effectiveCapabilities: duplexCapabilities(true), + }; + + const bridge = await startAdapterExecutionTargetPaperclipBridge({ + runId: "run-loss", + target, + runtimeRootDir: path.join(remoteCwd, ".paperclip-runtime", "codex"), + adapterKey: "codex", + hostApiToken: "real-run-jwt", + hostApiUrl: api.origin, + enableSandboxDuplexBridge: true, + duplexTelemetryRecorder: recorder, + }); + try { + expect(bridge?.env.PAPERCLIP_API_BRIDGE_MODE).toBe("duplex_v1"); + if (dispatchFirst) { + control.emitData( + encodeDuplexFrame({ + version: DUPLEX_FRAME_VERSION, + type: "request", + id: "req-loss", + method: "GET", + path: "/api/agents/me", + query: "", + headers: { authorization: "Bearer bridge-token" }, + body: "", + }), + ); + await waitForCondition( + () => control.written.length >= 1, + "the broker to write a duplex response frame", + 4000, + ); + } + // A malformed frame is a protocol failure. The broker records a terminal loss. + control.emitData("@@@ not a duplex frame @@@\n"); + await waitForCondition( + () => counters.some((c) => c.metric === DUPLEX_COUNTER_LOSS_TOTAL), + "the broker to record a loss counter", + 4000, + ); + const loss = counters.find((c) => c.metric === DUPLEX_COUNTER_LOSS_TOTAL); + expect(loss?.dimensions.loss_class).toBe(expectedClass); + expect(loss?.dimensions).toMatchObject({ transport: "duplex", outcome: "error" }); + assertOnlyFixedDimensionKeys(loss?.dimensions); + } finally { + await bridge?.stop(); + await api.close(); + } + }, 20000); + + it("keeps serving the request path when the telemetry recorder throws", async () => { + const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-duplex-guard-")); + cleanupDirs.push(rootDir); + const remoteCwd = path.join(rootDir, "workspace"); + await mkdir(remoteCwd, { recursive: true }); + const api = await startRecordingApiServer(); + const { runner, control } = makeDuplexSelectionRunner(); + const { recorder } = createRecordingDuplexRecorder({ failEvery: true }); + const target: AdapterSandboxExecutionTarget = { + kind: "remote", + transport: "sandbox", + providerKey: "daytona", + remoteCwd, + timeoutMs: 30_000, + runner, + effectiveCapabilities: duplexCapabilities(true), + }; + + const bridge = await startAdapterExecutionTargetPaperclipBridge({ + runId: "run-guard", + target, + runtimeRootDir: path.join(remoteCwd, ".paperclip-runtime", "codex"), + adapterKey: "codex", + hostApiToken: "real-run-jwt", + hostApiUrl: api.origin, + enableSandboxDuplexBridge: true, + duplexTelemetryRecorder: recorder, + }); + try { + // The throwing recorder never blocked the duplex selection. + expect(bridge?.env.PAPERCLIP_API_BRIDGE_MODE).toBe("duplex_v1"); + control.emitData( + encodeDuplexFrame({ + version: DUPLEX_FRAME_VERSION, + type: "request", + id: "req-guard", + method: "GET", + path: "/api/agents/me", + query: "", + headers: { authorization: "Bearer bridge-token" }, + body: "", + }), + ); + await waitForCondition( + () => control.written.length >= 1, + "the broker to write a duplex response frame", + 4000, + ); + // The request path still delivered a real host response. + expect(control.written[0]).toMatchObject({ type: "response", id: "req-guard", status: 200 }); + expect(api.requests).toHaveLength(1); + } finally { + await bridge?.stop(); + await api.close(); + } + }, 20000); + + it("keeps sentinel route, query, body, tokens, and provider errors off the duplex telemetry and logs", async () => { + const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-duplex-redact-")); + cleanupDirs.push(rootDir); + const remoteCwd = path.join(rootDir, "workspace"); + await mkdir(remoteCwd, { recursive: true }); + const api = await startRecordingApiServer(); + + const ROUTE_SENTINEL = "sentinelroute8f21"; + const QUERY_SENTINEL = "sentinelquery3d90"; + const BODY_SENTINEL = "sentinelbodya17c"; + const BRIDGE_TOKEN_SENTINEL = "sentinelbridgetok55e2"; + const AGENT_TOKEN_SENTINEL = "sentinelagenttoke91b4"; + const PROVIDER_ERROR_SENTINEL = "sentinelprovidererr7a3d"; + const sentinels = [ + ROUTE_SENTINEL, + QUERY_SENTINEL, + BODY_SENTINEL, + BRIDGE_TOKEN_SENTINEL, + AGENT_TOKEN_SENTINEL, + PROVIDER_ERROR_SENTINEL, + ]; + + // A runner whose channel throws a provider error on the response write, so the + // broker records a stream-failure loss carrying the sentinel message. + const base = createLocalSandboxRunner(); + const control = { emitData: (_chunk: string) => {} }; + const openDuplexChannel = async (openInput: { + command: readonly string[]; + }): Promise => { + const joined = openInput.command.join(" "); + const nonce = /PAPERCLIP_BRIDGE_NONCE='([^']*)'/.exec(joined)?.[1] ?? ""; + let dataListener: ((chunk: string) => void) | null = null; + const channel: CommandManagedDuplexChannel = { + write(_data: string): void { + // Every response write fails with a provider error carrying the sentinel. + throw new Error(`provider write failed: ${PROVIDER_ERROR_SENTINEL}`); + }, + onData(listener: (chunk: string) => void): void { + dataListener = listener; + control.emitData = (chunk) => dataListener?.(chunk); + setImmediate(() => dataListener?.(`${JSON.stringify({ version: 1, type: "ready", nonce })}\n`)); + }, + onExit(_listener: (exit: { exitCode: number | null }) => void): void {}, + stop(): void {}, + close(): Promise { + return Promise.resolve(); + }, + }; + return channel; + }; + const runner = { ...base, openDuplexChannel }; + + const logLines: string[] = []; + const { recorder, spans, counters, events } = createRecordingDuplexRecorder(); + const target: AdapterSandboxExecutionTarget = { + kind: "remote", + transport: "sandbox", + providerKey: "daytona", + remoteCwd, + timeoutMs: 30_000, + runner, + effectiveCapabilities: duplexCapabilities(true), + }; + + const previousDebug = process.env.PAPERCLIP_BRIDGE_DEBUG; + process.env.PAPERCLIP_BRIDGE_DEBUG = "1"; + let bridge: Awaited> = null; + try { + bridge = await startAdapterExecutionTargetPaperclipBridge({ + runId: "run-redact", + target, + runtimeRootDir: path.join(remoteCwd, ".paperclip-runtime", "codex"), + adapterKey: "codex", + hostApiToken: AGENT_TOKEN_SENTINEL, + hostApiUrl: api.origin, + enableSandboxDuplexBridge: true, + duplexTelemetryRecorder: recorder, + onLog: async (_stream, chunk) => { + logLines.push(chunk); + }, + }); + expect(bridge?.env.PAPERCLIP_API_BRIDGE_MODE).toBe("duplex_v1"); + + // A request that carries the sentinel route, query, body, and bridge token. + // The response write then fails with the sentinel provider error. + control.emitData( + encodeDuplexFrame({ + version: DUPLEX_FRAME_VERSION, + type: "request", + id: "req-redact", + method: "GET", + path: `/api/${ROUTE_SENTINEL}`, + query: `secret=${QUERY_SENTINEL}`, + headers: { authorization: `Bearer ${BRIDGE_TOKEN_SENTINEL}` }, + body: BODY_SENTINEL, + }), + ); + // Give the forward and the failing response write time to run and record a loss. + await waitForCondition( + () => counters.some((c) => c.metric === DUPLEX_COUNTER_LOSS_TOTAL), + "the broker to record a loss counter after the failed write", + 4000, + ); + + // Serialize every telemetry record and every log line, then assert that no + // sentinel reaches any of them on the duplex path. + const telemetryDump = JSON.stringify({ spans, counters, events }); + const logDump = logLines.join(""); + for (const sentinel of sentinels) { + expect(telemetryDump).not.toContain(sentinel); + expect(logDump).not.toContain(sentinel); + } + } finally { + if (previousDebug === undefined) delete process.env.PAPERCLIP_BRIDGE_DEBUG; + else process.env.PAPERCLIP_BRIDGE_DEBUG = previousDebug; + await bridge?.stop(); + await api.close(); + } + }, 20000); + + it("maps a sentinel provider key to the constant other across every sink", async () => { + const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-duplex-prov-")); + cleanupDirs.push(rootDir); + const remoteCwd = path.join(rootDir, "workspace"); + await mkdir(remoteCwd, { recursive: true }); + const api = await startRecordingApiServer(); + const { runner, control } = makeDuplexSelectionRunner(); + const { recorder, spans, counters, events } = createRecordingDuplexRecorder(); + const PROVIDER_SENTINEL = "sentinel-plugin-provider-key-9c2a"; + const target: AdapterSandboxExecutionTarget = { + kind: "remote", + transport: "sandbox", + providerKey: PROVIDER_SENTINEL, + remoteCwd, + timeoutMs: 30_000, + runner, + effectiveCapabilities: duplexCapabilities(true), + }; + + const bridge = await startAdapterExecutionTargetPaperclipBridge({ + runId: "run-prov", + target, + runtimeRootDir: path.join(remoteCwd, ".paperclip-runtime", "codex"), + adapterKey: "codex", + hostApiToken: "real-run-jwt", + hostApiUrl: api.origin, + enableSandboxDuplexBridge: true, + duplexTelemetryRecorder: recorder, + }); + try { + expect(bridge?.env.PAPERCLIP_API_BRIDGE_MODE).toBe("duplex_v1"); + control.emitData( + encodeDuplexFrame({ + version: DUPLEX_FRAME_VERSION, + type: "request", + id: "req-prov", + method: "GET", + path: "/api/agents/me", + query: "", + headers: { authorization: "Bearer bridge-token" }, + body: "", + }), + ); + await waitForCondition( + () => spans.some((s) => s.name === DUPLEX_SPAN_REQUEST), + "the broker to record a request span", + 4000, + ); + + // Every recorded provider dimension is the constant `other`, never the key. + const allDimensions = [ + ...spans.map((s) => s.dimensions), + ...counters.map((c) => c.dimensions), + ...events.map((e) => e.dimensions), + ]; + expect(allDimensions.length).toBeGreaterThan(0); + for (const dimensions of allDimensions) { + expect(dimensions.provider).toBe("other"); + } + // The raw key reaches no sink. + const telemetryDump = JSON.stringify({ spans, counters, events }); + expect(telemetryDump).not.toContain(PROVIDER_SENTINEL); + } finally { + await bridge?.stop(); + await api.close(); + } + }, 20000); }); // One decoded stdout frame from the generated duplex gateway. The gateway writes diff --git a/packages/adapter-utils/src/execution-target.ts b/packages/adapter-utils/src/execution-target.ts index 5c75157fbe2..ce16c24e26d 100644 --- a/packages/adapter-utils/src/execution-target.ts +++ b/packages/adapter-utils/src/execution-target.ts @@ -51,6 +51,11 @@ import { decodeDuplexLine, type DuplexRequestFrame, } from "./duplex-frame-codec.js"; +import { + createDuplexTelemetry, + type DuplexFallbackReason, + type DuplexTelemetryRecorder, +} from "./duplex-telemetry.js"; import { createSshCommandManagedRuntimeRunner, parseSshRemoteExecutionSpec, runSshCommand, shellQuote } from "./ssh.js"; import { ensureCommandResolvable, @@ -2288,6 +2293,24 @@ type DuplexReadinessResult = | { ok: true } | { ok: false; reason: DuplexReadinessFailure }; +/** + * Map a readiness failure to the fixed fallback reason. Each readiness failure + * maps to exactly one reason from the closed telemetry set, so the fallback + * counter and the transport event carry only an approved value. + */ +function duplexReadinessFallbackReason(reason: DuplexReadinessFailure): DuplexFallbackReason { + switch (reason) { + case "protocol_contamination": + return "contaminated"; + case "nonce_mismatch": + return "ready_nonce_mismatch"; + case "timeout": + return "ready_timeout"; + case "channel_exit": + return "ready_invalid"; + } +} + /** * 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 @@ -2481,6 +2504,12 @@ export async function startAdapterExecutionTargetPaperclipBridge(input: { // request's execs group under one wrapper span. When it is absent, the request // work runs under the run parent with no wrapper span. runtimeSpan?: RuntimeSpanRunner; + // The injected recorder for the fixed duplex observability surface. The factory + // binds it to a provider-scoped telemetry facade, which records the channel-open + // span, the request span, the guarded counters, and the transport event. The + // default is a no-op recorder, so the surface stays inert until the host injects + // a real recorder. + duplexTelemetryRecorder?: DuplexTelemetryRecorder | null; }): Promise { if (!adapterExecutionTargetUsesPaperclipBridge(input.target)) { return null; @@ -2539,6 +2568,17 @@ export async function startAdapterExecutionTargetPaperclipBridge(input: { const bridgeAsset = await createSandboxCallbackBridgeAsset(); + // The provider-scoped telemetry facade for the fixed duplex observability + // surface. It maps the raw provider key through the allowlist one time, so no + // raw plugin key reaches a span, a counter, or the event. The default recorder + // is a no-op, so the facade is inert until the host injects a real recorder. + const duplexProviderKey = + "providerKey" in target ? target.providerKey ?? undefined : undefined; + const duplexTelemetry = createDuplexTelemetry({ + recorder: input.duplexTelemetryRecorder ?? undefined, + providerKey: duplexProviderKey, + }); + // PAPERCLIP_BRIDGE_DEBUG opts into verbose stdout logs of every bridge proxy // request/response. The query string is logged verbatim, so callers who pass // auth tokens or other sensitive values as query parameters should be aware @@ -2554,9 +2594,14 @@ export async function startAdapterExecutionTargetPaperclipBridge(input: { const forwardBridgeRequest = async ( request: { method: string; path: string; query: string; headers: Record; body: string }, signal?: AbortSignal, + options?: { suppressDebugLog?: boolean }, ): Promise<{ status: number; headers: Record; body: string }> => { const method = request.method.trim().toUpperCase() || "GET"; - if (bridgeDebugEnabled) { + // The per-request debug log prints the method, the path, and the query. The + // duplex path suppresses it, so no route or query rides a log line there. The + // file path keeps the existing behavior. + const emitDebugLog = bridgeDebugEnabled && options?.suppressDebugLog !== true; + if (emitDebugLog) { await onLog( "stdout", `[paperclip] Bridge proxy ${method} ${request.path}${request.query ? `?${request.query}` : ""}\n`, @@ -2580,7 +2625,7 @@ export async function startAdapterExecutionTargetPaperclipBridge(input: { ...(method === "GET" || method === "HEAD" ? {} : { body: request.body }), signal: forwardSignal, }); - if (bridgeDebugEnabled) { + if (emitDebugLog) { await onLog( "stdout", `[paperclip] Bridge proxy response ${response.status} for ${method} ${request.path}${request.query ? `?${request.query}` : ""}\n`, @@ -2605,7 +2650,19 @@ export async function startAdapterExecutionTargetPaperclipBridge(input: { "effectiveCapabilities" in target && target.effectiveCapabilities?.duplexCommandStream === true; const openDuplexChannel = runner.openDuplexChannel?.bind(runner); + // Record the pre-attempt fallback for a file-bridge selection that opens no + // channel. `gate_off` marks the kill switch off; `capability_absent` marks the + // capability or the runner method absent. A later channel-open failure records + // its own fallback through the channel-open attempt below. + if (!duplexRequested) { + duplexTelemetry.recordFallback("gate_off"); + } else if (!capabilityGranted || typeof openDuplexChannel !== "function") { + duplexTelemetry.recordFallback("capability_absent"); + } if (duplexRequested && capabilityGranted && typeof openDuplexChannel === "function") { + // Begin the channel-open attempt. The block reports exactly one terminal: + // `ready` on success, or `fallback(reason)` on an open or a readiness failure. + const duplexChannelOpen = duplexTelemetry.startChannelOpen(); const readinessTimeoutMs = typeof input.duplexReadinessTimeoutMs === "number" && Number.isFinite(input.duplexReadinessTimeoutMs) && @@ -2658,13 +2715,15 @@ export async function startAdapterExecutionTargetPaperclipBridge(input: { env: gatewayEnv, }); channel = await openDuplexChannel({ command }); - } catch (error) { + } catch { // The channel never opened, so no request could carry the bridge token. - // Fall through to the file bridge below. + // Fall through to the file bridge below. The log line names no raw provider + // error, so no provider error rides a log line on the duplex path. if (channel) await closeDuplexChannelWithinBudget(channel, DEFAULT_DUPLEX_CLEANUP_BUDGET_MS); + duplexChannelOpen.fallback("open_failed"); await onLog( "stderr", - `[paperclip] Could not open the sandbox duplex channel: ${error instanceof Error ? error.message : String(error)}. Using the file bridge.\n`, + "[paperclip] Could not open the sandbox duplex channel. Using the file bridge.\n", ); channel = null; } @@ -2675,8 +2734,11 @@ export async function startAdapterExecutionTargetPaperclipBridge(input: { if (!readiness.ok) { // Fail closed. Close the partial channel inside a bounded budget, then // select the file bridge. The broker never started, so no request that - // carries the bridge token reached the channel or any endpoint. + // carries the bridge token reached the channel or any endpoint. The reason + // is a fixed enum, so it rides the log line and the fallback telemetry + // with no raw value. await closeDuplexChannelWithinBudget(channel, DEFAULT_DUPLEX_CLEANUP_BUDGET_MS); + duplexChannelOpen.fallback(duplexReadinessFallbackReason(readiness.reason)); await onLog( "stderr", `[paperclip] Sandbox duplex readiness failed (${readiness.reason}). Using the file bridge.\n`, @@ -2702,13 +2764,16 @@ export async function startAdapterExecutionTargetPaperclipBridge(input: { body: JSON.stringify({ error: denialReason }), }; } - return forwardBridgeRequest(request, options.signal); - }, - logger: (message) => { - void onLog("stderr", `[paperclip] ${message}\n`); + // Suppress the per-request debug log on the duplex path, so no route + // or query rides a log line here. + return forwardBridgeRequest(request, options.signal, { suppressDebugLog: true }); }, + // The duplex path emits only the fixed transport telemetry. It passes no + // free-form logger, so no raw provider error rides a log line here. + telemetry: duplexTelemetry, }); broker.start(); + duplexChannelOpen.ready(); await onLog( "stdout", "[paperclip] Sandbox duplex transport ready; serving the host-assigned origin.\n", @@ -2726,6 +2791,11 @@ export async function startAdapterExecutionTargetPaperclipBridge(input: { // provider session remains when the caller releases the lease. await broker.close(); broker.stop(); + // A channel that did not reach the `closed` state may leave a live + // provider session, so record one session leak. + if (broker.state !== "closed") { + duplexTelemetry.recordSessionLeak(); + } await bridgeAsset.cleanup(); }, };