From 2d90f9684a721d133e061ec43b13bfc32d6f79de Mon Sep 17 00:00:00 2001 From: Eran Date: Sat, 5 Sep 2026 05:53:27 +0800 Subject: [PATCH] fix(responses): recover agent_message encrypted-content rejections instead of adapter_eof A codex-app thread whose history carries subagent agent_message items with backend-minted encrypted_content parts fails with the exact upstream rejection "Encrypted function output content could not be decrypted or decoded." once the serving identity changes. The ChatGPT backend reports that failure as a 200 SSE stream holding response.created and a bare error event, then EOF with no terminal, so the turn surfaced to Codex as a misleading adapter_eof with the real error hidden. - Treat a zero-output error SSE event carrying the exact decryption rejection as a retryable preflight terminal; error events no longer commit a stream as output. - Detect and strip encrypted_content parts in agent_message content[] (in addition to function_call_output/custom_tool_call_output output[]) during the existing one-shot opaque-blob recovery rebuild, replacing them with an omission marker. - Extract the flat message of stream error events so relay failed-tail responses surface the real upstream error instead of adapter_eof when recovery is exhausted. Verified end-to-end on the reporter thread: the turn now completes with recoveryKinds=[opaque-blob-rejection] on the second send instead of adapter_eof. --- src/lib/errors.ts | 7 +- src/server/relay-eager.ts | 26 +- src/server/relay.ts | 26 +- .../responses/combo-stream-preflight.ts | 28 +- src/server/responses/core.ts | 196 ++++++++- tests/passthrough-abort.test.ts | 4 +- tests/responses-opaque-blob-recovery.test.ts | 404 ++++++++++++++++++ tests/sse-failed-tail.test.ts | 26 ++ 8 files changed, 692 insertions(+), 25 deletions(-) diff --git a/src/lib/errors.ts b/src/lib/errors.ts index 624917507c..04213628aa 100644 --- a/src/lib/errors.ts +++ b/src/lib/errors.ts @@ -18,7 +18,12 @@ export function upstreamErrorMessageFromPayload(payload: unknown): string | unde const message = json.error?.message ?? json.last_error?.message ?? json.response?.error?.message - ?? json.response?.incomplete_details?.message; + ?? json.response?.incomplete_details?.message + // The Responses stream error event carries a flat message (type/code/message), + // unlike the response.failed envelope the branches above already cover. + ?? ((json as { type?: unknown }).type === "error" + ? (json as unknown as { message?: unknown }).message + : undefined); return typeof message === "string" ? message : undefined; } diff --git a/src/server/relay-eager.ts b/src/server/relay-eager.ts index f389a00d5b..be90e8b21c 100644 --- a/src/server/relay-eager.ts +++ b/src/server/relay-eager.ts @@ -79,6 +79,8 @@ export type EagerRelayOptions = { postCancelDrainMs?: number; /** Post-cancel discard-drain byte bound. Default 32 MiB. */ postCancelDrainBytes?: number; + /** Last known upstream failure to preserve when EOF would otherwise become adapter_eof. */ + upstreamError?: string; /** Injectable clock for tests. */ now?: () => number; }; @@ -108,6 +110,24 @@ export function relaySseEagerBounded( const terminalEncoder = new TextEncoder(); const adapterEofFrame = adapterEofIncompleteFrame(terminalEncoder); const terminalSentinel = doneFrame(terminalEncoder); + const upstreamErrorFrame = opts?.upstreamError === undefined + ? adapterEofFrame + : terminalEncoder.encode(`event: response.failed\ndata: ${JSON.stringify({ + type: "response.failed", + response: { + status: "failed", + error: { + type: "upstream_error", + code: "upstream_server_error", + message: opts.upstreamError, + }, + last_error: { + type: "upstream_error", + code: "upstream_server_error", + message: opts.upstreamError, + }, + }, + })}\n\n`); const terminalBoundary = createSseTerminalOutputBoundary(); const activeRewrite: SseBlockRewrite | undefined = hooks.rewriteBlocks ?? (hooks.rewritePayload ? payloadRewriteAsBlockRewrite(hooks.rewritePayload) : undefined); @@ -282,12 +302,12 @@ export function relaySseEagerBounded( } else if (!hooks.sawTerminal() && !cancelled && !upstream.signal.aborted) { // A clean 200 EOF without a Responses terminal must be visible to // Codex as one incomplete turn, followed by the normal sentinel. - queuedBytes += adapterEofFrame.byteLength + terminalSentinel.byteLength; + queuedBytes += upstreamErrorFrame.byteLength + terminalSentinel.byteLength; try { - controllerRef?.enqueue(adapterEofFrame); + controllerRef?.enqueue(upstreamErrorFrame); controllerRef?.enqueue(terminalSentinel); } catch { /* client already gone */ } - syntheticKind = "incomplete"; + syntheticKind = opts?.upstreamError === undefined ? "incomplete" : "failed"; } break; } diff --git a/src/server/relay.ts b/src/server/relay.ts index a523ffd7e5..f128444d96 100644 --- a/src/server/relay.ts +++ b/src/server/relay.ts @@ -260,10 +260,29 @@ export function relaySseWithFailedTail( body: ReadableStream, upstream: AbortController, onClientGone?: (reason?: unknown) => void, + opts?: { upstreamError?: string }, ): ReadableStream { const reader = body.getReader(); const encoder = new TextEncoder(); const terminalBoundary = createSseTerminalOutputBoundary(); + const failedTailPayload = opts?.upstreamError === undefined + ? FAILED_TAIL_FALLBACK_PAYLOAD + : JSON.stringify({ + type: "response.failed", + response: { + status: "failed", + error: { + type: "upstream_error", + code: "upstream_server_error", + message: opts.upstreamError, + }, + last_error: { + type: "upstream_error", + code: "upstream_server_error", + message: opts.upstreamError, + }, + }, + }); let closed = false; const relayChunk = ( controller: ReadableStreamDefaultController, @@ -306,8 +325,11 @@ export function relaySseWithFailedTail( // A clean upstream EOF is still a failed Responses turn when no // protocol terminal arrived. Make that state explicit so Codex // does not treat HTTP 200 + bare EOF as a retryable disconnect. - const incomplete = adapterEofIncompleteFrame(encoder); - controller.enqueue(incomplete); + controller.enqueue(encoder.encode( + opts?.upstreamError === undefined + ? `event: response.incomplete\ndata: ${ADAPTER_EOF_INCOMPLETE_PAYLOAD}\n\n` + : `event: response.failed\ndata: ${failedTailPayload}\n\n`, + )); controller.enqueue(doneFrame(encoder)); } terminalBoundary.dispose(); diff --git a/src/server/responses/combo-stream-preflight.ts b/src/server/responses/combo-stream-preflight.ts index 3856c32b98..bf239f12b1 100644 --- a/src/server/responses/combo-stream-preflight.ts +++ b/src/server/responses/combo-stream-preflight.ts @@ -30,6 +30,23 @@ const RETRYABLE_ZERO_OUTPUT_INCOMPLETE_REASONS = new Set([ "upstream_stall_timeout", ]); +// The ChatGPT backend reports an undecryptable replayed blob as a bare error SSE +// event (not response.failed) before any output; with zero committed output that +// event is exactly as replayable as a response.failed terminal. +const ENCRYPTED_FUNCTION_OUTPUT_REJECTION_MESSAGE = + "Encrypted function output content could not be decrypted or decoded."; + +function errorEventMessage(payload: Record): string | undefined { + const direct = payload.message; + if (typeof direct === "string") return direct; + const nested = payload.error; + if (nested !== null && typeof nested === "object" && !Array.isArray(nested)) { + const message = (nested as { message?: unknown }).message; + if (typeof message === "string") return message; + } + return undefined; +} + function retryableZeroOutputTerminal(payload: unknown): boolean { if (!payload || typeof payload !== "object" || Array.isArray(payload)) return false; const event = payload as { @@ -37,6 +54,9 @@ function retryableZeroOutputTerminal(payload: unknown): boolean { response?: { incomplete_details?: { reason?: unknown } }; }; if (event.type === "response.failed") return true; + if (event.type === "error") { + return errorEventMessage(event) === ENCRYPTED_FUNCTION_OUTPUT_REJECTION_MESSAGE; + } if (event.type !== "response.incomplete") return false; const reason = event.response?.incomplete_details?.reason; return typeof reason === "string" && RETRYABLE_ZERO_OUTPUT_INCOMPLETE_REASONS.has(reason); @@ -51,6 +71,9 @@ export function comboStreamPayloadCommitsOutput(payload: unknown): boolean { if (!payload || typeof payload !== "object" || Array.isArray(payload)) return true; const type = (payload as { type?: unknown }).type; if (typeof type !== "string") return true; + // An error event carries no client-visible output; treating it as committing + // would pin a child to a turn that already failed before producing anything. + if (type === "error") return false; return !PRE_OUTPUT_CONTROL_EVENTS.has(type) && !TERMINAL_EVENTS.has(type); } @@ -180,7 +203,10 @@ export async function preflightComboStreamResponse( inspector.feed(retained); } - if ((terminalStatus === "failed" || terminalStatus === "incomplete") + // A bare error event is not a protocol terminal (terminalStatus stays undefined), + // so its exact-message retryable match doubles as the terminal evidence. + if ((terminalStatus === "failed" || terminalStatus === "incomplete" + || retryableTerminalPayload?.type === "error") && !outputCommitted && retryableTerminalPayload) { await reader.cancel("retrying zero-output combo stream terminal").catch(() => undefined); return { kind: "failed", response: failedTerminalResponse(response, retryableTerminalPayload, logCtx) }; diff --git a/src/server/responses/core.ts b/src/server/responses/core.ts index cb1e00678e..f2fb447f3c 100644 --- a/src/server/responses/core.ts +++ b/src/server/responses/core.ts @@ -629,28 +629,97 @@ const OPAQUE_RESPONSES_INPUT_TYPES = new Set([ "compaction_summary", "context_compaction", ]); +const ENCRYPTED_FUNCTION_OUTPUT_REJECTION = "Encrypted function output content could not be decrypted or decoded."; +const FUNCTION_OUTPUT_TYPES = new Set(["function_call_output", "custom_tool_call_output"]); +// codex-app subagent results replay as agent_message items whose content parts may carry +// backend-minted encrypted_content; the ChatGPT backend decrypts them in its function-output +// path, so a cross-identity replay of those parts produces ENCRYPTED_FUNCTION_OUTPUT_REJECTION. +const AGENT_MESSAGE_TYPE = "agent_message"; + +function encryptedFunctionOutputParts(output: unknown): boolean { + return Array.isArray(output) && output.some(part => ( + part !== null + && typeof part === "object" + && !Array.isArray(part) + && (part as { type?: unknown }).type === "encrypted_content" + && typeof (part as { encrypted_content?: unknown }).encrypted_content === "string" + && (part as { encrypted_content: string }).encrypted_content.length > 0 + )); +} -function outboundResponsesBodyCarriesOpaqueBlob(bodyText: string | undefined): boolean { - if (!bodyText) return false; +function outboundResponsesInput(bodyText: string | undefined): unknown[] | undefined { + if (!bodyText) return undefined; try { const body = JSON.parse(bodyText) as unknown; - if (!body || typeof body !== "object" || Array.isArray(body)) return false; + if (!body || typeof body !== "object" || Array.isArray(body)) return undefined; const input = (body as { input?: unknown }).input; - if (!Array.isArray(input)) return false; - return input.some(item => { - if (!item || typeof item !== "object" || Array.isArray(item)) return false; - const candidate = item as { type?: unknown; encrypted_content?: unknown }; - return typeof candidate.type === "string" - && OPAQUE_RESPONSES_INPUT_TYPES.has(candidate.type) - && typeof candidate.encrypted_content === "string" - && candidate.encrypted_content.length > 0; - }); + return Array.isArray(input) ? input : undefined; + } catch { + return undefined; + } +} + +function outboundResponsesBodyCarriesEncryptedFunctionOutput(bodyText: string | undefined): boolean { + const input = outboundResponsesInput(bodyText); + if (!input) return false; + return input.some(item => { + if (item === null || typeof item !== "object" || Array.isArray(item)) return false; + const candidate = item as { type?: unknown; output?: unknown; content?: unknown }; + const type = String(candidate.type ?? ""); + if (FUNCTION_OUTPUT_TYPES.has(type) && encryptedFunctionOutputParts(candidate.output)) return true; + return type === AGENT_MESSAGE_TYPE && encryptedFunctionOutputParts(candidate.content); + }); +} + +function outboundResponsesBodyCarriesOpaqueBlob(bodyText: string | undefined): boolean { + const input = outboundResponsesInput(bodyText); + if (!input) return false; + return input.some(item => { + if (!item || typeof item !== "object" || Array.isArray(item)) return false; + const candidate = item as { type?: unknown; encrypted_content?: unknown; output?: unknown }; + if ( + typeof candidate.type === "string" + && OPAQUE_RESPONSES_INPUT_TYPES.has(candidate.type) + && typeof candidate.encrypted_content === "string" + && candidate.encrypted_content.length > 0 + ) return true; + if ( + typeof candidate.type === "string" + && FUNCTION_OUTPUT_TYPES.has(candidate.type) + && encryptedFunctionOutputParts(candidate.output) + ) return true; + return candidate.type === AGENT_MESSAGE_TYPE + && encryptedFunctionOutputParts((candidate as { content?: unknown }).content); + }); +} + +function isEncryptedFunctionOutputRejection(bodyText: string): boolean { + if (bodyText.trim() === ENCRYPTED_FUNCTION_OUTPUT_REJECTION) return true; + try { + const payload = JSON.parse(bodyText) as unknown; + if (!payload || typeof payload !== "object" || Array.isArray(payload)) return false; + const record = payload as { detail?: unknown; message?: unknown; error?: unknown }; + if (record.detail === ENCRYPTED_FUNCTION_OUTPUT_REJECTION) return true; + if (record.message === ENCRYPTED_FUNCTION_OUTPUT_REJECTION) return true; + if (record.error === ENCRYPTED_FUNCTION_OUTPUT_REJECTION) return true; + return record.error !== null + && typeof record.error === "object" + && !Array.isArray(record.error) + && (record.error as { message?: unknown }).message === ENCRYPTED_FUNCTION_OUTPUT_REJECTION; } catch { return false; } } function isSelfIdentifiedOpaqueBlobRejection(bodyText: string): boolean { + if (isEncryptedFunctionOutputRejection(bodyText)) return true; + try { + if (upstreamErrorMessageFromPayload(JSON.parse(bodyText) as unknown) === ENCRYPTED_FUNCTION_OUTPUT_REJECTION) { + return true; + } + } catch { + /* invalid JSON bodies fall through to the exact nested envelope checks */ + } try { const payload = JSON.parse(bodyText) as unknown; if (!payload || typeof payload !== "object" || Array.isArray(payload)) return false; @@ -695,8 +764,13 @@ export function shouldAttemptOpaqueBlobRecovery(args: { errorBody: string; alreadyAttempted: boolean; }): boolean { - return args.status >= 400 - && args.status < 500 + const acceptedStatus = (args.status >= 400 && args.status < 500) + || ( + args.status === 502 + && outboundResponsesBodyCarriesEncryptedFunctionOutput(args.outboundBody) + && isEncryptedFunctionOutputRejection(args.errorBody) + ); + return acceptedStatus && args.adapterName === "openai-responses" && !args.alreadyAttempted && outboundResponsesBodyCarriesOpaqueBlob(args.outboundBody) @@ -712,7 +786,7 @@ async function opaqueBlobRejectionBodyForRecovery( ): Promise { if ( response.status < 400 - || response.status >= 500 + || (response.status >= 500 && response.status !== 502) || adapterName !== "openai-responses" || alreadyAttempted || !outboundResponsesBodyCarriesOpaqueBlob(outboundBody) @@ -789,6 +863,50 @@ function normalizeUpstreamErrorText(text: string, fallback: string): NormalizedU function prepareOpaqueBlobRecovery(parsed: OcxParsedRequest): void { parsed._stripReasoningEncryptedContent = true; + const rawBody = parsed._rawBody; + if (!rawBody || typeof rawBody !== "object" || Array.isArray(rawBody)) return; + const input = (rawBody as { input?: unknown }).input; + if (!Array.isArray(input)) return; + const stripEncryptedParts = (parts: unknown[]): unknown[] => { + let changed = false; + const stripped = parts.map(part => { + if ( + part !== null + && typeof part === "object" + && !Array.isArray(part) + && (part as { type?: unknown }).type === "encrypted_content" + && typeof (part as { encrypted_content?: unknown }).encrypted_content === "string" + && (part as { encrypted_content: string }).encrypted_content.length > 0 + ) { + changed = true; + return { type: "input_text", text: "[encrypted content omitted]" }; + } + return part; + }); + return changed ? stripped : parts; + }; + const strippedInput = input.map(item => { + if (!item || typeof item !== "object" || Array.isArray(item)) return item; + const record = item as Record; + const type = String(record.type ?? ""); + if (FUNCTION_OUTPUT_TYPES.has(type) && Array.isArray(record.output)) { + const output = stripEncryptedParts(record.output); + return output !== record.output ? { ...record, output } : item; + } + if (type === AGENT_MESSAGE_TYPE && Array.isArray(record.content)) { + const content = stripEncryptedParts(record.content); + return content !== record.content ? { ...record, content } : item; + } + return item; + }); + parsed._rawBody = { ...(rawBody as Record), input: strippedInput }; +} + +function resetStreamedOpaqueBlobLogContext(logCtx: RequestLogContext): void { + delete logCtx.upstreamError; + delete logCtx.terminalHttpStatus; + delete logCtx.terminalErrorCode; + delete logCtx.terminalIncompleteReason; } type OpaqueBlobRecoveryGuard = { attempted: boolean }; @@ -4609,6 +4727,38 @@ async function handleResponsesInner( upstreamResponse = opaqueBlobRecovery.response; continue passthroughRecovery; } + + const recoveryContentType = upstreamResponse.headers.get("content-type")?.toLowerCase() ?? ""; + const streamedFunctionOutputCandidate = upstreamResponse.ok + && !!upstreamResponse.body + && (recoveryContentType.includes("text/event-stream") || (!recoveryContentType && parsed.stream)) + && !opaqueBlobRecoveryGuard.attempted + && outboundResponsesBodyCarriesEncryptedFunctionOutput(request.body); + if (streamedFunctionOutputCandidate) { + const preflightLog: RequestLogContext = { model: logCtx.model, provider: logCtx.provider }; + const preflight = await preflightComboStreamResponse(upstreamResponse, preflightLog); + upstreamResponse = preflight.response; + if (preflight.kind === "failed") { + const streamedOpaqueRecovery = await attemptOpaqueBlobRecovery({ + response: upstreamResponse, + outboundBody: request.body, + adapterName: adapter.name, + parsed, + guard: opaqueBlobRecoveryGuard, + signal: upstream.signal, + }, rebuildAndRefetch); + if (streamedOpaqueRecovery.kind === "failed") return streamedOpaqueRecovery.response; + if (streamedOpaqueRecovery.kind === "recovered") { + resetStreamedOpaqueBlobLogContext(logCtx); + upstreamResponse = streamedOpaqueRecovery.response; + continue passthroughRecovery; + } + logCtx.upstreamError = preflightLog.upstreamError; + logCtx.terminalHttpStatus = preflightLog.terminalHttpStatus; + logCtx.terminalErrorCode = preflightLog.terminalErrorCode; + logCtx.terminalIncompleteReason = preflightLog.terminalIncompleteReason; + } + } break; } const headers = sanitizePassthroughHeaders(upstreamResponse.headers); @@ -4903,7 +5053,14 @@ async function handleResponsesInner( }, onClientCancel: () => options.onNativePassthroughCancel?.(), onDone: () => unregisterTurn(turnAc), - }, inlineEagerRewrite ? { rewriteBudget: translatorBudget } : undefined); + }, inlineEagerRewrite + ? { + rewriteBudget: translatorBudget, + ...(logCtx.upstreamError === undefined ? {} : { upstreamError: logCtx.upstreamError }), + } + : logCtx.upstreamError === undefined + ? undefined + : { upstreamError: logCtx.upstreamError }); // When selected, this relay closes response.completed even if upstream // keeps the connection alive. Marked Codex WS traffic, Windows // forced-rewrite traffic, and Darwin explicit eager traffic apply @@ -4982,7 +5139,12 @@ async function handleResponsesInner( const rewrittenBody = clientBlockRewrite !== undefined ? relaySseWithBlockRewrite(nativeBody, clientBlockRewrite, translatorBudget) : nativeBody; - const clientBody = relaySseWithFailedTail(rewrittenBody, upstream, reason => clientGone.abort(reason)); + const clientBody = relaySseWithFailedTail( + rewrittenBody, + upstream, + reason => clientGone.abort(reason), + { upstreamError: logCtx.upstreamError }, + ); return markNativePassthroughSseResponse(new Response(clientBody, { status: upstreamResponse.status, headers, diff --git a/tests/passthrough-abort.test.ts b/tests/passthrough-abort.test.ts index 1a6aae29cd..760378a136 100644 --- a/tests/passthrough-abort.test.ts +++ b/tests/passthrough-abort.test.ts @@ -76,7 +76,9 @@ describe("passthrough relayWithAbort (RC2, passthrough path)", () => { expect(sseBranch).toContain("win32EagerRewrite"); expect(sseBranch).toContain("rewriteBlocks: clientBlockRewrite"); // Elsewhere the failed-tail relay converts mid-stream resets into a clean response.failed. - expect(sseBranch).toContain("relaySseWithFailedTail(rewrittenBody, upstream"); + expect(sseBranch).toContain("relaySseWithFailedTail("); + expect(sseBranch).toContain("rewrittenBody"); + expect(sseBranch).toContain("upstreamError: logCtx.upstreamError"); expect(sseBranch).toContain("new Response(clientBody"); expect(sseBranch).toContain("markNativePassthroughSseResponse"); // #314/phase 100 two-platform contract: the real core gate delegates to the diff --git a/tests/responses-opaque-blob-recovery.test.ts b/tests/responses-opaque-blob-recovery.test.ts index 277561ba22..24b3f86da3 100644 --- a/tests/responses-opaque-blob-recovery.test.ts +++ b/tests/responses-opaque-blob-recovery.test.ts @@ -19,6 +19,8 @@ import { removeTreeWithRetry } from "./helpers/remove-tree"; const originalFetch = globalThis.fetch; const originalOpenCodexHome = process.env.OPENCODEX_HOME; const BLOB = "provider-minted-opaque-state"; +const FUNCTION_OUTPUT_BLOB = `g${"A".repeat(127)}`; +const FUNCTION_OUTPUT_DECRYPT_MESSAGE = "Encrypted function output content could not be decrypted or decoded."; const OPENAI_BLOB_ERROR = JSON.stringify({ error: { message: "The encrypted content could not be verified.", @@ -34,6 +36,13 @@ const CHATGPT_UNVERIFIABLE_BLOB_ERROR = JSON.stringify({ code: null, }, }); +const CHATGPT_FUNCTION_OUTPUT_DECRYPT_ERROR = JSON.stringify({ + error: { + message: FUNCTION_OUTPUT_DECRYPT_MESSAGE, + type: "server_error", + code: null, + }, +}); const XAI_DECODE_ERROR = JSON.stringify({ code: "invalid-argument", error: "Could not decode the compaction blob: invalid payload", @@ -91,6 +100,58 @@ function serializedOutboundWithBlob(): string { return JSON.stringify({ model: "model-a", input: reasoningReplayInput() }); } +function functionOutputReplayInput(): Array> { + return [ + { + type: "function_call", + call_id: "call-encrypted-output", + name: "browser_capture", + arguments: "{}", + }, + { + type: "function_call_output", + call_id: "call-encrypted-output", + output: [ + { type: "encrypted_content", encrypted_content: FUNCTION_OUTPUT_BLOB }, + { type: "input_text", text: "visible tool output" }, + { type: "input_image", image_url: "data:image/png;base64,AAAA", detail: "high" }, + ], + }, + { + type: "message", + role: "user", + content: [{ type: "input_text", text: "continue" }], + }, + ]; +} + +function serializedOutboundWithEncryptedFunctionOutput(): string { + return JSON.stringify({ model: "model-a", input: functionOutputReplayInput() }); +} + +function agentMessageReplayInput(): Array> { + return [ + { + type: "agent_message", + author: "/root/child_task", + recipient: "/root", + content: [ + { type: "input_text", text: "Message Type: MESSAGE\nTask name: /root\nSender: /root/child_task\nPayload:" }, + { type: "encrypted_content", encrypted_content: FUNCTION_OUTPUT_BLOB }, + ], + }, + { + type: "message", + role: "user", + content: [{ type: "input_text", text: "continue" }], + }, + ]; +} + +function serializedOutboundWithEncryptedAgentMessage(): string { + return JSON.stringify({ model: "model-a", input: agentMessageReplayInput() }); +} + function config(): OcxConfig { return { defaultProvider: "first", @@ -137,6 +198,97 @@ function requestWithIdentityHeaders( }); } +function functionOutputRequest(stream = false): Request { + return new Request("http://localhost/v1/responses", { + method: "POST", + headers: { + "content-type": "application/json", + "x-codex-parent-thread-id": "thread-encrypted-function-output", + }, + body: JSON.stringify({ + model: "first/model-a", + stream, + store: false, + input: functionOutputReplayInput(), + }), + }); +} + +function agentMessageRequest(stream = false): Request { + return new Request("http://localhost/v1/responses", { + method: "POST", + headers: { + "content-type": "application/json", + "x-codex-parent-thread-id": "thread-encrypted-agent-message", + }, + body: JSON.stringify({ + model: "first/model-a", + stream, + store: false, + input: agentMessageReplayInput(), + }), + }); +} + +function streamedFunctionOutputDecryptFailure(): Response { + const failed = { + type: "response.failed", + response: { + id: "resp-function-output-failed", + status: "failed", + error: { + message: FUNCTION_OUTPUT_DECRYPT_MESSAGE, + type: "server_error", + code: "upstream_server_error", + }, + }, + }; + return new Response(`event: response.failed\ndata: ${JSON.stringify(failed)}\n\ndata: [DONE]\n\n`, { + status: 200, + headers: { "content-type": "text/event-stream" }, + }); +} + +// The observed ChatGPT production shape: response.created, then a bare error +// event carrying the decryption rejection, then EOF with no terminal event. +function streamedFunctionOutputDecryptErrorEvent(): Response { + const created = { + type: "response.created", + response: { id: "resp-function-output-error-event", status: "in_progress" }, + }; + const errorEvent = { + type: "error", + error: { + type: "server_error", + code: "upstream_server_error", + message: FUNCTION_OUTPUT_DECRYPT_MESSAGE, + }, + }; + return new Response( + `event: response.created\ndata: ${JSON.stringify(created)}\n\nevent: error\ndata: ${JSON.stringify(errorEvent)}\n\n`, + { + status: 200, + headers: { "content-type": "text/event-stream" }, + }, + ); +} + +function streamedSuccess(id: string): Response { + const completed = { + type: "response.completed", + response: { + id, + status: "completed", + model: "model-a", + output: [], + }, + }; + return new Response(`event: response.completed\ndata: ${JSON.stringify(completed)}\n\ndata: [DONE]\n\n`, { + status: 200, + headers: { "content-type": "text/event-stream" }, + }); +} + function rejection(body = OPENAI_BLOB_ERROR): Response { return new Response(body, { status: 400, @@ -202,9 +354,261 @@ describe("opaque blob recovery trigger", () => { }), })).toBe(false); }); + + test("accepts the exact ChatGPT 502 rejection only when function output carries encrypted content", () => { + const base = { + status: 502, + adapterName: "openai-responses", + outboundBody: serializedOutboundWithEncryptedFunctionOutput(), + errorBody: CHATGPT_FUNCTION_OUTPUT_DECRYPT_ERROR, + alreadyAttempted: false, + }; + + expect(shouldAttemptOpaqueBlobRecovery(base)).toBe(true); + expect(shouldAttemptOpaqueBlobRecovery({ ...base, status: 500 })).toBe(false); + expect(shouldAttemptOpaqueBlobRecovery({ + ...base, + errorBody: JSON.stringify({ error: { message: "Bad gateway" } }), + })).toBe(false); + expect(shouldAttemptOpaqueBlobRecovery({ + ...base, + outboundBody: JSON.stringify({ model: "model-a", input: [{ type: "message", role: "user" }] }), + })).toBe(false); + expect(shouldAttemptOpaqueBlobRecovery({ ...base, alreadyAttempted: true })).toBe(false); + }); + + test("accepts the exact ChatGPT 502 rejection when an agent_message content part carries encrypted content", () => { + const base = { + status: 502, + adapterName: "openai-responses", + outboundBody: serializedOutboundWithEncryptedAgentMessage(), + errorBody: CHATGPT_FUNCTION_OUTPUT_DECRYPT_ERROR, + alreadyAttempted: false, + }; + + expect(shouldAttemptOpaqueBlobRecovery(base)).toBe(true); + expect(shouldAttemptOpaqueBlobRecovery({ + ...base, + outboundBody: JSON.stringify({ + model: "model-a", + input: [{ type: "agent_message", content: [{ type: "input_text", text: "plain" }] }], + }), + })).toBe(false); + }); }); describe("opaque blob recovery through /v1/responses", () => { + test("recovers a zero-output streamed function-output decrypt failure before client relay", async () => { + const outbound: Array> = []; + globalThis.fetch = (async (_input: RequestInfo | URL, init?: RequestInit) => { + outbound.push(JSON.parse(String(init?.body)) as Record); + return outbound.length === 1 + ? streamedFunctionOutputDecryptFailure() + : streamedSuccess("resp-stream-function-output-recovered"); + }) as typeof fetch; + + const response = await handleResponses(functionOutputRequest(true), config(), { model: "", provider: "" }); + const body = await response.text(); + + expect(response.status).toBe(200); + expect(body).toContain("response.completed"); + expect(body).not.toContain(FUNCTION_OUTPUT_DECRYPT_MESSAGE); + expect(outbound).toHaveLength(2); + const retriedInput = outbound.at(1)?.input as Array> | undefined; + expect(retriedInput?.at(1)).toEqual({ + type: "function_call_output", + call_id: "call-encrypted-output", + output: [ + { type: "input_text", text: "[encrypted content omitted]" }, + { type: "input_text", text: "visible tool output" }, + { type: "input_image", image_url: "data:image/png;base64,AAAA", detail: "high" }, + ], + }); + }); + + test("retries a ChatGPT function-output decrypt failure once with an omission marker", async () => { + const outbound: Array> = []; + globalThis.fetch = (async (_input: RequestInfo | URL, init?: RequestInit) => { + outbound.push(JSON.parse(String(init?.body)) as Record); + return outbound.length <= 3 + ? new Response(CHATGPT_FUNCTION_OUTPUT_DECRYPT_ERROR, { + status: 502, + headers: { "content-type": "application/json" }, + }) + : success("resp-function-output-recovered"); + }) as typeof fetch; + const logCtx: RequestLogContext = { model: "", provider: "" }; + + const response = await handleResponses(functionOutputRequest(), config(), logCtx); + expect(response.status).toBe(200); + await response.text(); + + expect(outbound).toHaveLength(4); + const firstInput = outbound.at(0)?.input as Array> | undefined; + const retriedInput = outbound.at(3)?.input as Array> | undefined; + expect(firstInput?.at(1)).toEqual(functionOutputReplayInput().at(1)); + expect(retriedInput?.at(0)).toEqual(functionOutputReplayInput().at(0)); + expect(retriedInput?.at(1)).toEqual({ + type: "function_call_output", + call_id: "call-encrypted-output", + output: [ + { type: "input_text", text: "[encrypted content omitted]" }, + { type: "input_text", text: "visible tool output" }, + { type: "input_image", image_url: "data:image/png;base64,AAAA", detail: "high" }, + ], + }); + expect(retriedInput?.at(2)).toEqual(functionOutputReplayInput().at(2)); + expect(logCtx.activeAttempt?.sendCount).toBe(4); + expect(logCtx.activeAttempt?.recoveryKinds).toEqual(["transient-5xx", "opaque-blob-rejection"]); + }); + + test("surfaces a repeated function-output decrypt rejection after one sanitized rebuild", async () => { + const logCtx: RequestLogContext = { model: "", provider: "" }; + const outbound: Array> = []; + globalThis.fetch = (async (_input: RequestInfo | URL, init?: RequestInit) => { + outbound.push(JSON.parse(String(init?.body)) as Record); + return new Response(CHATGPT_FUNCTION_OUTPUT_DECRYPT_ERROR, { + status: 502, + headers: { "content-type": "application/json" }, + }); + }) as typeof fetch; + + const response = await handleResponses(functionOutputRequest(), config(), logCtx); + expect(response.status).toBe(502); + const body = await response.json() as { error?: { message?: string } }; + expect(body.error?.message).toBe(FUNCTION_OUTPUT_DECRYPT_MESSAGE); + + expect(outbound).toHaveLength(6); + const initialInput = outbound.at(0)?.input as Array> | undefined; + const finalInput = outbound.at(-1)?.input as Array> | undefined; + expect(initialInput?.at(1)).toEqual(functionOutputReplayInput().at(1)); + expect(finalInput?.at(1)).toEqual({ + type: "function_call_output", + call_id: "call-encrypted-output", + output: [ + { type: "input_text", text: "[encrypted content omitted]" }, + { type: "input_text", text: "visible tool output" }, + { type: "input_image", image_url: "data:image/png;base64,AAAA", detail: "high" }, + ], + }); + }); + + test("keeps a repeated streamed function-output rejection visible after one sanitized rebuild", async () => { + const outbound: Array> = []; + globalThis.fetch = (async (_input: RequestInfo | URL, init?: RequestInit) => { + outbound.push(JSON.parse(String(init?.body)) as Record); + return streamedFunctionOutputDecryptFailure(); + }) as typeof fetch; + const logCtx: RequestLogContext = { model: "", provider: "" }; + + const response = await handleResponses(functionOutputRequest(true), config(), logCtx); + const body = await response.text(); + + expect(response.status).toBe(200); + expect(body).toContain("response.failed"); + expect(body).toContain(FUNCTION_OUTPUT_DECRYPT_MESSAGE); + expect(logCtx.upstreamError).toBe(FUNCTION_OUTPUT_DECRYPT_MESSAGE); + expect(outbound).toHaveLength(2); + const finalInput = outbound.at(1)?.input as Array> | undefined; + expect(finalInput?.at(1)).toEqual({ + type: "function_call_output", + call_id: "call-encrypted-output", + output: [ + { type: "input_text", text: "[encrypted content omitted]" }, + { type: "input_text", text: "visible tool output" }, + { type: "input_image", image_url: "data:image/png;base64,AAAA", detail: "high" }, + ], + }); + }); + + test("retries a ChatGPT agent-message decrypt failure once with an omission marker", async () => { + const outbound: Array> = []; + globalThis.fetch = (async (_input: RequestInfo | URL, init?: RequestInit) => { + outbound.push(JSON.parse(String(init?.body)) as Record); + return outbound.length <= 3 + ? new Response(CHATGPT_FUNCTION_OUTPUT_DECRYPT_ERROR, { + status: 502, + headers: { "content-type": "application/json" }, + }) + : success("resp-agent-message-recovered"); + }) as typeof fetch; + const logCtx: RequestLogContext = { model: "", provider: "" }; + + const response = await handleResponses(agentMessageRequest(), config(), logCtx); + expect(response.status).toBe(200); + await response.text(); + + expect(outbound).toHaveLength(4); + const retriedInput = outbound.at(3)?.input as Array> | undefined; + expect(retriedInput?.at(0)).toEqual({ + type: "agent_message", + author: "/root/child_task", + recipient: "/root", + content: [ + { type: "input_text", text: "Message Type: MESSAGE\nTask name: /root\nSender: /root/child_task\nPayload:" }, + { type: "input_text", text: "[encrypted content omitted]" }, + ], + }); + expect(retriedInput?.at(1)).toEqual(agentMessageReplayInput().at(1)); + expect(logCtx.activeAttempt?.recoveryKinds).toEqual(["transient-5xx", "opaque-blob-rejection"]); + }); + + test("recovers a zero-output streamed agent-message decrypt failure before client relay", async () => { + const outbound: Array> = []; + globalThis.fetch = (async (_input: RequestInfo | URL, init?: RequestInit) => { + outbound.push(JSON.parse(String(init?.body)) as Record); + return outbound.length === 1 + ? streamedFunctionOutputDecryptFailure() + : streamedSuccess("resp-stream-agent-message-recovered"); + }) as typeof fetch; + + const response = await handleResponses(agentMessageRequest(true), config(), { model: "", provider: "" }); + const body = await response.text(); + + expect(response.status).toBe(200); + expect(body).toContain("response.completed"); + expect(body).not.toContain(FUNCTION_OUTPUT_DECRYPT_MESSAGE); + expect(outbound).toHaveLength(2); + const retriedInput = outbound.at(1)?.input as Array> | undefined; + expect(retriedInput?.at(0)).toEqual({ + type: "agent_message", + author: "/root/child_task", + recipient: "/root", + content: [ + { type: "input_text", text: "Message Type: MESSAGE\nTask name: /root\nSender: /root/child_task\nPayload:" }, + { type: "input_text", text: "[encrypted content omitted]" }, + ], + }); + }); + + test("recovers a zero-output error-event decrypt failure before client relay", async () => { + const outbound: Array> = []; + globalThis.fetch = (async (_input: RequestInfo | URL, init?: RequestInit) => { + outbound.push(JSON.parse(String(init?.body)) as Record); + return outbound.length === 1 + ? streamedFunctionOutputDecryptErrorEvent() + : streamedSuccess("resp-stream-error-event-recovered"); + }) as typeof fetch; + + const response = await handleResponses(agentMessageRequest(true), config(), { model: "", provider: "" }); + const body = await response.text(); + + expect(response.status).toBe(200); + expect(body).toContain("response.completed"); + expect(body).not.toContain(FUNCTION_OUTPUT_DECRYPT_MESSAGE); + expect(outbound).toHaveLength(2); + const retriedInput = outbound.at(1)?.input as Array> | undefined; + expect(retriedInput?.at(0)).toEqual({ + type: "agent_message", + author: "/root/child_task", + recipient: "/root", + content: [ + { type: "input_text", text: "Message Type: MESSAGE\nTask name: /root\nSender: /root/child_task\nPayload:" }, + { type: "input_text", text: "[encrypted content omitted]" }, + ], + }); + }); + test("#2247 strips reasoning and compaction ciphertext before a pooled thread moves accounts", async () => { const outbound: Array<{ accountId: string | null; body: Record }> = []; globalThis.fetch = (async (_input: RequestInfo | URL, init?: RequestInit) => { diff --git a/tests/sse-failed-tail.test.ts b/tests/sse-failed-tail.test.ts index 7be3e288ec..aed01fa542 100644 --- a/tests/sse-failed-tail.test.ts +++ b/tests/sse-failed-tail.test.ts @@ -153,6 +153,32 @@ describe("relaySseWithFailedTail", () => { expect(out.endsWith("data: [DONE]\n\n")).toBe(true); }); + test("clean EOF after a recorded upstream error reports that error instead of adapter_eof", async () => { + const upstream = new AbortController(); + const src = sourceStream(['data: {"type":"response.in_progress"}\n\n']); + const out = await drain(relaySseWithFailedTail(src, upstream, undefined, { + upstreamError: "The usage limit has been reached", + })); + + expect(out).toContain("event: response.failed"); + expect(out).toContain("The usage limit has been reached"); + expect(out).not.toContain('"reason":"adapter_eof"'); + expect(out.endsWith("data: [DONE]\n\n")).toBe(true); + }); + + test("clean EOF after an upstream error keeps the same failed payload through eager relay", async () => { + const upstream = new AbortController(); + const src = sourceStream(['data: {"type":"response.in_progress"}\n\n']); + const out = await drain(relaySseEagerBounded(src, upstream, parityHooks, { + upstreamError: "The usage limit has been reached", + })); + + expect(out).toContain("event: response.failed"); + expect(out).toContain("The usage limit has been reached"); + expect(out).not.toContain('"reason":"adapter_eof"'); + expect(out.endsWith("data: [DONE]\n\n")).toBe(true); + }); + test("translator overflow failed tail preserves translation_buffer_limit", async () => { const upstream = new AbortController(); const error = new TranslatorBudgetExceededError("live_transient", 32 * 1024 * 1024);