From dc7ce1f79085b36ad8964e8112f386ac623650e1 Mon Sep 17 00:00:00 2001 From: Yumi Date: Mon, 7 Sep 2026 19:09:02 -0600 Subject: [PATCH] feat(debug): trace adapter and bridge streams without raw content Extracted and adapted from fork commit b12ea6956366af0b5c80c34c45e7b6c8adb739f4. Co-authored-by: Yumi Co-authored-by: SB Yoon <44089734+yansigit@users.noreply.github.com> --- .../src/content/docs/reference/cli/agents.md | 3 + src/bridge.ts | 97 +++++++++++++++++ src/images/loop.ts | 9 +- src/lib/debug.ts | 42 +++++++ src/server/responses/core.ts | 103 +++++++++++++++--- src/web-search/loop.ts | 9 +- tests/adapters/bridge.test.ts | 91 +++++++++++++++- ...rminal-continuation-owner-rotation.test.ts | 84 +++++++++++++- tests/images/loop.test.ts | 32 +++++- tests/lib/debug.test.ts | 43 +++++++- tests/web-search/web-search.test.ts | 47 +++++++- 11 files changed, 537 insertions(+), 23 deletions(-) diff --git a/docs-site/src/content/docs/reference/cli/agents.md b/docs-site/src/content/docs/reference/cli/agents.md index 4b95d1bcd2..013333f253 100644 --- a/docs-site/src/content/docs/reference/cli/agents.md +++ b/docs-site/src/content/docs/reference/cli/agents.md @@ -152,6 +152,9 @@ 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. + + 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`. diff --git a/src/bridge.ts b/src/bridge.ts index 20e7c3fe09..147b6f47b4 100644 --- a/src/bridge.ts +++ b/src/bridge.ts @@ -43,6 +43,7 @@ import { type TranslatorBudget, type TranslatorBufferKind, } from "./lib/translator-budget"; +import { debugFingerprint, debugStreamDiagnostic, type DebugStreamDiagnosticContext } from "./lib/debug"; function uuid(): string { return crypto.randomUUID().replace(/-/g, ""); @@ -204,6 +205,88 @@ interface OutputItem { export type ResponsesTerminalStatus = "completed" | "failed" | "incomplete"; +export interface BridgeDiagnosticSequence { value: number } + +export interface BridgeDiagnosticContext extends DebugStreamDiagnosticContext { + sequence?: BridgeDiagnosticSequence; +} + +export function adapterEventDiagnosticDetails(event: AdapterEvent): Record { + 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), + ); +} + export function bridgeToResponsesSSE( events: AsyncIterable, modelId: string, @@ -264,6 +347,8 @@ export function bridgeToResponsesSSE( setInterval: (handler: () => void, ms: number) => unknown; clearInterval: (id: unknown) => void; }; + /** Internal, opt-in structural stream diagnostics. */ + diagnostic?: BridgeDiagnosticContext; }, ): ReadableStream { const replayCacheScope = options?.replayCacheScope; @@ -372,6 +457,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. @@ -932,6 +1018,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. diff --git a/src/images/loop.ts b/src/images/loop.ts index 7d4855f91b..31bc7aec3c 100644 --- a/src/images/loop.ts +++ b/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"; @@ -279,6 +279,8 @@ export interface ImageBridgeDeps { onCompletedResponse?: (response: Record, 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; } /** @@ -650,6 +652,10 @@ export async function runWithImageBridge(deps: ImageBridgeDeps): Promise deps.onUsage?.(usage), } : {}), ...(deps.onCompletedResponse ? { onCompletedResponse: deps.onCompletedResponse } : {}), + ...(deps.diagnostic ? { diagnostic: deps.diagnostic } : {}), }, ); return new Response(sse, { headers: SSE_HEADERS }); diff --git a/src/lib/debug.ts b/src/lib/debug.ts index 5d4b54b050..2f42134d2b 100644 --- a/src/lib/debug.ts +++ b/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, +): 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, + }); +} diff --git a/src/server/responses/core.ts b/src/server/responses/core.ts index 133ed9cdbc..34d3c8004b 100644 --- a/src/server/responses/core.ts +++ b/src/server/responses/core.ts @@ -1,6 +1,6 @@ import type { Server } from "bun"; import { randomUUID } from "node:crypto"; -import { bridgeToResponsesSSE, buildResponseJSON, formatErrorResponse, type ResponsesTerminalStatus } from "../../bridge"; +import { adapterEventDiagnosticDetails, bridgeToResponsesSSE, buildResponseJSON, formatErrorResponse, type BridgeDiagnosticContext, type BridgeDiagnosticSequence, type ResponsesTerminalStatus } from "../../bridge"; import { formatPassthroughUpstreamError } from "./passthrough-error"; import { createResponsesFieldBackfillBlockRewrite, @@ -87,7 +87,8 @@ import { pickComboTargetWithWait, targetKey, } from "../../combos"; -import { isInjectionDebugEnabled } from "../../lib/debug-settings"; +import { isDebugEnabled, isInjectionDebugEnabled } from "../../lib/debug-settings"; +import { debugStreamDiagnostic } from "../../lib/debug"; import { CYBER_POLICY_ERROR_CODE, CYBER_POLICY_FALLBACK_MESSAGE, @@ -429,6 +430,34 @@ import { preflightComboStreamResponse } from "./combo-stream-preflight"; // already-committed event boundary and can replay custom adapter work. const runTurnAdapterSseResponses = new WeakSet(); +function diagnoseAdapterEvents( + events: AsyncIterable, + adapterName: string, + requestId: string | undefined, + logCtx: RequestLogContext, + state: BridgeDiagnosticSequence, +): AsyncIterable { + if (!requestId) return events; + return (async function* () { + for await (const event of events) { + const attempt = logCtx.activeAttempt; + debugStreamDiagnostic( + { + requestId, + adapterName, + ...(attempt?.ordinal !== undefined ? { attempt: attempt.ordinal } : {}), + ...(attempt?.recoveryKinds.at(-1) !== undefined ? { recovery: attempt.recoveryKinds.at(-1) } : {}), + }, + "adapter", + ++state.value, + event.type, + adapterEventDiagnosticDetails(event), + ); + yield event; + } + })(); +} + /** * Adapters whose continuation state must survive Codex's store:false requests. */ @@ -4227,6 +4256,23 @@ async function handleResponsesInner( ); if (passiveSubjectId) logCtx.activeAttempt.labRouteSubjectId = passiveSubjectId; } + const diagnosticRequestId = isDebugEnabled() ? randomUUID() : undefined; + const adapterDiagnosticState: BridgeDiagnosticSequence = { value: 0 }; + const diagnosticContext: BridgeDiagnosticContext | undefined = diagnosticRequestId + ? { requestId: diagnosticRequestId, adapterName: adapter.name, sequence: adapterDiagnosticState } + : undefined; + const noteDiagnosticAttempt = ( + attempt: RequestLogContext["activeAttempt"], + inputEstimate: number | undefined, + recovery?: AttemptRecoveryKind, + adapterName?: string, + ): void => { + noteAttemptSend(attempt, inputEstimate, recovery); + if (!diagnosticContext) return; + diagnosticContext.attempt = attempt?.ordinal; + diagnosticContext.recovery = recovery; + if (adapterName) diagnosticContext.adapterName = adapterName; + }; const isPassthrough = "passthrough" in adapter && !!adapter.passthrough; const rawInput = (parsed._rawBody as { input?: unknown }).input; @@ -6145,7 +6191,8 @@ async function handleResponsesInner( ...(vidPlan ? { videoPlan: vidPlan } : {}), forwardHeaders: selectedForwardHeaders, onAttemptSend: (recovery?: AttemptRecoveryKind) => - noteAttemptSend(logCtx.activeAttempt, logCtx.usageLogInputTokens, recovery), + noteDiagnosticAttempt(logCtx.activeAttempt, logCtx.usageLogInputTokens, recovery, adapter.name), + ...(diagnosticContext ? { diagnostic: diagnosticContext } : {}), abortSignal: options.abortSignal, maxRounds: imgPlan && vidPlan ? clampImageMaxRounds(Math.min(config.images?.maxRounds ?? 3, config.images?.videoMaxRounds ?? 2)) @@ -6253,7 +6300,8 @@ async function handleResponsesInner( recordAdapterTier(logCtx, request); }, onAttemptSend: (recovery?: AttemptRecoveryKind) => - noteAttemptSend(logCtx.activeAttempt, logCtx.usageLogInputTokens, recovery), + noteDiagnosticAttempt(logCtx.activeAttempt, logCtx.usageLogInputTokens, recovery, adapter.name), + ...(diagnosticContext ? { diagnostic: diagnosticContext } : {}), onUsage: usage => { logCtx.usageFromBridge = true; if (usage) { @@ -6329,11 +6377,16 @@ async function handleResponsesInner( pacingSlotAcquired = false, ): Promise => { try { + if (diagnosticContext) { + diagnosticContext.recovery = recovery; + diagnosticContext.attempt = logCtx.activeAttempt?.ordinal; + diagnosticContext.adapterName = runTurnAdapter.name; + } if (!pacingSlotAcquired) { await waitForProviderRequestSlot(route.providerName, route.provider, route.modelId, runTurnAbort.signal); } await refreshRunTurnSelection(); - noteAttemptSend(logCtx.activeAttempt, logCtx.usageLogInputTokens, recovery); + noteDiagnosticAttempt(logCtx.activeAttempt, logCtx.usageLogInputTokens, recovery, runTurnAdapter.name); const runTurnProviderFetch = providerFetch( route.provider, options.codexWsRuntimeIdentity, @@ -6458,13 +6511,15 @@ async function handleResponsesInner( onBacklogExceeded: () => runTurnAbort.abort(), }); void runTurnAttempt(retryQueue, "empty-completion"); - return retryQueue.stream(); + return diagnoseAdapterEvents(retryQueue.stream(), runTurnAdapter.name, diagnosticRequestId, logCtx, adapterDiagnosticState); }; const { toolNsMap, declaredToolNames, toolParameterSchemas, freeformToolNames, toolSearchToolNames } = toolBridgeMaps; if (parsed.stream) { void runTurn(); - let eventSource: AsyncIterable = queue.stream(); + let eventSource: AsyncIterable = diagnoseAdapterEvents( + queue.stream(), runTurnAdapter.name, diagnosticRequestId, logCtx, adapterDiagnosticState, + ); if (route.provider.authMode === "oauth" || (genericFailoverAccountId && isGenericOAuthFailoverEnabled(config, route.providerName))) { // Preflight holds only heartbeats and the first meaningful event. A first-event 429 can be // replayed transparently; after any output reaches the bridge, a later error stays terminal. @@ -6513,6 +6568,7 @@ async function handleResponsesInner( // grok-build's strict decoder dies on the typed response.heartbeat frame; its // eventsource layer tolerates comment keep-alives. Codex needs the opposite. ...(logCtx.surface === "grok" ? { heartbeatStyle: "comment" as const } : {}), + ...(diagnosticContext ? { diagnostic: diagnosticContext } : {}), onUsage: usage => { // Raw adapter usage, pre wire-normalization: the bridged SSE now always carries // zero-default detail objects, so provenance must come from here (cache_detail_missing). @@ -7335,8 +7391,13 @@ async function handleResponsesInner( // Optional recovery label for same-target / failover continuation sends. const replayKind: AttemptRecoveryKind | undefined = recoveryKind; try { + if (diagnosticContext) { + diagnosticContext.recovery = replayKind; + diagnosticContext.attempt = logCtx.activeAttempt?.ordinal; + diagnosticContext.adapterName = activeAdapter.name; + } if (activeAdapter.fetchResponse) { - noteAttemptSend(logCtx.activeAttempt, continuationEstimate, replayKind); + noteDiagnosticAttempt(logCtx.activeAttempt, continuationEstimate, replayKind, activeAdapter.name); await waitForProviderRequestSlot(route.providerName, route.provider, nextParsed.modelId, upstream.signal); return await activeAdapter.fetchResponse(builtContinuationRequest, { abortSignal: upstream.signal, @@ -7357,7 +7418,7 @@ async function handleResponsesInner( : fetchWithResetRetry; return await fetchContinuationWithRetryPolicy( recovery => { - noteAttemptSend(logCtx.activeAttempt, continuationEstimate, recovery ?? replayKind); + noteDiagnosticAttempt(logCtx.activeAttempt, continuationEstimate, recovery ?? replayKind, activeAdapter.name); return fetchWithHeaderTimeout( builtContinuationRequest.url, applyUpstreamRecoveryInit({ @@ -7608,7 +7669,13 @@ async function handleResponsesInner( const detachContinuationBodyGuard = cancelBodyOnAbort(response.body, upstream.signal); try { if (nextParsed.stream) { - yield* activeAdapter.parseStream(response, translatorBudget, logCtx.activeTierMetadata); + yield* diagnoseAdapterEvents( + activeAdapter.parseStream(response, translatorBudget, logCtx.activeTierMetadata), + activeAdapter.name, + diagnosticRequestId, + logCtx, + adapterDiagnosticState, + ); } else if (activeAdapter.parseResponse) { yield* await activeAdapter.parseResponse(response, translatorBudget, logCtx.activeTierMetadata); } else { @@ -7640,11 +7707,18 @@ async function handleResponsesInner( }; if (parsed.stream) { - const initialEventStream = activeAdapter.parseStream( - upstreamResponse, - translatorBudget, - logCtx.activeTierMetadata, + const initialEventStream = diagnoseAdapterEvents( + activeAdapter.parseStream(upstreamResponse, translatorBudget, logCtx.activeTierMetadata), + activeAdapter.name, + diagnosticRequestId, + logCtx, + adapterDiagnosticState, ); + if (diagnosticContext) { + diagnosticContext.adapterName = activeAdapter.name; + diagnosticContext.attempt = logCtx.activeAttempt?.ordinal; + diagnosticContext.recovery = logCtx.activeAttempt?.recoveryKinds.at(-1); + } const eventStream = terminalGuardEnabled ? guardTerminalEventStream({ parsed, @@ -7680,6 +7754,7 @@ async function handleResponsesInner( ...(routedCompaction ? { compaction: true } : {}), // Same grok-surface split as the runTurn branch above. ...(logCtx.surface === "grok" ? { heartbeatStyle: "comment" as const } : {}), + ...(diagnosticContext ? { diagnostic: diagnosticContext } : {}), onUsage: usage => { // Raw adapter usage, pre wire-normalization (see the runTurn branch above). logCtx.usageFromBridge = true; diff --git a/src/web-search/loop.ts b/src/web-search/loop.ts index 0c957e1c17..f0852b6d8b 100644 --- a/src/web-search/loop.ts +++ b/src/web-search/loop.ts @@ -3,7 +3,7 @@ import type { AdapterEvent, OcxMessage, OcxParsedRequest, OcxProviderConfig, Ocx 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 { runWebSearch, type SidecarOutcome, type SidecarOutcomeRecorder, type SidecarSettings } from "./executor"; import { runAnthropicWebSearch } from "./anthropic-executor"; import { runXaiWebSearch, type XaiSearchOptions } from "./xai-executor"; @@ -323,6 +323,8 @@ export interface WebSearchLoopDeps { retryOn429Policy?: Required | null; /** Called only when the final bridged Responses stream reaches completed or incomplete. */ onCompletedResponse?: (response: Record) => void; + /** Internal, opt-in structural stream diagnostics shared with the final bridge. */ + diagnostic?: BridgeDiagnosticContext; } /** @@ -612,6 +614,10 @@ export async function runWithWebSearch(deps: WebSearchLoopDeps): Promise { @@ -33,7 +35,94 @@ async function collectSse(stream: ReadableStream): Promise<{ event?: }); } +const initialDebugEnv = process.env.OCX_DEBUG; + describe("Responses bridge reasoning and usage parity", () => { + afterEach(() => { + resetDebugSettingsForTests(); + resetDebugLogBufferForTests(); + if (initialDebugEnv === undefined) delete process.env.OCX_DEBUG; + else process.env.OCX_DEBUG = initialDebugEnv; + }); + + test("bridge diagnostics classify each adapter event once without changing the wire", async () => { + process.env.OCX_DEBUG = "1"; + const error = spyOn(console, "error").mockImplementation(() => {}); + try { + const events: AdapterEvent[] = [ + { type: "assistant_boundary" }, + { type: "text_delta", text: "fixture reasoning and secret" }, + { type: "tool_call_start", id: "call-1", name: "secret_tool" }, + { type: "tool_call_delta", arguments: '{"secret":"argument"}' }, + { type: "tool_call_end" }, + { type: "done" }, + ]; + const frames = await collectSse(bridgeToResponsesSSE(replay(events), "routed/model", undefined, undefined, undefined, undefined, undefined, { + diagnostic: { requestId: "req-1", adapterName: "openai-chat", attempt: 2, recovery: "empty-completion" }, + })); + const lines = getDebugLogEntries().map(entry => entry.line).filter(line => line.includes("\"stage\":\"bridge\"")); + expect(lines).toHaveLength(events.length); + expect(lines.filter(line => line.includes('"eventType":"assistant_boundary"'))).toHaveLength(1); + expect(lines.filter(line => line.includes('"eventType":"tool_call_start"'))).toHaveLength(1); + expect(lines.filter(line => line.includes('"eventType":"tool_call_delta"'))).toHaveLength(1); + expect(lines.every(line => !line.includes("fixture reasoning and secret") && !line.includes("secret_tool") && !line.includes("secret"))).toBe(true); + expect(frames.filter(frame => frame.event === "response.completed")).toHaveLength(1); + expect(frames.find(frame => frame.event === "response.output_text.delta")?.data).toMatchObject({ delta: "fixture reasoning and secret" }); + } finally { + error.mockRestore(); + } + }); + + test("incomplete diagnostics fingerprint upstream-controlled reasons instead of logging them", async () => { + process.env.OCX_DEBUG = "1"; + const error = spyOn(console, "error").mockImplementation(() => {}); + const reason = "upstream err.message fixture-secret"; + try { + await collectSse(bridgeToResponsesSSE(replay([ + { type: "incomplete", reason }, + ]), "routed/model", undefined, undefined, undefined, undefined, undefined, { + diagnostic: { requestId: "req-incomplete", adapterName: "openai-responses", attempt: 1 }, + })); + const line = getDebugLogEntries().map(entry => entry.line).find(entry => entry.includes('"stage":"bridge"')) ?? ""; + expect(line).not.toContain(reason); + expect(line).toContain(`"reasonByteLength":${Buffer.byteLength(reason)}`); + expect(line).toContain('"reasonFingerprint"'); + } finally { + error.mockRestore(); + } + }); + + test("diagnostics fingerprint arbitrary upstream error codes and stop reasons", async () => { + process.env.OCX_DEBUG = "1"; + const error = spyOn(console, "error").mockImplementation(() => {}); + const code = "Bearer upstream-code-secret@example.test"; + const stopReason = "provider-stop-secret-account-123"; + try { + await collectSse(bridgeToResponsesSSE(replay([ + { type: "error", message: "failed", code }, + ]), "routed/model", undefined, undefined, undefined, undefined, undefined, { + diagnostic: { requestId: "req-upstream-fields", adapterName: "openai-chat" }, + })); + const errorLine = getDebugLogEntries().map(entry => entry.line) + .find(line => line.includes('"stage":"bridge"') && line.includes('"eventType":"error"')) ?? ""; + await collectSse(bridgeToResponsesSSE(replay([ + { type: "done", stopReason }, + ]), "routed/model", undefined, undefined, undefined, undefined, undefined, { + diagnostic: { requestId: "req-upstream-fields", adapterName: "openai-chat" }, + })); + const doneLine = getDebugLogEntries().map(entry => entry.line) + .find(line => line.includes('"stage":"bridge"') && line.includes('"eventType":"done"')) ?? ""; + expect(errorLine).not.toContain(code); + expect(errorLine).toContain(`"codeByteLength":${Buffer.byteLength(code)}`); + expect(errorLine).toContain('"codeFingerprint"'); + expect(doneLine).not.toContain(stopReason); + expect(doneLine).toContain(`"stopReasonByteLength":${Buffer.byteLength(stopReason)}`); + expect(doneLine).toContain('"stopReasonFingerprint"'); + } finally { + error.mockRestore(); + } + }); + test("first-output callback fires once on first non-empty delta (heartbeat/empty skipped)", async () => { let firstOutputs = 0; await collectSse(bridgeToResponsesSSE(replay([ diff --git a/tests/adapters/terminal-continuation-owner-rotation.test.ts b/tests/adapters/terminal-continuation-owner-rotation.test.ts index 458f8237b6..fc28a472c2 100644 --- a/tests/adapters/terminal-continuation-owner-rotation.test.ts +++ b/tests/adapters/terminal-continuation-owner-rotation.test.ts @@ -1,4 +1,4 @@ -import { afterEach, beforeEach, describe, expect, mock, test } from "bun:test"; +import { afterEach, beforeEach, describe, expect, mock, spyOn, test } from "bun:test"; import { mkdtempSync} from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; @@ -16,6 +16,8 @@ import type { OcxParsedRequest, OcxProviderConfig, } from "../../src/types"; +import { getDebugLogEntries, resetDebugLogBufferForTests } from "../../src/lib/debug-log-buffer"; +import { resetDebugSettingsForTests } from "../../src/lib/debug-settings"; import { removeTreeWithRetry } from "../helpers/remove-tree"; interface BuildObservation { @@ -24,6 +26,7 @@ interface BuildObservation { } let builds: BuildObservation[] = []; +const PREVIOUS_DEBUG = process.env.OCX_DEBUG; function eventsForPhase(phase: string): AdapterEvent[] { if (phase === "seed") { @@ -46,6 +49,16 @@ function eventsForPhase(phase: string): AdapterEvent[] { }, ]; } + if (phase === "final") { + return [ + { type: "text_delta", text: "completed after connection reset" }, + { + type: "done", + stopReason: "end_turn", + providerState: { kiro: { conversationId: "private-final" } }, + }, + ]; + } if (phase === "rotated") { return [ { type: "text_delta", text: "completed on the rotated key" }, @@ -127,9 +140,78 @@ describe("terminal continuation provider-owner rotation", () => { else process.env.OPENCODEX_HOME = previousHome; clearKeyCooldowns(); clearResponseStateForTests(); + resetDebugSettingsForTests(); + resetDebugLogBufferForTests(); + if (PREVIOUS_DEBUG === undefined) delete process.env.OCX_DEBUG; + else process.env.OCX_DEBUG = PREVIOUS_DEBUG; removeTreeWithRetry(testHome); }); + test("continuation connection-reset recovery labels adapter and bridge diagnostics", async () => { + process.env.OCX_DEBUG = "1"; + const error = spyOn(console, "error").mockImplementation(() => {}); + const key = "key-continuation-000111222333"; + const config: OcxConfig = { + port: 0, + defaultProvider: "owned", + providers: { + owned: { + adapter: "test-terminal-owned", + baseUrl: "https://owned-terminal.test/v1", + authMode: "key", + apiKey: key, + terminalContinuationGuard: true, + replayTransientFailures: true, + }, + }, + } as OcxConfig; + saveConfig(config); + let sends = 0; + globalThis.fetch = (async (_input, init) => { + sends += 1; + if (sends === 1) return new Response("", { headers: { "x-test-phase": "plan" } }); + if (sends === 2) { + const reset = new Error("socket reset fixture"); + (reset as Error & { code?: string }).code = "ECONNRESET"; + throw reset; + } + expect(new Headers(init?.headers).get("authorization")).toBe(`Bearer ${key}`); + return new Response("", { headers: { "x-test-phase": "final" } }); + }) as typeof fetch; + try { + const response = await handleResponses( + new Request("http://localhost/v1/responses", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + model: "owned/model", + input: "Please modify the file now", + stream: true, + tools: [{ type: "function", name: "read_file", description: "read", parameters: { type: "object" } }], + }), + }), + config, + { model: "", provider: "" }, + ); + expect(response.status).toBe(200); + await response.text(); + const lines = getDebugLogEntries().map(entry => entry.line); + const adapter = lines.find(line => + line.includes('"stage":"adapter"') + && line.includes('"eventType":"text_delta"') + && line.includes('"recovery":"connection-reset"')) ?? ""; + const bridge = lines.find(line => + line.includes('"stage":"bridge"') + && line.includes('"eventType":"text_delta"') + && line.includes('"recovery":"connection-reset"')) ?? ""; + expect(adapter).toContain('"recovery":"connection-reset"'); + expect(bridge).toContain('"recovery":"connection-reset"'); + expect(sends).toBe(3); + } finally { + error.mockRestore(); + } + }); + test("429 rotation fences inherited state and persists the rotated owner", async () => { const keyA = "key-alpha-000111222333"; const keyB = "key-beta-444555666777"; diff --git a/tests/images/loop.test.ts b/tests/images/loop.test.ts index 01db03bf0c..976e230b90 100644 --- a/tests/images/loop.test.ts +++ b/tests/images/loop.test.ts @@ -1,4 +1,4 @@ -import { afterAll, beforeAll, beforeEach, describe, expect, mock, test } from "bun:test"; +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, mock, spyOn, test } from "bun:test"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { randomUUID } from "node:crypto"; @@ -7,8 +7,11 @@ import type { AdapterEvent, OcxParsedRequest } from "../../src/types"; import type { ImageBridgePlan, ImageCallResult } from "../../src/images/types"; import type { ImageBridgeDeps } from "../../src/images/loop"; import { createTestTranslatorBudget } from "../helpers/translator-budget"; +import { getDebugLogEntries, resetDebugLogBufferForTests } from "../../src/lib/debug-log-buffer"; +import { resetDebugSettingsForTests } from "../../src/lib/debug-settings"; const PREV_HOME = process.env.OPENCODEX_HOME; +const PREV_DEBUG = process.env.OCX_DEBUG; let runWithImageBridgeProduction: typeof import("../../src/images/loop")["runWithImageBridge"]; let clampImageMaxRounds: typeof import("../../src/images/loop")["clampImageMaxRounds"]; let DEFAULT_MAX_ROUNDS: typeof import("../../src/images/loop")["DEFAULT_MAX_ROUNDS"]; @@ -52,6 +55,12 @@ function runWithImageBridge( }); } afterAll(() => { if (PREV_HOME === undefined) delete process.env.OPENCODEX_HOME; else process.env.OPENCODEX_HOME = PREV_HOME; mock.restore(); }); +afterEach(() => { + resetDebugSettingsForTests(); + resetDebugLogBufferForTests(); + if (PREV_DEBUG === undefined) delete process.env.OCX_DEBUG; + else process.env.OCX_DEBUG = PREV_DEBUG; +}); // --- Mock adapter: yields canned events per iteration from a queue --- let streamQueue: AdapterEvent[][] = []; @@ -104,6 +113,27 @@ async function runAndGetSSE(streams: AdapterEvent[][], fulfill?: ImageCallResult } describe("runWithImageBridge", () => { + test("routed image streams carry adapter and bridge diagnostics", async () => { + process.env.OCX_DEBUG = "1"; + const error = spyOn(console, "error").mockImplementation(() => {}); + try { + streamQueue = [[{ type: "text_delta", text: "image diagnostic secret" }, { type: "done" }]]; + const response = await runWithImageBridge({ + parsed: makeParsed(), + adapter: mockAdapter, + plan, + diagnostic: { requestId: "image-diagnostic", adapterName: "test" }, + }); + await response.text(); + const lines = getDebugLogEntries().map(entry => entry.line); + expect(lines.some(line => line.includes('"stage":"adapter"') && line.includes('"eventType":"text_delta"'))).toBe(true); + expect(lines.some(line => line.includes('"stage":"bridge"') && line.includes('"eventType":"text_delta"'))).toBe(true); + expect(lines.every(line => !line.includes("image diagnostic secret"))).toBe(true); + } finally { + error.mockRestore(); + } + }); + test("translator overflow remains typed through the image loop and bridge", async () => { const sse = await runAndGetSSE([[ { diff --git a/tests/lib/debug.test.ts b/tests/lib/debug.test.ts index 6ce7359d2d..a5eb8d1c7c 100644 --- a/tests/lib/debug.test.ts +++ b/tests/lib/debug.test.ts @@ -3,7 +3,7 @@ import { appendDebugLogLine, debugBufferMetrics, getDebugLogEntries, resetDebugL import { ResourceAdmissionError, RETAINED_TRUNCATION_MARKER, retainedUtf8Bytes } from "../../src/lib/admission"; import { getInjectionDebugLogEntries, injectionDebugLog, resetInjectionDebugLogBufferForTests } from "../../src/lib/injection-debug-log"; import { markActivity, activityBreadcrumb } from "../../src/lib/sidecar-tracker"; -import { debugDroppedFrame, debugProviderDiagnostic } from "../../src/lib/debug"; +import { debugDroppedFrame, debugFingerprint, debugProviderDiagnostic, debugStreamDiagnostic } from "../../src/lib/debug"; import { resetDebugSettingsForTests, setDebugSettings } from "../../src/lib/debug-settings"; describe("debug frame logging", () => { @@ -75,6 +75,47 @@ describe("debug frame logging", () => { } }); + test("debug fingerprints are process-stable, content-free, and debug-gated", () => { + delete process.env.OCX_DEBUG; + expect(debugFingerprint("fixture reasoning and secret")).toBeUndefined(); + + process.env.OCX_DEBUG = "1"; + const same = debugFingerprint("fixture reasoning and secret"); + const again = debugFingerprint("fixture reasoning and secret"); + const different = debugFingerprint("different fixture"); + expect(same).toMatch(/^[0-9a-f]{64}$/); + expect(again).toBe(same); + expect(different).toMatch(/^[0-9a-f]{64}$/); + expect(different).not.toBe(same); + expect(getDebugLogEntries()).toHaveLength(0); + }); + + test("stream diagnostics are structural and omit content", () => { + process.env.OCX_DEBUG = "1"; + const error = spyOn(console, "error").mockImplementation(() => {}); + try { + debugStreamDiagnostic( + { requestId: "req-1", adapterName: "openai-chat" }, + "adapter", + 3, + "text_delta", + { byteLength: 21, fingerprint: debugFingerprint("fixture reasoning and secret"), attempt: 2, recovery: "empty-completion" }, + ); + const line = getDebugLogEntries()[0]?.line ?? ""; + expect(line).toContain("[ocx:openai-chat:stream]"); + expect(line).toContain('"stage":"adapter"'); + expect(line).toContain('"sequence":3'); + expect(line).toContain('"eventType":"text_delta"'); + expect(line).toContain('"byteLength":21'); + expect(line).toContain('"fingerprint"'); + expect(line).toContain('"attempt":2'); + expect(line).toContain('"recovery":"empty-completion"'); + expect(line).not.toContain("fixture reasoning and secret"); + } finally { + error.mockRestore(); + } + }); + test("debugProviderDiagnostic emits when enabled via runtime settings API", () => { delete process.env.OCX_DEBUG; setDebugSettings({ debug: true }); diff --git a/tests/web-search/web-search.test.ts b/tests/web-search/web-search.test.ts index ce8f6e0f8a..a5e9ccc780 100644 --- a/tests/web-search/web-search.test.ts +++ b/tests/web-search/web-search.test.ts @@ -1,4 +1,4 @@ -import { afterEach, describe, expect, test } from "bun:test"; +import { afterEach, describe, expect, spyOn, test } from "bun:test"; import { parseRequest } from "../../src/responses/parser"; import { planWebSearch, shouldResolveOpenAiWebSearchSidecar, webSearchStallTimeoutSec } from "../../src/web-search"; import { runWithWebSearch as runWithWebSearchProduction, type WebSearchLoopDeps } from "../../src/web-search/loop"; @@ -13,6 +13,8 @@ import type { AdapterFetchContext, ProviderAdapter } from "../../src/adapters/ba import type { OcxMessage, OcxParsedRequest } from "../../src/types"; import { fakeChatGptJwt } from "../helpers/fake-chatgpt-jwt"; import { createTestTranslatorBudget } from "../helpers/translator-budget"; +import { getDebugLogEntries, resetDebugLogBufferForTests } from "../../src/lib/debug-log-buffer"; +import { resetDebugSettingsForTests } from "../../src/lib/debug-settings"; import { withUpstreamHttpVersion } from "../../src/lib/upstream-http-version"; /** @@ -396,7 +398,47 @@ describe("web-search sidecar planning", () => { }); const originalFetch = globalThis.fetch; -afterEach(() => { globalThis.fetch = originalFetch; }); +const originalDebug = process.env.OCX_DEBUG; +afterEach(() => { + globalThis.fetch = originalFetch; + resetDebugSettingsForTests(); + resetDebugLogBufferForTests(); + if (originalDebug === undefined) delete process.env.OCX_DEBUG; + else process.env.OCX_DEBUG = originalDebug; +}); + +test("routed web-search streams carry adapter and bridge diagnostics", async () => { + process.env.OCX_DEBUG = "1"; + const error = spyOn(console, "error").mockImplementation(() => {}); + try { + const adapter: ProviderAdapter = { + name: "diagnostic-search", + buildRequest: () => ({ url: "https://routed.test/v1", method: "POST", headers: {}, body: "{}" }), + fetchResponse: async () => new Response("wire", { status: 200 }), + async *parseStream() { + yield { type: "text_delta", text: "search diagnostic secret" }; + yield { type: "done" }; + }, + }; + const response = await runWithWebSearch({ + parsed: parseRequest({ model: "routed/model", input: "hi", stream: true, tools: [{ type: "web_search" }] }), + adapter, + forwardProvider, + hostedTool: { type: "web_search" }, + selectedForwardHeaders: new Headers({ authorization: "Bearer token" }), + settings: { model: "gpt-5.6-luna", reasoning: "low", timeoutMs: 30_000 }, + maxSearches: 1, + diagnostic: { requestId: "search-diagnostic", adapterName: "diagnostic-search" }, + }); + await response.text(); + const lines = getDebugLogEntries().map(entry => entry.line); + expect(lines.some(line => line.includes('"stage":"adapter"') && line.includes('"eventType":"text_delta"'))).toBe(true); + expect(lines.some(line => line.includes('"stage":"bridge"') && line.includes('"eventType":"text_delta"'))).toBe(true); + expect(lines.every(line => !line.includes("search diagnostic secret"))).toBe(true); + } finally { + error.mockRestore(); + } +}); test("issue #2885 — Zhipu-shaped web-search routing preserves the provider HTTP version pin", async () => { let routedProtocol: string | undefined; @@ -2651,4 +2693,3 @@ describe("connection-reset recovery parity on the web-search legs", () => { expect(typeof attempts[1]!.body).toBe("string"); }); }); -