fix(gateway): keep interrupted SSE streams health-neutral

Terminal-missing HTTP SSE upstream closes still finalize the request and attempt as upstream_stream_error, but no longer demote bridge health or open proxy_stream routing circuits.

This prevents interrupted 200 OK streams from exhausting otherwise eligible backends into no_eligible_backend while keeping idle timeouts and explicit upstream failures on the existing failure path.

Author: masterkain

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

@@ -199,6 +199,15 @@ defmodule CodexPooler.Gateway.Runtime.Finalization.Streaming do
     record_health_failure(code, code, context)
   end
 
+  defp record_stream_failure_health(
+         :upstream_stream_interrupted,
+         "upstream_stream_error",
+         nil,
+         _headers,
+         _context
+       ),
+       do: :ok
+
   defp record_stream_failure_health(reason, code, nil, _headers, context) do
     record_health_failure(reason, code, context)
   end

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

@@ -12,8 +12,10 @@ defmodule CodexPooler.Gateway.Runtime.Streaming.StreamLifecycleTest do
   alias CodexPooler.Accounting.{Attempt, Request}
   alias CodexPooler.FakeUpstream
   alias CodexPooler.Gateway.Payloads.RequestOptions
+  alias CodexPooler.Gateway.Persistence.{BridgeDemotion, RoutingCircuitState}
   alias CodexPooler.Gateway.Routing.{BridgeRing, RoutePlanInput}
   alias CodexPooler.Gateway.Runtime.Dispatch.{Context, ResponseContext}
+  alias CodexPooler.Gateway.Runtime.Finalization.Streaming
   alias CodexPooler.Gateway.Runtime.Streaming.OpenAIStreamCollector
   alias CodexPooler.Gateway.Runtime.Streaming.StreamLifecycle
   alias CodexPooler.Gateway.Service
@@ -248,6 +250,56 @@ defmodule CodexPooler.Gateway.Runtime.Streaming.StreamLifecycleTest do
     assert_pre_first_stall_finalized!(setup, "partial frame stall")
   end
 
+  test "terminal-missing upstream SSE close fails request without poisoning route health" do
+    {setup, _first_upstream, _second_upstream} =
+      stream_retry_setup(
+        FakeUpstream.sse_stream([]),
+        FakeUpstream.sse_stream([])
+      )
+
+    {:ok, auth} = Access.authenticate_authorization_header(setup.authorization)
+    payload = payload(setup)
+    request_options = request_options(auth, payload, setup)
+
+    assert {:ok, reserved} =
+             Accounting.reserve(auth, setup.model, payload, %{
+               endpoint: @endpoint_path,
+               transport: "http_sse",
+               correlation_id:
+                 "upstream-stream-interrupted-#{System.unique_integer([:positive])}",
+               request_metadata: %{}
+             })
+
+    assert {:ok, attempt} = Accounting.create_attempt(reserved.request, setup.assignment)
+
+    context =
+      retry_context(setup, auth, request_options, reserved.request,
+        candidates: [{setup.assignment, setup.identity}],
+        attempt: attempt
+      )
+
+    response_context = %ResponseContext{context: context, response: %Req.Response{status: 200}}
+
+    assert {:ok, _finalized} =
+             Streaming.finalize_failure(
+               "event: response.created\n\n",
+               :upstream_stream_interrupted,
+               response_context
+             )
+
+    assert [request] = Repo.all(from(r in Request, where: r.id == ^reserved.request.id))
+    assert request.status == "failed"
+    assert request.last_error_code == "upstream_stream_error"
+
+    assert [attempt] = Repo.all(from(a in Attempt, where: a.request_id == ^request.id))
+    assert attempt.status == "failed"
+    assert attempt.network_error_code == "upstream_stream_error"
+    assert attempt.response_metadata["error_kind"] == "stream_interrupted"
+
+    assert Repo.all(from(d in BridgeDemotion)) == []
+    assert Repo.all(from(c in RoutingCircuitState)) == []
+  end
+
   defp retry_context(setup, auth, request_options, request, opts \\ []) do
     candidates =
       Keyword.get(opts, :candidates, [

test/codex_pooler_web/controllers/runtime/backend_codex_controller_test.exs

@@ -1272,6 +1272,53 @@ defmodule CodexPoolerWeb.Runtime.BackendCodexControllerTest do
     assert_turn_state_not_persisted!(setup, response_turn_state)
   end
 
+  test "POST /v1/responses terminal-missing SSE close does not poison route health",
+       %{conn: conn} do
+    upstream =
+      start_upstream(
+        FakeUpstream.sse_stream(
+          [
+            {"response.created",
+             %{
+               "type" => "response.created",
+               "response" => %{"id" => "resp_public_terminal_missing"}
+             }}
+          ],
+          done: false
+        )
+      )
+
+    setup = gateway_setup(upstream)
+
+    conn =
+      conn
+      |> auth(setup)
+      |> post("/v1/responses", %{
+        "model" => setup.model.exposed_model_id,
+        "input" => "synthetic public terminal-missing stream request",
+        "stream" => true
+      })
+
+    assert conn.status == 200
+    assert conn.resp_body =~ "event: response.failed\n"
+    assert conn.resp_body =~ ~s("code":"upstream_stream_error")
+
+    assert [request] = Repo.all(from(r in Request, where: r.pool_id == ^setup.pool.id))
+    assert request.endpoint == "/backend-api/codex/responses"
+    assert request.transport == "http_sse"
+    assert request.status == "failed"
+    assert request.last_error_code == "upstream_stream_error"
+
+    assert [attempt] = Repo.all(from(a in Attempt, where: a.request_id == ^request.id))
+    assert attempt.status == "failed"
+    assert attempt.upstream_status_code == 200
+    assert attempt.network_error_code == "upstream_stream_error"
+    assert attempt.response_metadata["error_kind"] == "stream_interrupted"
+
+    assert Repo.all(from(d in BridgeDemotion)) == []
+    assert Repo.all(from(c in RoutingCircuitState)) == []
+  end
+
   test "POST /backend-api/codex/responses sends trusted Responses Lite marker from selected model metadata",
        %{conn: conn} do
     upstream =