fix(gateway): persist safe public responses stream summaries

Record bounded public Responses stream summaries through relay finalization and accounting metadata so local OpenCode /v1/responses requests can be correlated without storing raw stream content.

Harden the accounting sanitizer for public_openai_responses_stream by allowlisting map fields, finite terminal labels, and malformed non-map values.

Verification: mix precommit (3003 tests); mix quality; focused accounting metadata suite (8 tests); localhost:4000 OpenCode smoke evidence.

Author: masterkain

lib/codex_pooler/accounting/metadata.ex

@@ -14,6 +14,30 @@ defmodule CodexPooler.Accounting.Metadata do
   @usage_not_applicable "not_applicable"
   @redacted "[REDACTED]"
   @sensitive_key_fragments ~w(api_key apikey authorization bearer token access_token refresh_token upstream_token upstream_secret cookie set-cookie secret password prompt messages input output completion content raw_request raw_response body payload file filename audio image transcript transcription upload_url download_url sas_url signed_url auth_json chatgpt_account_id)
+  @public_openai_responses_stream_modes ~w(normalized passthrough)
+  @public_openai_responses_stream_terminal_values ~w(completed failed incomplete)
+  @public_openai_responses_stream_boolean_keys ~w(
+    created_seen
+    visible_seen
+    terminal_seen
+    synthetic_terminal_sent
+    passthrough_seen
+  )
+  @public_openai_responses_stream_counter_keys ~w(
+    delta_count
+    delta_bytes
+    text_done_count
+    text_done_bytes
+    item_done_count
+    source_chunk_count
+    stream_bytes
+    relay_bytes
+  )
+  @public_openai_responses_stream_terminal_keys ~w(
+    finish_class
+    terminal_kind
+    terminal_status
+  )
   @sensitive_exact_keys MapSet.new([
                           "analytics",
                           "arc",
@@ -297,6 +321,9 @@ defmodule CodexPooler.Accounting.Metadata do
       normalized == "payload_compression" ->
         sanitize_payload_compression_map(value)
 
+      normalized == "public_openai_responses_stream" ->
+        sanitize_public_openai_responses_stream_map(value)
+
       sensitive_key?(normalized) ->
         @redacted
 
@@ -311,26 +338,67 @@ defmodule CodexPooler.Accounting.Metadata do
   end
 
   defp sanitize_value(value, key) when is_list(value) do
-    if sensitive_key?(key) do
-      @redacted
-    else
-      Enum.map(value, &sanitize_value(&1, key))
+    cond do
+      public_openai_responses_stream_key?(key) -> %{}
+      sensitive_key?(key) -> @redacted
+      true -> Enum.map(value, &sanitize_value(&1, key))
     end
   end
 
   defp sanitize_value(value, key) when is_binary(value) do
     cond do
+      public_openai_responses_stream_key?(key) -> %{}
       sensitive_key?(key) -> @redacted
       sensitive_binary?(value) -> @redacted
       true -> value
     end
   end
 
-  defp sanitize_value(value, _key), do: value
+  defp sanitize_value(value, key) do
+    if public_openai_responses_stream_key?(key), do: %{}, else: value
+  end
 
   defp sanitize_payload_compression_map(value) when is_map(value),
     do: RequestCompressionMetadata.sanitize_map(value)
 
+  defp sanitize_public_openai_responses_stream_map(value) when is_map(value) do
+    Enum.reduce(value, %{}, fn {child_key, child_value}, summary ->
+      key = normalize_key(child_key)
+
+      case sanitize_public_openai_responses_stream_field(key, child_value) do
+        :drop -> summary
+        sanitized_value -> Map.put(summary, key, sanitized_value)
+      end
+    end)
+  end
+
+  defp sanitize_public_openai_responses_stream_field("schema_version", value)
+       when is_integer(value) and value > 0,
+       do: value
+
+  defp sanitize_public_openai_responses_stream_field("mode", value)
+       when value in @public_openai_responses_stream_modes,
+       do: value
+
+  defp sanitize_public_openai_responses_stream_field(key, value)
+       when key in @public_openai_responses_stream_boolean_keys and is_boolean(value),
+       do: value
+
+  defp sanitize_public_openai_responses_stream_field(key, value)
+       when key in @public_openai_responses_stream_counter_keys and is_integer(value) and
+              value >= 0,
+       do: value
+
+  defp sanitize_public_openai_responses_stream_field(key, value)
+       when key in @public_openai_responses_stream_terminal_keys do
+    if value in @public_openai_responses_stream_terminal_values, do: value, else: nil
+  end
+
+  defp sanitize_public_openai_responses_stream_field(_key, _value), do: :drop
+
+  defp public_openai_responses_stream_key?(key),
+    do: normalize_key(key) == "public_openai_responses_stream"
+
   defp sensitive_binary?(value) do
     String.match?(value, ~r/sk-cxp-[a-f0-9]{12}-[A-Za-z0-9_-]+/) or
       String.match?(value, ~r/(?i)bearer\s+[A-Za-z0-9._~+\/-]+=*/) or

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

@@ -194,6 +194,12 @@ defmodule CodexPooler.Gateway.Runtime.Finalization do
     to: Streaming,
     as: :finalize_success
 
+  @spec finalize_stream_success(binary(), ResponseContext.t(), callbacks(), term()) ::
+          stream_finalization_result()
+  defdelegate finalize_stream_success(body, response_context, callbacks, stream_state),
+    to: Streaming,
+    as: :finalize_success
+
   @spec record_retryable_first_event_stream_failure(
           binary(),
           stream_failure(),
@@ -221,6 +227,12 @@ defmodule CodexPooler.Gateway.Runtime.Finalization do
     to: Streaming,
     as: :finalize_failure
 
+  @spec finalize_stream_failure(binary(), term(), ResponseContext.t(), term()) ::
+          stream_finalization_result()
+  defdelegate finalize_stream_failure(body, reason, response_context, stream_state),
+    to: Streaming,
+    as: :finalize_failure
+
   @spec stream_error_code(term()) :: String.t()
   defdelegate stream_error_code(reason), to: Streaming, as: :error_code
 

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

@@ -3,6 +3,7 @@ defmodule CodexPooler.Gateway.Runtime.Finalization.Metadata do
 
   alias CodexPooler.Gateway.Payloads.DebugPayloadSummary
   alias CodexPooler.Gateway.Payloads.RequestOptions
+  alias CodexPooler.Gateway.Runtime.Streaming.DownstreamStream
   alias CodexPooler.Gateway.Transports.BoundedResponseBody
   alias CodexPooler.Quotas.Evidence.CodexParsers.RateLimitReachedType
 
@@ -11,6 +12,27 @@ defmodule CodexPooler.Gateway.Runtime.Finalization.Metadata do
     "/backend-api/codex/responses/compact"
   ]
 
+  @public_openai_responses_stream_keys ~w(
+    schema_version
+    mode
+    created_seen
+    visible_seen
+    delta_count
+    delta_bytes
+    text_done_count
+    text_done_bytes
+    item_done_count
+    terminal_seen
+    terminal_kind
+    terminal_status
+    finish_class
+    synthetic_terminal_sent
+    source_chunk_count
+    stream_bytes
+    relay_bytes
+    passthrough_seen
+  )
+
   @spec response_metadata(Req.Response.t(), String.t() | nil, RequestOptions.t() | map()) ::
           map()
   def response_metadata(response, error_kind, opts) do
@@ -80,6 +102,13 @@ defmodule CodexPooler.Gateway.Runtime.Finalization.Metadata do
     |> Map.put("stream_error_code", failure.code)
   end
 
+  @spec merge_stream_state_metadata(map(), term()) :: map()
+  def merge_stream_state_metadata(metadata, state) when is_map(metadata) do
+    Map.merge(metadata, public_openai_responses_stream_metadata(state))
+  end
+
+  def merge_stream_state_metadata(metadata, _state), do: metadata
+
   @spec maybe_put_masked_error_metadata(map(), String.t() | nil, String.t()) :: map()
   def maybe_put_masked_error_metadata(metadata, upstream_code, code)
       when is_binary(upstream_code) and upstream_code != code do
@@ -177,6 +206,19 @@ defmodule CodexPooler.Gateway.Runtime.Finalization.Metadata do
 
   defp maybe_put_websocket_frame_headers(metadata, _headers), do: metadata
 
+  defp public_openai_responses_stream_metadata(state) do
+    case DownstreamStream.public_openai_responses_stream_metadata(state) do
+      %{"public_openai_responses_stream" => summary} when is_map(summary) ->
+        %{
+          "public_openai_responses_stream" =>
+            Map.take(summary, @public_openai_responses_stream_keys)
+        }
+
+      _metadata ->
+        %{}
+    end
+  end
+
   @spec maybe_put_backend_turn_state_response_header(
           [{String.t(), String.t()}],
           Req.Response.t(),

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

@@ -28,7 +28,14 @@ defmodule CodexPooler.Gateway.Runtime.Finalization.Streaming do
 
   @spec finalize_success(binary(), ResponseContext.t(), callbacks()) ::
           finalization_result()
-  def finalize_success(body, %ResponseContext{context: context, response: response}, callbacks) do
+  @spec finalize_success(binary(), ResponseContext.t(), callbacks(), term()) ::
+          finalization_result()
+  def finalize_success(
+        body,
+        %ResponseContext{context: context, response: response},
+        callbacks,
+        stream_state \\ nil
+      ) do
     %{
       reserved: reserved,
       attempt: attempt,
@@ -44,7 +51,9 @@ defmodule CodexPooler.Gateway.Runtime.Finalization.Streaming do
            SettlementAttrs.success(
              context,
              response.status,
-             Metadata.response_metadata(response, nil, request_options),
+             response
+             |> Metadata.response_metadata(nil, request_options)
+             |> Metadata.merge_stream_state_metadata(stream_state),
              started: started
            )
          ) do
@@ -135,7 +144,13 @@ defmodule CodexPooler.Gateway.Runtime.Finalization.Streaming do
   end
 
   @spec finalize_failure(binary(), term(), ResponseContext.t()) :: finalization_result()
-  def finalize_failure(body, reason, %ResponseContext{context: context, response: response}) do
+  @spec finalize_failure(binary(), term(), ResponseContext.t(), term()) :: finalization_result()
+  def finalize_failure(
+        body,
+        reason,
+        %ResponseContext{context: context, response: response},
+        stream_state \\ nil
+      ) do
     code = error_code(reason)
     terminal_failure = terminal_failure_reason(reason)
 
@@ -144,6 +159,7 @@ defmodule CodexPooler.Gateway.Runtime.Finalization.Streaming do
       attempt_metadata =
         response
         |> Metadata.response_metadata("stream_interrupted", context.request_options)
+        |> Metadata.merge_stream_state_metadata(stream_state)
         |> Metadata.maybe_put_masked_error_metadata(
           terminal_failure && terminal_failure.upstream_code,
           code