test: cover safety-buffering stream metadata

Author: masterkain

docs-site/src/content/docs/reference/runtime-routes.mdx

@@ -211,6 +211,7 @@ account helpers, analytics posting, thread-goal helpers, memory summaries,
 search helpers, realtime helper calls, safety helper calls, identity JWKS, or
 reset-credit consume operations. Configure Codex clients by pointing
 `model_providers.*.base_url` at `/backend-api/codex`.
+Supported Responses stream event metadata is relayed on model-provider streams as upstream sends it. Codex Pooler does not synthesize app-server notifications such as `turn/safetyBuffering/updated`.
 
 ## Backend bridge and usage routes
 

test/codex_pooler/gateway/request_compression/token_counter_test.exs

@@ -31,6 +31,23 @@ defmodule CodexPooler.Gateway.RequestCompression.TokenCounterTest do
       assert {:ok, 51, %{encoding: "cl100k_base"}} = TokenCounter.count("gpt-4-turbo", output)
     end
 
+    test "counts literal special-token-looking strings as ordinary text" do
+      content = """
+      tool output can contain <|endoftext|>
+      diff markers can contain <|fim_prefix|>, <|fim_middle|>, and <|fim_suffix|>
+      """
+
+      assert {:ok, o200k_count, %{encoding: "o200k_base"}} =
+               TokenCounter.count("gpt-4o", content)
+
+      assert o200k_count > 0
+
+      assert {:ok, cl100k_count, %{encoding: "cl100k_base"}} =
+               TokenCounter.count("gpt-4-turbo", content)
+
+      assert cl100k_count > 0
+    end
+
     test "matches oracle counts for unicode and diff-like content" do
       unicode = "unicode sample: ciao mondo, こんにちは, Привет, مرحبا, 😀"
 

test/codex_pooler/gateway/transports/stream_protocol_test.exs

@@ -93,6 +93,35 @@ defmodule CodexPooler.Gateway.Transports.Streaming.StreamProtocolTest do
                |> StreamProtocol.decode_sse_data()
     end
 
+    test "preserves safety-buffering metadata on public Responses stream events" do
+      state = StreamProtocol.public_openai_responses_stream_state()
+
+      event =
+        sse_event("response.output_text.delta", %{
+          "type" => "response.output_text.delta",
+          "delta" => "visible synthetic safety-buffered text",
+          "safety_buffering" => %{
+            "model" => "safety-buffering-model-sentinel",
+            "use_cases" => ["cyber"],
+            "reasons" => ["user-risk-sentinel"]
+          }
+        })
+
+      assert {chunk, _state} =
+               StreamProtocol.normalize_public_openai_responses_sse_data(event, state)
+
+      assert [%{"event" => "response.output_text.delta", "data" => data}] =
+               public_sse_events(chunk)
+
+      assert data["delta"] == "visible synthetic safety-buffered text"
+
+      assert data["safety_buffering"] == %{
+               "model" => "safety-buffering-model-sentinel",
+               "use_cases" => ["cyber"],
+               "reasons" => ["user-risk-sentinel"]
+             }
+    end
+
     test "synthesizes missing reasoning and message output item ids" do
       state = StreamProtocol.public_openai_responses_stream_state()
 

test/codex_pooler/gateway/transports/websocket_codec_test.exs

@@ -141,6 +141,37 @@ defmodule CodexPooler.Gateway.Transports.Streaming.WebsocketCodecTest do
       refute Process.get({:websocket_sse_buffer, request_id})
     end
 
+    test "preserves safety-buffering metadata from upstream SSE frames" do
+      request_id = "websocket-safety-buffering"
+
+      sse =
+        "event: response.output_text.delta\n" <>
+          "data: " <>
+          Jason.encode!(%{
+            "type" => "response.output_text.delta",
+            "delta" => "visible synthetic safety-buffered text",
+            "safety_buffering" => %{
+              "model" => "safety-buffering-model-sentinel",
+              "use_cases" => ["cyber"],
+              "reasons" => ["user-risk-sentinel"]
+            }
+          }) <>
+          "\n\n"
+
+      assert {[message], ""} = WebsocketCodec.stream_messages(request_id, sse, "")
+
+      assert %{
+               "type" => "response.output_text.delta",
+               "safety_buffering" => safety_buffering
+             } = Jason.decode!(message)
+
+      assert safety_buffering == %{
+               "model" => "safety-buffering-model-sentinel",
+               "use_cases" => ["cyber"],
+               "reasons" => ["user-risk-sentinel"]
+             }
+    end
+
     test "drops oversized incomplete SSE buffers instead of retaining them" do
       attach_stream_buffer_telemetry()
       request_id = "websocket-buffer-oversized"

test/codex_pooler_web/controllers/runtime/backend_codex_controller_test.exs

@@ -6615,6 +6615,62 @@ defmodule CodexPoolerWeb.Runtime.BackendCodexControllerTest do
     assert_turn_state_not_persisted!(setup, response_turn_state)
   end
 
+  test "POST /backend-api/codex/responses relays stream safety-buffering metadata without persisting it",
+       %{conn: conn} do
+    safety_buffering = %{
+      "model" => "safety-buffering-model-sentinel",
+      "use_cases" => ["cyber"],
+      "reasons" => ["user-risk-sentinel"]
+    }
+
+    upstream =
+      start_upstream(
+        FakeUpstream.sse_stream([
+          {"response.output_text.delta",
+           %{
+             "type" => "response.output_text.delta",
+             "delta" => "visible synthetic safety-buffered text",
+             "safety_buffering" => safety_buffering
+           }},
+          {"response.completed",
+           %{
+             "type" => "response.completed",
+             "response" => %{
+               "id" => "resp_backend_stream_safety_buffering",
+               "usage" => %{"input_tokens" => 4, "output_tokens" => 3, "total_tokens" => 7}
+             }
+           }}
+        ])
+      )
+
+    setup = gateway_setup(upstream)
+
+    conn =
+      conn
+      |> auth(setup)
+      |> post("/backend-api/codex/responses", %{
+        "model" => setup.model.exposed_model_id,
+        "input" => "synthetic streaming safety-buffering relay request",
+        "stream" => true
+      })
+
+    assert conn.resp_body =~ "event: response.output_text.delta\n"
+    assert conn.resp_body =~ ~s("safety_buffering":)
+    assert conn.resp_body =~ "safety-buffering-model-sentinel"
+    assert conn.resp_body =~ "user-risk-sentinel"
+    assert conn.resp_body =~ "data: [DONE]\n\n"
+
+    assert [request] = Repo.all(from(r in Request, where: r.pool_id == ^setup.pool.id))
+    assert request.status == "succeeded"
+
+    assert [attempt] = Repo.all(from(a in Attempt, where: a.request_id == ^request.id))
+    assert attempt.status == "succeeded"
+
+    metadata_text = inspect({request.request_metadata, attempt.response_metadata})
+    refute metadata_text =~ "safety-buffering-model-sentinel"
+    refute metadata_text =~ "user-risk-sentinel"
+  end
+
   test "POST /backend-api/codex/responses/compact keeps opencode continuity headers local without forwarding",
        %{
          conn: conn