fix(responses): recover agent_message encrypted-content rejections instead of adapter_eof

docs-site/src/content/docs/guides/sub-agent-surface.md

@@ -158,6 +158,19 @@ for the full trust boundary and configuration.
 Combo routing remains unchanged and continues to consider only canonical native ChatGPT targets for
 encrypted tasks.
 
+## Rejected encrypted history
+
+An upstream Responses server can reject encrypted parts in an earlier `agent_message`
+with `Encrypted function output content could not be decrypted or decoded.`. Before
+any output is committed, opencodex replaces those parts with `[encrypted content omitted]`
+and rebuilds the request once. The surrounding readable content stays intact; the
+omitted content is not decrypted or recovered by this retry.
+
+If the rebuilt request receives another bare SSE `error` followed by EOF, both relay
+modes preserve the error message in a `response.failed` terminal instead of reporting
+`adapter_eof`. Other upstream `response.failed` events remain SSE failures. This history
+recovery does not change the encrypted v2 task-delivery restrictions described above.
+
 ## Changing the mode
 
 ### GUI

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;
 }
 

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;
         }

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<BoundedSseFrameBuffer["feed"]>,
@@ -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<Uint8Array>,
   upstream: AbortController,
   onClientGone?: (reason?: unknown) => void,
+  opts?: { upstreamError?: string },
 ): ReadableStream<Uint8Array> {
   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();

src/server/responses/combo-stream-preflight.ts

@@ -1,4 +1,5 @@
 import type { ResponsesTerminalStatus } from "../../bridge";
+import { ENCRYPTED_FUNCTION_OUTPUT_REJECTION, upstreamErrorMessageFromPayload } from "../../lib/errors";
 import type { RequestLogContext } from "../request-log";
 import { createSseInspector } from "../relay";
 import { MAX_CLIENT_SSE_FRAME_BYTES } from "../sse-frame-buffer";
@@ -37,6 +38,9 @@ function retryableZeroOutputTerminal(payload: unknown): boolean {
     response?: { incomplete_details?: { reason?: unknown } };
   };
   if (event.type === "response.failed") return true;
+  if (event.type === "error") {
+    return upstreamErrorMessageFromPayload(event) === ENCRYPTED_FUNCTION_OUTPUT_REJECTION;
+  }
   if (event.type !== "response.incomplete") return false;
   const reason = event.response?.incomplete_details?.reason;
   return typeof reason === "string" && RETRYABLE_ZERO_OUTPUT_INCOMPLETE_REASONS.has(reason);
@@ -51,6 +55,9 @@ export function comboStreamPayloadCommitsOutput(payload: unknown): boolean {
   if (!payload || typeof payload !== "object" || Array.isArray(payload)) return true;
   const type = (payload as { type?: unknown }).type;
   if (typeof type !== "string") return true;
+  // An error event carries no client-visible output; treating it as committing
+  // would pin a child to a turn that already failed before producing anything.
+  if (type === "error") return false;
   return !PRE_OUTPUT_CONTROL_EVENTS.has(type) && !TERMINAL_EVENTS.has(type);
 }
 
@@ -135,6 +142,7 @@ export type ComboStreamPreflightResult =
 export async function preflightComboStreamResponse(
   response: Response,
   logCtx: RequestLogContext,
+  retryableTerminal: (payload: unknown) => boolean = retryableZeroOutputTerminal,
 ): Promise<ComboStreamPreflightResult> {
   const contentType = response.headers.get("content-type")?.toLowerCase() ?? "";
   if (!response.ok || !response.body || !contentType.includes("text/event-stream")) {
@@ -152,7 +160,7 @@ export async function preflightComboStreamResponse(
     onParsedPayload: payload => {
       if (comboStreamPayloadCommitsOutput(payload)) outputCommitted = true;
       if (!payload || typeof payload !== "object" || Array.isArray(payload)) return;
-      if (retryableZeroOutputTerminal(payload)) {
+      if (retryableTerminal(payload)) {
         retryableTerminalPayload = payload as Record<string, unknown>;
       }
     },
@@ -180,7 +188,10 @@ export async function preflightComboStreamResponse(
         inspector.feed(retained);
       }
 
-      if ((terminalStatus === "failed" || terminalStatus === "incomplete")
+      // A bare error event is not a protocol terminal (terminalStatus stays undefined),
+      // so its exact-message retryable match doubles as the terminal evidence.
+      if ((terminalStatus === "failed" || terminalStatus === "incomplete"
+        || retryableTerminalPayload?.type === "error")
         && !outputCommitted && retryableTerminalPayload) {
         await reader.cancel("retrying zero-output combo stream terminal").catch(() => undefined);
         return { kind: "failed", response: failedTerminalResponse(response, retryableTerminalPayload, logCtx) };