Merge d7656fb774 into 8d6b8dab3d

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

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.
 

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})