fix(gateway): accept trailing terminal response SSE events

Recognize complete public /v1/responses terminal events even when the upstream closes without the final SSE blank-line separator, while keeping nonterminal partial buffers conservative.

Add metadata-only request-log diagnostics and regression coverage for trailing terminal stream settlement.

Author: masterkain

lib/codex_pooler/accounting/request_logs.ex

@@ -404,7 +404,7 @@ defmodule CodexPooler.Accounting.RequestLogs do
 
   @spec maybe_put_transport_failure(map(), Attempt.t()) :: map()
   defp maybe_put_transport_failure(projection, %Attempt{} = attempt) do
-    case transport_failure_projection(attempt.response_metadata) do
+    case transport_failure_projection(attempt) do
       transport_failure when map_size(transport_failure) > 0 ->
         Map.put(projection, :transport_failure, transport_failure)
 
@@ -413,9 +413,24 @@ defmodule CodexPooler.Accounting.RequestLogs do
     end
   end
 
-  @spec transport_failure_projection(term()) :: map()
-  defp transport_failure_projection(response_metadata) do
-    response_metadata
+  @spec transport_failure_projection(Attempt.t()) :: map()
+  defp transport_failure_projection(%Attempt{} = attempt) do
+    metadata = Accounting.sanitize_metadata(attempt.response_metadata)
+
+    metadata
+    |> persisted_transport_failure()
+    |> case do
+      transport_failure when map_size(transport_failure) > 0 ->
+        transport_failure
+
+      _transport_failure ->
+        stream_interruption_transport_failure(attempt, metadata)
+    end
+    |> safe_transport_failure()
+  end
+
+  defp persisted_transport_failure(metadata) do
+    metadata
     |> Accounting.sanitize_metadata()
     |> case do
       %{"transport_failure" => transport_failure} when is_map(transport_failure) ->
@@ -427,9 +442,96 @@ defmodule CodexPooler.Accounting.RequestLogs do
       _metadata ->
         %{}
     end
-    |> safe_transport_failure()
   end
 
+  defp stream_interruption_transport_failure(
+         %Attempt{} = attempt,
+         %{"error_kind" => "stream_interrupted"} = metadata
+       ) do
+    cond do
+      attempt.network_error_code == "client_disconnected" ->
+        %{
+          "reason_class" => "downstream_client_disconnect",
+          "reason" => "client_disconnected",
+          "phase" => "send_payload",
+          "pre_visible_output" => pre_visible_output?(metadata),
+          "terminal_seen" => false,
+          "text_frame_count" => stream_text_frame_count(metadata)
+        }
+
+      attempt.network_error_code == "downstream_stream_error" ->
+        %{
+          "reason_class" => "downstream_stream_error",
+          "reason" => "downstream_write_failed",
+          "phase" => "send_payload",
+          "pre_visible_output" => pre_visible_output?(metadata),
+          "terminal_seen" => false,
+          "text_frame_count" => stream_text_frame_count(metadata)
+        }
+
+      attempt.network_error_code == "stream_idle_timeout" ->
+        %{
+          "reason_class" => "upstream_stream_idle_timeout",
+          "reason" => "idle_timeout",
+          "phase" => "receive_timeout",
+          "pre_visible_output" => pre_visible_output?(metadata),
+          "terminal_seen" => false,
+          "text_frame_count" => stream_text_frame_count(metadata)
+        }
+
+      present?(terminal_stream_error_code(metadata)) ->
+        %{
+          "reason_class" => "upstream_terminal_failure",
+          "reason" => terminal_stream_error_code(metadata),
+          "phase" => "receive",
+          "pre_visible_output" => pre_visible_output?(metadata),
+          "terminal_seen" => true,
+          "text_frame_count" => stream_text_frame_count(metadata)
+        }
+
+      attempt.network_error_code == "upstream_stream_error" ->
+        %{
+          "reason_class" => "upstream_stream_interrupted",
+          "reason" => "closed_before_terminal",
+          "phase" => "upstream_close",
+          "pre_visible_output" => pre_visible_output?(metadata),
+          "terminal_seen" => false,
+          "text_frame_count" => stream_text_frame_count(metadata)
+        }
+
+      true ->
+        %{
+          "reason_class" => "stream_interrupted",
+          "reason" => "interrupted",
+          "phase" => "receive",
+          "pre_visible_output" => pre_visible_output?(metadata),
+          "terminal_seen" => false,
+          "text_frame_count" => stream_text_frame_count(metadata)
+        }
+    end
+  end
+
+  defp stream_interruption_transport_failure(_attempt, _metadata), do: %{}
+
+  defp terminal_stream_error_code(metadata) do
+    Map.get(metadata, "stream_error_code") || Map.get(metadata, "upstream_error_code")
+  end
+
+  defp pre_visible_output?(metadata), do: stream_text_frame_count(metadata) == 0
+
+  defp stream_text_frame_count(metadata) do
+    metadata
+    |> Map.get("stream_text_frame_count")
+    |> non_negative_integer()
+    |> case do
+      count when is_integer(count) -> count
+      nil -> default_stream_text_frame_count(metadata)
+    end
+  end
+
+  defp default_stream_text_frame_count(%{"error_kind" => "stream_interrupted"}), do: 1
+  defp default_stream_text_frame_count(_metadata), do: 0
+
   @spec safe_transport_failure(map()) :: map()
   defp safe_transport_failure(transport_failure) when is_map(transport_failure) do
     %{}

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

