fix: preserve response incomplete terminal semantics

Treat ordinary OpenAI Responses response.incomplete terminals as delivered incomplete outcomes instead of upstream failures. Keep error-coded incomplete terminals, stale continuation anchors, stream truncation, and client disconnects on the existing failure or retry paths. Align public /v1 streaming, websocket, non-stream collection, chat completions accounting, request logs, and docs around the shared classifier.

Author: masterkain

docs-site/src/content/docs/clients/openai-compatible.mdx

@@ -127,6 +127,8 @@ The `/v1` surface is compatibility over Codex routing, not a separate OpenAI eng
 
 `POST /v1/responses` lifts system and developer input-message text into top-level `instructions` before dispatching to Codex-compatible work. Streaming `/v1/responses` and `/v1/chat/completions` return early upstream terminal errors as the first public SSE event or data chunk, without synthetic assistant/success prefixes. Non-streaming failures remain OpenAI-shaped JSON errors.
 
+Ordinary OpenAI Responses incomplete terminals are preserved. If an upstream returns `response.incomplete` with `status: "incomplete"` for output limits or content filtering and no embedded error, `/v1/responses` streaming and websocket clients receive that incomplete terminal instead of a synthetic failure. Error-coded incomplete terminals, such as context overflow or stale continuation anchors, are still returned through the sanitized failure path.
+
 Accepted `POST /v1/responses` tool definitions are narrow. OpenAI Responses remote MCP tool definitions are rejected before dispatch.
 
 ## Request compression behavior

docs-site/src/content/docs/operators/request-logs.mdx

@@ -118,6 +118,8 @@ The table shows the latest matching rows, with a hard page size limit.
 
 Use status and errors together. A completed row with token usage usually means the request reached an upstream and settled successfully. A failed row may still show route, latency, or partial attempt metadata. A rejected row usually means admission or policy failed before upstream work.
 
+OpenAI Responses `response.incomplete` can be a successful delivered terminal response. For example, an upstream may stop at `max_output_tokens` or a content filter and still return safe usage metadata. Those rows stay `succeeded` and should not be treated as upstream health failures. Error-coded incomplete terminals, stale continuation anchors, stream truncation, and client disconnects still appear as failures or interruptions with sanitized error codes.
+
 Common causes to inspect are:
 
 1. Pool API key paused, revoked, expired, or attached to the wrong Pool
@@ -134,7 +136,7 @@ The Errors column should stay safe to copy into operator tickets. It should not
 
 Usage lines summarize request accounting, including request count context, token totals, cached input tokens, and estimated cost when pricing is available. Treat these as operational attribution and capacity evidence. They are not a replacement for provider billing records.
 
-If usage is blank, check the request status first. A request can be admitted without final usage if it failed before settlement, if pricing was not matched, or if accounting metadata was not available for that route.
+If usage is blank, check the request status first. A request can be admitted without final usage if it failed before settlement, if pricing was not matched, or if accounting metadata was not available for that route. A delivered incomplete response with missing upstream usage remains `usage_unknown` and does not get an invented settled cost from the reservation estimate.
 
 ## Operational Checklist
 

lib/codex_pooler/gateway/openai_compatibility/responses/sse.ex

@@ -9,7 +9,15 @@ defmodule CodexPooler.Gateway.OpenAICompatibility.Responses.SSE do
   def response_from_sse(body) when is_binary(body) do
     case Jason.decode(body) do
       {:ok, %{} = decoded} ->
-        {:ok, Map.put_new(decoded, "object", "response")}
+        response = Map.put_new(decoded, "object", "response")
+
+        case response_terminal_type(response) do
+          nil ->
+            {:ok, response}
+
+          type ->
+            terminal_error(%{"type" => type, "response" => response}, response) || {:ok, response}
+        end
 
       _error ->
         body |> decoded_sse_events() |> response_from_sse_events()
@@ -49,11 +57,22 @@ defmodule CodexPooler.Gateway.OpenAICompatibility.Responses.SSE do
     )
   end
 
-  defp terminal_error(_event, %{"status" => status})
-       when status in ["completed", "in_progress", "incomplete"],
-       do: nil
-
   defp terminal_error(event, response) do
+    event_type = event["type"] || response_terminal_type(response)
+
+    case StreamProtocol.terminal_outcome(event_type, event) do
+      {:ok, %{kind: kind}} when kind in [:completed, :incomplete] ->
+        nil
+
+      {:ok, %{kind: :failed, failure: failure}} ->
+        terminal_failure_error(event, response, failure)
+
+      _outcome ->
+        terminal_failure_error(event, response, nil)
+    end
+  end
+
+  defp terminal_failure_error(event, response, failure) do
     error = response["error"] || event["error"]
 
     case error do
