fix(openai-compat): preserve nested response failure codes

Normalize public Responses terminal failures so redacted safe error codes are available under response.error for streaming clients while keeping provider messages generic.

Cover transport, runtime relay, collected response, and compatibility matrix behavior.

Author: masterkain

lib/codex_pooler/gateway/runtime/streaming/downstream_stream.ex

@@ -199,7 +199,7 @@ defmodule CodexPooler.Gateway.Runtime.Streaming.DownstreamStream do
     state
     |> maybe_put_public_openai_responses_response_id(decoded)
     |> maybe_mark_public_openai_responses_data_seen(event)
-    |> maybe_mark_public_openai_responses_terminal_seen(event)
+    |> maybe_mark_public_openai_responses_terminal_seen(event_type, decoded)
   end
 
   defp normalize_codex_responses_stream_data(data, endpoint, opts, state) when is_binary(data) do
@@ -300,8 +300,8 @@ defmodule CodexPooler.Gateway.Runtime.Streaming.DownstreamStream do
     end
   end
 
-  defp maybe_mark_public_openai_responses_terminal_seen(state, event) do
-    case StreamProtocol.terminal_outcome_event(event) do
+  defp maybe_mark_public_openai_responses_terminal_seen(state, event_type, decoded) do
+    case StreamProtocol.terminal_outcome(event_type, decoded) do
       {:ok, %{kind: :failed, failure: failure}} ->
         state
         |> Map.put(:public_openai_responses_terminal_seen?, true)

lib/codex_pooler/gateway/transports/streaming/stream_protocol/public_responses.ex

@@ -485,6 +485,12 @@ defmodule CodexPooler.Gateway.Transports.Streaming.StreamProtocol.PublicResponse
     )
   end
 
+  defp normalize_response_error(
+         %{"error" => %{} = public_error, "response" => %{} = response} = decoded
+       ) do
+    Map.put(decoded, "response", Map.put(response, "error", public_error))
+  end
+
   defp normalize_response_error(decoded), do: decoded
 
   defp normalize_terminal_error(error) do

test/codex_pooler/gateway/runtime/streaming/downstream_stream_test.exs

@@ -447,6 +447,42 @@ defmodule CodexPooler.Gateway.Runtime.Streaming.DownstreamStreamTest do
       assert failure.event_type == "response.failed"
     end
 