@@ -15,6 +15,16 @@ defmodule CodexPooler.Gateway.Transports.Streaming.StreamProtocol.PublicResponse
   @spec new_state() :: state()
   def new_state, do: %{buffer: "", created?: false, text_delta?: false, passthrough?: false}
 
+  @terminal_buffer_markers [
+    "data: [DONE]",
+    "response.completed",
+    "response.failed",
+    "response.incomplete",
+    "event: error",
+    ~s("type":"error"),
+    ~s("type": "error")
+  ]
+
   @spec normalize_data(binary(), state()) :: {binary(), state()}
   def normalize_data(data, %{passthrough?: true} = state) when is_binary(data) do
     normalize_passthrough_data(data, state)
@@ -25,6 +35,9 @@ defmodule CodexPooler.Gateway.Transports.Streaming.StreamProtocol.PublicResponse
     {blocks, buffer} = StreamProtocol.complete_sse_blocks(buffered_data, bounded?: false)
 
     cond do
+      terminal_buffer?(buffer) ->
+        normalize_blocks(blocks ++ [buffer], "", state)
+
       blocks == [] and StreamProtocol.oversized_incomplete_sse_block?(buffer) ->
         record_oversized_incomplete(byte_size(buffered_data))
         {buffered_data, %{state | buffer: "", passthrough?: true}}
@@ -267,12 +280,37 @@ defmodule CodexPooler.Gateway.Transports.Streaming.StreamProtocol.PublicResponse
 
   defp clean_string(_value), do: nil
 
+  @spec terminal_buffer?(binary()) :: boolean()
+  defp terminal_buffer?(""), do: false
+
+  defp terminal_buffer?(buffer) when is_binary(buffer) do
+    if terminal_buffer_candidate?(buffer) do
+      done_marker?(buffer) or decoded_terminal_buffer?(buffer)
+    else
+      false
+    end
+  end
+
+  @spec terminal_buffer_candidate?(binary()) :: boolean()
+  defp terminal_buffer_candidate?(buffer) do
+    Enum.any?(@terminal_buffer_markers, &String.contains?(buffer, &1))
+  end
+
+  @spec done_marker?(binary()) :: boolean()
+  defp done_marker?(block), do: String.trim(block) == "data: [DONE]"
+
+  @spec decoded_terminal_buffer?(binary()) :: boolean()
+  defp decoded_terminal_buffer?(buffer) do
+    {event_type, decoded} = stream_block_event(buffer)
+
+    decoded != %{} and terminal_event?(event_type || decoded_string(decoded, "type"))
+  end
+
   defp stream_terminal?(blocks) do
     Enum.any?(blocks, fn block ->
       {event_type, decoded} = stream_block_event(block)
 
-      terminal_event?(event_type || decoded_string(decoded, "type")) or
-        block == "data: [DONE]"
+      terminal_event?(event_type || decoded_string(decoded, "type")) or done_marker?(block)
     end)
   end
 

test/codex_pooler/accounting/request_logs_test.exs

@@ -1967,6 +1967,145 @@ defmodule CodexPooler.Accounting.RequestLogsTest do
     refute inspect(log.debug) =~ "Bearer sk-example-hidden"
   end
 