@@ -70,10 +89,18 @@ defmodule CodexPooler.Gateway.OpenAICompatibility.Responses.SSE do
          )}
 
       _other ->
-        {:error, Error.reason(502, "upstream_error", "upstream response failed")}
+        {:error, Error.reason(502, failure_code(failure), "upstream request failed")}
     end
   end
 
+  defp failure_code(%{code: code}) when is_binary(code), do: code
+  defp failure_code(_failure), do: "upstream_error"
+
+  defp response_terminal_type(%{"status" => "completed"}), do: "response.completed"
+  defp response_terminal_type(%{"status" => "failed"}), do: "response.failed"
+  defp response_terminal_type(%{"status" => "incomplete"}), do: "response.incomplete"
+  defp response_terminal_type(_response), do: nil
+
   defp maybe_backfill_output(%{"output" => output} = response, _events)
        when is_list(output) and output != [],
        do: response

lib/codex_pooler/gateway/runtime/dispatch/websocket_attempt.ex

@@ -7,6 +7,7 @@ defmodule CodexPooler.Gateway.Runtime.Dispatch.WebsocketAttempt do
   alias CodexPooler.Gateway.Runtime.Dispatch.ResponseContext
   alias CodexPooler.Gateway.Runtime.Finalization
   alias CodexPooler.Gateway.Runtime.Finalization.{AttemptSettlement, Metadata}
+  alias CodexPooler.Gateway.Transports.Streaming.StreamProtocol
   alias CodexPooler.Gateway.Transports.Streaming.WebsocketCodec
   alias CodexPooler.Gateway.Transports.UpstreamDispatch
   alias CodexPooler.Gateway.Transports.UpstreamDispatch.Request, as: DispatchRequest
@@ -81,19 +82,19 @@ defmodule CodexPooler.Gateway.Runtime.Dispatch.WebsocketAttempt do
           failure
         )
 
-      {:ok, %{terminal: "response.completed"} = response} ->
-        Finalization.finalize_completed_websocket_response(
-          context,
+      {:ok, %{terminal: terminal} = response} ->
+        finalization =
           response
           |> Map.put(:started, started)
-          |> Map.put(:callbacks, callbacks)
-        )
+          |> maybe_put_websocket_callbacks(callbacks)
 
