test(responses): cover comparison ID ownership and wire decoding

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

tests/integration/authorization/test_responses_id_ownership.py

@@ -3,7 +3,7 @@ from __future__ import annotations
 import json
 import os
 import uuid
-from collections.abc import Callable
+from collections.abc import Callable, Mapping
 from pathlib import Path
 from typing import Final, Literal
 from urllib.parse import urlsplit
@@ -16,7 +16,7 @@ from integration._support.process import owned_proxy
 from integration._support.responses_vendor import same_response
 from integration._support.wire import Reply, Request, Wire, wire_server
 from integration._support.worker_readiness import wait_for_worker_readiness
-from openai.types.responses import ResponseInputParam
+from openai.types.responses import ResponseCompletedEvent, ResponseInputParam
 from pydantic import BaseModel, ConfigDict, JsonValue, TypeAdapter
 
 pytestmark: Final = pytest.mark.shares_worker_state
@@ -138,6 +138,203 @@ def _every_id_route(client: openai.OpenAI, model: str, response_id: str) -> tupl
     )
 
 
+def _comparison_body(response_id: str, nested: bool) -> Mapping[str, object]:
+    options: Final = {"comparison_response_id": response_id}
+    if nested:
+        return {
+            "prompt_cache_options": {"mode": "explicit", "ttl": "12h"},
+            "extra_body": {"prompt_cache_options": options},
+        }
+    return {"prompt_cache_options": options}
+
+
+@pytest.mark.parametrize(
+    ("prefix", "nested", "stream"),
+    (("/v1", False, False), ("", True, False), ("/openai/v1", False, True), ("/v1", True, True)),
+)
+def test_regression_comparison_response_id_owner_repeated_requests_reach_raw_provider_id(
+    gateway: Gateway, prefix: str, nested: bool, stream: bool
+) -> None:
+    raw_id: Final = f"resp_comparison_{uuid.uuid4().hex}"
+
+    def respond(request: Request) -> Reply:
+        if request.target == "/v1/models":
+            return _model_discovery_reply()
+        assert (request.method, request.target) == ("POST", "/v1/responses")
+        body: Final = _JSON_OBJECT.validate_json(request.body)
+        if body["input"] == "create comparison response":
+            assert body == {"model": _MODEL, "input": "create comparison response"}, request.body
+            return Reply(body=_response(raw_id))
+        assert body == {
+            "model": _MODEL,
+            "input": "compare owned response",
+            "stream": stream,
+            "prompt_cache_options": {"comparison_response_id": raw_id},
+        }, request.body
+        if stream:
+            return Reply(
+                content_type="text/event-stream",
+                chunks=(
+                    b"data: "
+                    + json.dumps({"type": "response.completed", "response": json.loads(_response(raw_id))}).encode()
+                    + b"\n\n",
+                    b"data: [DONE]\n\n",
+                ),
+            )
+        return Reply(body=_response(raw_id))
+
+    with wire_server(respond) as wire, gateway.scenario() as scenario:
+        model: Final = _deployment(scenario, wire)
+        team: Final = scenario.team(models=[model])
+        user: Final = scenario.member(team)
+        with _client(gateway, _key(scenario, team, user, model), prefix) as client:
+            created: Final = client.responses.create(model=model, input="create comparison response")
+            assert created.id != raw_id, created.model_dump_json()
+            assert len(_contract_requests(wire)) == 1
+
+            def compare_again() -> None:
+                result: Final = client.responses.create(
+                    model=model,
+                    input="compare owned response",
+                    stream=stream,
+                    extra_body=_comparison_body(created.id, nested),
+                )
+                if isinstance(result, openai.Stream):
+                    events: Final = tuple(result)
+                    assert isinstance(events[-1], ResponseCompletedEvent), events
+                    assert events[-1].response.status == "completed", events
+                else:
+                    assert result.status == "completed", result.model_dump_json()
+
+            for _ in range(2):
+                compare_again()
+            assert len(_contract_requests(wire)) == 2
+
+
+@pytest.mark.parametrize("nested", (False, True))
+@pytest.mark.parametrize("stranger", ("user", "team"))
+def test_regression_comparison_response_id_other_identity_is_denied_before_dispatch(
+    gateway: Gateway, nested: bool, stranger: str
+) -> None:
+    raw_id: Final = f"resp_comparison_{uuid.uuid4().hex}"
+
+    def respond(request: Request) -> Reply:
+        if request.target == "/v1/models":
+            return _model_discovery_reply()
+        return Reply(body=_response(raw_id))
+
+    with wire_server(respond) as wire, gateway.scenario() as scenario:
+        model: Final = _deployment(scenario, wire)
+        team_a: Final = scenario.team(models=[model])
+        team_b: Final = scenario.team(models=[model])
+        user_a: Final = scenario.member(team_a)
+        user_b: Final = scenario.member(team_a)
+        gateway.post("/team/member_add", {"team_id": team_b, "member": {"role": "user", "user_id": user_a}})
+        with (
+            _client(gateway, _key(scenario, team_a, user_a, model), "/v1") as owner,
+            _client(
+                gateway,
+                _key(
+                    scenario, team_a if stranger == "user" else team_b, user_b if stranger == "user" else user_a, model
+                ),
+                "/v1",
+            ) as other,
+        ):
+            created: Final = owner.responses.create(model=model, input="create private comparison")
+            assert created.id != raw_id
+            assert len(_contract_requests(wire)) == 1
+            _refused(
+                lambda: other.responses.create(
+                    model=model, input="compare another identity", extra_body=_comparison_body(created.id, nested)
+                ),
+                _OTHER_USER_DETAIL if stranger == "user" else _OTHER_TEAM_DETAIL,
+            )
+            assert _contract_requests(wire) == (), "A cross-identity comparison reached the provider"
+
+
+@pytest.mark.parametrize("nested", (False, True))
+@pytest.mark.parametrize("comparison_id", ("resp_unmanaged_comparison", "resp_!!!not-base64", "cmF3X3ZlbmRvcl9pZA=="))
+def test_guard_unmanaged_comparison_response_ids_pass_through_without_opt_in(
+    gateway: Gateway, nested: bool, comparison_id: str
+) -> None:
+    def respond(request: Request) -> Reply:
+        if request.target == "/v1/models":
+            return _model_discovery_reply()
+        assert (request.method, request.target) == ("POST", "/v1/responses")
+        assert _JSON_OBJECT.validate_json(request.body) == {
+            "model": _MODEL,
+            "input": "unmanaged comparison",
+            "prompt_cache_options": {"comparison_response_id": comparison_id},
+        }, request.body
+        return Reply(body=_response(f"resp_result_{uuid.uuid4().hex}"))
+
+    with wire_server(respond) as wire, gateway.scenario() as scenario:
+        model: Final = _deployment(scenario, wire)
+        team: Final = scenario.team(models=[model])
+        user: Final = scenario.member(team)
+        with _client(gateway, _key(scenario, team, user, model), "/v1") as client:
+            compared: Final = client.responses.create(
+                model=model, input="unmanaged comparison", extra_body=_comparison_body(comparison_id, nested)
+            )
+            assert compared.status == "completed", compared.model_dump_json()
+            assert len(_contract_requests(wire)) == 1
+
+
+@pytest.mark.parametrize("effective_owner", (False, True), ids=("guard-raw", "regression-owned"))
+def test_comparison_extra_body_ignores_foreign_top_level(gateway: Gateway, effective_owner: bool) -> None:
+    foreign_raw_id: Final = f"resp_ignored_{uuid.uuid4().hex}"
+    owner_raw_id: Final = f"resp_effective_{uuid.uuid4().hex}"
+    unmanaged_id: Final = "resp_raw_effective_comparison"
+
+    def respond(request: Request) -> Reply:
+        if request.target == "/v1/models":
+            return _model_discovery_reply()
+        assert (request.method, request.target) == ("POST", "/v1/responses")
+        body: Final = _JSON_OBJECT.validate_json(request.body)
+        if body["input"] in ("seed ignored comparison", "seed effective comparison"):
+            assert body == {"model": _MODEL, "input": body["input"]}, request.body
+            return Reply(body=_response(foreign_raw_id if body["input"] == "seed ignored comparison" else owner_raw_id))
+        assert body == {
+            "model": _MODEL,
+            "input": "compare effective options",
+            "prompt_cache_options": {"comparison_response_id": owner_raw_id if effective_owner else unmanaged_id},
+        }, request.body
+        return Reply(body=_response(owner_raw_id))
+
+    with wire_server(respond) as wire, gateway.scenario() as scenario:
+        model: Final = _deployment(scenario, wire)
+        team: Final = scenario.team(models=[model])
+        owner_user: Final = scenario.member(team)
+        foreign_user: Final = scenario.member(team)
+        with (
+            _client(gateway, _key(scenario, team, owner_user, model), "/v1") as owner,
+            _client(gateway, _key(scenario, team, foreign_user, model), "/v1") as foreign,
+        ):
+            foreign_response: Final = foreign.responses.create(model=model, input="seed ignored comparison")
+            assert foreign_response.id != foreign_raw_id
+            owner_response: Final = owner.responses.create(model=model, input="seed effective comparison")
+            assert owner_response.id != owner_raw_id
+            assert len(_contract_requests(wire)) == 2
+            compared: Final = owner.responses.create(
+                model=model,
+                input="compare effective options",
+                extra_body={
+                    "prompt_cache_options": {
+                        "comparison_response_id": foreign_response.id,
+                        "mode": "explicit",
+                        "ttl": "12h",
+                    },
+                    "extra_body": {
+                        "prompt_cache_options": {
+                            "comparison_response_id": owner_response.id if effective_owner else unmanaged_id
+                        }
+                    },
+                },
+            )
+            assert compared.status == "completed", compared.model_dump_json()
+            assert len(_contract_requests(wire)) == 1
+
+
 @pytest.mark.parametrize("prefix", _ALIASES)
 def test_response_ids_are_scoped_to_the_issuing_user_and_team(gateway: Gateway, prefix: str) -> None:
     raw_id: Final = f"resp_owned_{uuid.uuid4().hex}"

