fix: preserve request option values on invalid updates

Author: masterkain

lib/codex_pooler/gateway/payloads/request_options.ex

@@ -624,8 +624,12 @@ defmodule CodexPooler.Gateway.Payloads.RequestOptions do
   defp maybe_put_websocket_owner_update(updates, key, value, normalizer \\ & &1)
   defp maybe_put_websocket_owner_update(updates, _key, nil, _normalizer), do: updates
 
-  defp maybe_put_websocket_owner_update(updates, key, value, normalizer),
-    do: Map.put(updates, key, normalizer.(value))
+  defp maybe_put_websocket_owner_update(updates, key, value, normalizer) do
+    case normalizer.(value) do
+      nil -> updates
+      value -> Map.put(updates, key, value)
+    end
+  end
 
   defp websocket_owner_forwarder_opts(values) when is_list(values), do: values
   defp websocket_owner_forwarder_opts(_value), do: []
@@ -680,6 +684,13 @@ defmodule CodexPooler.Gateway.Payloads.RequestOptions do
 
   defp forwarded_headers(_headers), do: []
 
+  defp forwarded_headers_update(headers) do
+    case forwarded_headers(headers) do
+      [] -> nil
+      headers -> headers
+    end
+  end
+
   defp safe_client_request_id(value) when is_binary(value) do
     value
     |> String.trim()
@@ -772,9 +783,9 @@ defmodule CodexPooler.Gateway.Payloads.RequestOptions do
 
   defp normalize_continuity_updates(updates) do
     updates
-    |> normalize_update(:bridge_owner_lease_ttl_seconds, &optional_positive_integer/1)
-    |> normalize_update(:reconnect_window_seconds, &optional_non_negative_integer/1)
-    |> normalize_update(:session_header_source, &normalized_session_header_source/1)
+    |> normalize_optional_update(:bridge_owner_lease_ttl_seconds, &optional_positive_integer/1)
+    |> normalize_optional_update(:reconnect_window_seconds, &optional_non_negative_integer/1)
+    |> normalize_optional_update(:session_header_source, &normalized_session_header_source/1)
   end
 
   defp normalize_transport_updates(updates, %WebsocketOwnerContext{} = current_owner) do
@@ -792,7 +803,11 @@ defmodule CodexPooler.Gateway.Payloads.RequestOptions do
       ])
 
     transport_updates =
-      normalize_update(transport_updates, :forwarded_metadata_headers, &forwarded_headers/1)
+      normalize_optional_update(
+        transport_updates,
+        :forwarded_metadata_headers,
+        &forwarded_headers_update/1
+      )
 
     if map_size(owner_updates) == 0 do
       transport_updates
@@ -807,22 +822,25 @@ defmodule CodexPooler.Gateway.Payloads.RequestOptions do
 
   defp normalize_file_bridge_updates(updates) do
     updates
-    |> normalize_update(:forwarded_headers, &forwarded_headers/1)
-    |> normalize_update(:finalize_retry_timeout_ms, &optional_non_negative_integer/1)
-    |> normalize_update(:finalize_retry_interval_ms, &optional_non_negative_integer/1)
+    |> normalize_optional_update(:forwarded_headers, &forwarded_headers_update/1)
+    |> normalize_optional_update(:finalize_retry_timeout_ms, &optional_non_negative_integer/1)
+    |> normalize_optional_update(:finalize_retry_interval_ms, &optional_non_negative_integer/1)
   end
 
   defp normalize_runtime_updates(updates) do
