From ff5105c5e7e5b0f8f0d3938a5bc4533787558788 Mon Sep 17 00:00:00 2001 From: t Date: Sun, 6 Sep 2026 16:03:36 +0900 Subject: [PATCH 01/14] docs: lock current-dev opaque recovery carry boundaries --- .../040_opaque_recovery.md | 25 +++++++++++++++++++ 1 file changed, 25 insertions(+) diff --git a/devlog/_plan/260906_release_244_followups/040_opaque_recovery.md b/devlog/_plan/260906_release_244_followups/040_opaque_recovery.md index 9176f52cfc..b0a7cefde5 100644 --- a/devlog/_plan/260906_release_244_followups/040_opaque_recovery.md +++ b/devlog/_plan/260906_release_244_followups/040_opaque_recovery.md @@ -22,3 +22,28 @@ Existing maintainer CHANGES_REQUESTED targeted older 2d90f9684 reader race; inde Owner explicitly requests stacked PR workflow; use this relay foundation before combo-recovery and Grok terminal integration as an integration-validation stack, even though fixes are independently useful. Each layer remains independently tested via exact-head ci.yml runtime/gates. Security analysis stays scratch until public diff; no live Kiro. + +## Current-dev carry amendment (2026-09-06) + +The carry starts at adb696197 after the task-input, Kiro and fixture layers. +The source remains 2396829bd. The current core rewrite order also contains +tool-search restoration and function completion repair; preserve both and the +shared prompt-cache cohort field. Source review is not current-head approval. + +Default two-argument preflight callers keep their previous event classification. +Only the explicitly supplied exact decrypt predicate may make a matching bare +error replayable; unrelated errors still commit the stream, and an existing +unrelated response.failed stays an SSE terminal. The retry predicate accepts +only error/failed/incomplete envelopes, never output events carrying a message. +The new failed tail uses existing redactSecretString before the 512-character +limit. Test bounded synthesized messages in tee and eager paths with synthetic +credential canaries; retain original upstream frame passthrough semantics. + +This cohesive carry exceeds the default 500-line review size because the source +includes a large request-level regression matrix. Keep source and regression +commits distinct inside this one layer, with independent protocol/security review; +splitting the tests into a later PR would leave recovery unverified. Existing +large core/relay files retain their current ownership for this bounded carry: +no export moves or broad refactor amid replay/cancellation changes. A new generic +retry abstraction or core extraction would enlarge the behavior under review. +Remote CI verifies all source and tests together; no local suite/typecheck/build. From fd1737865bfcf3d2e554e31c3b05b8c169b0705b Mon Sep 17 00:00:00 2001 From: t Date: Sun, 6 Sep 2026 16:04:43 +0900 Subject: [PATCH 02/14] docs: preserve missing-content-type Responses preflight parity --- .../260906_release_244_followups/040_opaque_recovery.md | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/devlog/_plan/260906_release_244_followups/040_opaque_recovery.md b/devlog/_plan/260906_release_244_followups/040_opaque_recovery.md index b0a7cefde5..e3f994d7a9 100644 --- a/devlog/_plan/260906_release_244_followups/040_opaque_recovery.md +++ b/devlog/_plan/260906_release_244_followups/040_opaque_recovery.md @@ -47,3 +47,10 @@ large core/relay files retain their current ownership for this bounded carry: no export moves or broad refactor amid replay/cancellation changes. A new generic retry abstraction or core extraction would enlarge the behavior under review. Remote CI verifies all source and tests together; no local suite/typecheck/build. + +The existing core recognizes successful streaming Responses without Content-Type. +Keep that parity in the new preflight through an explicit fourth options argument +allowMissingContentType, enabled only by the same core streaming condition; default +combo callers still require text/event-stream. Add missing-header recovery and +non-SSE refusal controls. This avoids a source-PR gap where core selected recovery +but its preflight returned early solely because the header was absent. From 3b8cf8a8f147d45edaf3d359b3726da7d5a27f84 Mon Sep 17 00:00:00 2001 From: t Date: Sun, 6 Sep 2026 16:07:32 +0900 Subject: [PATCH 03/14] fix(responses): recover exact encrypted output rejection once Carry #3535 onto current dev, preserve default combo replay boundaries and handle missing Content-Type only on the existing native streaming path. Co-authored-by: yxr1995-maker <257504378+yxr1995-maker@users.noreply.github.com> --- .../content/docs/guides/sub-agent-surface.md | 13 ++ src/lib/errors.ts | 10 +- src/server/relay-eager.ts | 13 +- src/server/relay.ts | 29 ++- .../responses/combo-stream-preflight.ts | 22 +- src/server/responses/core.ts | 194 ++++++++++++++++-- structure/04_transports-and-sidecars.md | 14 ++ 7 files changed, 267 insertions(+), 28 deletions(-) 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 81905c4ac1..0ef2648b80 100644 --- a/docs-site/src/content/docs/guides/sub-agent-surface.md +++ b/docs-site/src/content/docs/guides/sub-agent-surface.md @@ -172,6 +172,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..e33486915a 100644 --- a/src/server/responses/combo-stream-preflight.ts +++ b/src/server/responses/combo-stream-preflight.ts @@ -135,9 +135,13 @@ export type ComboStreamPreflightResult = export async function preflightComboStreamResponse( response: Response, logCtx: RequestLogContext, + retryableTerminal: (payload: unknown) => boolean = retryableZeroOutputTerminal, + options?: { allowMissingContentType?: boolean }, ): Promise { const contentType = response.headers.get("content-type")?.toLowerCase() ?? ""; - if (!response.ok || !response.body || !contentType.includes("text/event-stream")) { + const isEventStream = contentType.includes("text/event-stream") + || (!contentType && options?.allowMissingContentType === true); + if (!response.ok || !response.body || !isEventStream) { return { kind: "accepted", response }; } @@ -150,11 +154,14 @@ export async function preflightComboStreamResponse( const inspector = createSseInspector({ logCtx, onParsedPayload: payload => { - if (comboStreamPayloadCommitsOutput(payload)) outputCommitted = true; + const retryable = retryableTerminal(payload); + const matchedBareError = retryable && payload !== null && typeof payload === "object" + && !Array.isArray(payload) && (payload as { type?: unknown }).type === "error"; + // Only an explicit caller predicate may opt a known bare error into replay. + // Default combo classification still commits unknown/error events. + if (comboStreamPayloadCommitsOutput(payload) && !matchedBareError) outputCommitted = true; if (!payload || typeof payload !== "object" || Array.isArray(payload)) return; - if (retryableZeroOutputTerminal(payload)) { - retryableTerminalPayload = payload as Record; - } + if (retryable) retryableTerminalPayload = payload as Record; }, onTerminal: status => { terminalStatus = status; }, }); @@ -180,7 +187,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 d1c3e8f1e0..25fcc4df0f 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"; @@ -641,28 +642,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; @@ -707,8 +776,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) @@ -724,7 +798,7 @@ async function opaqueBlobRejectionBodyForRecovery( ): Promise { if ( response.status < 400 - || response.status >= 500 + || (response.status >= 500 && response.status !== 502) || adapterName !== "openai-responses" || alreadyAttempted || !outboundResponsesBodyCarriesOpaqueBlob(outboundBody) @@ -801,6 +875,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 }; @@ -4888,6 +5006,44 @@ 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 => { + if (!payload || typeof payload !== "object" || Array.isArray(payload)) return false; + const type = (payload as { type?: unknown }).type; + return (type === "error" || type === "response.failed" || type === "response.incomplete") + && upstreamErrorMessageFromPayload(payload) === ENCRYPTED_FUNCTION_OUTPUT_REJECTION; + }, { allowMissingContentType: !recoveryContentType && parsed.stream }); + 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); @@ -5187,6 +5343,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 @@ -5269,7 +5426,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/structure/04_transports-and-sidecars.md b/structure/04_transports-and-sidecars.md index 4ea5e30404..1e8ffa1bde 100644 --- a/structure/04_transports-and-sidecars.md +++ b/structure/04_transports-and-sidecars.md @@ -393,6 +393,20 @@ Native passthrough SSE has TWO shapes, selected per request in inspection side-effect set (shared `createSseInspector` factory in `relay.ts`) including the #44 late-terminal semantics. +Both client readers also retain a bounded, redacted message from a bare upstream +`error` event. If EOF arrives without a real Responses terminal, they synthesize +one `response.failed` with that message instead of replacing it with `adapter_eof`. +The delivering reader owns this evidence; an asynchronous tee inspection branch +cannot reliably supply it before EOF. Existing real terminals remain authoritative. + +Native Responses may rebuild once when encrypted function/custom-tool output or +agent-message content receives the exact known decrypt rejection before output +commits. Recovery replaces only encrypted parts with an omission marker, preserves +the raw request object used by continuation persistence guards, and uses the same +adapter and cancellation path. A missing Content-Type is allowed only under the +existing successful streaming condition. Default combo preflight classification +is unchanged; only the native recovery caller supplies the exact error predicate. + Both shapes carry the inbound caller-abort signal separately from the turn/shutdown controller. A caller-driven read rejection is 499/client_cancel without pool penalty; a genuine upstream reset remains synthetic 502. An already received terminal, including From de0d22f3036be3e6db713646ac1304686593a12b Mon Sep 17 00:00:00 2001 From: t Date: Sun, 6 Sep 2026 16:14:57 +0900 Subject: [PATCH 04/14] test(responses): cover opaque recovery and headerless streaming --- tests/responses/passthrough-abort.test.ts | 4 +- .../responses-opaque-blob-recovery.test.ts | 578 ++++++++++++++++++ 2 files changed, 581 insertions(+), 1 deletion(-) 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..b9c99e1993 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,105 @@ 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 decryptStreamResponse(wire: string, contentType: string | null): Response { + // A string body would implicitly add text/plain even when headers are omitted. + const response = new Response(new TextEncoder().encode(wire), { + status: 200, + ...(contentType === null ? {} : { headers: { "content-type": contentType } }), + }); + expect(response.headers.get("content-type")).toBe(contentType); + return response; +} + +function streamedFunctionOutputDecryptFailure(contentType: string | null = "text/event-stream"): 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 decryptStreamResponse(`event: response.failed\ndata: ${JSON.stringify(failed)}\n\ndata: [DONE]\n\n`, contentType); +} + +// 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, + contentType: string | null = "text/event-stream", +): 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 decryptStreamResponse( + `event: response.created\ndata: ${JSON.stringify(created)}\n\nevent: error\ndata: ${JSON.stringify(errorEvent)}\n\n`, + contentType, + ); +} + +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 +364,425 @@ 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("recovers a missing-Content-Type 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(null) + : streamedSuccess("resp-missing-ct-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("recovers a missing-Content-Type 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(false, null) + : streamedSuccess("resp-missing-ct-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("absent Content-Type decrypt stream does not recover a non-stream request", async () => { + let sends = 0; + globalThis.fetch = Object.assign(async () => { + sends += 1; + return streamedFunctionOutputDecryptFailure(null); + }, { preconnect: originalFetch.preconnect }); + + const response = await handleResponses(functionOutputRequest(false), config(), { model: "", provider: "" }); + const body = await response.text(); + + expect(sends).toBe(1); + expect(body).toContain(FUNCTION_OUTPUT_DECRYPT_MESSAGE); + expect(body).not.toContain("response.completed"); + }); + + for (const contentType of ["application/json", "text/plain"] as const) { + test(`refuses non-SSE ${contentType} streamed decrypt recovery`, async () => { + let sends = 0; + globalThis.fetch = Object.assign(async () => { + sends += 1; + return streamedFunctionOutputDecryptFailure(contentType); + }, { preconnect: originalFetch.preconnect }); + + const response = await handleResponses(functionOutputRequest(true), config(), { model: "", provider: "" }); + const body = await response.text(); + + expect(sends).toBe(1); + expect(body).toContain(FUNCTION_OUTPUT_DECRYPT_MESSAGE); + expect(body).not.toContain("response.completed"); + }); + } + 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) => { From 840e4c0d6625ee26595d991ede8bd362fd2412a4 Mon Sep 17 00:00:00 2001 From: t Date: Sun, 6 Sep 2026 16:17:01 +0900 Subject: [PATCH 05/14] test(responses): guard preflight opt-in and safe failed tails --- tests/responses/sse-failed-tail.test.ts | 114 +++++++++++++++ tests/routing/combo-stream-preflight.test.ts | 141 +++++++++++++++++++ 2 files changed, 255 insertions(+) diff --git a/tests/responses/sse-failed-tail.test.ts b/tests/responses/sse-failed-tail.test.ts index 914818e27b..f21301ffb7 100644 --- a/tests/responses/sse-failed-tail.test.ts +++ b/tests/responses/sse-failed-tail.test.ts @@ -238,6 +238,32 @@ describe("relaySseWithFailedTail", () => { expect(eager.split("data: [DONE]").length - 1).toBe(1); }); + 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.each([ [ "clean EOF without a terminal", @@ -307,4 +333,92 @@ describe("relaySseWithFailedTail", () => { } } }); + + const CREDENTIAL_CANARY = "sk-testCANARY9live"; + const OVERLONG_SUFFIX = "x".repeat(600); + const BARE_ERROR_MESSAGE = "upstream failed " + CREDENTIAL_CANARY + " " + OVERLONG_SUFFIX; + const EXPECTED_SYNTHETIC_MESSAGE = ("upstream failed [REDACTED] " + OVERLONG_SUFFIX) + .slice(0, MAX_TAIL_ERROR_MESSAGE_CHARS); + + const sseDataFrame = (payload: unknown): string => "data: " + JSON.stringify(payload) + "\n\n"; + + const synthesizedTail = (out: string, original: string): string => { + expect(out.startsWith(original)).toBe(true); + return out.slice(original.length); + }; + + const synthesizedFailedPayload = (tail: string): { + type: string; + response: { status: string; error: { type: string; code: string; message: string } }; + } => { + const dataLine = tail.split("event: response.failed\ndata: ")[1]?.split("\n")[0]; + if (!dataLine) throw new Error("missing synthesized response.failed payload"); + return JSON.parse(dataLine) as { + type: string; + response: { status: string; error: { type: string; code: string; message: string } }; + }; + }; + + test.each([ + [ + "tee", + "flat", + (src: ReadableStream) => relaySseWithFailedTail(src, new AbortController()), + { type: "error", message: BARE_ERROR_MESSAGE }, + ], + [ + "tee", + "nested", + (src: ReadableStream) => relaySseWithFailedTail(src, new AbortController()), + { type: "error", error: { message: BARE_ERROR_MESSAGE } }, + ], + [ + "eager", + "flat", + (src: ReadableStream) => relaySseEagerBounded(src, new AbortController(), parityHooks), + { type: "error", message: BARE_ERROR_MESSAGE }, + ], + [ + "eager", + "nested", + (src: ReadableStream) => relaySseEagerBounded(src, new AbortController(), parityHooks), + { type: "error", error: { message: BARE_ERROR_MESSAGE } }, + ], + ] as const)("%s %s bare error synthesizes a redacted capped failed tail", async (_mode, _shape, relay, payload) => { + const original = sseDataFrame({ type: "response.in_progress" }) + sseDataFrame(payload); + const out = await drain(relay(sourceStream([original]))); + const tail = synthesizedTail(out, original); + const parsed = synthesizedFailedPayload(tail); + + expect(original).toContain(CREDENTIAL_CANARY); + expect(out.slice(0, original.length)).toBe(original); + expect(tail).toContain("event: response.failed"); + expect(tail).not.toContain(CREDENTIAL_CANARY); + expect(parsed.type).toBe("response.failed"); + expect(parsed.response.status).toBe("failed"); + expect(parsed.response.error.code).toBe("upstream_server_error"); + expect(parsed.response.error.message).toBe(EXPECTED_SYNTHETIC_MESSAGE); + expect(parsed.response.error.message).toHaveLength(MAX_TAIL_ERROR_MESSAGE_CHARS); + expect(parsed.response.error.message).not.toContain(CREDENTIAL_CANARY); + expect(terminalEvents(tail)).toEqual(["response.failed"]); + expect(doneEvents(out)).toHaveLength(1); + expect(doneEvents(tail)).toHaveLength(1); + expect(tail.endsWith("data: [DONE]\n\n")).toBe(true); + expect(out).not.toContain('"reason":"adapter_eof"'); + }); + + test.each(["tee", "eager"] as const)("%s existing terminal wins over a preceding bare error and does not duplicate DONE", async (mode) => { + const original = sseDataFrame({ type: "error", message: BARE_ERROR_MESSAGE }) + + 'event: response.completed\ndata: {"type":"response.completed","response":{"status":"completed"}}\n\n' + + "data: [DONE]\n\n"; + const relay = mode === "tee" + ? (src: ReadableStream) => relaySseWithFailedTail(src, new AbortController()) + : (src: ReadableStream) => relaySseEagerBounded(src, new AbortController(), parityHooks); + const out = await drain(relay(sourceStream([original]))); + + expect(out).toBe(original); + expect(terminalEvents(out)).toEqual(["response.completed"]); + expect(doneEvents(out)).toHaveLength(1); + expect(out).not.toContain("event: response.failed"); + }); }); diff --git a/tests/routing/combo-stream-preflight.test.ts b/tests/routing/combo-stream-preflight.test.ts index 97c927e391..4c7b20c35e 100644 --- a/tests/routing/combo-stream-preflight.test.ts +++ b/tests/routing/combo-stream-preflight.test.ts @@ -256,4 +256,145 @@ describe("combo stream preflight", () => { }); expect(JSON.stringify(body)).not.toContain("provider_trace_id"); }); + + const DECRYPT_REJECTION = + "Encrypted function output content could not be decrypted or decoded."; + + const exactDecryptRetryable = (payload: unknown): boolean => { + if (!payload || typeof payload !== "object" || Array.isArray(payload)) return false; + const event = payload as { + type?: unknown; + message?: unknown; + error?: { message?: unknown }; + response?: { error?: { message?: unknown } }; + }; + if (event.type !== "error" && event.type !== "response.failed" && event.type !== "response.incomplete") { + return false; + } + const message = event.error?.message + ?? event.response?.error?.message + ?? (event.type === "error" ? event.message : undefined); + return message === DECRYPT_REJECTION; + }; + + test("default 2-arg preflight commits a bare error, including exact decrypt, and preserves bytes", async () => { + expect(comboStreamPayloadCommitsOutput({ type: "error" })).toBe(true); + for (const payload of [ + { type: "error", message: "unrelated upstream busy" }, + { type: "error", message: DECRYPT_REJECTION }, + { type: "error", error: { message: DECRYPT_REJECTION } }, + ]) { + const source = sse( + { type: "response.created", response: { id: "r1", status: "in_progress" } }, + payload, + ); + const expected = await source.clone().text(); + const result = await preflightComboStreamResponse(source, { model: "m1", provider: "a" }); + expect(result.kind).toBe("accepted"); + expect(await result.response.text()).toBe(expected); + } + }); + + test("explicit 3-arg decrypt predicate converts a pre-output bare error into a failed terminal", async () => { + const source = sse( + { type: "response.created", response: { id: "r1", status: "in_progress" } }, + { type: "error", message: DECRYPT_REJECTION }, + ); + const original = await source.clone().text(); + const result = await preflightComboStreamResponse( + source, + { model: "m1", provider: "a" }, + exactDecryptRetryable, + ); + + expect(result.kind).toBe("failed"); + expect(result.response.status).toBe(502); + expect(result.response.headers.get("content-type")).toContain("application/json"); + expect(await result.response.text()).not.toBe(original); + }); + + test("an unrelated error followed by a matching failed terminal does not retry", async () => { + const source = sse( + { type: "response.created", response: { id: "r1", status: "in_progress" } }, + { type: "error", message: "unrelated upstream busy" }, + { + type: "response.failed", + response: { + status: "failed", + error: { type: "server_error", message: DECRYPT_REJECTION }, + }, + }, + ); + const expected = await source.clone().text(); + const result = await preflightComboStreamResponse( + source, + { model: "m1", provider: "a" }, + exactDecryptRetryable, + ); + + expect(result.kind).toBe("accepted"); + expect(await result.response.text()).toBe(expected); + }); + + test("output before a decrypt bare error does not retry", async () => { + const source = sse( + { type: "response.created", response: { id: "r1", status: "in_progress" } }, + { type: "response.output_text.delta", delta: "visible" }, + { type: "error", message: DECRYPT_REJECTION }, + ); + const expected = await source.clone().text(); + const result = await preflightComboStreamResponse( + source, + { model: "m1", provider: "a" }, + exactDecryptRetryable, + ); + + expect(result.kind).toBe("accepted"); + expect(await result.response.text()).toBe(expected); + }); + + test("default missing content-type is refused, and allowMissingContentType accepts only an absent type", async () => { + const payloads = [ + { type: "response.created", response: { id: "r1", status: "in_progress" } }, + { type: "error", message: DECRYPT_REJECTION }, + ]; + const body = payloads.map(payload => "data: " + JSON.stringify(payload) + "\n\n").join(""); + const encoded = () => new TextEncoder().encode(body); + const missingTypeResponse = () => { + const headers = new Headers(); + headers.delete("content-type"); + const response = new Response(encoded(), { headers }); + response.headers.delete("content-type"); + return response; + }; + + const missing = missingTypeResponse(); + expect(missing.headers.get("content-type")).toBeNull(); + const missingDefault = await preflightComboStreamResponse(missing, { model: "m1", provider: "a" }); + expect(missingDefault.kind).toBe("accepted"); + expect(await missingDefault.response.text()).toBe(body); + + const allowedMissingSource = missingTypeResponse(); + expect(allowedMissingSource.headers.get("content-type")).toBeNull(); + const allowedMissing = await preflightComboStreamResponse( + allowedMissingSource, + { model: "m1", provider: "a" }, + exactDecryptRetryable, + { allowMissingContentType: true }, + ); + expect(allowedMissing.kind).toBe("failed"); + expect(allowedMissing.response.status).toBe(502); + + for (const contentType of ["application/json", "text/plain"]) { + const source = new Response(encoded(), { headers: { "content-type": contentType } }); + const result = await preflightComboStreamResponse( + source, + { model: "m1", provider: "a" }, + exactDecryptRetryable, + { allowMissingContentType: true }, + ); + expect(result.kind).toBe("accepted"); + expect(await result.response.text()).toBe(body); + } + }); }); From b73809f7e9c57438ad9858a2599d02e355cda4a5 Mon Sep 17 00:00:00 2001 From: t Date: Sun, 6 Sep 2026 16:17:32 +0900 Subject: [PATCH 06/14] docs: record bounded opaque recovery integration evidence --- .../041_opaque_recovery_implementation.md | 20 +++++++++++++++++++ .../content/docs/guides/sub-agent-surface.md | 4 ++-- 2 files changed, 22 insertions(+), 2 deletions(-) create mode 100644 devlog/_plan/260906_release_244_followups/041_opaque_recovery_implementation.md diff --git a/devlog/_plan/260906_release_244_followups/041_opaque_recovery_implementation.md b/devlog/_plan/260906_release_244_followups/041_opaque_recovery_implementation.md new file mode 100644 index 0000000000..a2dbe6f54c --- /dev/null +++ b/devlog/_plan/260906_release_244_followups/041_opaque_recovery_implementation.md @@ -0,0 +1,20 @@ +# Opaque recovery implementation evidence + +Source 3b8cf8a8f carries PR3535 with a narrowly scoped preflight opt-in. The default +combo event classifier is unchanged; only a matched bare error supplied by the +native decrypt caller is replayable. Headerless streaming is an explicit option +under the existing core condition. Client-reader error evidence is redacted and +bounded before a failed tail is synthesized; real terminals remain authoritative. + +Independent plan audit accepted the scoped seam. Independent source/security +review passed: exact 502 gate, one sanitized rebuild, raw-body object identity, +no replay after visible output, cancellation and current rewrite ordering remain. +The source contributor is credited in the carry commit and PR. + +Regression commits cover native function and agent-message history, repeated +flat/nested errors, both relay shapes, unrelated errors and default combo byte +preservation, output commitment, missing-header and wrong-media-type controls, +and bounded synthesized-message redaction. The headerless fixture uses bytes +and asserts the absence of Content-Type because a string body supplies text/plain. +No local test suite, typecheck, build or live Kiro request was run. Final evidence +comes from hosted CI on the complete PR head and a fresh independent review. 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 0ef2648b80..da33f30d4f 100644 --- a/docs-site/src/content/docs/guides/sub-agent-surface.md +++ b/docs-site/src/content/docs/guides/sub-agent-surface.md @@ -174,8 +174,8 @@ 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 +An upstream Responses server can reject encrypted parts in earlier function/custom-tool +output or `agent_message` content 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. From 73a69e6619a24bbcec655dedb482434f0c57bb5d Mon Sep 17 00:00:00 2001 From: t Date: Sun, 6 Sep 2026 16:42:29 +0900 Subject: [PATCH 07/14] docs: plan opaque preflight transport and inspection finality repair --- .../000_plan.md | 21 ++++++++ .../010_failure_boundaries.md | 53 +++++++++++++++++++ 2 files changed, 74 insertions(+) create mode 100644 devlog/_plan/260906_opaque_transport_finality/000_plan.md create mode 100644 devlog/_plan/260906_opaque_transport_finality/010_failure_boundaries.md diff --git a/devlog/_plan/260906_opaque_transport_finality/000_plan.md b/devlog/_plan/260906_opaque_transport_finality/000_plan.md new file mode 100644 index 0000000000..3ff99fa965 --- /dev/null +++ b/devlog/_plan/260906_opaque_transport_finality/000_plan.md @@ -0,0 +1,21 @@ +# Opaque preflight transport and terminal outcomes + +Class C4. Mandatory parent-PR review repair under the existing authorized release +chain; work phase opaque-transport-finality, criterion c-2. Parent #3753 remains +open/draft at b73809f7e, child #3754 remains open/draft at f5c88beb9 with its parent +base restored. No parent merge occurred. The original #3535 was briefly closed +by an out-of-order follow-up, immediately reopened, and its comment corrected. +No completion, approval or release gate is waived. + +Public review references: PRRT_kwDOS-0Gi86fqEUo (preflight read failure escapes) +and PRRT_kwDOS-0Gi86fqEUq (tee EOF reports incomplete despite failed client tail). +The earlier full CI and independent reviews did not cover these paths. The +unfinished combo cycle is preserved and must consume the repaired parent before +its final verification. All execution remains hosted; no local suite/typecheck/ +build or live Kiro request. + +Implementation is one bounded failure-contract unit in 010_failure_boundaries.md. +Update the existing parent PR, run exact-head CI, cascade its commit into #3754, +and require fresh composed CI and review before bottom-up integration. Do not +close an original or retarget a child as a side effect of an unverified merge: +verify each preceding command and actual merged state before dependent actions. diff --git a/devlog/_plan/260906_opaque_transport_finality/010_failure_boundaries.md b/devlog/_plan/260906_opaque_transport_finality/010_failure_boundaries.md new file mode 100644 index 0000000000..23dbc3a59f --- /dev/null +++ b/devlog/_plan/260906_opaque_transport_finality/010_failure_boundaries.md @@ -0,0 +1,53 @@ +# Preserve preflight read failures and inspection finality + +## Current ownership + +Core selects native encrypted-output candidates and awaits combo-stream-preflight +before exposing headers. The preflight owns a bounded retained prefix and one +reader; replayBufferedResponse already emits that prefix and forwards later read +errors. Client relays own synthetic failed tails. consumeForInspection owns the +independent tee terminal callback used by native account health. The shared SSE +inspector reports real terminals and exposes parsed payload callbacks. + +## Planned change + +- src/server/responses/combo-stream-preflight.ts: native-only replayReadErrors + option, default false. Catch only reader.read rejection; opted-in callers get + an accepted reconstructed stream retaining the bounded prefix and the errored + reader. Default combo callers preserve their prior throw behavior. Do not retry + or classify a read reset as a decrypt rejection, swallow it, or grow buffers. +- src/server/responses/core.ts: enable that option only on the native opaque + preflight. After its await, caller abort takes the existing cancellation cleanup + path before any replay/rebuild. Other read failures reach the normal mid-stream + relay and inspection path, not a connect-phase error classifier. +- src/server/relay.ts: reuse a bounded/redacted bare-error message helper at the + client boundary and within consumeForInspection's parsed-payload callback. + Keep that evidence local to this reader rather than borrowing stale log state. + At clean EOF without a real terminal, a witnessed bare error reports failed + using the shared terminal HTTP mapper; an error-free EOF remains incomplete. + Preserve the caller's parsed-payload callback. Real terminals and cancellation + retain precedence; no extra terminal callback or healthy-account reset. + +## Rejected alternatives and scope + +A blanket core catch mapped as a connect error can misclassify an already-started +response's account outcome. Globally replaying all preflight errors changes combo +behavior. Reporting failure at the first bare error would override a later real +terminal. Borrowing the client relay's mutable state revives tee scheduling races. +Use the existing preflight/relay ownership and callback seams instead; no new +public inspector method, provider policy or retry budget. + +## Verification + +Existing native request fixtures add created-then-reset and created-then-caller- +abort cases: no uncaught handleResponses rejection, no sanitize resend, normal +failed stream or 499 cancellation and appropriate attempt/terminal metadata. +Run tee/eager variants where selected by the existing harness. Preflight tests +prove default read-error behavior is unchanged and native opt-in preserves prefix +and exact failure. Inspection/account-health fixtures cover flat/nested bare +errors at EOF, prior failure/avoidance not cleared, real-terminal precedence, +error-free EOF compatibility and cancellation neutrality. Existing redaction, +byte bounds, no-persistence and one-shot recovery tests remain. + +Independent plan/source/final review; exact parent and cascaded child hosted +Linux/macOS/gates CI. Final Windows six-shard and release gates remain mandatory. From 15b6d14747c61a372cfbdd422bc45289ed121839 Mon Sep 17 00:00:00 2001 From: t Date: Sun, 6 Sep 2026 16:43:56 +0900 Subject: [PATCH 08/14] docs: preserve original preflight read rejection without cancellation --- .../260906_opaque_transport_finality/010_failure_boundaries.md | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/devlog/_plan/260906_opaque_transport_finality/010_failure_boundaries.md b/devlog/_plan/260906_opaque_transport_finality/010_failure_boundaries.md index 23dbc3a59f..a4f93b4f81 100644 --- a/devlog/_plan/260906_opaque_transport_finality/010_failure_boundaries.md +++ b/devlog/_plan/260906_opaque_transport_finality/010_failure_boundaries.md @@ -14,7 +14,8 @@ inspector reports real terminals and exposes parsed payload callbacks. - src/server/responses/combo-stream-preflight.ts: native-only replayReadErrors option, default false. Catch only reader.read rejection; opted-in callers get an accepted reconstructed stream retaining the bounded prefix and the errored - reader. Default combo callers preserve their prior throw behavior. Do not retry + reader. Never cancel that errored reader: its original rejection must survive + the replay into relay/inspection. Default combo callers preserve their prior throw behavior. Do not retry or classify a read reset as a decrypt rejection, swallow it, or grow buffers. - src/server/responses/core.ts: enable that option only on the native opaque preflight. After its await, caller abort takes the existing cancellation cleanup From 3e1e6114db3dacad451eca540c242f5ca2281e6e Mon Sep 17 00:00:00 2001 From: t Date: Sun, 6 Sep 2026 16:45:03 +0900 Subject: [PATCH 09/14] fix(responses): preserve preflight resets and tee failure outcomes --- src/server/relay.ts | 26 ++++++++++++++----- .../responses/combo-stream-preflight.ts | 13 ++++++++-- src/server/responses/core.ts | 6 ++++- 3 files changed, 36 insertions(+), 9 deletions(-) diff --git a/src/server/relay.ts b/src/server/relay.ts index 37e4dbc7fe..a483d88f20 100644 --- a/src/server/relay.ts +++ b/src/server/relay.ts @@ -160,6 +160,13 @@ export function upstreamErrorTailFrame(encoder: TextEncoder, message: string): U })}\n\n`); } +function boundedBareUpstreamErrorMessage(payload: unknown): string | undefined { + if (!payload || typeof payload !== "object" || Array.isArray(payload) + || (payload as { type?: unknown }).type !== "error") return undefined; + const message = upstreamErrorMessageFromPayload(payload); + return message ? redactSecretString(message).slice(0, MAX_TAIL_ERROR_MESSAGE_CHARS) : undefined; +} + export type SseTerminalOutputBoundary = { feed(chunk: Uint8Array): Uint8Array; finish(): Uint8Array; @@ -198,10 +205,8 @@ export function createSseTerminalOutputBoundary(): SseTerminalOutputBoundary { 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 message = boundedBareUpstreamErrorMessage(parsed); + if (message !== undefined) upstreamError = message; const policyError = parsed !== undefined && isPolicyRewriteType(parsed) ? cyberPolicyTerminalError(parsed) : undefined; @@ -1379,11 +1384,16 @@ export function consumeForInspection( options?: InspectionConsumerOptions, ): void { const reader = body.getReader(); + let bareUpstreamError: string | undefined; const inspector = (options?.inspectorFactory ?? createSseInspector)({ onTerminal, logCtx, onCompletedResponse, - onParsedPayload: options?.onParsedPayload, + onParsedPayload: payload => { + const message = boundedBareUpstreamErrorMessage(payload); + if (message !== undefined) bareUpstreamError = message; + options?.onParsedPayload?.(payload); + }, onFirstOutput, pinCompletedResponseIdToFirstSeen: options?.pinCompletedResponseIdToFirstSeen, }); @@ -1397,7 +1407,11 @@ export function consumeForInspection( onCleanEof: () => { if (!inspector.reported()) { if (logCtx) logCtx.terminalSource = "synthetic"; - onTerminal("incomplete"); + if (bareUpstreamError !== undefined) { + onTerminal("failed", httpStatusForRequestLogTerminal("failed", logCtx)); + } else { + onTerminal("incomplete"); + } } }, onReadError: () => { diff --git a/src/server/responses/combo-stream-preflight.ts b/src/server/responses/combo-stream-preflight.ts index e33486915a..8c9ed6ebe5 100644 --- a/src/server/responses/combo-stream-preflight.ts +++ b/src/server/responses/combo-stream-preflight.ts @@ -136,7 +136,7 @@ export async function preflightComboStreamResponse( response: Response, logCtx: RequestLogContext, retryableTerminal: (payload: unknown) => boolean = retryableZeroOutputTerminal, - options?: { allowMissingContentType?: boolean }, + options?: { allowMissingContentType?: boolean; replayReadErrors?: boolean }, ): Promise { const contentType = response.headers.get("content-type")?.toLowerCase() ?? ""; const isEventStream = contentType.includes("text/event-stream") @@ -168,7 +168,16 @@ export async function preflightComboStreamResponse( try { for (;;) { - const next = await reader.read(); + let next: ReadableStreamReadResult; + try { + next = await reader.read(); + } catch (error) { + if (!options?.replayReadErrors) throw error; + // The native relay still owns post-header transport failures. Preserve + // the bounded prefix and the errored reader; cancelling it here would + // erase the failure before either client relay or inspection sees it. + return { kind: "accepted", response: replayBufferedResponse(response, reader, buffered) }; + } if (next.done) { inspector.finish(); } else { diff --git a/src/server/responses/core.ts b/src/server/responses/core.ts index 25fcc4df0f..583f7f8184 100644 --- a/src/server/responses/core.ts +++ b/src/server/responses/core.ts @@ -5021,7 +5021,11 @@ async function handleResponsesInner( const type = (payload as { type?: unknown }).type; return (type === "error" || type === "response.failed" || type === "response.incomplete") && upstreamErrorMessageFromPayload(payload) === ENCRYPTED_FUNCTION_OUTPUT_REJECTION; - }, { allowMissingContentType: !recoveryContentType && parsed.stream }); + }, { + allowMissingContentType: !recoveryContentType && parsed.stream, + replayReadErrors: true, + }); + if (options.abortSignal?.aborted) return transportFailureResponse(options.abortSignal.reason); upstreamResponse = preflight.response; if (preflight.kind === "failed") { const streamedOpaqueRecovery = await attemptOpaqueBlobRecovery({ From 812f7af1a1d6ba7a21f14b793ffc33c30fbbfdcc Mon Sep 17 00:00:00 2001 From: t Date: Sun, 6 Sep 2026 16:46:02 +0900 Subject: [PATCH 10/14] docs: align native error delivery and account outcome semantics --- structure/04_transports-and-sidecars.md | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/structure/04_transports-and-sidecars.md b/structure/04_transports-and-sidecars.md index 1e8ffa1bde..f97664f317 100644 --- a/structure/04_transports-and-sidecars.md +++ b/structure/04_transports-and-sidecars.md @@ -397,7 +397,12 @@ Both client readers also retain a bounded, redacted message from a bare upstream `error` event. If EOF arrives without a real Responses terminal, they synthesize one `response.failed` with that message instead of replacing it with `adapter_eof`. The delivering reader owns this evidence; an asynchronous tee inspection branch -cannot reliably supply it before EOF. Existing real terminals remain authoritative. +cannot reliably supply it before EOF. Inspection independently applies the same +bare-error rule when EOF arrives, so account health records failure instead of +clearing avoidance as if the turn had succeeded. Existing real terminals and +caller cancellation retain precedence on both branches. Native recovery preflight +also preserves a rejected body reader and its bounded prefix for the normal +mid-stream failure path; it does not turn that rejection into a decrypt retry. Native Responses may rebuild once when encrypted function/custom-tool output or agent-message content receives the exact known decrypt rejection before output From f0cdcb2866d905d0b05f292de625045c9f7414d3 Mon Sep 17 00:00:00 2001 From: t Date: Sun, 6 Sep 2026 16:49:36 +0900 Subject: [PATCH 11/14] test(responses): preserve preflight read resets and client aborts --- .../responses-opaque-blob-recovery.test.ts | 101 ++++++++++++++++++ tests/routing/combo-stream-preflight.test.ts | 68 +++++++++++- 2 files changed, 168 insertions(+), 1 deletion(-) diff --git a/tests/responses/responses-opaque-blob-recovery.test.ts b/tests/responses/responses-opaque-blob-recovery.test.ts index b9c99e1993..da3b62598b 100644 --- a/tests/responses/responses-opaque-blob-recovery.test.ts +++ b/tests/responses/responses-opaque-blob-recovery.test.ts @@ -783,6 +783,107 @@ describe("opaque blob recovery through /v1/responses", () => { }); } + for (const streamMode of ["legacy-tee", "eager-relay"] as const) { + test(`created-then-reset streamed function-output does not sanitize or resend (${streamMode})`, async () => { + const created = { + type: "response.created", + response: { id: "resp-function-output-reset", status: "in_progress" }, + }; + const prefix = new TextEncoder().encode( + `event: response.created +data: ${JSON.stringify(created)} + +`, + ); + const readError = new Error("upstream stream reset"); + const outbound: Array> = []; + globalThis.fetch = Object.assign(async (_input: RequestInfo | URL, init?: RequestInit) => { + outbound.push(JSON.parse(String(init?.body)) as Record); + let sentPrefix = false; + return new Response(new ReadableStream({ + pull(controller) { + if (!sentPrefix) { + sentPrefix = true; + controller.enqueue(prefix); + return; + } + return Promise.reject(readError); + }, + }), { status: 200, headers: { "content-type": "text/event-stream" } }); + }, { preconnect: originalFetch.preconnect }) as typeof fetch; + + const logCtx: RequestLogContext = { model: "", provider: "" }; + const response = await handleResponses(functionOutputRequest(true), { + ...config(), streamMode, + }, logCtx); + const body = await response.text(); + expect(response.status).toBe(200); + expect(body).toContain("response.failed"); + expect(body).toContain('"code":"upstream_reset"'); + expect(body).not.toContain('"reason":"adapter_eof"'); + expect(outbound).toHaveLength(1); + const sentInput = outbound.at(0)?.input as Array> | undefined; + expect(sentInput?.at(1)).toEqual(functionOutputReplayInput().at(1)); + expect(JSON.stringify(sentInput)).toContain("encrypted_content"); + }); + + test(`created-then-abort streamed function-output returns 499 without resend (${streamMode})`, async () => { + const created = { + type: "response.created", + response: { id: "resp-function-output-abort", status: "in_progress" }, + }; + const prefix = new TextEncoder().encode( + `event: response.created +data: ${JSON.stringify(created)} + +`, + ); + const abort = new AbortController(); + let fetchSignal: AbortSignal | undefined; + let sawCreated!: () => void; + const createdStarted = new Promise(resolve => { sawCreated = resolve; }); + const outbound: Array> = []; + globalThis.fetch = Object.assign(async (_input: RequestInfo | URL, init?: RequestInit) => { + outbound.push(JSON.parse(String(init?.body)) as Record); + fetchSignal = init?.signal ?? undefined; + let sentPrefix = false; + return new Response(new ReadableStream({ + pull(controller) { + if (!sentPrefix) { + sentPrefix = true; + controller.enqueue(prefix); + sawCreated(); + return new Promise((_resolve, reject) => { + const fail = () => reject(fetchSignal?.reason ?? new Error("aborted")); + if (fetchSignal?.aborted) { + fail(); + return; + } + fetchSignal?.addEventListener("abort", fail, { once: true }); + }); + } + }, + }), { status: 200, headers: { "content-type": "text/event-stream" } }); + }, { preconnect: originalFetch.preconnect }) as typeof fetch; + + const logCtx: RequestLogContext = { model: "", provider: "" }; + const pending = handleResponses(functionOutputRequest(true), { + ...config(), streamMode, + }, logCtx, { abortSignal: abort.signal }); + await createdStarted; + expect(fetchSignal).toBeDefined(); + abort.abort(); + expect(fetchSignal?.aborted).toBe(true); + const response = await pending; + expect(response.status).toBe(499); + const body = await response.json() as { error?: { code?: string; type?: string } }; + expect(body.error?.code ?? body.error?.type).toBe("client_cancelled"); + expect(outbound).toHaveLength(1); + const sentInput = outbound.at(0)?.input as Array> | undefined; + expect(sentInput?.at(1)).toEqual(functionOutputReplayInput().at(1)); + }); + } + 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/routing/combo-stream-preflight.test.ts b/tests/routing/combo-stream-preflight.test.ts index 4c7b20c35e..b1ce38b8eb 100644 --- a/tests/routing/combo-stream-preflight.test.ts +++ b/tests/routing/combo-stream-preflight.test.ts @@ -1,4 +1,4 @@ -import { describe, expect, test } from "bun:test"; +import { describe, expect, spyOn, test } from "bun:test"; import { comboStreamPayloadCommitsOutput, preflightComboStreamResponse, @@ -13,6 +13,41 @@ const sse = (...payloads: unknown[]): Response => new Response( const preflightChunkLimit = Math.max(1, Math.ceil(MAX_CLIENT_SSE_FRAME_BYTES / 1024)); +function prefixThenReadError(prefix: Uint8Array, error: Error): { + response: Response; + cancelSpy: () => ReturnType | undefined; +} { + let sentPrefix = false; + let cancelSpy: ReturnType | undefined; + const stream = new ReadableStream({ + pull(controller) { + if (!sentPrefix) { + sentPrefix = true; + controller.enqueue(prefix); + return; + } + return Promise.reject(error); + }, + }); + const originalGetReader = stream.getReader.bind(stream); + stream.getReader = (() => { + const reader = originalGetReader(); + cancelSpy = spyOn(reader, "cancel"); + return reader; + }) as ReadableStream["getReader"]; + return { + response: new Response(stream, { headers: { "content-type": "text/event-stream" } }), + cancelSpy: () => cancelSpy, + }; +} + +const createdPrefix = new TextEncoder().encode(`data: ${JSON.stringify({ + type: "response.created", + response: { id: "r1", status: "in_progress" }, +})} + +`); + describe("combo stream preflight", () => { test("keeps only lifecycle preamble replayable and treats unknown output conservatively", () => { expect(comboStreamPayloadCommitsOutput({ type: "response.created" })).toBe(false); @@ -397,4 +432,35 @@ describe("combo stream preflight", () => { expect(await result.response.text()).toBe(body); } }); + + test("default reader.read rejection still throws and does not cancel the reader", async () => { + const readError = new Error("preflight-read-reset"); + const source = prefixThenReadError(createdPrefix, readError); + await expect(preflightComboStreamResponse(source.response, { model: "m1", provider: "a" })) + .rejects.toBe(readError); + expect(source.cancelSpy()).toBeDefined(); + expect(source.cancelSpy()!.mock.calls).toHaveLength(0); + }); + + test("replayReadErrors accepts a reconstructed prefix and the same reader.read error", async () => { + const readError = new Error("preflight-read-reset"); + const source = prefixThenReadError(createdPrefix, readError); + const result = await preflightComboStreamResponse( + source.response, + { model: "m1", provider: "a" }, + undefined, + { replayReadErrors: true }, + ); + expect(result.kind).toBe("accepted"); + expect(source.cancelSpy()).toBeDefined(); + expect(source.cancelSpy()!.mock.calls).toHaveLength(0); + const reader = result.response.body!.getReader(); + const first = await reader.read(); + expect(first.done).toBe(false); + expect(first.value).toEqual(createdPrefix); + await expect(reader.read()).rejects.toBe(readError); + expect(source.cancelSpy()).toBeDefined(); + expect(source.cancelSpy()!.mock.calls).toHaveLength(0); + }); + }); From 4112efc3340af83e42b7f4ee8093e1c62815bfde Mon Sep 17 00:00:00 2001 From: t Date: Sun, 6 Sep 2026 16:51:20 +0900 Subject: [PATCH 12/14] docs: include semantic failure usage marker parity --- .../010_failure_boundaries.md | 13 +++++++++++++ 1 file changed, 13 insertions(+) diff --git a/devlog/_plan/260906_opaque_transport_finality/010_failure_boundaries.md b/devlog/_plan/260906_opaque_transport_finality/010_failure_boundaries.md index a4f93b4f81..ccc1552b5b 100644 --- a/devlog/_plan/260906_opaque_transport_finality/010_failure_boundaries.md +++ b/devlog/_plan/260906_opaque_transport_finality/010_failure_boundaries.md @@ -52,3 +52,16 @@ byte bounds, no-persistence and one-shot recovery tests remain. Independent plan/source/final review; exact parent and cascaded child hosted Linux/macOS/gates CI. Final Windows six-shard and release gates remain mandatory. + +## Usage-marker parity amendment + +Source review confirms the account-health blocker is closed by failed EOF. The +existing eager callback still labels every synthetic failure as streamAborted, +though a clean EOF after an explicit upstream error is a semantic failure, not a +body-read reset (PersistedUsageAttempt documents that distinction). Criterion c-2 +also requires usage outcome parity, so include this small related correction: +relay-eager passes optional upstream_error provenance only for that clean-EOF tail; +core records its semantic failed status without streamAborted. Ordinary reset +callbacks retain their one-argument shape, 502 and streamAborted. Add request-level +tee/eager assertions for repeated bare errors versus actual reset; do not infer +this marker from a stale log message or change real-terminal precedence. From b86021b9f6307a75c88767bfd90c04d2ff5550b0 Mon Sep 17 00:00:00 2001 From: t Date: Sun, 6 Sep 2026 16:52:43 +0900 Subject: [PATCH 13/14] fix(responses): align semantic failure accounting across relay modes --- src/server/relay-eager.ts | 9 +- src/server/responses/core.ts | 5 +- tests/codex-integration/codex-routing.test.ts | 84 ++++++++++++++ .../responses-opaque-blob-recovery.test.ts | 24 +++- .../consume-for-inspection-cancel.test.ts | 108 ++++++++++++++++++ 5 files changed, 225 insertions(+), 5 deletions(-) diff --git a/src/server/relay-eager.ts b/src/server/relay-eager.ts index 151a53ac82..7844c65886 100644 --- a/src/server/relay-eager.ts +++ b/src/server/relay-eager.ts @@ -64,7 +64,7 @@ export type EagerRelayHooks = { /** True once inspection has reported a protocol terminal (inspector.reported). */ sawTerminal: () => boolean; /** Record a synthetic terminal (caller decides incomplete vs failed-502). */ - onSynthetic: (kind: "incomplete" | "failed") => void; + onSynthetic: (kind: "incomplete" | "failed", reason?: "upstream_error") => void; /** Client cancelled and NO terminal arrived within the drain bounds. */ onClientCancel: () => void; /** Exactly once, after the producer fully stops (unregisterTurn parity). */ @@ -242,6 +242,7 @@ export function relaySseEagerBounded( const producer = async () => { let syntheticKind: "incomplete" | "failed" | null = null; + let syntheticReason: "upstream_error" | undefined; let deliveryFallbackSent = false; let priorRewriteFailure = false; let priorRewriteError: unknown; @@ -316,6 +317,7 @@ export function relaySseEagerBounded( controllerRef?.enqueue(terminalSentinel); } catch { /* client already gone */ } syntheticKind = upstreamError === undefined ? "incomplete" : "failed"; + syntheticReason = upstreamError === undefined ? undefined : "upstream_error"; } break; } @@ -456,7 +458,10 @@ export function relaySseEagerBounded( frameBufferBytes = 0; } terminalBoundary.dispose(); - if (syntheticKind && canDeliver()) hooks.onSynthetic(syntheticKind); + if (syntheticKind && canDeliver()) { + if (syntheticReason === undefined) hooks.onSynthetic(syntheticKind); + else hooks.onSynthetic(syntheticKind, syntheticReason); + } if (cancelled && !hooks.sawTerminal()) { hooks.onClientCancel(); } diff --git a/src/server/responses/core.ts b/src/server/responses/core.ts index 583f7f8184..bfe28dd309 100644 --- a/src/server/responses/core.ts +++ b/src/server/responses/core.ts @@ -5330,11 +5330,14 @@ async function handleResponsesInner( ...(clientBlockRewrite ? { rewriteBlocks: clientBlockRewrite } : {}), - onSynthetic: kind => { + onSynthetic: (kind, reason) => { if (!reportNativeTerminal) return; if (kind === "incomplete") { logCtx.terminalSource = "synthetic"; reportNativeTerminal("incomplete"); + } else if (reason === "upstream_error") { + logCtx.terminalSource = "synthetic"; + reportNativeTerminal("failed", logCtx.terminalHttpStatus ?? 502); } else { logCtx.transportPhase = "mid_stream"; logCtx.terminalSource = "synthetic"; diff --git a/tests/codex-integration/codex-routing.test.ts b/tests/codex-integration/codex-routing.test.ts index d4179ff447..bc6561e99e 100644 --- a/tests/codex-integration/codex-routing.test.ts +++ b/tests/codex-integration/codex-routing.test.ts @@ -1339,6 +1339,90 @@ describe("codex routing", () => { expect(getCodexUpstreamHealth("a")).toBeNull(); }); + test("flat bare error at inspection EOF records failed 502 without clearing avoidance", async () => { + const config = makeConfig(); + updateAccountQuota("a", 10); + updateAccountQuota("b", 10); + const now = 1_800_000_000_000; + recordCodexUpstreamOutcome(config, "a", 503, { now }); + recordCodexUpstreamOutcome(config, "a", 503, { now: now + 1 }); + recordCodexUpstreamOutcome(config, "a", 503, { now: now + 2 }); + expect(isCodexAccountSoftAvoided("a", now + 2)).toBe(true); + expect(getCodexUpstreamHealth("a")?.consecutiveFailures).toBe(3); + const terminals: Array<[string, number | undefined]> = []; + const encoder = new TextEncoder(); + const stream = new ReadableStream({ + start(controller) { + controller.enqueue(encoder.encode("data: " + JSON.stringify({ + type: "error", + message: "provider reset", + }) + "\n\n")); + controller.close(); + }, + }); + + await new Promise(resolve => { + consumeForInspection(stream, (status, override) => { + terminals.push([status, override]); + recordCodexUpstreamOutcome( + config, + "a", + status === "failed" ? (override ?? 502) : 200, + { now: now + 3, threadId: "bare-error-flat" }, + ); + }, undefined, resolve); + }); + + expect(terminals).toEqual([["failed", 502]]); + expect(isCodexAccountSoftAvoided("a", now + 3)).toBe(true); + expect(getCodexUpstreamHealth("a")).toMatchObject({ + consecutiveFailures: 4, + lastFailureStatus: 502, + }); + }); + + test("nested bare error at inspection EOF records failed 502 without clearing avoidance", async () => { + const config = makeConfig(); + updateAccountQuota("a", 10); + updateAccountQuota("b", 10); + const now = 1_800_000_000_000; + recordCodexUpstreamOutcome(config, "a", 503, { now }); + recordCodexUpstreamOutcome(config, "a", 503, { now: now + 1 }); + recordCodexUpstreamOutcome(config, "a", 503, { now: now + 2 }); + expect(isCodexAccountSoftAvoided("a", now + 2)).toBe(true); + expect(getCodexUpstreamHealth("a")?.consecutiveFailures).toBe(3); + const terminals: Array<[string, number | undefined]> = []; + const encoder = new TextEncoder(); + const stream = new ReadableStream({ + start(controller) { + controller.enqueue(encoder.encode("data: " + JSON.stringify({ + type: "error", + error: { message: "nested provider reset" }, + }) + "\n\n")); + controller.close(); + }, + }); + + await new Promise(resolve => { + consumeForInspection(stream, (status, override) => { + terminals.push([status, override]); + recordCodexUpstreamOutcome( + config, + "a", + status === "failed" ? (override ?? 502) : 200, + { now: now + 3, threadId: "bare-error-nested" }, + ); + }, undefined, resolve); + }); + + expect(terminals).toEqual([["failed", 502]]); + expect(isCodexAccountSoftAvoided("a", now + 3)).toBe(true); + expect(getCodexUpstreamHealth("a")).toMatchObject({ + consecutiveFailures: 4, + lastFailureStatus: 502, + }); + }); + test("transient cooldown escalates to 2m, 10m, then the 30m cap", () => { const config = makeConfig(); const now = 1_800_000_000_000; diff --git a/tests/responses/responses-opaque-blob-recovery.test.ts b/tests/responses/responses-opaque-blob-recovery.test.ts index da3b62598b..cf26431381 100644 --- a/tests/responses/responses-opaque-blob-recovery.test.ts +++ b/tests/responses/responses-opaque-blob-recovery.test.ts @@ -647,10 +647,21 @@ describe("opaque blob recovery through /v1/responses", () => { sends += 1; return streamedFunctionOutputDecryptErrorEvent(flat); }, { preconnect: originalFetch.preconnect }); + const logCtx: RequestLogContext = { model: "", provider: "" }; + const terminals: string[] = []; + let markTerminal!: () => void; + const terminal = new Promise(resolve => { markTerminal = resolve; }); const response = await handleResponses(agentMessageRequest(true), { ...config(), streamMode, - }, { model: "", provider: "" }); + }, logCtx, { onNativePassthroughTerminal: status => { + terminals.push(status); + markTerminal(); + } }); const body = await response.text(); + await terminal; + expect(terminals).toEqual(["failed"]); + expect(logCtx.activeAttempt).toBeDefined(); + expect(logCtx.activeAttempt?.streamAborted).not.toBe(true); expect(sends).toBe(2); expect(body).toContain(FUNCTION_OUTPUT_DECRYPT_MESSAGE); expect(body).not.toContain("adapter_eof"); @@ -813,10 +824,19 @@ data: ${JSON.stringify(created)} }, { preconnect: originalFetch.preconnect }) as typeof fetch; const logCtx: RequestLogContext = { model: "", provider: "" }; + const terminals: string[] = []; + let markTerminal!: () => void; + const terminal = new Promise(resolve => { markTerminal = resolve; }); const response = await handleResponses(functionOutputRequest(true), { ...config(), streamMode, - }, logCtx); + }, logCtx, { onNativePassthroughTerminal: status => { + terminals.push(status); + markTerminal(); + } }); const body = await response.text(); + await terminal; + expect(terminals).toEqual(["failed"]); + expect(logCtx.activeAttempt?.streamAborted).toBe(true); expect(response.status).toBe(200); expect(body).toContain("response.failed"); expect(body).toContain('"code":"upstream_reset"'); diff --git a/tests/server/consume-for-inspection-cancel.test.ts b/tests/server/consume-for-inspection-cancel.test.ts index 166642d4fc..681b91a717 100644 --- a/tests/server/consume-for-inspection-cancel.test.ts +++ b/tests/server/consume-for-inspection-cancel.test.ts @@ -317,3 +317,111 @@ describe("inspection consumer teardown", () => { expect(metadataSpy.disposes()).toBe(1); }); }); + +function errorFrame(payload: Record): Uint8Array { + return encoder.encode("data: " + JSON.stringify(payload) + "\n\n"); +} + +describe("consumeForInspection bare-error EOF finality", () => { + test("custom onParsedPayload still runs for a witnessed bare error", async () => { + const source = controlledStream(); + const parsed: unknown[] = []; + const terminals: string[] = []; + const done = new Promise(resolve => { + consumeForInspection( + source.stream, + status => terminals.push(status), + undefined, + resolve, + undefined, + undefined, + undefined, + undefined, + { onParsedPayload: payload => parsed.push(payload) }, + ); + }); + + source.push(errorFrame({ type: "error", message: "flat reset" })); + source.close(); + await done; + + expect(parsed).toEqual([{ type: "error", message: "flat reset" }]); + expect(terminals).toEqual(["failed"]); + }); + + test("a real completed terminal after a bare error still wins", async () => { + const source = controlledStream(); + const terminals: string[] = []; + const completed: unknown[] = []; + const done = new Promise(resolve => { + consumeForInspection( + source.stream, + status => terminals.push(status), + undefined, + resolve, + undefined, + undefined, + response => completed.push(response), + ); + }); + + source.push(errorFrame({ type: "error", error: { message: "nested reset" } })); + source.push(completedFrame("after-error")); + source.close(); + await done; + + expect(terminals).toEqual(["completed"]); + expect(completed).toHaveLength(1); + }); + + test("stale logCtx.upstreamError without a bare error remains incomplete", async () => { + const source = controlledStream(); + const logCtx: RequestLogContext = { model: "m", provider: "p", upstreamError: "stale borrowed failure" }; + let terminalStatus: string | null = null; + const done = new Promise(resolve => { + consumeForInspection( + source.stream, + status => { terminalStatus = status; }, + undefined, + resolve, + logCtx, + ); + }); + + source.push(encoder.encode("data: {\"type\":\"response.output_item.added\"}\n\n")); + source.close(); + await done; + + expect(terminalStatus).toBe("incomplete"); + }); + + test("cancellation after a bare error stays neutral", async () => { + const source = controlledStream(); + const ac = new AbortController(); + let terminals = 0; + let cancels = 0; + let markParsed!: () => void; + const parsed = new Promise(resolve => { markParsed = resolve; }); + const done = new Promise(resolve => { + consumeForInspection( + source.stream, + () => { terminals += 1; }, + ac.signal, + resolve, + undefined, + () => { cancels += 1; }, + undefined, + undefined, + { onParsedPayload: () => markParsed() }, + ); + }); + + source.push(errorFrame({ type: "error", message: "reset then cancel" })); + await parsed; + ac.abort(); + await done; + + expect(terminals).toBe(0); + expect(cancels).toBe(1); + }); +}); From cd6d4d346e3765cca85bd79675b7c3896da0822b Mon Sep 17 00:00:00 2001 From: t Date: Sun, 6 Sep 2026 16:56:10 +0900 Subject: [PATCH 14/14] fix(types): infer the configured stream reader result --- src/server/responses/combo-stream-preflight.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/server/responses/combo-stream-preflight.ts b/src/server/responses/combo-stream-preflight.ts index 8c9ed6ebe5..bbc90d0ca1 100644 --- a/src/server/responses/combo-stream-preflight.ts +++ b/src/server/responses/combo-stream-preflight.ts @@ -168,7 +168,7 @@ export async function preflightComboStreamResponse( try { for (;;) { - let next: ReadableStreamReadResult; + let next: Awaited>; try { next = await reader.read(); } catch (error) {