tests/integration/providers/test_responses_core_wire.py

@@ -3,8 +3,9 @@ from __future__ import annotations
 import json
 import os
 import uuid
-from collections.abc import Callable
-from typing import Final
+from collections.abc import AsyncIterator, Callable, Iterator
+from copy import deepcopy
+from typing import Final, Protocol, runtime_checkable
 
 import openai
 import pytest
@@ -14,6 +15,9 @@ from integration._support.worker_readiness import wait_for_worker_readiness
 from openai.types.responses import ResponseCompletedEvent
 from pydantic import JsonValue, TypeAdapter
 
+import litellm
+from litellm.types.llms.openai import ResponsesAPIResponse
+
 _MODEL: Final = "gpt-5"
 _API_KEY: Final = "synthetic-responses-key"
 _JSON_OBJECT: Final = TypeAdapter(dict[str, JsonValue])
@@ -21,6 +25,16 @@ _PROXY_WORKERS: Final = max(1, int(os.environ.get("INTEGRATION_PROXY_WORKERS", "
 _ALIASES: Final = ("/v1", "", "/openai/v1")
 
 
+@runtime_checkable
+class _AsyncResponseEvents(Protocol):
+    def __aiter__(self) -> AsyncIterator[object]: ...
+
+
+@runtime_checkable
+class _ResponseEvents(Protocol):
+    def __iter__(self) -> Iterator[object]: ...
+
+
 def _deployments_in_group(gateway: Gateway, group: str) -> int:
     entries: Final = gateway.get("/model/info")["data"]
     assert isinstance(entries, list), entries
@@ -69,6 +83,157 @@ def _stream(identity: str) -> tuple[bytes, ...]:
     ) + (b"data: [DONE]\n\n",)
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize("asynchronous", (False, True), ids=("sync", "async"))
+@pytest.mark.parametrize("stream", (False, True), ids=("nonstream", "stream"))
+@pytest.mark.parametrize("extra", (False, True), ids=("top-level", "extra-body-precedence"))
+async def test_regression_sdk_comparison_response_id_is_raw_on_the_wire(
+    asynchronous: bool, stream: bool, extra: bool
+) -> None:
+    raw_id: Final = f"resp_sdk_comparison_{uuid.uuid4().hex}"
+
+    def respond(request: Request) -> Reply:
+        assert (request.method, request.target) == ("POST", "/v1/responses")
+        body: Final = _JSON_OBJECT.validate_json(request.body)
+        if body["input"] == "seed comparison":
+            assert body == {"model": _MODEL, "input": "seed comparison"}, request.body
+            return Reply(body=_response(raw_id))
+        assert body == {
+            "model": _MODEL,
+            "input": "compare response",
+            "stream": stream,
+            "previous_response_id": raw_id,
+            "prompt_cache_options": {"comparison_response_id": raw_id},
+        }, request.body
+        return (
+            Reply(content_type="text/event-stream", chunks=_stream(raw_id)) if stream else Reply(body=_response(raw_id))
+        )
+
+    with wire_server(respond) as wire:
+        seeded: Final = litellm.responses(
+            model=f"openai/{_MODEL}", input="seed comparison", api_base=f"{wire.url}/v1", api_key=_API_KEY
+        )
+        assert isinstance(seeded, ResponsesAPIResponse), seeded
+        assert seeded.id != raw_id, seeded
+        params: Final = {
+            "model": f"openai/{_MODEL}",
+            "input": "compare response",
+            "stream": stream,
+            "previous_response_id": seeded.id,
+            "api_base": f"{wire.url}/v1",
+            "api_key": _API_KEY,
+            "num_retries": 0,
+            **(
+                {
+                    "prompt_cache_options": {"mode": "explicit", "ttl": "12h"},
+                    "extra_body": {"prompt_cache_options": {"comparison_response_id": seeded.id}},
+                }
+                if extra
+                else {"prompt_cache_options": {"comparison_response_id": seeded.id}}
+            ),
+        }
+        original: Final = deepcopy(params)
+        result: Final = await litellm.aresponses(**params) if asynchronous else litellm.responses(**params)
+        if stream:
+            if asynchronous:
+                assert isinstance(result, _AsyncResponseEvents), result
+                async_events: Final = tuple([event async for event in result])
+                async_completed: Final = ResponseCompletedEvent.model_validate(async_events[-1])
+                assert async_completed.response.status == "completed", async_events
+            else:
+                assert isinstance(result, _ResponseEvents), result
+                sync_events: Final = tuple(result)
+                sync_completed: Final = ResponseCompletedEvent.model_validate(sync_events[-1])
+                assert sync_completed.response.status == "completed", sync_events
+        else:
+            assert isinstance(result, ResponsesAPIResponse), result
+            assert result.status == "completed", result
+        assert params == original, "The caller's comparison options were mutated"
+        assert len(wire.drain()) == 2
+
+
+@pytest.mark.parametrize("extra", (False, True))
+@pytest.mark.parametrize("comparison_id", ("resp_custom_vendor_id", "resp_!!!not-base64", "cmF3X3ZlbmRvcl9pZA=="))
+def test_guard_sdk_custom_host_preserves_unmanaged_comparison_ids(extra: bool, comparison_id: str) -> None:
+    def respond(request: Request) -> Reply:
+        assert (request.method, request.target) == ("POST", "/v1/responses")
+        assert _JSON_OBJECT.validate_json(request.body) == {
+            "model": _MODEL,
+            "input": "custom comparison",
+            "prompt_cache_options": {"comparison_response_id": comparison_id, "mode": "implicit", "ttl": "30m"},
+        }, request.body
+        return Reply(body=_response("resp_custom_result"))
+
+    with wire_server(respond) as wire:
+        options: Final = {
+            "comparison_response_id": comparison_id,
+            "mode": "implicit",
+            "ttl": "30m",
+        }
+        params: Final = (
+            {"extra_body": {"prompt_cache_options": options}} if extra else {"prompt_cache_options": options}
+        )
+        original: Final = deepcopy(params)
+        result: Final = litellm.responses(
+            model=f"openai/{_MODEL}", input="custom comparison", api_base=f"{wire.url}/v1", api_key=_API_KEY, **params
+        )
+        assert isinstance(result, ResponsesAPIResponse), result
+        assert result.status == "completed", result
+        assert params == original
+        assert len(wire.drain()) == 1
+
+
+@pytest.mark.parametrize("provider", ("azure", "xai"))
+@pytest.mark.parametrize("extra", (False, True))
+def test_guard_inherited_responses_providers_preserve_comparison_wrappers(provider: str, extra: bool) -> None:
+    raw_id: Final = f"resp_inherited_{uuid.uuid4().hex}"
+
+    def seed(request: Request) -> Reply:
+        assert _JSON_OBJECT.validate_json(request.body) == {"model": _MODEL, "input": "seed inherited comparison"}
+        return Reply(body=_response(raw_id))
+
+    with wire_server(seed) as seed_wire:
+        seeded: Final = litellm.responses(
+            model=f"openai/{_MODEL}",
+            input="seed inherited comparison",
+            api_base=f"{seed_wire.url}/v1",
+            api_key=_API_KEY,
+        )
+        assert isinstance(seeded, ResponsesAPIResponse), seeded
+        assert seeded.id != raw_id
+        assert len(seed_wire.drain()) == 1
+
+    def respond(request: Request) -> Reply:
+        assert request.method == "POST", request
+        expected_target: Final = (
+            "/openai/responses?api-version=2025-03-01-preview" if provider == "azure" else "/v1/responses"
+        )
+        assert request.target == expected_target, request.target
+        assert _JSON_OBJECT.validate_json(request.body) == {
+            "model": _MODEL,
+            "input": "inherited comparison",
+            "prompt_cache_options": {"comparison_response_id": seeded.id},
+        }, request.body
+        return Reply(body=_response(raw_id))
+
+    with wire_server(respond) as wire:
+        options: Final = {"comparison_response_id": seeded.id}
+        params: Final = (
+            {"extra_body": {"prompt_cache_options": options}} if extra else {"prompt_cache_options": options}
+        )
+        result: Final = litellm.responses(
+            model=f"{provider}/{_MODEL}",
+            input="inherited comparison",
+            api_base=wire.url if provider == "azure" else f"{wire.url}/v1",
+            api_key=_API_KEY,
+            api_version="2025-03-01-preview" if provider == "azure" else None,
+            **params,
+        )
+        assert isinstance(result, ResponsesAPIResponse), result
+        assert result.status == "completed", result
+        assert len(wire.drain()) == 1
+
+
 def _register_in_group(scenario: Scenario, wire: Wire, group: str, order: int | None = None) -> str:
     existing: Final = _deployments_in_group(scenario.gateway, group)
     created: Final = scenario.gateway.post(

tests/unit/test_responses_id_security.py

@@ -4,11 +4,16 @@ Tests for ResponsesIDSecurity hook.
 Tests the security hook that prevents user B from seeing response from user A.
 """
 
+from collections.abc import Mapping
+from typing import Final
 from unittest.mock import MagicMock, patch
 
 import pytest
 from fastapi import HTTPException
+from pydantic import TypeAdapter
 
+from litellm.caching.caching import DualCache
+from litellm.proxy._types import UserAPIKeyAuth
 from litellm.proxy.hooks.responses_id_security import (
     ResponsesIDSecurity,
     _streams_response_ids,
@@ -854,6 +859,57 @@ def _issue_managed_id(hook, owner, provider_response_id=_FABRICATED_PROVIDER_RES
     return issued.id
 
 
+def _comparison_options(data: object, nested: bool) -> Mapping[str, str]:
+    request: Final = TypeAdapter(dict[str, object]).validate_python(data)
+    container: Final = TypeAdapter(dict[str, object]).validate_python(request["extra_body"]) if nested else request
+    return TypeAdapter(dict[str, str]).validate_python(container["prompt_cache_options"])
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("nested", (False, True))
+@pytest.mark.parametrize("stranger", ("user", "team"))
+async def test_regression_comparison_id_reentry_keeps_ownership_after_owner_retry(
+    salt_key_env: str, nested: bool, stranger: str
+) -> None:
+    hook: Final = ResponsesIDSecurity(general_settings_reader=lambda: {}, signing_key_reader=lambda: salt_key_env)
+    owner: Final = UserAPIKeyAuth(user_id="owner-user", team_id="owner-team")
+    caller: Final = UserAPIKeyAuth(
+        user_id="stranger-user" if stranger == "user" else owner.user_id,
+        team_id="stranger-team" if stranger == "team" else owner.team_id,
+    )
+    cache: Final = DualCache()
+    issued: Final = ResponsesAPIResponse.model_validate(
+        await hook.async_post_call_success_hook(  # pyright: ignore[reportUnknownMemberType]  # Legacy dict boundary
+            data={},
+            user_api_key_dict=owner,
+            response=ResponsesAPIResponse(
+                id=_FABRICATED_PROVIDER_RESPONSE_ID, created_at=1234567890, output=[], status="completed"
+            ),
+        )
+    )
+    assert issued.id != _FABRICATED_PROVIDER_RESPONSE_ID
+    options: Final = {"comparison_response_id": issued.id}
+    request: Final = {"extra_body": {"prompt_cache_options": options}} if nested else {"prompt_cache_options": options}
+    first: Final = TypeAdapter(dict[str, object]).validate_python(
+        await hook.async_pre_call_hook(  # pyright: ignore[reportUnknownMemberType]  # Legacy dict boundary
+            user_api_key_dict=owner, cache=cache, data=request, call_type="aresponses"
+        )
+    )
+    retried: Final = TypeAdapter(dict[str, object]).validate_python(
+        await hook.async_pre_call_hook(  # pyright: ignore[reportUnknownMemberType]  # Legacy dict boundary
+            user_api_key_dict=owner, cache=cache, data=request, call_type="aresponses"
+        )
+    )
+    with pytest.raises(HTTPException) as refusal:
+        await hook.async_pre_call_hook(  # pyright: ignore[reportUnknownMemberType]  # Legacy dict boundary
+            user_api_key_dict=caller, cache=cache, data=request, call_type="aresponses"
+        )
+    assert refusal.value.status_code == 403
+    assert f"not associated with the {stranger}" in refusal.value.detail
+    assert _comparison_options(first, nested) == {"comparison_response_id": _FABRICATED_PROVIDER_RESPONSE_ID}
+    assert _comparison_options(retried, nested) == {"comparison_response_id": _FABRICATED_PROVIDER_RESPONSE_ID}
+
+
 class TestUnrecognizedResponseIdIsRejected:
     """An id this proxy never issued carries no owner, so it must not reach the provider."""