Preserve namespaced Codex tools across external model calls

apps/desktop/assets/codex-bridge/test_tool_namespaces.py

@@ -0,0 +1,66 @@
+"""Namespace routing regressions: leaf-only calls cannot dispatch collaboration tools."""
+import copy
+import json
+import unittest
+from test_model_switch import env
+
+
+class ToolNamespaces(unittest.TestCase):
+    def request(self):
+        return {"tools": [
+            {"type": "namespace", "name": "collaboration", "tools": [{"type": "function", "name": "spawn_agent", "parameters": {"type": "object", "properties": {}}}]},
+            {"type": "namespace", "name": "other", "tools": [{"type": "function", "name": "spawn_agent"}]},
+            {"type": "namespace", "name": "functions", "tools": [{"type": "custom", "name": "apply_patch"}]},
+            {"type": "function", "name": "plain"}],
+            "input": [{"type": "function_call", "namespace": "collaboration", "name": "spawn_agent", "id": "fc_a", "call_id": "call_a", "arguments": "{}"},
+                      {"type": "function_call_output", "call_id": "call_a", "output": "child_123"}]}
+
+    def test_namespace_calls_round_trip_at_all_stream_stages(self):
+        req = self.request(); original = copy.deepcopy(req); names = {}
+        custom = env["flatten_request"](req, names)
+        alias = req["tools"][0]["name"]
+        self.assertNotEqual(alias, req["tools"][1]["name"])
+        self.assertLessEqual(len(alias), 64)
+        self.assertEqual(req["input"][0]["name"], alias)
+        self.assertNotIn("namespace", req["input"][0])
+        self.assertEqual(req["input"][1], original["input"][1])
+        adapter = env["SseRewriter"](custom, tool_names=names)
+        item = {"type": "function_call", "name": alias, "id": "fc_test", "call_id": "call_test", "arguments": "{}"}
+        for stage in ["response.output_item.added", "response.output_item.done"]:
+            restored = adapter._fix({"type": stage, "item": dict(item)})["item"]
+            self.assertEqual((restored["namespace"], restored["name"]), ("collaboration", "spawn_agent"))
+            self.assertEqual(restored["call_id"], "call_test")
+        restored = adapter._fix({"type": "response.completed", "response": {"output": [dict(item)]}})["response"]["output"][0]
+        self.assertEqual((restored["namespace"], restored["name"]), ("collaboration", "spawn_agent"))
+
+    def test_custom_tool_namespace_and_ids_survive_chunked_sse(self):
+        req = self.request(); names = {}; custom = env["flatten_request"](req, names)
+        alias = req["tools"][2]["name"]
+        item = {"type": "function_call", "name": alias, "id": "fc_patch", "call_id": "call_patch", "arguments": json.dumps({"input": "patch text"})}
+        raw = ("data: " + json.dumps({"type": "response.output_item.done", "item": item}) + "\n\n").encode()
+        a = env["SseRewriter"](custom, tool_names=names)
+        output = a.feed(raw[:31]) + a.feed(raw[31:]) + a.flush()
+        restored = json.loads(next(l[6:] for l in output.decode().splitlines() if l.startswith("data: ")))["item"]
+        self.assertEqual((restored["namespace"], restored["name"], restored["type"]), ("functions", "apply_patch", "custom_tool_call"))
+        self.assertEqual(restored["id"], "ctc_patch")
+        self.assertEqual(restored["input"], "patch text")
+
+    def test_plain_tools_stay_plain_and_history_not_mutated(self):
+        req = self.request(); original = copy.deepcopy(req); old_item = req["input"][0]
+        env["flatten_request"](req, {})
+        self.assertEqual(req["tools"][-1]["name"], "plain")
+        self.assertEqual(old_item, original["input"][0])
+        plain = {"type": "function_call", "name": "plain", "arguments": "{}"}
+        self.assertEqual(env["SseRewriter"](set())._fix({"item": plain})["item"], plain)
+
+    def test_additional_tools_long_names_and_tool_choice(self):
+        namespace = "provider_" * 20
+        req = {"tools": [], "input": [{"type": "additional_tools", "tools": [{"type": "namespace", "name": namespace, "tools": [{"type": "function", "name": "read"}]}]}], "tool_choice": {"type": "function", "namespace": namespace, "name": "read"}}
+        names = {}; env["flatten_request"](req, names)
+        alias = req["tools"][0]["name"]
+        self.assertLessEqual(len(alias), 64)
+        self.assertEqual(req["tool_choice"], {"type": "function", "name": alias})
+        self.assertEqual(req["input"], [])
+        self.assertEqual(names[alias], (namespace, "read"))
+
+if __name__ == "__main__": unittest.main()