diff --git a/app/adapters/responses_adapter.py b/app/adapters/responses_adapter.py index 6dbd900..487983e 100644 --- a/app/adapters/responses_adapter.py +++ b/app/adapters/responses_adapter.py @@ -47,6 +47,18 @@ def _text_format_to_response_format(fmt) -> dict | None: def responses_request_to_chat(body: dict) -> dict: """Convert Responses input, instructions and tools to a Chat request.""" + # Collect declared tools from both top-level tools and input additional_tools items. + raw_tools = [] + if isinstance(body.get("tools"), list): + raw_tools.extend(body["tools"]) + inp = body.get("input", []) + if isinstance(inp, list): + for item in inp: + if isinstance(item, dict) and item.get("type") == "additional_tools" and isinstance(item.get("tools"), list): + raw_tools.extend(item["tools"]) + + registry, chat_tools = _build_tool_registry_and_chat_tools(raw_tools) + messages: list[dict] = [] # instructions → system message @@ -55,11 +67,10 @@ def responses_request_to_chat(body: dict) -> dict: messages.append({"role": "system", "content": instructions}) # input → messages - inp = body.get("input", []) if isinstance(inp, str): messages.append({"role": "user", "content": inp}) elif isinstance(inp, list): - messages.extend(_convert_input_items(inp)) + messages.extend(_convert_input_items(inp, tool_registry=registry)) # Build the Chat request body. chat: dict[str, Any] = {"messages": messages, "stream": True} @@ -68,12 +79,12 @@ def responses_request_to_chat(body: dict) -> dict: if "model" in body: chat["model"] = body["model"] - # Normalize function tool definitions. - tools = body.get("tools") - if tools: - chat["tools"] = _convert_tools_for_chat(tools) + if chat_tools: + chat["tools"] = chat_tools + if registry.upstream_to_identity: + chat["_tool_registry"] = registry.to_dict() if "tool_choice" in body: - chat["tool_choice"] = body["tool_choice"] + chat["tool_choice"] = _convert_tool_choice_for_chat(body["tool_choice"], tool_registry=registry) # Forward supported parameters. for key in ("temperature", "top_p", "stop", "seed", @@ -102,7 +113,7 @@ def responses_request_to_chat(body: dict) -> dict: return chat -def _convert_input_items(items: list) -> list[dict]: +def _convert_input_items(items: list, tool_registry: ToolRegistry | None = None) -> list[dict]: """Convert input items and merge adjacent assistant messages with tool calls.""" messages: list[dict] = [] # Buffer adjacent assistant text and function calls. @@ -127,6 +138,17 @@ def _flush_assistant(): item_type = item.get("type") role = item.get("role", "") + # Ignore additional_tools in message sequence. + if item_type == "additional_tools": + continue + + # Agent message from multi-agent collaboration. + if item_type == "agent_message": + _flush_assistant() + content = _extract_content(item.get("content", "")) + messages.append({"role": "user", "content": content}) + continue + # Untyped role messages if item_type is None and role in ("user", "system", "developer"): _flush_assistant() @@ -165,11 +187,14 @@ def _flush_assistant(): raise ValueError("function_call.arguments must be a JSON string") if pending_assistant_content is None: pending_assistant_content = "" + raw_name = item.get("name", "") + ns = item.get("namespace") + mapped_name = tool_registry.get_upstream_name(ns, raw_name) if tool_registry else raw_name pending_tool_calls.append({ "id": item.get("call_id", item.get("id", _rand_id("call_"))), "type": "function", "function": { - "name": item.get("name", ""), + "name": mapped_name, "arguments": arguments, }, }) @@ -208,6 +233,10 @@ def _extract_content(content) -> str | list[dict]: kind = p.get("type") if kind in ("input_text", "text", "output_text"): parts.append({"type": "text", "text": p.get("text", "")}) + elif kind == "encrypted_content": + text = p.get("encrypted_content") or p.get("text", "") + if isinstance(text, str) and text: + parts.append({"type": "text", "text": text}) elif kind == "input_image": if p.get("file_id"): raise ValueError("Responses input_image file_id is not supported; provide image_url instead") @@ -242,28 +271,169 @@ def _extract_output_text(content_parts: list) -> str | list[dict]: return "".join(texts) -def _convert_tools_for_chat(tools: list) -> list: - """Convert Responses tool definitions to Chat function objects.""" - result = [] +def _sanitize_schema(schema: Any) -> Any: + """Recursively strip 'encrypted' client-only marker from parameter schemas.""" + if isinstance(schema, dict): + return {k: _sanitize_schema(v) for k, v in schema.items() if k != "encrypted"} + if isinstance(schema, list): + return [_sanitize_schema(item) for item in schema] + return schema + + +class ToolRegistry: + """Bidirectional mapping between Responses (namespace, name) and upstream Chat function names.""" + + def __init__(self, mappings: dict[str, dict[str, Any]] | None = None): + self.upstream_to_identity: dict[str, tuple[str | None, str]] = {} + self.identity_to_upstream: dict[tuple[str | None, str], str] = {} + self._name_to_identities: dict[str, list[tuple[str | None, str]]] = {} + + if mappings: + for up_name, ident in mappings.items(): + ns = ident.get("namespace") if isinstance(ident, dict) else None + nm = ident.get("name", up_name) if isinstance(ident, dict) else up_name + self._record(up_name, ns, nm) + + def _record(self, upstream_name: str, namespace: str | None, name: str) -> None: + ident = (namespace, name) + self.upstream_to_identity[upstream_name] = ident + self.identity_to_upstream[ident] = upstream_name + self._name_to_identities.setdefault(name, []).append(ident) + + def register(self, namespace: str | None, name: str) -> str: + """Register a tool identity (namespace, name) and return its unique upstream name.""" + ident = (namespace, name) + if ident in self.identity_to_upstream: + return self.identity_to_upstream[ident] + + if namespace is None: + upstream_name = name + else: + base = f"{namespace}__{name}" + candidate = base + counter = 1 + while candidate in self.upstream_to_identity: + candidate = f"{base}_{counter}" + counter += 1 + upstream_name = candidate + + self._record(upstream_name, namespace, name) + return upstream_name + + def get_identity(self, upstream_name: str) -> tuple[str | None, str]: + """Map upstream function name back to (namespace, name). Never guess by prefix.""" + if upstream_name in self.upstream_to_identity: + return self.upstream_to_identity[upstream_name] + return None, upstream_name + + def get_upstream_name(self, namespace: str | None, name: str) -> str: + """Map (namespace, name) to upstream name, with fallback for historical calls.""" + ident = (namespace, name) + if ident in self.identity_to_upstream: + return self.identity_to_upstream[ident] + if namespace is None: + if (None, name) in self.identity_to_upstream: + return self.identity_to_upstream[(None, name)] + matches = self._name_to_identities.get(name, []) + if len(matches) == 1: + return self.identity_to_upstream[matches[0]] + return name + return f"{namespace}__{name}" + + def to_dict(self) -> dict: + return { + up_name: {"namespace": ident[0], "name": ident[1]} + for up_name, ident in self.upstream_to_identity.items() + } + + @classmethod + def from_dict(cls, data: dict | None) -> ToolRegistry: + if isinstance(data, cls): + return data + if not isinstance(data, dict): + return cls() + return cls(data) + + +def _collect_raw_tools(tools: list, current_ns: str = "") -> list[tuple[str | None, str, dict]]: + """Recursively collect (namespace, name, tool_def) from Responses tools.""" + collected = [] for t in tools: if not isinstance(t, dict): continue - if t.get("type") != "function": - continue - # Already in Chat format. - if "function" in t: - result.append(t) + tool_type = t.get("type") + if tool_type == "namespace" and isinstance(t.get("tools"), list): + ns_name = t.get("name", "") + full_ns = f"{current_ns}.{ns_name}" if current_ns else ns_name + collected.extend(_collect_raw_tools(t["tools"], full_ns)) + elif tool_type == "function" or "function" in t: + fn = t.get("function") if isinstance(t.get("function"), dict) else t + name = fn.get("name") or t.get("name") + if isinstance(name, str) and name: + collected.append((current_ns or None, name, t)) + return collected + + +def _format_chat_tool(upstream_name: str, tool_def: dict) -> dict: + """Format tool definition for Chat Completions, ensuring sanitized schema.""" + if "function" in tool_def: + tool_obj = json.loads(json.dumps(tool_def)) + fn_dict = tool_obj.get("function") + if isinstance(fn_dict, dict): + fn_dict["name"] = upstream_name + if "parameters" in fn_dict: + fn_dict["parameters"] = _sanitize_schema(fn_dict["parameters"]) + return tool_obj + + fn: dict[str, Any] = {"name": upstream_name} + if "description" in tool_def: + fn["description"] = tool_def["description"] + if "parameters" in tool_def: + fn["parameters"] = _sanitize_schema(tool_def["parameters"]) + if "strict" in tool_def: + fn["strict"] = tool_def["strict"] + return {"type": "function", "function": fn} + + +def _build_tool_registry_and_chat_tools(raw_tools: list) -> tuple[ToolRegistry, list[dict]]: + """Build ToolRegistry and convert declared tools into unique Chat function tools.""" + collected = _collect_raw_tools(raw_tools) + registry = ToolRegistry() + result = [] + + # Register plain tools first so their upstream name strictly preserves original name. + plain = [item for item in collected if item[0] is None] + namespaced = [item for item in collected if item[0] is not None] + + seen_identities: set[tuple[str | None, str]] = set() + for ns, name, tool_def in plain + namespaced: + ident = (ns, name) + if ident in seen_identities: continue - # Nest the flat Responses function fields. - fn: dict[str, Any] = {"name": t.get("name", "")} - if "description" in t: - fn["description"] = t["description"] - if "parameters" in t: - fn["parameters"] = t["parameters"] - if "strict" in t: - fn["strict"] = t["strict"] - result.append({"type": "function", "function": fn}) - return result + seen_identities.add(ident) + upstream_name = registry.register(ns, name) + result.append(_format_chat_tool(upstream_name, tool_def)) + + return registry, result + + +def _convert_tools_for_chat(tools: list) -> list: + """Backwards-compatible helper for converting Responses tools.""" + _, chat_tools = _build_tool_registry_and_chat_tools(tools) + return chat_tools + + +def _convert_tool_choice_for_chat(choice: Any, tool_registry: ToolRegistry | None = None) -> Any: + """Convert Responses tool_choice to Chat Completions format.""" + if not isinstance(choice, dict): + return choice + fn_obj = choice.get("function") if isinstance(choice.get("function"), dict) else choice + name = fn_obj.get("name") if isinstance(fn_obj, dict) else None + if not isinstance(name, str) or not name.strip(): + return choice + namespace = fn_obj.get("namespace", choice.get("namespace")) if isinstance(fn_obj, dict) else choice.get("namespace") + mapped_name = tool_registry.get_upstream_name(namespace, name) if tool_registry else name + return {"type": "function", "function": {"name": mapped_name}} # --------------------------------------------------------------------------- @@ -275,7 +445,9 @@ class ResponsesStreamConverter: def __init__(self, model: str = "unknown", parallel_tool_calls: bool = True, *, realtime: bool = False, budget: StreamOutputBudget | None = None, - tool_states: dict | None = None, declared_names=None): + tool_states: dict | None = None, declared_names=None, + tool_registry: ToolRegistry | dict | None = None, + tool_namespaces: dict[str, str] | None = None): self.resp_id = _rand_id("resp_") self.msg_id = _rand_id("msg_") self.model = model @@ -284,6 +456,13 @@ def __init__(self, model: str = "unknown", parallel_tool_calls: bool = True, *, self._budget = budget if budget is not None else StreamOutputBudget(0) self._tool_states = tool_states self._local_tool_states: dict[int, dict] = {} + if isinstance(tool_registry, ToolRegistry): + self._tool_registry = tool_registry + elif isinstance(tool_registry, dict): + self._tool_registry = ToolRegistry.from_dict(tool_registry) + else: + self._tool_registry = None + self._tool_namespaces = dict(tool_namespaces or {}) self._declared_names = frozenset( value for value in (declared_names or ()) if isinstance(value, str) and value) self.created_at = int(time.time()) @@ -698,14 +877,26 @@ def _msg_item(self, status: str = "in_progress", empty: bool = False) -> dict: } def _fc_item(self, tc: dict, status: str, *, include_arguments: bool = True) -> dict: - return { + raw_name = tc["name"] or (tc.get("state", {}).get("name") or "") + name = raw_name + ns = tc.get("namespace") + if not ns: + if self._tool_registry: + ns, name = self._tool_registry.get_identity(raw_name) + elif self._tool_namespaces and raw_name in self._tool_namespaces: + ns = self._tool_namespaces[raw_name] + + item: dict[str, Any] = { "type": "function_call", "id": tc["fc_id"], "call_id": tc["id"] or (tc.get("state", {}).get("id") or ""), - "name": tc["name"] or (tc.get("state", {}).get("name") or ""), + "name": name, "arguments": self._tool_arguments(tc) if include_arguments else "", "status": status, } + if ns: + item["namespace"] = ns + return item def _response_obj(self, status: str, incomplete_reason: str | None = None) -> dict: output = [] diff --git a/converter.py b/converter.py index 543c27a..b3dfdca 100644 --- a/converter.py +++ b/converter.py @@ -3644,10 +3644,15 @@ def attempt(routed, cred, headers, url): async def _nonstream_adapted(url, headers, body, model_name, t0, rid, cred, *, anthropic=False, payload=None, canonical=None, request=None, policy=None): policy = policy or _snapshot_stream_policy("messages" if anthropic else "responses", body) - converter = (AnthropicStreamConverter(model=model_name) if anthropic else ResponsesStreamConverter(model=model_name, parallel_tool_calls=body.get("parallel_tool_calls", True))) + tool_registry = body.get("_tool_registry") + converter = (AnthropicStreamConverter(model=model_name) if anthropic else + ResponsesStreamConverter(model=model_name, + parallel_tool_calls=body.get("parallel_tool_calls", True), + tool_registry=tool_registry)) async def fetch(routed, cred, headers, url): - return await _fetch_checked_chat(url, headers, routed, model_name, rid, cred, + routed_body = {k: v for k, v in routed.items() if not k.startswith("_")} + return await _fetch_checked_chat(url, headers, routed_body, model_name, rid, cred, filter_retry=True, max_collect_bytes=policy.max_collect_bytes) try: collected = await await_or_hangup( @@ -3671,6 +3676,7 @@ async def _stream_adapted(url, headers, body, model_name, t0, rid, cred=None, *, """Map protocol events while sharing connection, aggregation and failure handling.""" protocol = "messages" if anthropic else "responses" policy = body.pop(_REQUEST_POLICY_KEY, None) or _snapshot_stream_policy(protocol, body) + tool_registry = body.get("_tool_registry") state = {} tracker = None declared_names = _declared_tool_names(body) @@ -3684,14 +3690,17 @@ async def _stream_adapted(url, headers, body, model_name, t0, rid, cred=None, *, ResponsesStreamConverter(model=model_name, parallel_tool_calls=body.get("parallel_tool_calls", True), realtime=True, budget=budget, tool_states=tracker.tools, - declared_names=declared_names)) + declared_names=declared_names, + tool_registry=tool_registry)) else: converter = (AnthropicStreamConverter(model=model_name) if anthropic else ResponsesStreamConverter(model=model_name, - parallel_tool_calls=body.get("parallel_tool_calls", True))) + parallel_tool_calls=body.get("parallel_tool_calls", True), + tool_registry=tool_registry)) sent = False + upstream_body = {k: v for k, v in body.items() if not k.startswith("_")} upstream = _chat_sse_lines( - url, headers, body, model_name, t0, rid, cred, + url, headers, upstream_body, model_name, t0, rid, cred, policy=policy, tracker=tracker, state=state) try: try: diff --git a/tests/test_responses_adapter.py b/tests/test_responses_adapter.py index eab1ff2..57f6e41 100644 --- a/tests/test_responses_adapter.py +++ b/tests/test_responses_adapter.py @@ -623,6 +623,193 @@ def test_parallel_tool_calls_roundtrip(): print("✅ test_parallel_tool_calls_roundtrip") +def test_tool_registry_and_identity_mapping(): + """Verify ToolRegistry handles plain tools, namespaced tools, collision avoidance, and reversible lookup.""" + from app.adapters.responses_adapter import ToolRegistry + + registry = ToolRegistry() + # 1. Plain tool: must keep original name + name1 = registry.register(None, "collaboration__spawn_agent") + assert name1 == "collaboration__spawn_agent" + assert registry.get_identity("collaboration__spawn_agent") == (None, "collaboration__spawn_agent") + + # 2. Namespaced tool colliding with existing plain tool: must disambiguate safely + name2 = registry.register("collaboration", "spawn_agent") + assert name2 == "collaboration__spawn_agent_1" + assert registry.get_identity("collaboration__spawn_agent_1") == ("collaboration", "spawn_agent") + # Plain tool lookup still returns original identity + assert registry.get_identity("collaboration__spawn_agent") == (None, "collaboration__spawn_agent") + + # 3. Tool with __ in name inside namespace + name3 = registry.register("custom_ns", "exec__command") + assert name3 == "custom_ns__exec__command" + assert registry.get_identity("custom_ns__exec__command") == ("custom_ns", "exec__command") + + # 4. Tools with same name in different namespaces + name_ns1 = registry.register("ns1", "search") + name_ns2 = registry.register("ns2", "search") + assert name_ns1 == "ns1__search" + assert name_ns2 == "ns2__search" + assert registry.get_identity("ns1__search") == ("ns1", "search") + assert registry.get_identity("ns2__search") == ("ns2", "search") + + # 5. Serialization and round-trip + dumped = registry.to_dict() + restored = ToolRegistry.from_dict(dumped) + assert restored.get_identity("custom_ns__exec__command") == ("custom_ns", "exec__command") + assert restored.get_identity("collaboration__spawn_agent") == (None, "collaboration__spawn_agent") + assert restored.get_identity("collaboration__spawn_agent_1") == ("collaboration", "spawn_agent") + print("✅ test_tool_registry_and_identity_mapping") + + +def test_responses_multiagent_request_conversion_and_schema_sanitization(): + """Verify responses_request_to_chat converts tools, strips encrypted markers, and maps tool_choice and history.""" + req = { + "model": "gpt-4o", + "tools": [ + { + "type": "function", + "name": "collaboration__spawn_agent", + "description": "Standard unnamespaced tool", + "parameters": {"type": "object", "properties": {"prompt": {"type": "string"}}}, + }, + { + "type": "namespace", + "name": "collaboration", + "tools": [ + { + "type": "function", + "name": "spawn_agent", + "description": "Spawn an agent", + "parameters": { + "type": "object", + "properties": { + "message": {"type": "string", "encrypted": True}, + "task_name": {"type": "string"}, + "meta": { + "type": "object", + "properties": { + "secret": {"type": "string", "encrypted": True}, + "public": {"type": "string"}, + }, + }, + }, + }, + } + ], + }, + { + "type": "namespace", + "name": "ns1", + "tools": [{"type": "function", "name": "search", "parameters": {"type": "object"}}], + }, + { + "type": "namespace", + "name": "ns2", + "tools": [{"type": "function", "name": "search", "parameters": {"type": "object"}}], + }, + { + "type": "namespace", + "name": "custom_ns", + "tools": [{"type": "function", "name": "exec__command", "parameters": {"type": "object"}}], + }, + ], + "tool_choice": {"type": "function", "name": "spawn_agent", "namespace": "collaboration"}, + "input": [ + { + "type": "additional_tools", + "tools": [ + { + "type": "namespace", + "name": "extra", + "tools": [{"type": "function", "name": "ping", "parameters": {"type": "object"}}], + } + ], + }, + { + "type": "agent_message", + "author": "/root", + "recipient": "/root/worker", + "content": [ + {"type": "input_text", "text": "Task header\n"}, + {"type": "encrypted_content", "encrypted_content": "Sensitive payload text"}, + ], + }, + { + "type": "function_call", + "call_id": "call_hist_1", + "name": "spawn_agent", + "namespace": "collaboration", + "arguments": '{"task_name": "t1"}', + }, + { + "type": "function_call_output", + "call_id": "call_hist_1", + "output": "ok", + }, + ], + } + + chat = responses_request_to_chat(req) + chat_tools = chat.get("tools", []) + tool_names = [t["function"]["name"] for t in chat_tools] + + # 1. Verify plain tool preserved original name + assert "collaboration__spawn_agent" in tool_names + # 2. Verify namespaced tool disambiguated + assert "collaboration__spawn_agent_1" in tool_names + # 3. Verify same name under different namespaces preserved + assert "ns1__search" in tool_names + assert "ns2__search" in tool_names + # 4. Verify tool with __ in name inside namespace + assert "custom_ns__exec__command" in tool_names + # 5. Verify additional_tools registered + assert "extra__ping" in tool_names + + # 6. Verify parameter schema cleaning (no 'encrypted' key anywhere) + collab_tool = next(t for t in chat_tools if t["function"]["name"] == "collaboration__spawn_agent_1") + props = collab_tool["function"]["parameters"]["properties"] + assert "encrypted" not in props["message"] + assert "encrypted" not in props["meta"]["properties"]["secret"] + + # 7. Verify tool_choice mapped + assert chat.get("tool_choice") == {"type": "function", "function": {"name": "collaboration__spawn_agent_1"}} + + # 8. Verify messages conversion + messages = chat.get("messages", []) + # User message from agent_message with encrypted_content + user_msg = next(m for m in messages if m["role"] == "user") + assert "Task header\n" in user_msg["content"] + assert "Sensitive payload text" in user_msg["content"] + + # Assistant message from function_call + asst_msg = next(m for m in messages if m["role"] == "assistant") + assert asst_msg["tool_calls"][0]["function"]["name"] == "collaboration__spawn_agent_1" + + # 9. Verify stream converter output reconstruction + conv = ResponsesStreamConverter(model="test", tool_registry=chat.get("_tool_registry")) + + # Namespaced tool call + slot_collab = {"id": "c1", "name": "collaboration__spawn_agent_1", "args": "{}", "fc_id": "fc_1", "output_idx": 0, "emitted": False, "emitted_args_length": 0} + item_collab = conv._fc_item(slot_collab, "completed") + assert item_collab["name"] == "spawn_agent" + assert item_collab["namespace"] == "collaboration" + + # Plain tool call + slot_plain = {"id": "c2", "name": "collaboration__spawn_agent", "args": "{}", "fc_id": "fc_2", "output_idx": 1, "emitted": False, "emitted_args_length": 0} + item_plain = conv._fc_item(slot_plain, "completed") + assert item_plain["name"] == "collaboration__spawn_agent" + assert "namespace" not in item_plain + + # Custom namespace with __ in name + slot_custom = {"id": "c3", "name": "custom_ns__exec__command", "args": "{}", "fc_id": "fc_3", "output_idx": 2, "emitted": False, "emitted_args_length": 0} + item_custom = conv._fc_item(slot_custom, "completed") + assert item_custom["name"] == "exec__command" + assert item_custom["namespace"] == "custom_ns" + + print("✅ test_responses_multiagent_request_conversion_and_schema_sanitization") + + if __name__ == "__main__": test_simple_text_request() test_array_input_request() @@ -647,4 +834,6 @@ def test_parallel_tool_calls_roundtrip(): test_usage_maps_cached_tokens_and_omits_when_unknown() test_reasoning_effort_and_text_format_are_mapped() test_parallel_tool_calls_roundtrip() + test_tool_registry_and_identity_mapping() + test_responses_multiagent_request_conversion_and_schema_sanitization() print(f"\n🎉 All {22} tests passed!") diff --git a/tests/test_responses_multiagent.py b/tests/test_responses_multiagent.py new file mode 100644 index 0000000..d741019 --- /dev/null +++ b/tests/test_responses_multiagent.py @@ -0,0 +1,424 @@ +#!/usr/bin/env python3 +"""Interface-level regression tests for Responses multi-agent tools and namespace identity mapping.""" + +import json +import sys +from pathlib import Path +import unittest +from unittest.mock import patch + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +import httpx +from fastapi.testclient import TestClient + +import converter +from app import upstream_io + + +def _chat_completion_response(tool_calls=None, content="hello"): + """Build a standard non-streaming Chat completions JSON response.""" + msg = {"role": "assistant", "content": content} + if tool_calls is not None: + msg["tool_calls"] = tool_calls + return { + "id": "chatcmpl_test", + "object": "chat.completion", + "created": 1234567890, + "model": "auto", + "choices": [ + { + "index": 0, + "message": msg, + "finish_reason": "tool_calls" if tool_calls else "stop", + } + ], + "usage": {"prompt_tokens": 10, "completion_tokens": 20, "total_tokens": 30}, + } + + +def _chat_sse_stream(tool_calls=None, content=None): + """Build Chat SSE streaming lines for buffered or realtime streams.""" + lines = [] + # Initial chunk + lines.append("data: " + json.dumps({ + "id": "chatcmpl_chunk", + "object": "chat.completion.chunk", + "created": 1234567890, + "model": "auto", + "choices": [{"index": 0, "delta": {"role": "assistant"}, "finish_reason": None}], + }) + "\n\n") + + if content: + lines.append("data: " + json.dumps({ + "id": "chatcmpl_chunk", + "object": "chat.completion.chunk", + "created": 1234567890, + "model": "auto", + "choices": [{"index": 0, "delta": {"content": content}, "finish_reason": None}], + }) + "\n\n") + + if tool_calls: + for idx, tc in enumerate(tool_calls): + fn = tc.get("function", {}) + lines.append("data: " + json.dumps({ + "id": "chatcmpl_chunk", + "object": "chat.completion.chunk", + "created": 1234567890, + "model": "auto", + "choices": [{ + "index": 0, + "delta": { + "tool_calls": [{ + "index": idx, + "id": tc.get("id", f"call_{idx}"), + "type": "function", + "function": {"name": fn.get("name", ""), "arguments": fn.get("arguments", "{}")}, + }] + }, + "finish_reason": None, + }], + }) + "\n\n") + + lines.append("data: " + json.dumps({ + "id": "chatcmpl_chunk", + "object": "chat.completion.chunk", + "created": 1234567890, + "model": "auto", + "choices": [{"index": 0, "delta": {}, "finish_reason": "tool_calls"}], + "usage": {"prompt_tokens": 10, "completion_tokens": 20, "total_tokens": 30}, + }) + "\n\n") + else: + lines.append("data: " + json.dumps({ + "id": "chatcmpl_chunk", + "object": "chat.completion.chunk", + "created": 1234567890, + "model": "auto", + "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}], + "usage": {"prompt_tokens": 10, "completion_tokens": 20, "total_tokens": 30}, + }) + "\n\n") + + lines.append("data: [DONE]\n\n") + return "".join(lines).encode("utf-8") + + +def _parse_responses_sse_events(raw_text: str) -> list[dict]: + """Parse SSE event stream from /v1/responses.""" + events = [] + for block in raw_text.strip().split("\n\n"): + if not block: + continue + data_str = next((line[6:] for line in block.splitlines() if line.startswith("data: ")), None) + if data_str and data_str != "[DONE]": + events.append(json.loads(data_str)) + return events + + +class ResponsesMultiAgentInterfaceTests(unittest.TestCase): + def setUp(self): + self.enterContext(patch.dict(converter.CONFIG, { + "api_key": "", "cred": None, "cred_pool": None, "model_guard": False, + "max_images": 16, "image_policy": "truncate", "max_request_bytes": 32 * 1024 * 1024, + "log_body_limit": 65536, "log_path": None, "desensitize": False, "no_compact": False, + "stream_mode": "compatible", + })) + self.enterContext(patch.object(converter, "_cred_for", return_value=(None, {}))) + self.enterContext(patch.object(converter, "_log")) + self.enterContext(patch.object(converter, "_note_cred_status")) + + self.upstream_requests = [] + self.mock_response_generator = None + + def handle_request(request: httpx.Request): + self.upstream_requests.append(request) + if self.mock_response_generator: + return self.mock_response_generator(request) + return httpx.Response(200, json=_chat_completion_response()) + + real_client = httpx.AsyncClient + transport = httpx.MockTransport(handle_request) + self.enterContext(patch.object(upstream_io.httpx, "AsyncClient", + side_effect=lambda **kw: real_client(transport=transport, **kw))) + self.client = self.enterContext(TestClient(converter.app)) + + def _sample_multiagent_payload(self, stream: bool = False): + return { + "model": "auto", + "stream": stream, + "tools": [ + { + "type": "function", + "name": "collaboration__spawn_agent", + "description": "Standard ordinary tool without namespace", + "parameters": {"type": "object", "properties": {"task": {"type": "string"}}}, + }, + { + "type": "namespace", + "name": "ns1", + "tools": [ + { + "type": "function", + "name": "search", + "description": "Search in ns1", + "parameters": {"type": "object", "properties": {"q": {"type": "string"}}}, + } + ], + }, + { + "type": "namespace", + "name": "ns2", + "tools": [ + { + "type": "function", + "name": "search", + "description": "Search in ns2", + "parameters": {"type": "object", "properties": {"q": {"type": "string"}}}, + } + ], + }, + { + "type": "namespace", + "name": "custom_ns", + "tools": [ + { + "type": "function", + "name": "exec__command", + "description": "Tool whose name contains double underscores", + "parameters": { + "type": "object", + "properties": {"cmd": {"type": "string", "encrypted": True}}, + }, + } + ], + }, + ], + "input": [{"role": "user", "content": "Execute multi-agent subtasks"}], + } + + def test_nonstream_tools_isolation_identity_and_metadata_sanitation(self): + """Cover non-streaming mode: verify identity restoration, ordinary tool preservation, and metadata purity.""" + # 1. Test upstream call to ns1.search + tc_ns1 = [{"id": "call_1", "type": "function", "function": {"name": "ns1__search", "arguments": '{"q":"weather"}'}}] + self.mock_response_generator = lambda req: httpx.Response(200, content=_chat_sse_stream(tool_calls=tc_ns1)) + + res = self.client.post("/v1/responses", json=self._sample_multiagent_payload(stream=False)) + self.assertEqual(res.status_code, 200) + data = res.json() + output = data.get("output", []) + self.assertEqual(len(output), 1) + self.assertEqual(output[0]["name"], "search") + self.assertEqual(output[0]["namespace"], "ns1") + + # Verify internal metadata was not sent to upstream + upstream_req = self.upstream_requests[-1] + upstream_json = json.loads(upstream_req.content.decode("utf-8")) + self.assertNotIn("_tool_registry", upstream_json) + self.assertNotIn("_tool_namespaces", upstream_json) + self.assertFalse(any(k.startswith("_") for k in upstream_json.keys())) + + # Verify upstream tools contains both ns1__search and ns2__search + declared = [t["function"]["name"] for t in upstream_json.get("tools", [])] + self.assertIn("ns1__search", declared) + self.assertIn("ns2__search", declared) + self.assertIn("collaboration__spawn_agent", declared) + self.assertIn("custom_ns__exec__command", declared) + + # 2. Test upstream call to ns2.search + tc_ns2 = [{"id": "call_2", "type": "function", "function": {"name": "ns2__search", "arguments": '{"q":"news"}'}}] + self.mock_response_generator = lambda req: httpx.Response(200, content=_chat_sse_stream(tool_calls=tc_ns2)) + res2 = self.client.post("/v1/responses", json=self._sample_multiagent_payload(stream=False)) + self.assertEqual(res2.status_code, 200) + out2 = res2.json().get("output", []) + self.assertEqual(out2[0]["name"], "search") + self.assertEqual(out2[0]["namespace"], "ns2") + + # 3. Test upstream call to plain tool collaboration__spawn_agent (MUST NOT be altered) + tc_collab = [{"id": "call_3", "type": "function", "function": {"name": "collaboration__spawn_agent", "arguments": '{"task":"inspect"}'}}] + self.mock_response_generator = lambda req: httpx.Response(200, content=_chat_sse_stream(tool_calls=tc_collab)) + res3 = self.client.post("/v1/responses", json=self._sample_multiagent_payload(stream=False)) + self.assertEqual(res3.status_code, 200) + out3 = res3.json().get("output", []) + self.assertEqual(out3[0]["name"], "collaboration__spawn_agent") + self.assertNotIn("namespace", out3[0]) + + # 4. Test upstream call to tool with __ in name + tc_custom = [{"id": "call_4", "type": "function", "function": {"name": "custom_ns__exec__command", "arguments": '{"cmd":"ls"}'}}] + self.mock_response_generator = lambda req: httpx.Response(200, content=_chat_sse_stream(tool_calls=tc_custom)) + res4 = self.client.post("/v1/responses", json=self._sample_multiagent_payload(stream=False)) + self.assertEqual(res4.status_code, 200) + out4 = res4.json().get("output", []) + self.assertEqual(out4[0]["name"], "exec__command") + self.assertEqual(out4[0]["namespace"], "custom_ns") + + def test_buffered_stream_tools_isolation_identity_and_metadata_sanitation(self): + """Cover buffered streaming mode (compatible): verify events and tool identities.""" + tc = [ + {"id": "call_a", "type": "function", "function": {"name": "ns1__search", "arguments": '{"q":"a"}'}}, + {"id": "call_b", "type": "function", "function": {"name": "collaboration__spawn_agent", "arguments": '{"task":"b"}'}}, + {"id": "call_c", "type": "function", "function": {"name": "custom_ns__exec__command", "arguments": '{"cmd":"c"}'}}, + ] + self.mock_response_generator = lambda req: httpx.Response( + 200, + content=_chat_sse_stream(tool_calls=tc), + headers={"Content-Type": "text/event-stream"}, + ) + + with patch.dict(converter.CONFIG, {"stream_mode": "compatible"}): + res = self.client.post("/v1/responses", json=self._sample_multiagent_payload(stream=True)) + self.assertEqual(res.status_code, 200) + events = _parse_responses_sse_events(res.text) + + # Check output_item.done events + done_items = [e["item"] for e in events if e.get("type") == "response.output_item.done" and e.get("item", {}).get("type") == "function_call"] + self.assertEqual(len(done_items), 3) + + # 1. ns1.search + self.assertEqual(done_items[0]["name"], "search") + self.assertEqual(done_items[0]["namespace"], "ns1") + + # 2. collaboration__spawn_agent + self.assertEqual(done_items[1]["name"], "collaboration__spawn_agent") + self.assertNotIn("namespace", done_items[1]) + + # 3. custom_ns.exec__command + self.assertEqual(done_items[2]["name"], "exec__command") + self.assertEqual(done_items[2]["namespace"], "custom_ns") + + # Check response.completed terminal + completed = next(e for e in events if e.get("type") == "response.completed") + outputs = completed["response"]["output"] + self.assertEqual(len(outputs), 3) + self.assertEqual(outputs[0]["name"], "search") + self.assertEqual(outputs[0]["namespace"], "ns1") + self.assertEqual(outputs[1]["name"], "collaboration__spawn_agent") + self.assertNotIn("namespace", outputs[1]) + self.assertEqual(outputs[2]["name"], "exec__command") + self.assertEqual(outputs[2]["namespace"], "custom_ns") + + # Verify no internal metadata sent upstream + upstream_req = self.upstream_requests[-1] + upstream_json = json.loads(upstream_req.content.decode("utf-8")) + self.assertNotIn("_tool_registry", upstream_json) + self.assertNotIn("_tool_namespaces", upstream_json) + self.assertFalse(any(k.startswith("_") for k in upstream_json.keys())) + + def test_realtime_stream_tools_isolation_identity_and_metadata_sanitation(self): + """Cover realtime streaming mode: verify incremental events and tool identities.""" + tc = [ + {"id": "call_1", "type": "function", "function": {"name": "ns2__search", "arguments": '{"q":"live"}'}}, + {"id": "call_2", "type": "function", "function": {"name": "collaboration__spawn_agent", "arguments": '{"task":"live"}'}}, + {"id": "call_3", "type": "function", "function": {"name": "custom_ns__exec__command", "arguments": '{"cmd":"live"}'}}, + ] + self.mock_response_generator = lambda req: httpx.Response( + 200, + content=_chat_sse_stream(tool_calls=tc), + headers={"Content-Type": "text/event-stream"}, + ) + + with patch.dict(converter.CONFIG, {"stream_mode": "realtime"}): + res = self.client.post("/v1/responses", json=self._sample_multiagent_payload(stream=True)) + self.assertEqual(res.status_code, 200) + events = _parse_responses_sse_events(res.text) + + # Check output_item.added events + added_items = [e["item"] for e in events if e.get("type") == "response.output_item.added" and e.get("item", {}).get("type") == "function_call"] + self.assertEqual(len(added_items), 3) + + # 1. ns2.search + self.assertEqual(added_items[0]["name"], "search") + self.assertEqual(added_items[0]["namespace"], "ns2") + + # 2. collaboration__spawn_agent + self.assertEqual(added_items[1]["name"], "collaboration__spawn_agent") + self.assertNotIn("namespace", added_items[1]) + + # 3. custom_ns.exec__command + self.assertEqual(added_items[2]["name"], "exec__command") + self.assertEqual(added_items[2]["namespace"], "custom_ns") + + # Check response.completed terminal + completed = next(e for e in events if e.get("type") == "response.completed") + outputs = completed["response"]["output"] + self.assertEqual(len(outputs), 3) + self.assertEqual(outputs[0]["name"], "search") + self.assertEqual(outputs[0]["namespace"], "ns2") + self.assertEqual(outputs[1]["name"], "collaboration__spawn_agent") + self.assertNotIn("namespace", outputs[1]) + self.assertEqual(outputs[2]["name"], "exec__command") + self.assertEqual(outputs[2]["namespace"], "custom_ns") + + # Verify no internal metadata sent upstream + upstream_req = self.upstream_requests[-1] + upstream_json = json.loads(upstream_req.content.decode("utf-8")) + self.assertNotIn("_tool_registry", upstream_json) + self.assertNotIn("_tool_namespaces", upstream_json) + self.assertFalse(any(k.startswith("_") for k in upstream_json.keys())) + + def test_historical_tool_calls_and_tool_choice_integration(self): + """Verify historical function_call and tool_choice with namespace are correctly mapped to upstream.""" + payload = { + "model": "auto", + "stream": False, + "tools": [ + { + "type": "namespace", + "name": "collaboration", + "tools": [{"type": "function", "name": "spawn_agent", "parameters": {"type": "object"}}], + } + ], + "tool_choice": {"type": "function", "name": "spawn_agent", "namespace": "collaboration"}, + "input": [ + { + "type": "agent_message", + "content": [ + {"type": "input_text", "text": "Task: "}, + {"type": "encrypted_content", "encrypted_content": "Run analysis worker"}, + ], + }, + { + "type": "function_call", + "call_id": "call_hist_1", + "name": "spawn_agent", + "namespace": "collaboration", + "arguments": '{"task_name": "worker_1"}', + }, + { + "type": "function_call_output", + "call_id": "call_hist_1", + "output": "worker spawned", + }, + ], + } + + tc = [{"id": "call_resp_1", "type": "function", "function": {"name": "collaboration__spawn_agent", "arguments": '{"task_name":"worker_2"}'}}] + self.mock_response_generator = lambda req: httpx.Response(200, content=_chat_sse_stream(tool_calls=tc)) + res = self.client.post("/v1/responses", json=payload) + self.assertEqual(res.status_code, 200) + + upstream_req = self.upstream_requests[-1] + upstream_json = json.loads(upstream_req.content.decode("utf-8")) + + # 1. tool_choice mapped + self.assertEqual(upstream_json.get("tool_choice"), "required") + # And tools filtered to that single tool by _normalize_tool_choice + self.assertEqual(upstream_json["tools"][0]["function"]["name"], "collaboration__spawn_agent") + + # 2. Historical tool call in assistant message mapped + messages = upstream_json["messages"] + asst = next(m for m in messages if m["role"] == "assistant" and "tool_calls" in m) + self.assertEqual(asst["tool_calls"][0]["function"]["name"], "collaboration__spawn_agent") + + # 3. Encrypted content extracted into user message + user = next(m for m in messages if m["role"] == "user") + self.assertIn("Task: ", user["content"]) + self.assertIn("Run analysis worker", user["content"]) + + # 4. Response returned by gateway has namespace and name restored + out = res.json().get("output", []) + self.assertEqual(len(out), 1) + self.assertEqual(out[0]["name"], "spawn_agent") + self.assertEqual(out[0]["namespace"], "collaboration") + + +if __name__ == "__main__": + unittest.main()