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 <noreply@paperclip.ing>
This commit is contained in:
Priya RamanandPaperclip committed 2026-08-20 04:57:22 +00:00
1 parent 43f9d3c4a2
commit c881bc460c
4 files changed
+905 -12

No files matched your search

@@ -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<typeof setTimeout>;
responseTimer: ReturnType<typeof setTimeout>;
/** 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",
);
},
);
@@ -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<string> = new Set<string>([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" },
});
},
};
}
@@ -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<CommandManagedDuplexChannel> => {
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<void> {
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<ReturnType<typeof startAdapterExecutionTargetPaperclipBridge>> = 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
+80 -10
View File
@@ -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<AdapterExecutionTargetPaperclipBridgeHandle | null> {
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<string, string>; body: string },
signal?: AbortSignal,
options?: { suppressDebugLog?: boolean },
): Promise<{ status: number; headers: Record<string, string>; 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();
},
};