-    normalize_update(
+    normalize_optional_update(
       updates,
       :payload_compression,
       &RequestCompressionMetadata.runtime_metadata/1
     )
   end
 
-  defp normalize_update(updates, key, normalizer) do
+  defp normalize_optional_update(updates, key, normalizer) do
     if Map.has_key?(updates, key) do
-      Map.update!(updates, key, normalizer)
+      case normalizer.(Map.fetch!(updates, key)) do
+        nil -> Map.delete(updates, key)
+        value -> Map.put(updates, key, value)
+      end
     else
       updates
     end

test/codex_pooler/gateway/payloads/request_options_test.exs

@@ -1006,6 +1006,99 @@ defmodule CodexPooler.Gateway.Payloads.RequestOptionsTest do
       assert options.file_bridge.pool_upstream_assignment_id == "assignment-id"
     end
 
+    test "optional normalizers ignore invalid updates instead of clearing existing values" do
+      options =
+        RequestOptions.build(
+          %{
+            session_header_source: "session-id",
+            forwarded_headers: [{"x-codex-client", "fixture"}],
+            finalize_retry_timeout_ms: 500,
+            payload_compression: %{"attempted" => true, "status" => "compressed"},
+            websocket_owner_forwarding_enabled?: true,
+            websocket_owner_downstream: %{pid: self()},
+            websocket_owner_downstream_epoch: 3
+          },
+          "/backend-api/codex/responses",
+          %{"model" => "example-model"}
+        )
+
+      updated =
+        options
+        |> RequestOptions.put_continuity(
+          session_header_source: "x-unsafe-header",
+          reconnect_window_seconds: -1
+        )
+        |> RequestOptions.put_transport(
+          forwarded_metadata_headers: [{"x-codex-client", :invalid}],
+          websocket_owner_downstream_epoch: "stale"
+        )
+        |> RequestOptions.put_file_bridge(
+          forwarded_headers: [{"x-codex-client", :invalid}],
+          finalize_retry_timeout_ms: -1
+        )
+        |> RequestOptions.put_runtime_context(
+          payload_compression: %{"enabled" => true, "attempted" => false, "status" => "disabled"}
+        )
+
+      assert updated.continuity.session_header_source == "session-id"
+      assert updated.continuity.reconnect_window_seconds == nil
+      assert updated.transport.forwarded_metadata_headers == [{"x-codex-client", "fixture"}]
+      assert updated.transport.websocket_owner.downstream_epoch == 3
+      assert updated.file_bridge.forwarded_headers == [{"x-codex-client", "fixture"}]
+      assert updated.file_bridge.finalize_retry_timeout_ms == 500
+
+      assert updated.runtime.payload_compression == %{
+               "attempted" => true,
+               "status" => "compressed"
+             }
+    end
+
+    test "optional normalizers still accept explicit valid replacements" do
+      options =
+        RequestOptions.build(
+          %{
+            session_header_source: "session-id",
+            forwarded_headers: [{"x-codex-client", "fixture"}],
+            finalize_retry_timeout_ms: 500,
+            payload_compression: %{"attempted" => true, "status" => "compressed"},
+            websocket_owner_forwarding_enabled?: true,
+            websocket_owner_downstream_epoch: 3
+          },
+          "/backend-api/codex/responses",
+          %{"model" => "example-model"}
+        )
+
+      updated =
+        options
+        |> RequestOptions.put_continuity(session_header_source: "x-session-id")
+        |> RequestOptions.put_transport(
+          forwarded_metadata_headers: [{"x-openai-client-user-agent", "synthetic"}],
+          websocket_owner_downstream_epoch: 4
+        )
+        |> RequestOptions.put_file_bridge(
+          forwarded_headers: [{"user-agent", "codex_cli_rs/0.0.0"}],
+          finalize_retry_timeout_ms: 0
+        )
+        |> RequestOptions.put_runtime_context(
+          payload_compression: %{"attempted" => true, "status" => "no_change"}
+        )
+
+      assert updated.continuity.session_header_source == "x-session-id"
+
+      assert updated.transport.forwarded_metadata_headers == [
+               {"x-openai-client-user-agent", "synthetic"}
+             ]
+
+      assert updated.transport.websocket_owner.downstream_epoch == 4
+      assert updated.file_bridge.forwarded_headers == [{"user-agent", "codex_cli_rs/0.0.0"}]
+      assert updated.file_bridge.finalize_retry_timeout_ms == 0
+
+      assert updated.runtime.payload_compression == %{
+               "attempted" => true,
+               "status" => "no_change"
+             }
+    end
+
     test "reject unknown section fields" do
       options =
         RequestOptions.build(%{}, "/backend-api/codex/responses", %{"model" => "example-model"})