-      {:ok, response} ->
-        Finalization.finalize_terminal_websocket_response(
-          context,
-          Map.put(response, :started, started)
-        )
+        case websocket_terminal_outcome(terminal, Map.get(response, :body, "")) do
+          {:ok, %{kind: kind}} when kind in [:completed, :incomplete] ->
+            Finalization.finalize_completed_websocket_response(context, finalization)
+
+          _outcome ->
+            Finalization.finalize_terminal_websocket_response(context, finalization)
+        end
 
       {:error, response} ->
         Finalization.finalize_failed_websocket_response(
@@ -303,6 +304,19 @@ defmodule CodexPooler.Gateway.Runtime.Dispatch.WebsocketAttempt do
     end
   end
 
+  defp maybe_put_websocket_callbacks(%{terminal: terminal} = finalization, callbacks) do
+    case websocket_terminal_outcome(terminal, Map.get(finalization, :body, "")) do
+      {:ok, %{kind: kind}} when kind in [:completed, :incomplete] ->
+        Map.put(finalization, :callbacks, callbacks)
+
+      _outcome ->
+        finalization
+    end
+  end
+
+  defp websocket_terminal_outcome("response.completed", _body), do: {:ok, %{kind: :completed}}
+  defp websocket_terminal_outcome(_terminal, body), do: StreamProtocol.terminal_outcome(body)
+
   defp finalize_exhausted_auth_refresh(context, dispatch_request, response, failure, started) do
     case Map.get(response, :body, "") do
       "" ->

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

@@ -61,16 +61,19 @@ defmodule CodexPooler.Gateway.Runtime.Finalization.Websocket do
 
   @spec finalize_terminal(DispatchContext.t(), map()) :: {:ok, map()} | {:error, map()}
   def finalize_terminal(context, finalization) do
-    %{
-      body: body,
-      terminal: terminal,
-      status: status,
-      headers: headers,
-      started: started
-    } = finalization
+    %{body: body, terminal: terminal} = finalization
 
-    %{reserved: reserved, attempt: attempt, request_options: request_options} =
-      context
+    case websocket_terminal_outcome(terminal, body) do
+      {:ok, %{kind: kind}} when kind in [:completed, :incomplete] ->
+        finalize_completed(context, finalization)
+
+      _outcome ->
+        finalize_terminal_failure(context, finalization)
+    end
+  end
+
+  defp finalize_terminal_failure(context, finalization) do
+    %{body: body, terminal: terminal, headers: headers} = finalization
 
     upstream_code =
       Map.get(finalization, :upstream_error_code) ||
@@ -80,35 +83,67 @@ defmodule CodexPooler.Gateway.Runtime.Finalization.Websocket do
     websocket_frame_headers = Map.get(finalization, :websocket_frame_headers, %{})
     metadata_headers = headers ++ Map.to_list(websocket_frame_headers)
 
-    with :ok <- Streaming.record_terminal_health_failure(upstream_code, metadata_headers, context) do
-      case AttemptSettlement.finalize_partial_stream_failure(
-             reserved.request,
-             attempt,
-             ResponseUsage.from_websocket_body(body),
-             SettlementAttrs.partial_stream_failure(
-               context,
-               status,
-               code,
-               code,
-               metadata_headers
-               |> Metadata.websocket_response_metadata(
-                 code,
-                 request_options,
-                 websocket_frame_headers
-               )
-               |> Metadata.maybe_put_masked_error_metadata(upstream_code, code),
-               started: started
-             )
-           ) do
-        {:ok, _finalized} ->
-          {:ok, %{status: 200, headers: [], websocket_messages: []}}
-
-        {:error, gateway_error} ->
-          {:error, gateway_error}
-      end
+    attempt_metadata =
+      terminal_failure_metadata(
+        context,
+        metadata_headers,
+        websocket_frame_headers,
+        code,
+        upstream_code
+      )
+
+    case Streaming.record_terminal_health_failure(upstream_code, metadata_headers, context) do
+      :ok ->
+        settle_terminal_failure(context, finalization, body, code, attempt_metadata)
+
+      {:error, _gateway_error} = error ->
+        error
+    end
+  end
+
+  defp terminal_failure_metadata(
+         context,
+         metadata_headers,
+         websocket_frame_headers,
+         code,
+         upstream_code
+       ) do
+    metadata_headers
+    |> Metadata.websocket_response_metadata(
+      code,
+      context.request_options,
+      websocket_frame_headers
+    )
+    |> Metadata.maybe_put_masked_error_metadata(upstream_code, code)
+  end
+
+  defp settle_terminal_failure(context, finalization, body, code, attempt_metadata) do
+    %{reserved: reserved, attempt: attempt} = context
+
+    case AttemptSettlement.finalize_partial_stream_failure(
+           reserved.request,
+           attempt,
+           ResponseUsage.from_websocket_body(body),
+           SettlementAttrs.partial_stream_failure(
+             context,
+             finalization.status,
+             code,
+             code,
+             attempt_metadata,
+             started: finalization.started
+           )
+         ) do
+      {:ok, _finalized} ->
+        {:ok, %{status: 200, headers: [], websocket_messages: []}}
+
+      {:error, gateway_error} ->
+        {:error, gateway_error}
     end
   end
 
+  defp websocket_terminal_outcome("response.completed", _body), do: {:ok, %{kind: :completed}}
+  defp websocket_terminal_outcome(_terminal, body), do: StreamProtocol.terminal_outcome(body)
+
   @spec finalize_failed(DispatchContext.t(), map()) :: {:error, map()}
   def finalize_failed(context, %{reason: :client_disconnected} = finalization) do
     %{headers: headers, started: started} = finalization

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

@@ -35,9 +35,9 @@ defmodule CodexPooler.Gateway.Runtime.Streaming.StreamAttempt do
 
   defp classify_data_after_first_event(data) do
     classification =
-      case StreamProtocol.terminal_failure(data) do
-        {:ok, failure} -> {:write_terminal_failure, data, failure}
-        :error -> {:write, data}
+      case StreamProtocol.terminal_outcome(data) do
+        {:ok, %{kind: :failed, failure: failure}} -> {:write_terminal_failure, data, failure}
+        _outcome -> {:write, data}
       end
 
     {classification, %{classified?: true, buffer: ""}}
@@ -80,9 +80,9 @@ defmodule CodexPooler.Gateway.Runtime.Streaming.StreamAttempt do
     if StreamProtocol.internal_rate_limit_event?(event) do
       {:write, buffer}
     else
-      case StreamProtocol.terminal_failure_event(event) do
-        {:ok, failure} -> {:write_terminal_failure, buffer, failure}
-        nil -> {:write, buffer}
+      case StreamProtocol.terminal_outcome_event(event) do
+        {:ok, %{kind: :failed, failure: failure}} -> {:write_terminal_failure, buffer, failure}
+        _outcome -> {:write, buffer}
       end
     end
   end