diff --git a/litellm/llms/github_copilot/responses/collaboration.py b/litellm/llms/github_copilot/responses/collaboration.py new file mode 100644 index 00000000000..55d3dc118fc --- /dev/null +++ b/litellm/llms/github_copilot/responses/collaboration.py @@ -0,0 +1,264 @@ +from collections.abc import Iterator, Mapping, Sequence +from types import MappingProxyType +from typing import Final + +from pydantic import TypeAdapter, ValidationError + +_CLIENT_NAMESPACE: Final = "collaboration" +_UPSTREAM_NAMESPACE: Final = "codex_collaboration_plaintext" +_MESSAGE_ACTIONS: Final = frozenset({"spawn_agent", "send_message", "followup_task"}) +_INVENTORY_ITEMS: Final = ("additional_tools", "tool_search_output") +_CALL_ITEMS: Final = ("function_call", "custom_tool_call") +_IDENTITY_KEYS: Final = ("type", "namespace", "name", "call_id") +_RECORD: Final = TypeAdapter(Mapping[str, object]) +_SEQUENCE: Final = TypeAdapter(Sequence[object]) + + +def _sequence(value: object) -> Sequence[object]: + return _SEQUENCE.validate_python(value, strict=True) if isinstance(value, (list, tuple)) else () + + +def _record(value: object) -> Mapping[str, object] | None: + if not isinstance(value, Mapping): + return None + try: + return _RECORD.validate_python(value, strict=True) + except ValidationError: + return None + + +def _mapping(value: object) -> Mapping[str, object]: + return _record(value) or MappingProxyType({}) + + +def _inventory(request: Mapping[str, object]) -> Iterator[Mapping[str, object]]: + for tool in _sequence(request.get("tools")): + if (tool_record := _record(tool)) is not None: + yield tool_record + for item in _sequence(request.get("input")): + if (item_record := _record(item)) is not None and item_record.get("type") in _INVENTORY_ITEMS: + for tool in _sequence(item_record.get("tools")): + if (nested_tool := _record(tool)) is not None: + yield nested_tool + + +def _encrypted_action(tool: object) -> str | None: + record: Final = _record(tool) + if record is None or record.get("type") != "function": + return None + name: Final = record.get("name") + parameters: Final = _mapping(record.get("parameters")) + properties: Final = _mapping(parameters.get("properties")) + message: Final = _mapping(properties.get("message")) + if isinstance(name, str) and name in _MESSAGE_ACTIONS: + return name if message.get("type") == "string" and message.get("encrypted") is True else None + return None + + +def _actions(request: Mapping[str, object]) -> Iterator[str]: + for tool in _inventory(request): + if tool.get("type") == "namespace" and tool.get("name") == _CLIENT_NAMESPACE: + for child in _sequence(tool.get("tools")): + if (action := _encrypted_action(child)) is not None: + yield action + + +def _identity(value: Mapping[str, object], source: str, target: str) -> Mapping[str, object]: + namespace: Final = value.get("namespace") + name: Final = value.get("name") + if isinstance(name, str): + for separator in (".", "__"): + if name.startswith(source + separator): + if namespace is not None and namespace != source: + raise ValueError("Conflicting Copilot collaboration namespace and function name") + return {**value, "namespace": target, "name": name[len(source) + len(separator) :]} + return {**value, "namespace": target} if namespace == source else value + + +def _plaintext_tool(tool: object) -> object: + record: Final = _record(tool) + if record is None or _encrypted_action(record) is None: + return tool + parameters: Final = _mapping(record.get("parameters")) + properties: Final = _mapping(parameters.get("properties")) + message: Final = _mapping(properties.get("message")) + return { + **record, + "parameters": { + **parameters, + "properties": { + **properties, + "message": {key: value for key, value in message.items() if key != "encrypted"}, + }, + }, + } + + +def _tool(tool: object, source: str, target: str) -> object: + record: Final = _record(tool) + if record is None or record.get("type") != "namespace" or record.get("name") != source: + return tool + return { + **record, + "name": target, + **( + { + "tools": [ + _plaintext_tool(child) if source == _CLIENT_NAMESPACE else child + for child in _sequence(record.get("tools")) + ] + } + if isinstance(record.get("tools"), (list, tuple)) + else {} + ), + } + + +def _tools(value: object, source: str, target: str) -> object: + sequence: Final = _sequence(value) + return [_tool(tool, source, target) for tool in sequence] if sequence else value + + +def _choice(value: object, source: str, target: str) -> object: + record: Final = _record(value) + if record is None: + return value + if record.get("type") == "namespace" and record.get("name") == source: + return {**record, "name": target} + if record.get("type") == "allowed_tools": + return {**record, "tools": [_choice(tool, source, target) for tool in _sequence(record.get("tools"))]} + rewritten: Final = _identity(record, source, target) + name: Final = record.get("name") + if rewritten is not record and "namespace" not in record and isinstance(name, str): + separator: Final = "__" if name.startswith(source + "__") else "." + return {**record, "name": target + separator + str(rewritten["name"])} + return rewritten + + +def _project_item(value: object) -> object: + record: Final = _record(value) + if record is None: + return value + if record.get("type") in _INVENTORY_ITEMS and "tools" in record: + return {**record, "tools": _tools(record.get("tools"), _CLIENT_NAMESPACE, _UPSTREAM_NAMESPACE)} + return _identity(record, _CLIENT_NAMESPACE, _UPSTREAM_NAMESPACE) if record.get("type") in _CALL_ITEMS else value + + +def _project_input(value: object) -> object: + sequence: Final = _sequence(value) + return [_project_item(item) for item in sequence] if sequence else value + + +class CopilotPlaintextCollaboration: + def __init__(self) -> None: + self._actions: frozenset[str] = frozenset() + self._identities: Mapping[int, Mapping[str, object]] = MappingProxyType({}) + + @property + def active(self) -> bool: + return bool(self._actions) + + def prepare_request(self, request: Mapping[str, object]) -> Mapping[str, object]: + self._actions = frozenset(_actions(request)) + self._identities = MappingProxyType({}) + if not self.active: + return request + if ( + request.get("store") is True + or request.get("previous_response_id") is not None + or request.get("conversation") is not None + ): + raise ValueError("Copilot plaintext collaboration requires stateless Responses requests") + if any(tool.get("name") == _UPSTREAM_NAMESPACE for tool in _inventory(request)): + raise ValueError(f"The namespace {_UPSTREAM_NAMESPACE!r} is reserved by the Copilot adapter") + projected: Final = { + key: ( + _tools(value, _CLIENT_NAMESPACE, _UPSTREAM_NAMESPACE) + if key == "tools" + else _project_input(value) + if key == "input" + else _choice(value, _CLIENT_NAMESPACE, _UPSTREAM_NAMESPACE) + if key == "tool_choice" + else value + ) + for key, value in request.items() + } + return {**projected, "store": False} + + def _remember(self, value: Mapping[str, object], index: int | None, *, item: bool) -> Mapping[str, object]: + canonical: Final = _identity(value, _UPSTREAM_NAMESPACE, _UPSTREAM_NAMESPACE) + if index is None: + return canonical + known: Final = self._identities.get(index, MappingProxyType({})) + keys: Final = _IDENTITY_KEYS if item else _IDENTITY_KEYS[1:] + identity: Final = { + key: canonical[key] for key in keys if isinstance(canonical.get(key), str) and canonical.get(key) != "" + } + if any(key in known and known[key] != field for key, field in identity.items()): + raise ValueError("Conflicting Copilot collaboration identity for one output_index") + merged: Final = {**known, **identity} + self._identities = MappingProxyType({**self._identities, index: merged}) + return {**canonical, **{key: merged[key] for key in keys if key in merged}} + + def _restore_identity(self, value: Mapping[str, object]) -> Mapping[str, object]: + restored: Final = _identity(value, _UPSTREAM_NAMESPACE, _CLIENT_NAMESPACE) + name: Final = restored.get("name") + if restored is value or not isinstance(name, str) or name not in self._actions: + return restored + if "encrypted_function_args" in value and value["encrypted_function_args"] != []: + raise ValueError("Copilot returned encrypted arguments for a plaintext collaboration tool") + return {**restored, "encrypted_function_args": []} + + def _restore_item(self, value: object, index: int | None) -> object: + record: Final = _record(value) + if record is None: + return value + if record.get("type") not in _CALL_ITEMS and index not in self._identities: + return value + remembered: Final = self._remember(record, index, item=True) + return ( + self._restore_identity(remembered) + if remembered.get("type") == "function_call" + else _identity(remembered, _UPSTREAM_NAMESPACE, _CLIENT_NAMESPACE) + ) + + def restore_response(self, response: Mapping[str, object]) -> Mapping[str, object]: + if not self.active: + return response + return { + key: ( + self._restore_output(value) + if key == "output" + else _tools(value, _UPSTREAM_NAMESPACE, _CLIENT_NAMESPACE) + if key == "tools" + else _choice(value, _UPSTREAM_NAMESPACE, _CLIENT_NAMESPACE) + if key == "tool_choice" + else value + ) + for key, value in response.items() + } + + def _restore_output(self, value: object) -> object: + sequence: Final = _sequence(value) + return [self._restore_item(item, index) for index, item in enumerate(sequence)] if sequence else value + + def restore_event(self, event: Mapping[str, object]) -> Mapping[str, object]: + if not self.active: + return event + output_index: Final = event.get("output_index") + index: Final = output_index if isinstance(output_index, int) and not isinstance(output_index, bool) else None + restored: Final = { + key: ( + self._restore_item(value, index) + if key == "item" + else self.restore_response(response) + if key == "response" and (response := _record(value)) is not None + else value + ) + for key, value in event.items() + } + if event.get("type") == "response.function_call_arguments.done": + return self._restore_identity(self._remember(restored, index, item=False)) + if event.get("type") == "response.function_call_arguments.delta": + return _identity(restored, _UPSTREAM_NAMESPACE, _CLIENT_NAMESPACE) + return restored diff --git a/litellm/llms/github_copilot/responses/transformation.py b/litellm/llms/github_copilot/responses/transformation.py index d468889b626..e8ab2eba3e2 100644 --- a/litellm/llms/github_copilot/responses/transformation.py +++ b/litellm/llms/github_copilot/responses/transformation.py @@ -9,6 +9,7 @@ https://github.com/caozhiyuan/copilot-api """ import os +from collections.abc import Mapping from typing import ( TYPE_CHECKING, Any, @@ -16,6 +17,9 @@ from typing import ( cast, # noqa: TID251 # narrows untyped request dicts at the session boundary ) +import httpx +from pydantic import TypeAdapter + import litellm from litellm._logging import verbose_logger from litellm.constants import DEFAULT_MAX_RECURSE_DEPTH @@ -24,6 +28,7 @@ from litellm.llms.openai.responses.transformation import OpenAIResponsesAPIConfi from litellm.types.llms.openai import ( ResponseInputParam, ResponsesAPIOptionalRequestParams, + ResponsesAPIResponse, ResponsesAPIStreamingResponse, ) from litellm.types.router import GenericLiteLLMParams @@ -38,6 +43,9 @@ from ..common_utils import ( pin_session_authorization, ) from ..per_user_auth import require_github_copilot_user_session +from .collaboration import CopilotPlaintextCollaboration + +_RESPONSE_OBJECT: Final = TypeAdapter(Mapping[str, object]) if TYPE_CHECKING: from litellm.litellm_core_utils.litellm_logging import Logging as _LiteLLMLoggingObj @@ -97,13 +105,24 @@ class GithubCopilotResponsesAPIConfig(OpenAIResponsesAPIConfig): - Supports vision requests with special header - Requires X-Initiator header based on input analysis + GITHUB_COPILOT_PLAINTEXT_COLLABORATION=true opts Codex agent messages into + plaintext dispatch over HTTPS. It requires stateless requests carrying the + encrypted collaboration tool inventory and does not recover old ciphertext. + Reference: https://api.githubcopilot.com/ """ - def __init__(self) -> None: + def __init__(self, plaintext_collaboration: bool | None = None) -> None: super().__init__() self.authenticator = Authenticator() self._stream_item_ids_by_output_index: dict[int, str] = {} + enabled: Final = ( + os.getenv("GITHUB_COPILOT_PLAINTEXT_COLLABORATION", "").lower() == "true" + if plaintext_collaboration is None + else plaintext_collaboration + ) + self._collaboration = CopilotPlaintextCollaboration() if enabled else None + self._unprojected_request: Mapping[str, object] = {} @property def custom_llm_provider(self) -> LlmProviders: @@ -138,13 +157,66 @@ class GithubCopilotResponsesAPIConfig(OpenAIResponsesAPIConfig): parsed_chunk: dict, logging_obj: LiteLLMLoggingObj, ) -> ResponsesAPIStreamingResponse: - parsed_chunk = self._normalize_stream_item_id(parsed_chunk) + normalized: Final = self._normalize_stream_item_id(parsed_chunk) + restored: Final = ( + dict(self._collaboration.restore_event(_RESPONSE_OBJECT.validate_python(normalized))) + if self._collaboration is not None + else normalized + ) return super().transform_streaming_response( model=model, - parsed_chunk=parsed_chunk, + parsed_chunk=restored, logging_obj=logging_obj, ) + def transform_responses_api_request( + self, + model: str, + input: str | ResponseInputParam, + response_api_optional_request_params: Mapping[str, object], + litellm_params: GenericLiteLLMParams, + headers: Mapping[str, str], + ) -> dict[str, object]: # mutable-ok: the HTTP handler expects a mutable provider request + request: Final = super().transform_responses_api_request( + model=model, + input=input, + response_api_optional_request_params=dict(response_api_optional_request_params), + litellm_params=litellm_params, + headers=dict(headers), + ) + if self._collaboration is None: + return request + self._unprojected_request = _RESPONSE_OBJECT.validate_python(request) + return dict(self._collaboration.prepare_request(self._unprojected_request)) + + def transform_extra_body( + self, + extra_body: Mapping[str, object], + request: Mapping[str, object], + model: str, + litellm_params: GenericLiteLLMParams, + ) -> Mapping[str, object]: + if self._collaboration is None: + return super().transform_extra_body(extra_body, request, model, litellm_params) + merged: Final = self.normalize_responses_api_request_dict({**self._unprojected_request, **extra_body}) + return self._collaboration.prepare_request(_RESPONSE_OBJECT.validate_python(merged)) + + def transform_response_api_response( + self, + model: str, + raw_response: httpx.Response, + logging_obj: LiteLLMLoggingObj, + ) -> ResponsesAPIResponse: + response: Final = super().transform_response_api_response(model, raw_response, logging_obj) + if self._collaboration is None or not self._collaboration.active: + return response + restored: Final = ResponsesAPIResponse.model_validate( + self._collaboration.restore_response(_RESPONSE_OBJECT.validate_python(response.model_dump())) + ) + return response.model_copy( + update={"output": restored.output, "tools": restored.tools, "tool_choice": restored.tool_choice} + ) + def _normalize_stream_item_id(self, parsed_chunk: dict) -> dict: """Rewrite streamed item ids to one stable id per output_index. diff --git a/tests/unit/llms/github_copilot/responses/test_copilot_collaboration.py b/tests/unit/llms/github_copilot/responses/test_copilot_collaboration.py new file mode 100644 index 00000000000..220399b652e --- /dev/null +++ b/tests/unit/llms/github_copilot/responses/test_copilot_collaboration.py @@ -0,0 +1,468 @@ +import copy +import datetime +from typing import Final + +import httpx +import pytest +from pydantic import TypeAdapter + +from litellm.litellm_core_utils.litellm_logging import Logging +from litellm.llms.github_copilot.responses.collaboration import CopilotPlaintextCollaboration +from litellm.llms.github_copilot.responses.transformation import GithubCopilotResponsesAPIConfig +from litellm.types.router import GenericLiteLLMParams + +CLIENT_NAMESPACE: Final = "collaboration" +UPSTREAM_NAMESPACE: Final = "codex_collaboration_plaintext" +JSON_OBJECT: Final = TypeAdapter(dict[str, object]) +MESSAGE: Final = 'Keep all content, including unicode λ and "quotes".\nSecond line.' +ARGUMENTS: Final = '{"message":"hello","target":"child"}' + + +def _tools(namespace: str = CLIENT_NAMESPACE, encrypted: bool = True) -> list[dict[str, object]]: + return [ + { + "type": "namespace", + "name": namespace, + "description": "Agent communication", + "tools": [ + { + "type": "function", + "name": action, + "parameters": { + "type": "object", + "properties": { + "message": { + "type": "string", + "description": "Full message", + **({"encrypted": True} if encrypted else {}), + }, + "target": {"type": "string", "encrypted": True}, + }, + "required": ["message"], + }, + } + for action in ("spawn_agent", "send_message", "followup_task") + ] + + [{"type": "function", "name": "wait_agent", "parameters": {"type": "object"}}], + } + ] + + +def _call(namespace: str = UPSTREAM_NAMESPACE, name: str = "send_message") -> dict[str, object]: + return { + "type": "function_call", + "id": "fc_call", + "call_id": "call_1", + "name": name, + "namespace": namespace, + "arguments": ARGUMENTS, + "status": "completed", + "provider_extension": {"keep": True}, + } + + +def _response(output: list[dict[str, object]]) -> dict[str, object]: + return { + "id": "resp_1", + "object": "response", + "created_at": 1, + "status": "completed", + "model": "gpt-6-astra", + "output": output, + "usage": {"input_tokens": 9, "output_tokens": 4, "total_tokens": 13}, + "provider_extension": {"keep": True}, + } + + +def _logging(stream: bool = False) -> Logging: + return Logging( + model="gpt-6-astra", + messages="hello", + stream=stream, + call_type="responses", + start_time=datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc), + litellm_call_id="call_logging", + function_id="function_logging", + supports_correlation_logging=False, + ) + + +def _activate(config: GithubCopilotResponsesAPIConfig) -> dict[str, object]: + return config.transform_responses_api_request( + model="gpt-6-astra", + input="hello", + response_api_optional_request_params={"tools": _tools(), "store": False}, + litellm_params=GenericLiteLLMParams(), + headers={}, + ) + + +def test_request_projection_preserves_message_text_ciphertext_and_nonmessage_schema() -> None: + adapter: Final = CopilotPlaintextCollaboration() + history: Final = [ + {"role": "user", "content": MESSAGE}, + {**_call(CLIENT_NAMESPACE), "encrypted_function_args": [{"name": "message", "content": "old-ciphertext"}]}, + {"type": "function_call_output", "call_id": "call_1", "output": "unchanged result"}, + {"type": "agent_message", "encrypted_content": "historical-ciphertext"}, + {"type": "reasoning", "encrypted_content": "reasoning-ciphertext", "summary": []}, + {"type": "additional_tools", "tools": _tools()}, + {"type": "tool_search_output", "call_id": "search_1", "tools": _tools()}, + ] + request: Final = {"model": "gpt-6-astra", "tools": _tools(), "input": history, "store": False} + original: Final = copy.deepcopy(request) + + projected: Final = adapter.prepare_request(request) + + assert projected == { + **request, + "tools": _tools(UPSTREAM_NAMESPACE, encrypted=False), + "input": [ + history[0], + {**history[1], "namespace": UPSTREAM_NAMESPACE}, + *history[2:5], + {"type": "additional_tools", "tools": _tools(UPSTREAM_NAMESPACE, encrypted=False)}, + {"type": "tool_search_output", "call_id": "search_1", "tools": _tools(UPSTREAM_NAMESPACE, encrypted=False)}, + ], + } + assert request == original + + +@pytest.mark.parametrize("inventory_type", ["additional_tools", "tool_search_output"]) +def test_historical_tool_inventory_can_activate_projection(inventory_type: str) -> None: + adapter: Final = CopilotPlaintextCollaboration() + request: Final = {"input": [{"type": inventory_type, "tools": _tools()}]} + + assert adapter.prepare_request(request) == { + "input": [{"type": inventory_type, "tools": _tools(UPSTREAM_NAMESPACE, encrypted=False)}], + "store": False, + } + assert adapter.restore_response({"output": [_call()]}) == { + "output": [{**_call(CLIENT_NAMESPACE), "encrypted_function_args": []}] + } + + +@pytest.mark.parametrize( + "tool", + [ + {"type": "namespace", "name": "collaboration", "tools": [{"type": "function", "name": "spawn_agent"}]}, + *_tools(encrypted=False), + *_tools(namespace="ordinary_tools"), + { + "type": "namespace", + "name": "collaboration", + "tools": [ + { + "type": "function", + "name": "unrelated", + "parameters": {"properties": {"message": {"type": "string", "encrypted": True}}}, + } + ], + }, + ], +) +def test_unrelated_collaboration_schemas_and_unactivated_responses_are_identity_passthrough( + tool: dict[str, object], +) -> None: + adapter: Final = CopilotPlaintextCollaboration() + request: Final = {"tools": [tool], "input": [{"role": "user", "content": MESSAGE}]} + response: Final = {"output": [_call()]} + event: Final = {"type": "response.output_item.added", "output_index": 0, "item": _call()} + + assert adapter.prepare_request(request) is request + assert adapter.restore_response(response) is response + assert adapter.restore_event(event) is event + + +@pytest.mark.parametrize( + ("choice", "expected"), + [ + ({"type": "namespace", "name": CLIENT_NAMESPACE}, {"type": "namespace", "name": UPSTREAM_NAMESPACE}), + ( + {"type": "function", "namespace": CLIENT_NAMESPACE, "name": "send_message"}, + {"type": "function", "namespace": UPSTREAM_NAMESPACE, "name": "send_message"}, + ), + ( + {"type": "function", "name": "collaboration.spawn_agent"}, + {"type": "function", "name": "codex_collaboration_plaintext.spawn_agent"}, + ), + ( + {"type": "function", "name": "collaboration__followup_task"}, + {"type": "function", "name": "codex_collaboration_plaintext__followup_task"}, + ), + ], +) +def test_forced_and_allowed_tool_choices_follow_the_projected_inventory( + choice: dict[str, object], expected: dict[str, object] +) -> None: + adapter: Final = CopilotPlaintextCollaboration() + request: Final = {"tools": _tools(), "tool_choice": choice} + allowed: Final = { + "type": "allowed_tools", + "mode": "required", + "tools": [choice, {"type": "function", "name": "other"}], + } + + assert adapter.prepare_request(request) == { + "tools": _tools(UPSTREAM_NAMESPACE, encrypted=False), + "tool_choice": expected, + "store": False, + } + assert adapter.prepare_request({"tools": _tools(), "tool_choice": allowed}) == { + "tools": _tools(UPSTREAM_NAMESPACE, encrypted=False), + "tool_choice": {**allowed, "tools": [expected, {"type": "function", "name": "other"}]}, + "store": False, + } + + +@pytest.mark.parametrize( + "extra", + [ + {"store": True}, + {"previous_response_id": "resp_previous"}, + {"conversation": "conv_previous"}, + {"input": [{"type": "additional_tools", "tools": _tools(UPSTREAM_NAMESPACE)}]}, + ], +) +def test_active_projection_rejects_stateful_requests_and_alias_collisions(extra: dict[str, object]) -> None: + adapter: Final = CopilotPlaintextCollaboration() + + with pytest.raises(ValueError, match=r"stateless Responses|reserved by the Copilot adapter"): + adapter.prepare_request({"tools": _tools(), **extra}) + + +@pytest.mark.parametrize("separator", [".", "__"]) +def test_qualified_history_and_returned_names_keep_call_identifiers(separator: str) -> None: + adapter: Final = CopilotPlaintextCollaboration() + history: Final = { + "type": "function_call", + "id": "fc_previous", + "call_id": "call_previous", + "name": CLIENT_NAMESPACE + separator + "spawn_agent", + "arguments": ARGUMENTS, + } + + assert adapter.prepare_request({"tools": _tools(), "input": [history]}) == { + "tools": _tools(UPSTREAM_NAMESPACE, encrypted=False), + "input": [{**history, "namespace": UPSTREAM_NAMESPACE, "name": "spawn_agent"}], + "store": False, + } + assert adapter.restore_response( + {"output": [{**history, "name": UPSTREAM_NAMESPACE + separator + "spawn_agent"}]} + ) == {"output": [{**history, "namespace": CLIENT_NAMESPACE, "name": "spawn_agent", "encrypted_function_args": []}]} + + +def test_extra_body_tools_and_history_activate_projection_and_ordinary_replacement_disables_it() -> None: + config: Final = GithubCopilotResponsesAPIConfig(plaintext_collaboration=True) + primary: Final = config.transform_responses_api_request( + model="gpt-6-astra", + input="hello", + response_api_optional_request_params={"store": False}, + litellm_params=GenericLiteLLMParams(), + headers={}, + ) + overlay: Final = { + "tools": _tools(), + "tool_choice": {"type": "function", "namespace": CLIENT_NAMESPACE, "name": "send_message"}, + "input": [_call(CLIENT_NAMESPACE)], + } + + merged: Final = config.transform_extra_body(overlay, primary, "gpt-6-astra", GenericLiteLLMParams()) + + assert merged == { + **primary, + "tools": _tools(UPSTREAM_NAMESPACE, encrypted=False), + "tool_choice": {"type": "function", "namespace": UPSTREAM_NAMESPACE, "name": "send_message"}, + "input": [_call()], + } + restored: Final = config.transform_response_api_response( + "gpt-6-astra", httpx.Response(200, json=_response([_call()])), _logging() + ) + assert JSON_OBJECT.validate_json(restored.model_dump_json(exclude_none=True))["output"] == [ + {**_call(CLIENT_NAMESPACE), "encrypted_function_args": []} + ] + + replacement: Final = {"tools": _tools(encrypted=False)} + assert config.transform_extra_body(replacement, primary, "gpt-6-astra", GenericLiteLLMParams()) == { + **primary, + **replacement, + } + untouched: Final = config.transform_response_api_response( + "gpt-6-astra", httpx.Response(200, json=_response([_call(CLIENT_NAMESPACE)])), _logging() + ) + assert JSON_OBJECT.validate_json(untouched.model_dump_json(exclude_none=True))["output"] == [ + _call(CLIENT_NAMESPACE) + ] + + +@pytest.mark.parametrize( + "overlay", + [ + {"store": True}, + {"previous_response_id": "resp_previous"}, + {"conversation": {"id": "conv_previous"}}, + {"input": [{"type": "tool_search_output", "tools": _tools(UPSTREAM_NAMESPACE)}]}, + ], +) +def test_extra_body_cannot_bypass_stateless_and_namespace_guards(overlay: dict[str, object]) -> None: + config: Final = GithubCopilotResponsesAPIConfig(plaintext_collaboration=True) + primary: Final = _activate(config) + + with pytest.raises(ValueError, match=r"stateless Responses|reserved by the Copilot adapter"): + config.transform_extra_body(overlay, primary, "gpt-6-astra", GenericLiteLLMParams()) + + +def test_explicitly_disabled_provider_preserves_encryption_protocol() -> None: + config: Final = GithubCopilotResponsesAPIConfig(plaintext_collaboration=False) + raw: Final = httpx.Response(200, json=_response([_call(CLIENT_NAMESPACE)])) + + assert _activate(config)["tools"] == _tools() + parsed: Final = config.transform_response_api_response("gpt-6-astra", raw, _logging()) + assert JSON_OBJECT.validate_json(parsed.model_dump_json(exclude_none=True))["output"] == [_call(CLIENT_NAMESPACE)] + + +def test_provider_nonstream_restoration_survives_serialization_and_keeps_headers_and_usage() -> None: + config: Final = GithubCopilotResponsesAPIConfig(plaintext_collaboration=True) + logging: Final = _logging() + raw: Final = httpx.Response( + 200, + json={**_response([_call(), _call(name="wait_agent")]), "tools": _tools(UPSTREAM_NAMESPACE, encrypted=False)}, + headers={"x-request-id": "request_1", "x-ratelimit-remaining-requests": "10"}, + ) + assert _activate(config)["tools"] == _tools(UPSTREAM_NAMESPACE, encrypted=False) + + response: Final = config.transform_response_api_response("gpt-6-astra", raw, logging) + serialized: Final = JSON_OBJECT.validate_json(response.model_dump_json(exclude_none=True)) + + assert serialized["output"] == [ + {**_call(CLIENT_NAMESPACE), "encrypted_function_args": []}, + _call(CLIENT_NAMESPACE, name="wait_agent"), + ] + assert serialized["tools"] == _tools(encrypted=False) + assert serialized["provider_extension"] == {"keep": True} + assert response.usage is not None + assert (response.usage.input_tokens, response.usage.output_tokens, response.usage.total_tokens) == (9, 4, 13) + assert response.hidden_params["headers"] == dict(raw.headers) + assert response.hidden_params["additional_headers"] + assert logging.model_call_details["original_response"] == raw.text + + +def _stream(config: GithubCopilotResponsesAPIConfig, event: dict[str, object]) -> dict[str, object]: + return JSON_OBJECT.validate_json( + config.transform_streaming_response("gpt-6-astra", event, _logging(stream=True)).model_dump_json( + exclude_none=True + ) + ) + + +@pytest.mark.parametrize("identity", [{}, {"name": None, "namespace": None}, {"name": "", "namespace": ""}]) +def test_stream_restores_sparse_events_and_completed_snapshot_through_real_event_models( + identity: dict[str, object], +) -> None: + config: Final = GithubCopilotResponsesAPIConfig(plaintext_collaboration=True) + _activate(config) + added: Final = _stream(config, {"type": "response.output_item.added", "output_index": 0, "item": _call()}) + delta: Final = _stream( + config, + { + "type": "response.function_call_arguments.delta", + "output_index": 0, + "item_id": "fc_changed", + "namespace": UPSTREAM_NAMESPACE, + "name": "send_message", + "delta": ARGUMENTS[:12], + }, + ) + arguments: Final = _stream( + config, + { + "type": "response.function_call_arguments.done", + "output_index": 0, + "item_id": "fc_changed", + "arguments": ARGUMENTS, + **identity, + }, + ) + done: Final = _stream( + config, + { + "type": "response.output_item.done", + "output_index": 0, + "item": {"type": "function_call", "id": "fc_changed", "arguments": ARGUMENTS, **identity}, + }, + ) + completed: Final = _stream(config, {"type": "response.completed", "response": _response([_call()])}) + + assert added["item"] == {**_call(CLIENT_NAMESPACE), "encrypted_function_args": []} + assert delta == { + "type": "response.function_call_arguments.delta", + "output_index": 0, + "item_id": "fc_call", + "namespace": CLIENT_NAMESPACE, + "name": "send_message", + "delta": ARGUMENTS[:12], + } + assert arguments == { + "type": "response.function_call_arguments.done", + "output_index": 0, + "item_id": "fc_call", + "call_id": "call_1", + "arguments": ARGUMENTS, + "namespace": CLIENT_NAMESPACE, + "name": "send_message", + "encrypted_function_args": [], + } + assert done["item"] == { + "type": "function_call", + "id": "fc_call", + "call_id": "call_1", + "arguments": ARGUMENTS, + "name": "send_message", + "namespace": CLIENT_NAMESPACE, + "encrypted_function_args": [], + } + snapshot: Final = JSON_OBJECT.validate_python(completed["response"]) + assert snapshot["output"] == [{**_call(CLIENT_NAMESPACE), "encrypted_function_args": []}] + + +def test_custom_calls_keep_their_own_encryption_metadata_without_a_plaintext_dispatch_marker() -> None: + adapter: Final = CopilotPlaintextCollaboration() + adapter.prepare_request({"tools": _tools()}) + call: Final = { + "type": "custom_tool_call", + "call_id": "custom_1", + "namespace": UPSTREAM_NAMESPACE, + "name": "send_message", + "input": "opaque custom payload", + "encrypted_function_args": ["provider_metadata"], + } + restored: Final = {**call, "namespace": CLIENT_NAMESPACE} + + assert adapter.restore_response({"output": [call]}) == {"output": [restored]} + assert adapter.restore_event({"type": "response.output_item.done", "output_index": 0, "item": call}) == { + "type": "response.output_item.done", + "output_index": 0, + "item": restored, + } + + +@pytest.mark.parametrize("namespace", ["different_namespace", CLIENT_NAMESPACE]) +def test_stream_rejects_conflicting_tool_identity(namespace: str) -> None: + adapter: Final = CopilotPlaintextCollaboration() + adapter.prepare_request({"tools": _tools()}) + adapter.restore_event({"type": "response.output_item.added", "output_index": 0, "item": _call()}) + + with pytest.raises(ValueError, match="Conflicting Copilot collaboration identity"): + adapter.restore_event({"type": "response.output_item.done", "output_index": 0, "item": _call(namespace)}) + + +@pytest.mark.parametrize("encrypted_args", [["message"], [{"name": "message", "content": "ciphertext"}], None]) +def test_provider_never_labels_encrypted_arguments_as_plaintext(encrypted_args: object) -> None: + adapter: Final = CopilotPlaintextCollaboration() + adapter.prepare_request({"tools": _tools()}) + output: Final = {**_call(), "encrypted_function_args": encrypted_args} + + with pytest.raises(ValueError, match="encrypted arguments for a plaintext collaboration tool"): + adapter.restore_response({"output": [output]}) + + with pytest.raises(ValueError, match="encrypted arguments for a plaintext collaboration tool"): + adapter.restore_event({"type": "response.output_item.added", "output_index": 0, "item": output})