+    test "copies top-level public Responses terminal error into response failure" do
+      opts =
+        RequestOptions.build(
+          %{public_openai_responses_stream: true},
+          "/v1/responses",
+          %{"stream" => true}
+        )
+
+      state = DownstreamStream.initial_state(:relay, opts)
+
+      failed =
+        sse_event("response.failed", %{
+          "type" => "response.failed",
+          "response" => %{
+            "id" => "resp_public_failed_top_level_error",
+            "status" => "failed"
+          },
+          "error" => %{
+            "type" => "invalid_request_error",
+            "code" => "context_length_exceeded"
+          }
+        })
+
+      assert {chunk, state} =
+               DownstreamStream.normalize_data(failed, "/v1/responses", opts, state)
+
+      assert [%{"event" => "response.failed", "data" => data}] = public_sse_events(chunk)
+      assert data["error"]["code"] == "context_length_exceeded"
+
+      assert {:failed, failure} = DownstreamStream.terminal_outcome(state)
+      assert failure.event_type == "response.failed"
+
+      assert {data["response"]["error"]["code"], failure.code} ==
+               {"context_length_exceeded", "context_length_exceeded"}
+    end
+
     test "synthesizes a sanitized terminal failure with the observed public response id" do
       opts =
         RequestOptions.build(

test/codex_pooler/gateway/transports/stream_protocol_test.exs

@@ -51,6 +51,7 @@ defmodule CodexPooler.Gateway.Transports.Streaming.StreamProtocolTest do
       assert data["type"] == "response.failed"
       assert data["response"]["status"] == "failed"
       assert data["error"]["code"] == "context_length_exceeded"
+      assert data["response"]["error"]["code"] == "context_length_exceeded"
     end
   end
 
@@ -322,6 +323,32 @@ defmodule CodexPooler.Gateway.Transports.Streaming.StreamProtocolTest do
       refute chunk =~ "event: response.output_text.delta\n"
     end
 
+    test "adds redacted nested response error for top-level context overflow failures" do
+      state = StreamProtocol.public_openai_responses_stream_state()
+
+      failed =
+        sse_event("response.failed", %{
+          "type" => "response.failed",
+          "error" => %{
+            "type" => "invalid_request_error",
+            "code" => "context_length_exceeded",
+            "message" => "synthetic untrusted overflow detail"
+          },
+          "response" => %{"id" => "resp_context_overflow", "status" => "failed"}
+        })
+
+      assert {chunk, _state} =
+               StreamProtocol.normalize_public_openai_responses_sse_data(failed, state)
+
+      assert [%{"event" => "response.failed", "data" => data}] = public_sse_events(chunk)
+      assert data["error"]["code"] == "context_length_exceeded"
+      assert data["error"]["message"] == "upstream request failed"
+
+      assert %{"error" => response_error} = data["response"]
+      assert response_error["code"] == "context_length_exceeded"
+      assert response_error["message"] == "upstream request failed"
+    end
+
     test "emits early top-level error without synthetic success prefix" do
       state = StreamProtocol.public_openai_responses_stream_state()
 

test/codex_pooler_web/controllers/v1/responses_controller_test.exs

@@ -2748,6 +2748,62 @@ defmodule CodexPoolerWeb.V1.ResponsesControllerTest do
     assert FakeUpstream.count(upstream) == 1
   end
 
+  @tag :provider_invalid_request_redaction
+  test "POST /v1/responses collected SSE preserves safe terminal error codes", %{conn: conn} do
+    provider_message =
+      "provider 400 leaked https://provider.internal.example/context?key=sk-secret and prompt SENTINEL_COLLECT"
+
+    upstream_error =
+      provider_invalid_request_error("context_length_exceeded", provider_message, "input")
+
+    cases = [
+      {"top-level only",
+       %{
+         "type" => "response.failed",
+         "error" => upstream_error,
+         "response" => %{
+           "id" => "resp_v1_collect_top_level_error",
+           "status" => "failed"
+         }
+       }},
+      {"nested",
+       %{
+         "type" => "response.failed",
+         "response" => %{
+           "id" => "resp_v1_collect_nested_error",
+           "status" => "failed",
+           "error" => upstream_error
+         }
+       }}
+    ]
+
+    Enum.each(cases, fn {_label, terminal_event} ->
+      upstream = start_upstream(FakeUpstream.sse_stream([{"response.failed", terminal_event}]))
+      setup = gateway_setup(upstream)
+
+      response =
+        conn
+        |> recycle()
+        |> auth(setup)
+        |> post("/v1/responses", %{
+          "model" => setup.model.exposed_model_id,
+          "input" => "synthetic collected provider invalid request"
+        })
+
+      assert %{"error" => error} = json_response(response, 502)
+      assert error["message"] == "upstream request failed"
+      assert error["type"] == "server_error"
+      assert error["code"] == "context_length_exceeded"
+      refute Map.has_key?(error, "param")
+      refute response.resp_body =~ provider_message
+      refute response.resp_body =~ "provider.internal.example"
+      refute response.resp_body =~ "sk-secret"
+      refute response.resp_body =~ "SENTINEL_COLLECT"
+      refute response.resp_body =~ "input"
+      assert FakeUpstream.count(upstream) == 1
+    end)
+  end
+
   @tag :server_error_redaction
   test "POST /v1/responses SSE collection redacts safe-looking terminal 502 errors" do
     provider_message =
@@ -2780,6 +2836,38 @@ defmodule CodexPoolerWeb.V1.ResponsesControllerTest do
     refute inspect(error) =~ "provider_stack"
   end
 
+  @tag :server_error_redaction
+  test "POST /v1/responses SSE collection redacts top-level-only terminal errors" do
+    provider_message =
+      "provider failed at https://upstream.internal.example/internal/context?token=secret"
+
+    upstream_error =
+      provider_invalid_request_error("context_length_exceeded", provider_message, "input")
+
+    body =
+      "event: response.failed\n" <>
+        "data: " <>
+        Jason.encode!(%{
+          "type" => "response.failed",
+          "error" => upstream_error,
+          "response" => %{
+            "id" => "resp_v1_collect_top_level_only_failed",
+            "status" => "failed"
+          }
+        }) <>
+        "\n\n"
+
+    assert {:error, error} = Responses.response_from_sse(body)
+    assert error.status == 502
+    assert error.message == "upstream request failed"
+    assert error.code == "context_length_exceeded"
+    assert error.param == nil
+    refute inspect(error) =~ "provider failed"
+    refute inspect(error) =~ "upstream.internal.example"
+    refute inspect(error) =~ "/internal/context"
+    refute inspect(error) =~ "input"
+  end
+
   @tag :server_error_redaction
   test "POST /v1/responses SSE collection rejects failure-coded incomplete" do
     body =

test/support/compatibility_matrix.ex

@@ -292,7 +292,7 @@ defmodule CodexPooler.CompatibilityMatrix do
       future_routes: [],
       fixture: :v1_supported_surface,
       contract:
-        "OpenAI-compatible /v1 routes are default-on for pools, require bearer API-key auth, return OpenAI-shaped errors without anonymous local or CIDR bypasses, include narrow GET /v1/responses Responses websocket compatibility only, exclude broad /v1/realtime routes, keep POST /v1/responses/compact routed only to deterministic unsupported_endpoint with no upstream compact dispatch, reject OpenAI Responses remote MCP tool definitions before upstream dispatch in both top-level tools and nested additional_tools.tools locations with OpenAI-shaped invalid_request errors, consume continuity headers using the documented local precedence without forwarding session-id, x-session-id, or x-session-affinity upstream, fail closed for pinned /v1/responses continuations whose upstream account needs revoked-refresh-token reauthentication with the shared restart_with_full_context recovery guidance, allow prompt-cache routing locality only on POST responses and chat completions, accept Codex-native Responses web_search hosted tool shapes with boolean access flags while keeping web_search_preview type-only, accept Responses truncation auto and disabled locally without forwarding it upstream, lift Responses system/developer input-message text into top-level instructions, emit early public streaming terminal errors without synthetic success prefixes, emit sanitized response.failed upstream_stream_error when POST /v1/responses SSE has already exposed public Responses data and upstream closes before a Responses terminal event, keep that synthetic terminal limited to public HTTP SSE and preserve backend raw/websocket stream behavior, redact server-class/internal/upstream public /v1 errors while preserving invalid_request_error validation details, map Responses content_filter/content-filter incomplete reasons to chat finish_reason content_filter while other incomplete reasons remain length, forward structured tool-result/function_call_output payloads unchanged, translate chat-style role=tool continuation messages and Hermes assistant tool-call replays into Responses function_call/function_call_output input items before validation, accept safe Hermes assistant replay status values, drop known OMP function_call replay status fields before validation, translate OpenClaw assistant thinking replays before validation, and keep chat input fallback, Responses additional_tools support narrow and non-executable, and Responses namespace-tool support narrow"
+        "OpenAI-compatible /v1 routes are default-on for pools, require bearer API-key auth, return OpenAI-shaped errors without anonymous local or CIDR bypasses, include narrow GET /v1/responses Responses websocket compatibility only, exclude broad /v1/realtime routes, keep POST /v1/responses/compact routed only to deterministic unsupported_endpoint with no upstream compact dispatch, reject OpenAI Responses remote MCP tool definitions before upstream dispatch in both top-level tools and nested additional_tools.tools locations with OpenAI-shaped invalid_request errors, consume continuity headers using the documented local precedence without forwarding session-id, x-session-id, or x-session-affinity upstream, fail closed for pinned /v1/responses continuations whose upstream account needs revoked-refresh-token reauthentication with the shared restart_with_full_context recovery guidance, allow prompt-cache routing locality only on POST responses and chat completions, accept Codex-native Responses web_search hosted tool shapes with boolean access flags while keeping web_search_preview type-only, accept Responses truncation auto and disabled locally without forwarding it upstream, lift Responses system/developer input-message text into top-level instructions, emit early public streaming terminal errors without synthetic success prefixes, emit sanitized response.failed upstream_stream_error when POST /v1/responses SSE has already exposed public Responses data and upstream closes before a Responses terminal event, keep that synthetic terminal limited to public HTTP SSE and preserve backend raw/websocket stream behavior, redact server-class/internal/upstream public /v1 errors while preserving invalid_request_error validation details, preserve safe machine-readable codes for redacted public OpenAI-compatible Responses terminal failures in nested response.error through low-level public SSE normalization and the runtime streaming relay, keep top-level error code-aligned when Pooler emits one, map Responses content_filter/content-filter incomplete reasons to chat finish_reason content_filter while other incomplete reasons remain length, forward structured tool-result/function_call_output payloads unchanged, translate chat-style role=tool continuation messages and Hermes assistant tool-call replays into Responses function_call/function_call_output input items before validation, accept safe Hermes assistant replay status values, drop known OMP function_call replay status fields before validation, translate OpenClaw assistant thinking replays before validation, and keep chat input fallback, Responses additional_tools support narrow and non-executable, and Responses namespace-tool support narrow"
     },
     %{
       slug: :v1_unsupported_public_surface,
@@ -781,6 +781,14 @@ defmodule CodexPooler.CompatibilityMatrix do
         server_class_message: "upstream request failed",
         server_class_type: "server_error",
         server_class_code: ["safe_upstream_code", "upstream_error"],
+        responses_terminal_code_locations: [
+          "response.error.code",
+          "top_level_error.code_when_emitted"
+        ],
+        responses_terminal_stream_paths: [
+          "low_level_public_sse_normalization",
+          "runtime_streaming_relay"
+        ],
         preserves_invalid_request_error_details: true
       },
       chat_finish_reasons: %{