diff --git a/docs-site/src/content/docs/guides/sub-agent-surface.md b/docs-site/src/content/docs/guides/sub-agent-surface.md index 9e8ac4a894..b9cea5b0cb 100644 --- a/docs-site/src/content/docs/guides/sub-agent-surface.md +++ b/docs-site/src/content/docs/guides/sub-agent-surface.md @@ -158,6 +158,19 @@ for the full trust boundary and configuration. Combo routing remains unchanged and continues to consider only canonical native ChatGPT targets for encrypted tasks. +## Rejected encrypted history + +An upstream Responses server can reject encrypted parts in an earlier `agent_message` +with `Encrypted function output content could not be decrypted or decoded.`. Before +any output is committed, opencodex replaces those parts with `[encrypted content omitted]` +and rebuilds the request once. The surrounding readable content stays intact; the +omitted content is not decrypted or recovered by this retry. + +If the rebuilt request receives another bare SSE `error` followed by EOF, both relay +modes preserve the error message in a `response.failed` terminal instead of reporting +`adapter_eof`. Other upstream `response.failed` events remain SSE failures. This history +recovery does not change the encrypted v2 task-delivery restrictions described above. + ## Changing the mode ### GUI diff --git a/src/lib/errors.ts b/src/lib/errors.ts index 2ae0b9e6f0..8fb5126487 100644 --- a/src/lib/errors.ts +++ b/src/lib/errors.ts @@ -4,10 +4,15 @@ export interface OcxErrorPayload { code: string | null; } +export const ENCRYPTED_FUNCTION_OUTPUT_REJECTION = + "Encrypted function output content could not be decrypted or decoded."; + /** Canonical human-readable message paths used by Responses upstream failures. */ export function upstreamErrorMessageFromPayload(payload: unknown): string | undefined { if (!payload || typeof payload !== "object" || Array.isArray(payload)) return undefined; const json = payload as { + type?: unknown; + message?: unknown; error?: { message?: unknown }; last_error?: { message?: unknown }; response?: { @@ -18,7 +23,10 @@ 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.type === "error" ? json.message : undefined); return typeof message === "string" ? message : undefined; } diff --git a/src/server/relay-eager.ts b/src/server/relay-eager.ts index 655997b813..151a53ac82 100644 --- a/src/server/relay-eager.ts +++ b/src/server/relay-eager.ts @@ -29,6 +29,7 @@ import { createSseTerminalOutputBoundary, doneFrame, failedTailFrame, + upstreamErrorTailFrame, } from "./relay"; import { nextSseBlock, @@ -81,6 +82,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; }; @@ -303,12 +306,16 @@ export function relaySseEagerBounded( } else if (!hooks.sawTerminal() && canDeliver()) { // 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; + const upstreamError = terminalBoundary.upstreamError() ?? opts?.upstreamError; + const upstreamErrorFrame = upstreamError === undefined + ? adapterEofFrame + : upstreamErrorTailFrame(terminalEncoder, upstreamError); + queuedBytes += upstreamErrorFrame.byteLength + terminalSentinel.byteLength; try { - controllerRef?.enqueue(adapterEofFrame); + controllerRef?.enqueue(upstreamErrorFrame); controllerRef?.enqueue(terminalSentinel); } catch { /* client already gone */ } - syntheticKind = "incomplete"; + syntheticKind = upstreamError === undefined ? "incomplete" : "failed"; } break; } diff --git a/src/server/relay.ts b/src/server/relay.ts index 60b57ea025..37e4dbc7fe 100644 --- a/src/server/relay.ts +++ b/src/server/relay.ts @@ -5,6 +5,7 @@ import { CYBER_POLICY_FALLBACK_MESSAGE, isCyberPolicyCode, isCyberPolicyMessage, + upstreamErrorMessageFromPayload, } from "../lib/errors"; import { redactSecretString } from "../lib/redact"; import { isTranslatorBudgetExceededError } from "../lib/translator-budget"; @@ -147,11 +148,24 @@ export function failedTailFrame(encoder: TextEncoder, err: unknown): Uint8Array return encoder.encode(`\n\nevent: response.failed\ndata: ${payload}\n\n${DONE_SSE_FRAME_TEXT}`); } +export function upstreamErrorTailFrame(encoder: TextEncoder, message: string): Uint8Array { + const error = { + type: "upstream_error", + code: "upstream_server_error", + message: redactSecretString(message).slice(0, MAX_TAIL_ERROR_MESSAGE_CHARS), + }; + return encoder.encode(`event: response.failed\ndata: ${JSON.stringify({ + type: "response.failed", + response: { status: "failed", error, last_error: error }, + })}\n\n`); +} + export type SseTerminalOutputBoundary = { feed(chunk: Uint8Array): Uint8Array; finish(): Uint8Array; terminalSeen(): boolean; doneSeen(): boolean; + upstreamError(): string | undefined; dispose(): void; }; @@ -170,6 +184,7 @@ export function createSseTerminalOutputBoundary(): SseTerminalOutputBoundary { let done = false; let pendingDone: { block: Uint8Array; delimiter: Uint8Array } | null = null; let disposed = false; + let upstreamError: string | undefined; const processFrames = ( frames: ReturnType, @@ -181,6 +196,12 @@ export function createSseTerminalOutputBoundary(): SseTerminalOutputBoundary { const payload = sseDataPayload(decoder.decode(frame.block)); const isDone = payload === "[DONE]"; const parsed = payload === null ? undefined : parseSsePayload(payload); + // Observe on the client reader itself: a tee inspection branch may lag + // behind EOF, so its log context cannot determine the outgoing terminal. + if (parsed && typeof parsed === "object" && "type" in parsed && parsed.type === "error") { + const message = upstreamErrorMessageFromPayload(parsed); + if (message) upstreamError = redactSecretString(message).slice(0, MAX_TAIL_ERROR_MESSAGE_CHARS); + } const policyError = parsed !== undefined && isPolicyRewriteType(parsed) ? cyberPolicyTerminalError(parsed) : undefined; @@ -239,6 +260,7 @@ export function createSseTerminalOutputBoundary(): SseTerminalOutputBoundary { }, terminalSeen: () => terminal, doneSeen: () => done, + upstreamError: () => upstreamError, dispose() { if (disposed) return; disposed = true; @@ -260,6 +282,7 @@ export function relaySseWithFailedTail( body: ReadableStream, upstream: AbortController, onClientGone?: (reason?: unknown) => void, + opts?: { upstreamError?: string }, ): ReadableStream { const reader = body.getReader(); const encoder = new TextEncoder(); @@ -306,8 +329,10 @@ 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); + const upstreamError = terminalBoundary.upstreamError() ?? opts?.upstreamError; + controller.enqueue(upstreamError === undefined + ? adapterEofIncompleteFrame(encoder) + : upstreamErrorTailFrame(encoder, upstreamError)); 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..ad9596b741 100644 --- a/src/server/responses/combo-stream-preflight.ts +++ b/src/server/responses/combo-stream-preflight.ts @@ -1,4 +1,5 @@ import type { ResponsesTerminalStatus } from "../../bridge"; +import { ENCRYPTED_FUNCTION_OUTPUT_REJECTION, upstreamErrorMessageFromPayload } from "../../lib/errors"; import type { RequestLogContext } from "../request-log"; import { createSseInspector } from "../relay"; import { MAX_CLIENT_SSE_FRAME_BYTES } from "../sse-frame-buffer"; @@ -37,6 +38,9 @@ function retryableZeroOutputTerminal(payload: unknown): boolean { response?: { incomplete_details?: { reason?: unknown } }; }; if (event.type === "response.failed") return true; + if (event.type === "error") { + return upstreamErrorMessageFromPayload(event) === ENCRYPTED_FUNCTION_OUTPUT_REJECTION; + } 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 +55,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); } @@ -135,6 +142,7 @@ export type ComboStreamPreflightResult = export async function preflightComboStreamResponse( response: Response, logCtx: RequestLogContext, + retryableTerminal: (payload: unknown) => boolean = retryableZeroOutputTerminal, ): Promise { const contentType = response.headers.get("content-type")?.toLowerCase() ?? ""; if (!response.ok || !response.body || !contentType.includes("text/event-stream")) { @@ -152,7 +160,7 @@ export async function preflightComboStreamResponse( onParsedPayload: payload => { if (comboStreamPayloadCommitsOutput(payload)) outputCommitted = true; if (!payload || typeof payload !== "object" || Array.isArray(payload)) return; - if (retryableZeroOutputTerminal(payload)) { + if (retryableTerminal(payload)) { retryableTerminalPayload = payload as Record; } }, @@ -180,7 +188,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 0471617da0..9b12865128 100644 --- a/src/server/responses/core.ts +++ b/src/server/responses/core.ts @@ -263,6 +263,7 @@ import { import { redactSecretString, sanitizeLogMetadataString } from "../../lib/redact"; import { readBoundedResponseBody } from "../../lib/bounded-body"; import { + ENCRYPTED_FUNCTION_OUTPUT_REJECTION, isRateLimitOrQuotaFailureMessage, upstreamErrorMessageFromPayload, } from "../../lib/errors"; @@ -639,28 +640,96 @@ const OPAQUE_RESPONSES_INPUT_TYPES = new Set([ "compaction_summary", "context_compaction", ]); +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; @@ -705,8 +774,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) @@ -722,7 +796,7 @@ async function opaqueBlobRejectionBodyForRecovery( ): Promise { if ( response.status < 400 - || response.status >= 500 + || (response.status >= 500 && response.status !== 502) || adapterName !== "openai-responses" || alreadyAttempted || !outboundResponsesBodyCarriesOpaqueBlob(outboundBody) @@ -799,6 +873,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; + }); + Object.assign(rawBody, { input: strippedInput }); +} + +function resetStreamedOpaqueBlobLogContext(logCtx: RequestLogContext): void { + delete logCtx.upstreamError; + delete logCtx.terminalHttpStatus; + delete logCtx.terminalErrorCode; + delete logCtx.terminalIncompleteReason; } type OpaqueBlobRecoveryGuard = { attempted: boolean }; @@ -4870,6 +4988,39 @@ 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, + payload => upstreamErrorMessageFromPayload(payload) === ENCRYPTED_FUNCTION_OUTPUT_REJECTION); + 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); @@ -5166,6 +5317,7 @@ async function handleResponsesInner( }, { clientGoneSignal: options.abortSignal, ...(inlineEagerRewrite ? { rewriteBudget: translatorBudget } : {}), + ...(logCtx.upstreamError === undefined ? {} : { upstreamError: logCtx.upstreamError }), }); // When selected, this relay closes response.completed even if upstream // keeps the connection alive. Marked Codex WS traffic, Windows @@ -5248,7 +5400,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/responses/passthrough-abort.test.ts b/tests/responses/passthrough-abort.test.ts index ef41494213..6fdc468b1b 100644 --- a/tests/responses/passthrough-abort.test.ts +++ b/tests/responses/passthrough-abort.test.ts @@ -78,7 +78,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).toMatch( + /relaySseWithFailedTail\(\s*rewrittenBody,\s*upstream,\s*reason\s*=>\s*clientGone\.abort\(reason\),\s*\{\s*upstreamError:\s*logCtx\.upstreamError\s*\},\s*\)/, + ); 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/responses-opaque-blob-recovery.test.ts b/tests/responses/responses-opaque-blob-recovery.test.ts index 967909c154..2258ca2bc3 100644 --- a/tests/responses/responses-opaque-blob-recovery.test.ts +++ b/tests/responses/responses-opaque-blob-recovery.test.ts @@ -15,10 +15,14 @@ import { import type { RequestLogContext } from "../../src/server/request-log"; import type { OcxConfig } from "../../src/types"; import { removeTreeWithRetry } from "../helpers/remove-tree"; +import { markBodyNonPersistable, rememberResponseState, previousResponseProviderState } from "../../src/responses/state"; const originalFetch = globalThis.fetch; const originalOpenCodexHome = process.env.OPENCODEX_HOME; const BLOB = "provider-minted-opaque-state"; +// Synthetic Fernet-shaped data must survive the outbound ciphertext shape gate. +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 +38,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 +102,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 +200,98 @@ 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(flat = false): Response { + const created = { + type: "response.created", + response: { id: "resp-function-output-error-event", status: "in_progress" }, + }; + const error = { + type: "server_error", + code: "upstream_server_error", + message: FUNCTION_OUTPUT_DECRYPT_MESSAGE, + }; + const errorEvent = flat ? { ...error, type: "error" } : { + type: "error", + error, + }; + 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 +357,337 @@ 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]" }, + ], + }); + }); + + for (const streamMode of ["legacy-tee", "eager-relay"] as const) { + test(`preserves non-decrypt failed SSE with encrypted history (${streamMode})`, async () => { + const failed = { type: "response.failed", response: { + id: "resp-other-failure", status: "failed", output: [], + error: { type: "server_error", code: "unrelated_failure", message: "Other upstream failure" }, + } }; + const wire = `event: response.failed\ndata: ${JSON.stringify(failed)}\n\ndata: [DONE]\n\n`; + let sends = 0; + globalThis.fetch = Object.assign(async () => { + sends += 1; + return new Response(wire, { headers: { "content-type": "text/event-stream" } }); + }, { preconnect: originalFetch.preconnect }); + const response = await handleResponses(agentMessageRequest(true), { + ...config(), streamMode, + }, { model: "", provider: "" }); + expect(response.status).toBe(200); + expect(response.headers.get("content-type")).toContain("text/event-stream"); + expect(await response.text()).toBe(wire); + expect(sends).toBe(1); + }); + + for (const flat of [false, true]) { + test(`repeated bare decrypt errors terminate as failed (${streamMode}, flat=${flat})`, async () => { + let sends = 0; + globalThis.fetch = Object.assign(async () => { + sends += 1; + return streamedFunctionOutputDecryptErrorEvent(flat); + }, { preconnect: originalFetch.preconnect }); + const response = await handleResponses(agentMessageRequest(true), { + ...config(), streamMode, + }, { model: "", provider: "" }); + const body = await response.text(); + expect(sends).toBe(2); + expect(body).toContain(FUNCTION_OUTPUT_DECRYPT_MESSAGE); + expect(body).not.toContain("adapter_eof"); + expect(body.match(/^event: response.failed$/gm)).toHaveLength(1); + expect(body.match(/^data: \[DONE\]$/gm)).toHaveLength(1); + }); + } + } + + test("recovers a flat error event once and preserves the marked raw body identity", async () => { + const definition = ADAPTER_REGISTRY["openai-responses"]; + const originalCreate = definition.create; + const rawBodies: unknown[] = []; + const createSpy = spyOn(definition, "create").mockImplementation((provider, context) => { + const adapter = originalCreate(provider, context); + const buildRequest = adapter.buildRequest.bind(adapter); + adapter.buildRequest = (parsed, incoming) => { + rawBodies.push(parsed._rawBody); + if (rawBodies.length === 1) markBodyNonPersistable(parsed._rawBody); + return buildRequest(parsed, incoming); + }; + return adapter; + }); + let sends = 0; + globalThis.fetch = Object.assign(async () => { + sends += 1; + return sends === 1 ? streamedFunctionOutputDecryptErrorEvent(true) : streamedSuccess("resp-identity"); + }, { preconnect: originalFetch.preconnect }); + try { + const response = await handleResponses(agentMessageRequest(true), config(), { model: "", provider: "" }); + const body = await response.text(); + expect(body).toContain("response.completed"); + expect(body).not.toContain(FUNCTION_OUTPUT_DECRYPT_MESSAGE); + expect(sends).toBe(2); + expect(rawBodies).toHaveLength(2); + expect(rawBodies[1]).toBe(rawBodies[0]); + rememberResponseState(rawBodies[1], { id: "resp-marked-identity", status: "completed", output: [] }, + { cursor: { conversationId: "must-not-persist" } }, { force: true }); + expect(previousResponseProviderState("resp-marked-identity")).toBeUndefined(); + } finally { + createSpy.mockRestore(); + } + }); + 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/responses/sse-failed-tail.test.ts b/tests/responses/sse-failed-tail.test.ts index 914818e27b..a01ef55943 100644 --- a/tests/responses/sse-failed-tail.test.ts +++ b/tests/responses/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);