+  test "request log debug projection classifies interrupted HTTP SSE attempts" do
+    %{pool: pool, api_key: api_key} = active_api_key_fixture()
+    %{assignment: assignment} = upstream_assignment_fixture(pool)
+
+    request =
+      request_fixture(%{pool: pool, api_key: api_key}, %{
+        requested_model: "gpt-debug-http-sse-interrupted",
+        endpoint: "/backend-api/codex/responses",
+        transport: "http_sse",
+        status: "failed",
+        correlation_id: "debug-http-sse-interrupted",
+        response_status_code: 200,
+        request_metadata: %{
+          "codex_session_id" => "session-http-sse-interrupted",
+          "codex_session_key" => "session-key-http-sse-interrupted",
+          "request_body" => "raw prompt should stay hidden"
+        }
+      })
+      |> Ecto.Changeset.change(last_error_code: "upstream_stream_error")
+      |> Repo.update!()
+
+    attempt =
+      request
+      |> attempt_fixture(assignment, %{
+        status: "failed",
+        retryable: false,
+        upstream_status_code: 200,
+        usage_status: "usage_known"
+      })
+      |> Ecto.Changeset.change(%{
+        latency_ms: 654,
+        network_error_code: "upstream_stream_error",
+        error_message: "raw stream body must not enter debug",
+        response_metadata: %{
+          "error_kind" => "stream_interrupted",
+          "content_type" => "text/event-stream",
+          "status_code" => 200,
+          "raw_body" => "raw visible response body should stay hidden",
+          "raw_headers" => %{"authorization" => "Bearer sk-example-hidden"}
+        }
+      })
+      |> Repo.update!()
+
+    _turn =
+      debug_turn_fixture(pool, api_key, assignment, request, %{
+        status: "failed",
+        error_code: "upstream_stream_error",
+        final_attempt_id: attempt.id
+      })
+
+    assert %{items: [log], total: 1} = Accounting.list_request_logs(pool)
+    assert [attempt_debug] = log.debug.attempts
+
+    assert attempt_debug.transport_failure == %{
+             reason_class: "upstream_stream_interrupted",
+             reason: "closed_before_terminal",
+             phase: "upstream_close",
+             pre_visible_output: false,
+             terminal_seen: false,
+             text_frame_count: 1
+           }
+
+    refute inspect(log.debug) =~ "session-http-sse-interrupted"
+    refute inspect(log.debug) =~ "session-key-http-sse-interrupted"
+    refute inspect(log.debug) =~ "raw prompt should stay hidden"
+    refute inspect(log.debug) =~ "raw stream body"
+    refute inspect(log.debug) =~ "raw visible response body"
+    refute inspect(log.debug) =~ "Bearer sk-example-hidden"
+  end
+
+  test "request log debug projection classifies supported stream interruption families" do
+    %{pool: pool, api_key: api_key} = active_api_key_fixture()
+    %{assignment: assignment} = upstream_assignment_fixture(pool)
+
+    request =
+      request_fixture(%{pool: pool, api_key: api_key}, %{
+        requested_model: "gpt-debug-stream-families",
+        endpoint: "/backend-api/codex/responses",
+        transport: "http_sse",
+        status: "failed",
+        correlation_id: "debug-stream-families",
+        response_status_code: 200,
+        request_metadata: %{"codex_session_id" => "session-stream-families"}
+      })
+      |> Ecto.Changeset.change(last_error_code: "upstream_stream_error")
+      |> Repo.update!()
+
+    stream_family_attempt(request, assignment, 1, "client_disconnected", %{})
+
+    stream_family_attempt(request, assignment, 2, "upstream_stream_error", %{
+      "stream_error_code" => "server_error"
+    })
+
+    stream_family_attempt(request, assignment, 3, "stream_idle_timeout", %{})
+    stream_family_attempt(request, assignment, 4, "stream_parse_error", %{})
+
+    assert %{items: [log], total: 1} = Accounting.list_request_logs(pool)
+
+    attempts_by_number = Map.new(log.debug.attempts, &{&1.attempt_number, &1.transport_failure})
+
+    assert attempts_by_number[1] == %{
+             reason_class: "downstream_client_disconnect",
+             reason: "client_disconnected",
+             phase: "send_payload",
+             pre_visible_output: false,
+             terminal_seen: false,
+             text_frame_count: 1
+           }
+
+    assert attempts_by_number[2] == %{
+             reason_class: "upstream_terminal_failure",
+             reason: "server_error",
+             phase: "receive",
+             pre_visible_output: false,
+             terminal_seen: true,
+             text_frame_count: 1
+           }
+
+    assert attempts_by_number[3] == %{
+             reason_class: "upstream_stream_idle_timeout",
+             reason: "idle_timeout",
+             phase: "receive_timeout",
+             pre_visible_output: false,
+             terminal_seen: false,
+             text_frame_count: 1
+           }
+
+    assert attempts_by_number[4] == %{
+             reason_class: "stream_interrupted",
+             reason: "interrupted",
+             phase: "receive",
+             pre_visible_output: false,
+             terminal_seen: false,
+             text_frame_count: 1
+           }
+
+    refute inspect(log.debug) =~ "session-stream-families"
+  end
+
   test "request log debug projection reports failed request with linked open turn as mismatch" do
     %{pool: pool, api_key: api_key} = active_api_key_fixture()
     %{assignment: assignment} = upstream_assignment_fixture(pool)
@@ -2616,6 +2755,22 @@ defmodule CodexPooler.Accounting.RequestLogsTest do
     }
   end
 
+  defp stream_family_attempt(request, assignment, attempt_number, network_error_code, metadata) do
+    request
+    |> attempt_fixture(assignment, %{
+      attempt_number: attempt_number,
+      status: "failed",
+      retryable: false,
+      upstream_status_code: 200,
+      usage_status: "usage_unknown"
+    })
+    |> Ecto.Changeset.change(%{
+      network_error_code: network_error_code,
+      response_metadata: Map.merge(%{"error_kind" => "stream_interrupted"}, metadata)
+    })
+    |> Repo.update!()
+  end
+
   defp debug_turn_fixture(pool, api_key, assignment, request, attrs) do
     now = ~U[2026-05-26 00:00:00.000000Z]