feat(debug): add content-free adapter and bridge stream diagnostics

docs-site/src/content/docs/reference/cli/agents.md

@@ -178,6 +178,10 @@ ocx debug usage on|off|status|reset
 ocx debug usage logs [-f|--follow]
 ```
 
+Provider debug also records structural adapter/bridge stream events: sequence, attempt and recovery labels, byte counts, and process-local HMAC fingerprints. Text, reasoning, tool arguments, queries, and provider state are not included in these stream diagnostic records. Fingerprints change after a proxy restart; disable provider debug after collecting a reproduction.
+Raw credentials, account IDs, and request bodies are never recorded in stream diagnostics.
+
+
 With no scope, `ocx debug` prints usage and, when the proxy is stopped, the next-start environment
 defaults. Provider debug defaults from `OCX_DEBUG=1` (legacy `OCX_DEBUG_FRAMES=1` also works); usage
 debug defaults from `OPENCODEX_USAGE_DEBUG=1`.

src/bridge.ts

@@ -5,3 +5,5 @@ export { setOwnedBudgetAbandonedMsForTests } from "./bridge/internal";
 export { buildResponseJSON } from "./bridge/response-json";
 export { bridgeToResponsesSSE } from "./bridge/sse";
 export type { ResponsesTerminalStatus } from "./bridge/sse";
+export { adapterEventDiagnosticDetails, diagnoseAdapterEvent } from "./bridge/diagnostic";
+export type { BridgeDiagnosticContext, BridgeDiagnosticSequence } from "./bridge/diagnostic";

src/bridge/diagnostic.ts

@@ -0,0 +1,85 @@
+import type { AdapterEvent } from "../types";
+import { debugFingerprint, debugStreamDiagnostic, type DebugStreamDiagnosticContext } from "../lib/debug";
+
+export interface BridgeDiagnosticSequence { value: number }
+
+export interface BridgeDiagnosticContext extends DebugStreamDiagnosticContext {
+  sequence?: BridgeDiagnosticSequence;
+}
+
+export function adapterEventDiagnosticDetails(event: AdapterEvent): Record<string, unknown> {
+  switch (event.type) {
+    case "text_delta":
+      return { byteLength: Buffer.byteLength(event.text), fingerprint: debugFingerprint(event.text) };
+    case "thinking_delta":
+      return { byteLength: Buffer.byteLength(event.thinking), fingerprint: debugFingerprint(event.thinking) };
+    case "reasoning_raw_delta":
+      return { byteLength: Buffer.byteLength(event.text), fingerprint: debugFingerprint(event.text) };
+    case "thinking_signature":
+    case "redacted_thinking":
+    case "kiro_redacted_reasoning": {
+      const content = event.type === "thinking_signature" ? event.signature : event.data;
+      return { byteLength: Buffer.byteLength(content), fingerprint: debugFingerprint(content) };
+    }
+    case "tool_call_delta":
+      return { byteLength: Buffer.byteLength(event.arguments), fingerprint: debugFingerprint(event.arguments) };
+    case "tool_call_start":
+      return {
+        idByteLength: Buffer.byteLength(event.id),
+        idFingerprint: debugFingerprint(event.id),
+        nameByteLength: Buffer.byteLength(event.name),
+        nameFingerprint: debugFingerprint(event.name),
+      };
+    case "web_search_call_begin":
+      return { idByteLength: Buffer.byteLength(event.id), idFingerprint: debugFingerprint(event.id) };
+    case "web_search_call_end": {
+      const queries = JSON.stringify(event.queries);
+      return {
+        idByteLength: Buffer.byteLength(event.id),
+        idFingerprint: debugFingerprint(event.id),
+        byteLength: Buffer.byteLength(queries),
+        fingerprint: debugFingerprint(queries),
+        status: event.status,
+      };
+    }
+    case "error":
+      return {
+        byteLength: Buffer.byteLength(event.message),
+        fingerprint: debugFingerprint(event.message),
+        ...(event.status !== undefined ? { status: event.status } : {}),
+        ...(event.code !== undefined
+          ? { codeByteLength: Buffer.byteLength(event.code), codeFingerprint: debugFingerprint(event.code) }
+          : {}),
+        ...(event.retryable !== undefined ? { retryable: event.retryable } : {}),
+      };
+    case "incomplete":
+      return {
+        ...(event.message !== undefined ? { byteLength: Buffer.byteLength(event.message), fingerprint: debugFingerprint(event.message) } : {}),
+        reasonByteLength: Buffer.byteLength(event.reason),
+        reasonFingerprint: debugFingerprint(event.reason),
+        ...(event.retryable !== undefined ? { retryable: event.retryable } : {}),
+      };
+    case "done":
+      return {
+        ...(event.stopReason !== undefined
+          ? { stopReasonByteLength: Buffer.byteLength(event.stopReason), stopReasonFingerprint: debugFingerprint(event.stopReason) }
+          : {}),
+        ...(event.endTurn !== undefined ? { endTurn: event.endTurn } : {}),
+      };
+    default:
+      return {};
+  }
+}
+
+/** Emit one adapter-stage diagnostic while preserving one sequence across sidecar iterations. */
+export function diagnoseAdapterEvent(context: BridgeDiagnosticContext, event: AdapterEvent): void {
+  const sequence = context.sequence ??= { value: 0 };
+  debugStreamDiagnostic(
+    context,
+    "adapter",
+    ++sequence.value,
+    event.type,
+    adapterEventDiagnosticDetails(event),
+  );
+}
+

src/bridge/sse.ts

@@ -49,6 +49,8 @@ import {
 } from "../lib/translator-budget";
 import { adapterFailureFromEvent, emptyChunks, joinChunks, ownedBudgetAbandonedMs, responsesUsage, toolCallArgumentsUsable, uuid, webSearchAction } from "./internal";
 import type { OutputItem, StringChunks } from "./internal";
+import { adapterEventDiagnosticDetails, type BridgeDiagnosticContext } from "./diagnostic";
+import { debugStreamDiagnostic } from "../lib/debug";
 
 function sseEvent(name: string, data: Record<string, unknown>): string {
   return `event: ${name}\ndata: ${JSON.stringify(data)}\n\n`;
@@ -127,13 +129,15 @@ export function bridgeToResponsesSSE(
      */
     replayCacheScope?: OcxReasoningReplayScopeRef;
     /**
-     * Test seam for the wire/stall beat loop. Production omits this and uses the
-     * global timers; injecting here must not change scheduling semantics.
-     */
+    * Test seam for the wire/stall beat loop. Production omits this and uses the
+    * global timers; injecting here must not change scheduling semantics.
+    */
     timers?: {
       setInterval: (handler: () => void, ms: number) => unknown;
       clearInterval: (id: unknown) => void;
     };
+    /** Internal, opt-in structural stream diagnostics. */
+    diagnostic?: BridgeDiagnosticContext;
   },
 ): ReadableStream<Uint8Array> {
   const replayCacheScope = options?.replayCacheScope;
@@ -386,6 +390,7 @@ export function bridgeToResponsesSSE(
   };
   const responseId = options?.responseId ?? `resp_${uuid()}`;
   let seq = 0;
+  let diagnosticSequence = 0;
   // Set once the client is gone (cancel) or an enqueue throws on a torn-down controller, so we
   // never enqueue again and never throw a second time inside start() — the RC2 double-throw that
   // otherwise surfaced as proxy-side stream noise on every client disconnect.
@@ -941,6 +946,17 @@ export function bridgeToResponsesSSE(
           }
           if (next.done) { upstreamDone = true; break; }
           const event = next.value;
+          if (options?.diagnostic) {
+            debugStreamDiagnostic(
+              options.diagnostic,
+              "bridge",
+              options.diagnostic.sequence
+                ? ++options.diagnostic.sequence.value
+                : ++diagnosticSequence,
+              event.type,
+              adapterEventDiagnosticDetails(event),
+            );
+          }
           let terminalEvent = false;
           // Invisible adapter heartbeats (and buffered web-search progress) count as upstream
           // liveness only — they must not suppress wire keepalives that re-arm Codex idle timers.

src/images/loop.ts

@@ -18,7 +18,7 @@ import type { AdapterEvent, OcxMessage, OcxParsedRequest, OcxProviderContinuatio
 import { namespacedToolName, toolChoiceToolPredicate } from "../types";
 import { cloneProviderOpaqueToolCallMetadata } from "../responses/provider-opaque-metadata";
 import type { AttemptRecoveryKind } from "../usage/log";
-import { bridgeToResponsesSSE } from "../bridge";
+import { bridgeToResponsesSSE, diagnoseAdapterEvent, type BridgeDiagnosticContext } from "../bridge";
 import { clearableDeadline, idleDeadline } from "../lib/abort";
 import { readBoundedResponseBody } from "../lib/bounded-body";
 import { applyUpstreamRecoveryInit, fetchWithResetRetry, prepareSameTarget429Wait } from "../lib/upstream-retry";
@@ -332,6 +332,8 @@ export interface ImageBridgeDeps {
   onCompletedResponse?: (response: Record<string, unknown>, providerState?: OcxProviderContinuationState) => void;
   /** WebSocket Responses path only — leave response id empty for protocol compatibility. */
   forceEmptyResponseId?: boolean;
+  /** Internal, opt-in structural stream diagnostics shared with the final bridge. */
+  diagnostic?: BridgeDiagnosticContext;
 }
 
 /**
@@ -717,6 +719,12 @@ export async function runWithImageBridge(deps: ImageBridgeDeps): Promise<Respons
   const consumeIterationEvents = async function* (prepared: IterationResponse): AsyncGenerator<AdapterEvent, IterationSplit> {
     const events: AdapterEvent[] = prepared.collectedEvents ?? [];
     const iterationBudget = prepared.collectedEvents ? undefined : createIterationEventBudget();
+    if (!iterationBudget && deps.diagnostic) {
+      // Collected runTurn events skip the streaming parse loop below, so diagnose
+      // them here before terminal scanning and replay.
+      deps.diagnostic.adapterName = prepared.responseAdapter.name;
+      for (const event of events) diagnoseAdapterEvent(deps.diagnostic, event);
+    }
     try {
       if (iterationBudget) {
         const parse = prepared.responseAdapter.parseStream.bind(prepared.responseAdapter);
@@ -725,6 +733,10 @@ export async function runWithImageBridge(deps: ImageBridgeDeps): Promise<Respons
           inactivityTimeoutMs: stallTimeoutMs,
           translatorBudget,
         })) {
+          if (deps.diagnostic) {
+            deps.diagnostic.adapterName = prepared.responseAdapter.name;
+            diagnoseAdapterEvent(deps.diagnostic, event);
+          }
           if (event.type === "heartbeat") yield event;
           else {
             iterationBudget.retain(event);
@@ -1045,6 +1057,7 @@ export async function runWithImageBridge(deps: ImageBridgeDeps): Promise<Respons
         onUsage: (usage: OcxUsage | undefined) => deps.onUsage?.(usage),
       } : {}),
       ...(deps.onCompletedResponse ? { onCompletedResponse: deps.onCompletedResponse } : {}),
+      ...(deps.diagnostic ? { diagnostic: deps.diagnostic } : {}),
     },
   );
   return new Response(sse, { headers: SSE_HEADERS });

src/lib/debug.ts

@@ -1,7 +1,10 @@
+import { createHmac, randomBytes } from "node:crypto";
 import { appendDebugLogLine } from "./debug-log-buffer";
 import { isDebugEnabled } from "./debug-settings";
 import { redactSecrets } from "./redact";
 
+let debugFingerprintKey: Uint8Array | undefined;
+
 function emitDebugLine(line: string): void {
   if (!isDebugEnabled()) return;
   try {
@@ -29,3 +32,42 @@ export function debugProviderDiagnostic(adapter: string, event: string, details:
     /* diagnostics must never affect request handling */
   }
 }
+
+/** Process-local, content-free correlation aid for opt-in provider diagnostics. */
+export function debugFingerprint(value: string | Uint8Array): string | undefined {
+  if (!isDebugEnabled()) return undefined;
+  try {
+    debugFingerprintKey ??= randomBytes(32);
+    return createHmac("sha256", debugFingerprintKey).update(value).digest("hex");
+  } catch {
+    return undefined;
+  }
+}
+
+export interface DebugStreamDiagnosticContext {
+  requestId: string;
+  adapterName: string;
+  attempt?: number;
+  recovery?: string;
+}
+
+export type DebugStreamDiagnosticStage = "adapter" | "bridge";
+
+/** Emit one structural line for an adapter/bridge event without retaining its content. */
+export function debugStreamDiagnostic(
+  context: DebugStreamDiagnosticContext,
+  stage: DebugStreamDiagnosticStage,
+  sequence: number,
+  eventType: string,
+  details?: Record<string, unknown>,
+): void {
+  debugProviderDiagnostic(context.adapterName, "stream", {
+    stage,
+    sequence,
+    eventType,
+    ...(context.requestId !== undefined ? { requestId: context.requestId } : {}),
+    ...(context.attempt !== undefined ? { attempt: context.attempt } : {}),
+    ...(context.recovery !== undefined ? { recovery: context.recovery } : {}),
+    ...details,
+  });
+}

src/server/responses/adapter-continuation.ts

@@ -32,6 +32,7 @@ import {
 import { redactSecretString } from "../../lib/redact";
 import { resolveWireProtocolOverride } from "../adapter-resolve";
 import { bindRouteReasoningReplayScope } from "./core-replay";
+import { diagnoseAdapterEvents } from "./stream-diagnostics";
 import {
   ANTHROPIC_POOL_MAX_FAILOVERS_PER_REQUEST,
   rotateAnthropicAccountOn429,
@@ -86,6 +87,8 @@ export function createAdapterContinuations(
     | "applyFailoverSnapshot"
     | "replayOAuthCredentialSnapshot"
     | "noteRoutedAttemptSend"
+    | "streamDiagnostic"
+    | "noteDiagnosticAttemptSend"
   >,
   sidecarState: Pick<ResponsesSidecarAuth, "routedCompaction">,
   sendBudgetState: Pick<
@@ -196,7 +199,11 @@ export function createAdapterContinuations(
       const replayKind: AttemptRecoveryKind | undefined = recoveryKind;
       try {
         if (transportState.activeAdapter.fetchResponse) {
-          transportState.noteRoutedAttemptSend(continuationEstimate, replayKind);
+          transportState.noteDiagnosticAttemptSend(
+            continuationEstimate,
+            replayKind,
+            transportState.activeAdapter.name,
+          );
           await waitForProviderRequestSlot(route.providerName, route.provider, nextParsed.modelId, upstream.signal);
           return await transportState.activeAdapter.fetchResponse(builtContinuationRequest, {
             abortSignal: upstream.signal,
@@ -221,7 +228,11 @@ export function createAdapterContinuations(
           : fetchWithResetRetry;
         return await fetchContinuationWithRetryPolicy(
           recovery => {
-            transportState.noteRoutedAttemptSend(continuationEstimate, recovery ?? replayKind);
+            transportState.noteDiagnosticAttemptSend(
+              continuationEstimate,
+              recovery ?? replayKind,
+              transportState.activeAdapter.name,
+            );
             return fetchWithHeaderTimeout(
               builtContinuationRequest.url,
               applyUpstreamRecoveryInit({
@@ -534,7 +545,13 @@ export function createAdapterContinuations(
           response,
           upstream.signal,
           bodyInactivityMs,
-          guarded => transportState.activeAdapter.parseStream(guarded, translatorBudget, logCtx.activeTierMetadata),
+          guarded =>
+            diagnoseAdapterEvents(
+              transportState.activeAdapter.parseStream(guarded, translatorBudget, logCtx.activeTierMetadata),
+              () => transportState.activeAdapter.name,
+              transportState.streamDiagnostic,
+              logCtx,
+            ),
         );
       } else if (transportState.activeAdapter.parseResponse) {
         yield* await readResponseBodyWithInactivity(

src/server/responses/adapter-delivery.ts

@@ -20,6 +20,7 @@ import {
   ResponseBodyInactivityError,
 } from "../../lib/response-body-inactivity";
 import { resolveStallTimeoutSec } from "../../stall-timeout";
+import { diagnoseAdapterEvents } from "./stream-diagnostics";
 
 /** One responsibility of the Responses request pipeline; state owners are explicit. */
 export async function deliverAdapterResponse(
@@ -32,7 +33,7 @@ export async function deliverAdapterResponse(
     | "rememberKiroDeliveredFinalAnswer"
     | "responseStateOptions"
   >,
-  transportState: Pick<ResponsesTransport, "activeAdapter" | "bindKeyUsageFromBridge">,
+  transportState: Pick<ResponsesTransport, "activeAdapter" | "bindKeyUsageFromBridge" | "streamDiagnostic">,
   sidecarState: Pick<ResponsesSidecarAuth, "routedCompaction">,
   responseEffects: Pick<
     ResponsesEffects,
@@ -79,7 +80,13 @@ export async function deliverAdapterResponse(
           upstreamResponse,
           upstream.signal,
           bodyInactivityMs,
-          response => transportState.activeAdapter.parseStream(response, translatorBudget, logCtx.activeTierMetadata),
+        response =>
+          diagnoseAdapterEvents(
+            transportState.activeAdapter.parseStream(response, translatorBudget, logCtx.activeTierMetadata),
+            () => transportState.activeAdapter.name,
+            transportState.streamDiagnostic,
+            logCtx,
+          ),
         );
       } catch (error) {
         if (error instanceof ResponseBodyInactivityError) {
@@ -128,6 +135,7 @@ export async function deliverAdapterResponse(
       toolParameterSchemas,
         ...(options.onFirstOutput ? { onFirstOutput: options.onFirstOutput } : {}),
         ...(routedCompaction ? { compaction: true } : {}),
+        ...(transportState.streamDiagnostic ? { diagnostic: transportState.streamDiagnostic.context } : {}),
         // Same grok-surface split as the runTurn branch above.
         ...(logCtx.surface === "grok" ? { heartbeatStyle: "comment" as const } : {}),
         onUsage: usage => {

src/server/responses/request-transport.ts

@@ -65,6 +65,7 @@ import {
 } from "../request-log";
 import type { AttemptRecoveryKind } from "../../usage/log";
 import { resolvePassiveRouteSubjectId } from "../passive-route-linker";
+import { createStreamDiagnostic, type StreamDiagnostic } from "./stream-diagnostics";
 
 /** Owns live credential selection and adapter bindings for one request. */
 export async function prepareResponsesTransport(
@@ -253,6 +254,21 @@ export async function prepareResponsesTransport(
     if (usesApiKeyAccount(route.provider)) pendingKeySend = { estimate, recovery };
     else noteProviderAttemptSend(logCtx, route.providerName, route.provider, estimate, recovery);
   };
+  // Opt-in structural stream diagnostics shared by the adapter, continuation,
+  // sidecar, and bridge stages of this request. Created once so every stage
+  // appends to a single sequence; undefined unless provider debug is on.
+  const streamDiagnostic: StreamDiagnostic | undefined = createStreamDiagnostic(route.providerName);
+  const noteDiagnosticAttemptSend = (
+    estimate: number | undefined,
+    recovery?: AttemptRecoveryKind,
+    adapterName?: string,
+  ): void => {
+    noteRoutedAttemptSend(estimate, recovery);
+    if (!streamDiagnostic) return;
+    streamDiagnostic.context.attempt = logCtx.activeAttempt?.ordinal;
+    streamDiagnostic.context.recovery = recovery;
+    if (adapterName) streamDiagnostic.context.adapterName = adapterName;
+  };
   const commitKeyAttemptSend = (): void => {
     if (!usesApiKeyAccount(route.provider)) return;
     noteProviderAttemptSend(logCtx, route.providerName, route.provider,
@@ -796,6 +812,8 @@ export async function prepareResponsesTransport(
     refreshRunTurnAdapter,
     oauthDispatch,
     noteRoutedAttemptSend,
+    noteDiagnosticAttemptSend,
+    streamDiagnostic,
     commitKeyAttemptSend,
     bindKeyUsageFromBridge,
     anthropicSessionKey,