From 0f8adafa112c5efa2d55cfa9496b330711634a8f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=AE=89=E8=BE=B0?= Date: Wed, 4 Nov 2026 20:27:18 +0800 Subject: [PATCH 1/6] feat(interactions): add QCA managed agents --- .github/workflows/test-unit.yml | 1 + ...r_002_qca_managed_interaction_transport.md | 56 +++ litellm/interactions/main.py | 41 ++ litellm/llms/qca/__init__.py | 1 + litellm/llms/qca/interactions/__init__.py | 3 + litellm/llms/qca/interactions/handler.py | 369 ++++++++++++++++++ litellm/proxy/google_endpoints/endpoints.py | 4 - litellm/types/utils.py | 1 + tests/unit/interactions/test_main.py | 42 ++ .../llms/qca/interactions/test_handler.py | 183 +++++++++ 10 files changed, 697 insertions(+), 4 deletions(-) create mode 100644 docs/adrs/adr_002_qca_managed_interaction_transport.md create mode 100644 litellm/llms/qca/__init__.py create mode 100644 litellm/llms/qca/interactions/__init__.py create mode 100644 litellm/llms/qca/interactions/handler.py create mode 100644 tests/unit/llms/qca/interactions/test_handler.py diff --git a/.github/workflows/test-unit.yml b/.github/workflows/test-unit.yml index 0a490bb8f45..c075c2b114c 100644 --- a/.github/workflows/test-unit.yml +++ b/.github/workflows/test-unit.yml @@ -346,6 +346,7 @@ jobs: tests/unit/llms/pg_vector tests/unit/llms/pi tests/unit/llms/publicai + tests/unit/llms/qca tests/unit/llms/ragflow tests/unit/llms/recraft tests/unit/llms/reducto diff --git a/docs/adrs/adr_002_qca_managed_interaction_transport.md b/docs/adrs/adr_002_qca_managed_interaction_transport.md new file mode 100644 index 00000000000..97c0292be6a --- /dev/null +++ b/docs/adrs/adr_002_qca_managed_interaction_transport.md @@ -0,0 +1,56 @@ +# QCA managed interaction transport + +## Status + +Accepted. + +## Context + +QCA Cloud Agents exposes a session and event API rather than a single model inference request. A client creates a +session, appends a user message, then follows the session event stream until the agent returns to an idle or terminal +state. The Interactions API expects one logical request with a stable interaction ID, normalized steps, and an optional +stream. + +The generic Interactions provider transformation supports one upstream HTTP request. Making it orchestrate several QCA +requests would mix transport state with request and response transformation. Routing QCA through the Responses API +bridge would also lose session continuation and event resume semantics. + +## Decision + +QCA uses a dedicated Interactions handler. It owns session creation, user event submission, SSE resume with +`Last-Event-ID`, terminal state detection, and normalization into LiteLLM Interactions responses. + +An interaction ID contains the QCA session ID and accepted input event ID. Continuation reuses the session and submits a +new user event. The input event ID is also the SSE resume cursor so events emitted between event acceptance and stream +connection are not missed. + +QCA credit fields remain provider usage extras under the normalized response usage object. LiteLLM does not convert +credits into a currency cost because QCA does not publish a stable conversion owned by LiteLLM. + +The first implementation accepts async text interactions only. Unsupported tools, files, background execution, and +lifecycle operations fail before any upstream request. + +## Alternatives considered + +Extending the generic transformation was rejected because its one-request contract cannot represent session creation, +event submission, and streaming without provider-specific branches throughout the shared handler. + +Mapping QCA to chat completions or Responses was rejected because neither surface preserves the accepted event cursor or +the managed session lifecycle. + +## Consequences + +The provider has a small dedicated transport module while the public Interactions entry point keeps one provider +dispatch branch. Router deployments can expose QCA through an ordinary model alias with server-owned credentials and an +environment ID. + +The interaction ID is an opaque continuation token for clients. QCA lifecycle endpoints remain unsupported until their +semantics can be mapped without pretending a session delete or cancel is an interaction operation. + +## References + +QCA documents the session creation contract at and event +submission at . + +The event stream and resume cursor are documented at +and . diff --git a/litellm/interactions/main.py b/litellm/interactions/main.py index acd43c9b830..506dfe8d823 100644 --- a/litellm/interactions/main.py +++ b/litellm/interactions/main.py @@ -305,6 +305,47 @@ def create( local_vars.update(kwargs) optional_params: Final = InteractionsAPIRequestUtils.get_requested_interactions_api_optional_params(local_vars) + if custom_llm_provider == "qca": + from litellm.llms.qca.interactions import create_qca_interaction + + if not _is_async: + raise litellm.UnsupportedParamsError( + message="QCA interactions require litellm.interactions.acreate", + llm_provider="qca", + model=model, + ) + litellm_logging_obj.update_from_kwargs( # pyright: ignore[reportUnknownMemberType] # old API untyped + kwargs=kwargs, + model=model, + optional_params=dict(optional_params), + litellm_params={"litellm_call_id": litellm_call_id}, + custom_llm_provider=custom_llm_provider, + ) + return create_qca_interaction( + agent=agent or model, + input=input, + environment=environment, + previous_interaction_id=previous_interaction_id, + stream=stream, + api_key=litellm_params.api_key if isinstance(litellm_params.api_key, str) else None, + api_base=litellm_params.api_base, + timeout=timeout, + logging_obj=litellm_logging_obj, + unsupported_params={ + "tools": tools, + "system_instruction": system_instruction, + "generation_config": generation_config, + "store": store, + "background": background, + "response_modalities": response_modalities, + "response_format": response_format, + "response_mime_type": response_mime_type, + "agent_config": agent_config, + "extra_headers": extra_headers, + "extra_body": extra_body, + }, + ) + # Check if this is a bridge provider (litellm_responses) - similar to responses API # Either provider is explicitly "litellm_responses" or no config found (bridge to responses) if custom_llm_provider == "litellm_responses" or interactions_api_config is None: diff --git a/litellm/llms/qca/__init__.py b/litellm/llms/qca/__init__.py new file mode 100644 index 00000000000..8b137891791 --- /dev/null +++ b/litellm/llms/qca/__init__.py @@ -0,0 +1 @@ + diff --git a/litellm/llms/qca/interactions/__init__.py b/litellm/llms/qca/interactions/__init__.py new file mode 100644 index 00000000000..ff418e45e9c --- /dev/null +++ b/litellm/llms/qca/interactions/__init__.py @@ -0,0 +1,3 @@ +from litellm.llms.qca.interactions.handler import create_qca_interaction + +__all__ = ["create_qca_interaction"] diff --git a/litellm/llms/qca/interactions/handler.py b/litellm/llms/qca/interactions/handler.py new file mode 100644 index 00000000000..e94aaa283a4 --- /dev/null +++ b/litellm/llms/qca/interactions/handler.py @@ -0,0 +1,369 @@ +from collections.abc import AsyncIterator, Mapping +from datetime import datetime, timezone +from typing import Final +from urllib.parse import urlsplit + +import httpx +from pydantic import BaseModel, ConfigDict, TypeAdapter + +from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj +from litellm.llms.custom_httpx.sse_lines import aiter_sse_lines +from litellm.types.interactions import InteractionsAPIResponse, InteractionsAPIStreamingResponse + +QCA_DEFAULT_API_BASE: Final = "https://api.qoder.com/api/v1/cloud" +_JSON_OBJECT: Final = TypeAdapter(dict[str, object]) + + +class QCAEventContent(BaseModel): + model_config = ConfigDict(frozen=True, extra="ignore") + + type: str + text: str | None = None + + +class QCAStopReason(BaseModel): + model_config = ConfigDict(frozen=True, extra="ignore") + + type: str + + +class QCAEvent(BaseModel): + model_config = ConfigDict(frozen=True, extra="ignore") + + id: str + type: str + content: tuple[QCAEventContent, ...] = () + stop_reason: QCAStopReason | None = None + model_credits: float | None = None + sandbox_runtime_credits: float | None = None + total_credits: float | None = None + + +class QCASessionAgent(BaseModel): + model_config = ConfigDict(frozen=True, extra="ignore") + + id: str + + +class QCASession(BaseModel): + model_config = ConfigDict(frozen=True, extra="ignore") + + id: str + agent: str | QCASessionAgent | None = None + environment_id: str | None = None + + +class QCAEventEnvelope(BaseModel): + model_config = ConfigDict(frozen=True, extra="ignore") + + data: QCAEvent | None = None + event: QCAEvent | None = None + + +class QCAEventsPage(BaseModel): + model_config = ConfigDict(frozen=True, extra="ignore") + + data: tuple[QCAEvent, ...] + + +def _require_identifier(value: str | None, prefix: str, name: str) -> str: + if value is None or not value.startswith(prefix): + raise ValueError(f"QCA {name} must start with {prefix!r}") + return value + + +def _api_base(value: str | None) -> str: + base: Final = (value or QCA_DEFAULT_API_BASE).rstrip("/") + parsed: Final = urlsplit(base) + invalid: Final = ( + parsed.scheme not in {"http", "https"} + or not parsed.netloc + or parsed.username is not None + or bool(parsed.query) + or bool(parsed.fragment) + ) + if invalid: + raise ValueError("QCA api_base must be an absolute HTTP URL without credentials, query, or fragment") + return base + + +def _text_input(value: object) -> str: + if not isinstance(value, str) or not value: + raise ValueError("QCA interactions currently require a non-empty text input") + return value + + +def _previous_session_id(previous_interaction_id: str | None) -> str | None: + if previous_interaction_id is None: + return None + parts: Final = previous_interaction_id.split(".") + if len(parts) != 3 or parts[0] != "qca1": + raise ValueError("QCA previous_interaction_id is invalid") + _require_identifier(parts[2], "evt_", "event id") + return _require_identifier(parts[1], "sess_", "session id") + + +def _event_from_payload(payload: str) -> QCAEvent: + decoded: Final = _JSON_OBJECT.validate_json(payload) + envelope: Final = QCAEventEnvelope.model_validate(decoded) + return envelope.data or envelope.event or QCAEvent.model_validate(decoded) + + +def _accepted_event(payload: object) -> QCAEvent: + page: Final = QCAEventsPage.model_validate(payload) + if not page.data: + raise ValueError("QCA accepted the user message without returning its event") + return page.data[0] + + +def _interaction_id(session_id: str, input_event_id: str) -> str: + return f"qca1.{session_id}.{input_event_id}" + + +def _usage(event: QCAEvent) -> Mapping[str, object] | None: + if event.total_credits is None and event.model_credits is None and event.sandbox_runtime_credits is None: + return None + return { + "qca_session_credits": { + "model_credits": event.model_credits, + "sandbox_runtime_credits": event.sandbox_runtime_credits, + "total_credits": event.total_credits, + }, + "qca_usage_scope": "session_cumulative", + "qca_usage_is_final": event.type == "session.status_idle", + } + + +def _text_step(event: QCAEvent) -> Mapping[str, object] | None: + text: Final = "".join(content.text or "" for content in event.content if content.type == "text") + if event.type != "agent.message" or not text: + return None + return {"type": "message", "id": event.id, "role": "assistant", "content": [{"type": "text", "text": text}]} + + +def _terminal_status(event: QCAEvent) -> str | None: + if event.type in {"session.error", "session.deleted", "session.status_terminated"}: + return "failed" + if event.type != "session.status_idle": + return None + if event.stop_reason is not None and event.stop_reason.type == "requires_action": + return "requires_action" + return "completed" + + +def _normalized_event(event: QCAEvent) -> tuple[Mapping[str, object] | None, str | None]: + return _text_step(event), _terminal_status(event) + + +async def _stream_events(response: httpx.Response) -> AsyncIterator[QCAEvent]: + data_lines: tuple[str, ...] = () # rebind-ok: immutable SSE record accumulation + async for line in aiter_sse_lines(response.aiter_text()): + if line == "": + if data_lines: + yield _event_from_payload("\n".join(data_lines)) + data_lines = () + continue + if line.startswith("data:"): + data_lines = (*data_lines, line[5:].lstrip()) + if data_lines: + yield _event_from_payload("\n".join(data_lines)) + + +async def _normalized_stream( + response: httpx.Response, interaction_id: str, agent: str +) -> AsyncIterator[InteractionsAPIStreamingResponse]: + try: + yield InteractionsAPIStreamingResponse( + event_type="interaction.created", + interaction_id=interaction_id, + interaction={"id": interaction_id, "agent": agent, "status": "in_progress", "steps": []}, + ) + async for event in _stream_events(response): + step, status = _normalized_event(event) + if step is not None: + yield InteractionsAPIStreamingResponse( + event_type="step.delta", + interaction_id=interaction_id, + index=0, + delta=dict(step), + ) + if status is not None: + yield InteractionsAPIStreamingResponse( + event_type=f"interaction.{status}", + interaction_id=interaction_id, + interaction={ + "id": interaction_id, + "agent": agent, + "status": status, + "usage": _usage(event), + }, + ) + return + raise httpx.RemoteProtocolError("QCA event stream ended before the session reached a terminal state") + finally: + await response.aclose() + + +async def _closing_stream( + stream: AsyncIterator[InteractionsAPIStreamingResponse], client: httpx.AsyncClient +) -> AsyncIterator[InteractionsAPIStreamingResponse]: + try: + async for chunk in stream: + yield chunk + finally: + await client.aclose() + + +def _response_from_terminal( + terminal: InteractionsAPIStreamingResponse, + interaction_id: str, + agent: str, + steps: tuple[Mapping[str, object], ...], +) -> InteractionsAPIResponse: + if terminal.interaction is None: + raise httpx.RemoteProtocolError("QCA terminal event is missing its interaction") + status: Final = terminal.interaction.get("status") + usage_value: Final = terminal.interaction.get("usage") + usage: Final = _JSON_OBJECT.validate_python(usage_value) if isinstance(usage_value, dict) else None + return InteractionsAPIResponse( + id=interaction_id, + agent=agent, + status=status if isinstance(status, str) else None, + steps=[dict(step) for step in steps], + usage=usage, + ) + + +async def _observed_stream( + stream: AsyncIterator[InteractionsAPIStreamingResponse], + interaction_id: str, + agent: str, + logging_obj: LiteLLMLoggingObj, + start_time: datetime, +) -> AsyncIterator[InteractionsAPIStreamingResponse]: + steps: tuple[Mapping[str, object], ...] = () # rebind-ok: immutable response accumulation + async for chunk in stream: + if chunk.event_type == "step.delta" and chunk.delta is not None: + steps = (*steps, chunk.delta) + if chunk.event_type in {"interaction.completed", "interaction.requires_action", "interaction.failed"}: + await logging_obj.dispatch_success_handlers( # pyright: ignore[reportUnknownMemberType] # old API untyped + _response_from_terminal(chunk, interaction_id, agent, steps), + start_time=start_time, + end_time=datetime.now(timezone.utc), + cache_hit=None, + prefer_async_handlers=True, + ) + yield chunk + + +async def _collect_response( + stream: AsyncIterator[InteractionsAPIStreamingResponse], interaction_id: str, agent: str +) -> InteractionsAPIResponse: + steps: tuple[Mapping[str, object], ...] = () # rebind-ok: immutable response accumulation + terminal: InteractionsAPIStreamingResponse | None = None # rebind-ok: captures the terminal chunk + async for chunk in stream: + if chunk.event_type == "step.delta" and chunk.delta is not None: + steps = (*steps, chunk.delta) + if chunk.event_type in {"interaction.completed", "interaction.requires_action", "interaction.failed"}: + terminal = chunk + if terminal is None: + raise httpx.RemoteProtocolError("QCA event stream did not produce a terminal interaction") + return _response_from_terminal(terminal, interaction_id, agent, steps) + + +async def _session_id( + client: httpx.AsyncClient, + base: str, + headers: Mapping[str, str], + agent_id: str, + environment_id: str, + previous_session_id: str | None, +) -> str: + if previous_session_id is not None: + existing_response: Final = await client.get(f"{base}/sessions/{previous_session_id}", headers=headers) + existing_response.raise_for_status() + session: Final = QCASession.model_validate(existing_response.json()) + session_agent: Final = session.agent.id if isinstance(session.agent, QCASessionAgent) else session.agent + if session.id != previous_session_id or session_agent != agent_id or session.environment_id != environment_id: + raise ValueError("QCA previous_interaction_id does not match the configured agent and environment") + return previous_session_id + create_response: Final = await client.post( + f"{base}/sessions", + headers=headers, + json={"agent": agent_id, "environment_id": environment_id}, + ) + create_response.raise_for_status() + return _require_identifier(QCASession.model_validate(create_response.json()).id, "sess_", "session id") + + +async def create_qca_interaction( + *, + agent: str | None, + input: object, + environment: object, + previous_interaction_id: str | None, + stream: bool | None, + api_key: str | None, + api_base: str | None, + timeout: float | httpx.Timeout | None, + unsupported_params: Mapping[str, object], + logging_obj: LiteLLMLoggingObj, + client: httpx.AsyncClient | None = None, +) -> InteractionsAPIResponse | AsyncIterator[InteractionsAPIStreamingResponse]: + unsupported: Final = tuple(name for name, value in unsupported_params.items() if value not in (None, False, (), {})) + if unsupported: + raise ValueError(f"QCA interactions do not support {', '.join(sorted(unsupported))}") + agent_id: Final = _require_identifier(agent, "agent_", "agent id") + environment_id: Final = _require_identifier( + environment if isinstance(environment, str) else None, + "env_", + "environment id", + ) + token: Final = api_key + if not token: + raise ValueError("QCA api_key is required") + base: Final = _api_base(api_base) + text: Final = _text_input(input) + previous_session_id: Final = _previous_session_id(previous_interaction_id) + owns_client: Final = client is None + http_client: Final = client or httpx.AsyncClient(timeout=timeout) + start_time: Final = datetime.now(timezone.utc) + try: + headers: Final = {"Authorization": f"Bearer {token}", "Accept": "application/json"} + session_id: Final = await _session_id( + http_client, + base, + headers, + agent_id, + environment_id, + previous_session_id, + ) + event_response: Final = await http_client.post( + f"{base}/sessions/{session_id}/events", + headers=headers, + json={"events": [{"type": "user.message", "content": [{"type": "text", "text": text}]}]}, + ) + event_response.raise_for_status() + input_event: Final = _accepted_event(_JSON_OBJECT.validate_json(event_response.content)) + input_event_id: Final = _require_identifier(input_event.id, "evt_", "event id") + interaction_id: Final = _interaction_id(session_id, input_event_id) + stream_request: Final = http_client.build_request( + "GET", + f"{base}/sessions/{session_id}/events/stream", + headers={**headers, "Accept": "text/event-stream", "Last-Event-ID": input_event_id}, + ) + stream_response: Final = await http_client.send(stream_request, stream=True) + stream_response.raise_for_status() + normalized: Final = _normalized_stream(stream_response, interaction_id, agent_id) + if stream is True: + observed: Final = _observed_stream(normalized, interaction_id, agent_id, logging_obj, start_time) + return _closing_stream(observed, http_client) if owns_client else observed + result: Final = await _collect_response(normalized, interaction_id, agent_id) + except BaseException: + if owns_client: + await http_client.aclose() + raise + else: + if owns_client: + await http_client.aclose() + return result diff --git a/litellm/proxy/google_endpoints/endpoints.py b/litellm/proxy/google_endpoints/endpoints.py index c018f071b3c..12571bad0f0 100644 --- a/litellm/proxy/google_endpoints/endpoints.py +++ b/litellm/proxy/google_endpoints/endpoints.py @@ -446,10 +446,6 @@ async def create_interaction( data: Final = await read_request_body(request=request) - # Default to gemini provider for interactions - if "custom_llm_provider" not in data: - data["custom_llm_provider"] = "gemini" - processor: Final = ProxyBaseLLMRequestProcessing(data=data) try: return await processor.base_process_llm_request( diff --git a/litellm/types/utils.py b/litellm/types/utils.py index ee80b63f93b..0f672c5718e 100644 --- a/litellm/types/utils.py +++ b/litellm/types/utils.py @@ -4662,6 +4662,7 @@ class LlmProviders(str, Enum): CODESTRAL = "codestral" TEXT_COMPLETION_CODESTRAL = "text-completion-codestral" DASHSCOPE = "dashscope" + QCA = "qca" QWENCLOUD = "qwencloud" QWEN_AI_PLATFORM = "qwen_ai_platform" MODELSCOPE = "modelscope" diff --git a/tests/unit/interactions/test_main.py b/tests/unit/interactions/test_main.py index 5af5dd7d82a..ac3d4a8c199 100644 --- a/tests/unit/interactions/test_main.py +++ b/tests/unit/interactions/test_main.py @@ -12,6 +12,7 @@ from respx import MockRouter import litellm from litellm import interactions from litellm.integrations.custom_logger import CustomLogger +from litellm.types.interactions import InteractionsAPIResponse _JSON_OBJECT: Final = TypeAdapter(dict[str, JsonValue]) @@ -856,3 +857,44 @@ def test_create_streaming_hands_completed_interaction_to_success_callback( ) assert recorder.logged == (_GEMINI_COMPLETED_INTERACTION,) + + +@pytest.mark.asyncio +async def test_acreate_qca_runs_managed_session_flow(respx_mock: MockRouter) -> None: + base: Final = "https://qca.test/api/v1/cloud" + respx_mock.post(f"{base}/sessions").mock(return_value=httpx.Response(200, json={"id": "sess_public"})) + respx_mock.post(f"{base}/sessions/sess_public/events").mock( + return_value=httpx.Response(202, json={"data": [{"id": "evt_input", "type": "user.message"}]}) + ) + respx_mock.get(f"{base}/sessions/sess_public/events/stream").mock( + return_value=httpx.Response( + 200, + text=( + 'data: {"id":"evt_answer","type":"agent.message",' + '"content":[{"type":"text","text":"ready"}]}\n\n' + 'data: {"id":"evt_idle","type":"session.status_idle",' + '"stop_reason":{"type":"end_turn"}}\n\n' + ), + headers={"content-type": "text/event-stream"}, + ) + ) + + response: Final = await interactions.acreate( + model="qca/agent_public", + input="run", + environment="env_public", + api_key="qca-secret", + api_base=base, + ) + + assert isinstance(response, InteractionsAPIResponse) + assert response.id == "qca1.sess_public.evt_input" + assert response.status == "completed" + assert response.steps == [ + { + "type": "message", + "id": "evt_answer", + "role": "assistant", + "content": [{"type": "text", "text": "ready"}], + } + ] diff --git a/tests/unit/llms/qca/interactions/test_handler.py b/tests/unit/llms/qca/interactions/test_handler.py new file mode 100644 index 00000000000..d58e0e775be --- /dev/null +++ b/tests/unit/llms/qca/interactions/test_handler.py @@ -0,0 +1,183 @@ +import json +from typing import Final +from unittest.mock import AsyncMock, MagicMock + +import httpx +import pytest + +from litellm.llms.qca.interactions.handler import create_qca_interaction +from litellm.types.interactions import InteractionsAPIResponse + + +def _sse_event(payload: dict[str, object]) -> bytes: + return f"data: {json.dumps(payload)}\n\n".encode() + + +@pytest.mark.asyncio +async def test_create_qca_interaction_maps_session_events_to_completed_response() -> None: + requests: list[httpx.Request] = [] + + def route(request: httpx.Request) -> httpx.Response: + requests.append(request) + if request.url.path.endswith("/sessions"): + return httpx.Response(200, json={"id": "sess_test"}) + if request.url.path.endswith("/events"): + return httpx.Response(202, json={"data": [{"id": "evt_input", "type": "user.message"}]}) + body: Final = b"".join( + ( + _sse_event( + { + "id": "evt_answer", + "type": "agent.message", + "content": [{"type": "text", "text": "done"}], + } + ), + _sse_event( + { + "id": "evt_idle", + "type": "session.status_idle", + "stop_reason": {"type": "end_turn"}, + "total_credits": 3.5, + } + ), + ) + ) + return httpx.Response(200, content=body, headers={"content-type": "text/event-stream"}) + + async with httpx.AsyncClient(transport=httpx.MockTransport(route)) as client: + logging_obj: Final = MagicMock(dispatch_success_handlers=AsyncMock()) + response: Final = await create_qca_interaction( + agent="agent_test", + input="finish the task", + environment="env_test", + previous_interaction_id=None, + stream=False, + api_key="secret", + api_base="https://qca.test/api/v1/cloud", + timeout=30, + unsupported_params={}, + logging_obj=logging_obj, + client=client, + ) + + assert isinstance(response, InteractionsAPIResponse) + assert response.id == "qca1.sess_test.evt_input" + assert response.status == "completed" + assert response.steps == [ + { + "type": "message", + "id": "evt_answer", + "role": "assistant", + "content": [{"type": "text", "text": "done"}], + } + ] + assert response.usage == { + "qca_session_credits": { + "model_credits": None, + "sandbox_runtime_credits": None, + "total_credits": 3.5, + }, + "qca_usage_scope": "session_cumulative", + "qca_usage_is_final": True, + } + assert requests[-1].headers["last-event-id"] == "evt_input" + + +@pytest.mark.asyncio +async def test_create_qca_interaction_continues_session_and_streams_normalized_events() -> None: + requests: list[httpx.Request] = [] + + def route(request: httpx.Request) -> httpx.Response: + requests.append(request) + if request.method == "GET" and request.url.path.endswith("/sessions/sess_existing"): + return httpx.Response( + 200, + json={"id": "sess_existing", "agent": {"id": "agent_test"}, "environment_id": "env_test"}, + ) + if request.url.path.endswith("/events"): + return httpx.Response(202, json={"data": [{"id": "evt_next", "type": "user.message"}]}) + body: Final = b"".join( + ( + _sse_event( + { + "id": "evt_action", + "type": "session.status_idle", + "stop_reason": {"type": "requires_action"}, + } + ), + ) + ) + return httpx.Response(200, content=body, headers={"content-type": "text/event-stream"}) + + async with httpx.AsyncClient(transport=httpx.MockTransport(route)) as client: + logging_obj: Final = MagicMock(dispatch_success_handlers=AsyncMock()) + result: Final = await create_qca_interaction( + agent="agent_test", + input="continue", + environment="env_test", + previous_interaction_id="qca1.sess_existing.evt_old", + stream=True, + api_key="secret", + api_base="https://qca.test/api/v1/cloud", + timeout=30, + unsupported_params={}, + logging_obj=logging_obj, + client=client, + ) + assert not isinstance(result, InteractionsAPIResponse) + chunks: Final = [chunk async for chunk in result] + + assert [chunk.event_type for chunk in chunks] == ["interaction.created", "interaction.requires_action"] + assert chunks[-1].interaction_id == "qca1.sess_existing.evt_next" + assert all(not request.url.path.endswith("/sessions") for request in requests) + + +@pytest.mark.asyncio +async def test_create_qca_interaction_rejects_unsupported_features_before_network_call() -> None: + async with httpx.AsyncClient(transport=httpx.MockTransport(lambda _: httpx.Response(500))) as client: + logging_obj: Final = MagicMock(dispatch_success_handlers=AsyncMock()) + with pytest.raises(ValueError, match="background"): + await create_qca_interaction( + agent="agent_test", + input="hello", + environment="env_test", + previous_interaction_id=None, + stream=False, + api_key="secret", + api_base=None, + timeout=30, + unsupported_params={"background": True}, + logging_obj=logging_obj, + client=client, + ) + + +@pytest.mark.asyncio +async def test_create_qca_interaction_rejects_continuation_for_another_agent() -> None: + requests: list[httpx.Request] = [] + + def route(request: httpx.Request) -> httpx.Response: + requests.append(request) + return httpx.Response( + 200, + json={"id": "sess_existing", "agent": {"id": "agent_other"}, "environment_id": "env_test"}, + ) + + async with httpx.AsyncClient(transport=httpx.MockTransport(route)) as client: + logging_obj: Final = MagicMock(dispatch_success_handlers=AsyncMock()) + with pytest.raises(ValueError, match="configured agent and environment"): + await create_qca_interaction( + agent="agent_test", + input="continue", + environment="env_test", + previous_interaction_id="qca1.sess_existing.evt_old", + stream=False, + api_key="secret", + api_base=None, + timeout=30, + unsupported_params={}, + logging_obj=logging_obj, + client=client, + ) + + assert [request.method for request in requests] == ["GET"] From 82d8d8107416a16371adc3f509f4877797c967f0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=AE=89=E8=BE=B0?= Date: Thu, 5 Nov 2026 09:57:37 +0800 Subject: [PATCH 2/6] fix(interactions): secure Qoder managed sessions --- litellm/llms/qca/interactions/handler.py | 68 +++++++++-- litellm/proxy/hooks/responses_id_security.py | 18 ++- litellm/proxy/route_llm_request.py | 25 ++++ litellm/types/utils.py | 1 + tests/unit/interactions/test_main.py | 4 +- .../llms/qca/interactions/test_handler.py | 108 ++++++++++++++++-- tests/unit/proxy/test_route_llm_request.py | 29 ++++- tests/unit/test_responses_id_security.py | 63 ++++++++++ 8 files changed, 288 insertions(+), 28 deletions(-) diff --git a/litellm/llms/qca/interactions/handler.py b/litellm/llms/qca/interactions/handler.py index e94aaa283a4..f6e6fb82916 100644 --- a/litellm/llms/qca/interactions/handler.py +++ b/litellm/llms/qca/interactions/handler.py @@ -4,14 +4,16 @@ from typing import Final from urllib.parse import urlsplit import httpx -from pydantic import BaseModel, ConfigDict, TypeAdapter +from pydantic import BaseModel, ConfigDict, TypeAdapter, ValidationError +from litellm.exceptions import APIError, AuthenticationError, RateLimitError from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj from litellm.llms.custom_httpx.sse_lines import aiter_sse_lines from litellm.types.interactions import InteractionsAPIResponse, InteractionsAPIStreamingResponse QCA_DEFAULT_API_BASE: Final = "https://api.qoder.com/api/v1/cloud" _JSON_OBJECT: Final = TypeAdapter(dict[str, object]) +_JSON_OBJECTS: Final = TypeAdapter(list[dict[str, object]]) class QCAEventContent(BaseModel): @@ -116,6 +118,25 @@ def _accepted_event(payload: object) -> QCAEvent: return page.data[0] +def _provider_error(error: httpx.HTTPStatusError, model: str) -> Exception: + response: Final = error.response + message: Final = response.text or str(error) + match response.status_code: + case 401 | 403: + return AuthenticationError(message=message, llm_provider="qca", model=model, response=response) + case 429: + return RateLimitError(message=message, llm_provider="qca", model=model, response=response) + case status_code: + return APIError( + status_code=status_code, + message=message, + llm_provider="qca", + model=model, + request=response.request, + response=response, + ) + + def _interaction_id(session_id: str, input_event_id: str) -> str: return f"qca1.{session_id}.{input_event_id}" @@ -138,7 +159,19 @@ def _text_step(event: QCAEvent) -> Mapping[str, object] | None: text: Final = "".join(content.text or "" for content in event.content if content.type == "text") if event.type != "agent.message" or not text: return None - return {"type": "message", "id": event.id, "role": "assistant", "content": [{"type": "text", "text": text}]} + return {"type": "model_output", "content": [{"type": "text", "text": text}]} + + +def _text_delta(step: Mapping[str, object]) -> Mapping[str, object]: + try: + content: Final = _JSON_OBJECTS.validate_python(step.get("content")) + except ValidationError: + return {} + return content[0] if content else {} + + +def _step_from_delta(delta: Mapping[str, object]) -> Mapping[str, object]: + return {"type": "model_output", "content": [dict(delta)]} def _terminal_status(event: QCAEvent) -> str | None: @@ -185,7 +218,7 @@ async def _normalized_stream( event_type="step.delta", interaction_id=interaction_id, index=0, - delta=dict(step), + delta=dict(_text_delta(step)), ) if status is not None: yield InteractionsAPIStreamingResponse( @@ -234,6 +267,16 @@ def _response_from_terminal( ) +def _terminal_for_logging( + terminal: InteractionsAPIStreamingResponse, + interaction_id: str, + agent: str, + steps: tuple[Mapping[str, object], ...], +) -> InteractionsAPIStreamingResponse: + response: Final = _response_from_terminal(terminal, interaction_id, agent, steps) + return terminal.model_copy(update={"interaction": response.model_dump()}) + + async def _observed_stream( stream: AsyncIterator[InteractionsAPIStreamingResponse], interaction_id: str, @@ -244,14 +287,13 @@ async def _observed_stream( steps: tuple[Mapping[str, object], ...] = () # rebind-ok: immutable response accumulation async for chunk in stream: if chunk.event_type == "step.delta" and chunk.delta is not None: - steps = (*steps, chunk.delta) + steps = (*steps, _step_from_delta(chunk.delta)) if chunk.event_type in {"interaction.completed", "interaction.requires_action", "interaction.failed"}: - await logging_obj.dispatch_success_handlers( # pyright: ignore[reportUnknownMemberType] # old API untyped - _response_from_terminal(chunk, interaction_id, agent, steps), + await logging_obj.async_success_handler( + result=_terminal_for_logging(chunk, interaction_id, agent, steps), start_time=start_time, end_time=datetime.now(timezone.utc), cache_hit=None, - prefer_async_handlers=True, ) yield chunk @@ -263,7 +305,7 @@ async def _collect_response( terminal: InteractionsAPIStreamingResponse | None = None # rebind-ok: captures the terminal chunk async for chunk in stream: if chunk.event_type == "step.delta" and chunk.delta is not None: - steps = (*steps, chunk.delta) + steps = (*steps, _step_from_delta(chunk.delta)) if chunk.event_type in {"interaction.completed", "interaction.requires_action", "interaction.failed"}: terminal = chunk if terminal is None: @@ -310,7 +352,11 @@ async def create_qca_interaction( logging_obj: LiteLLMLoggingObj, client: httpx.AsyncClient | None = None, ) -> InteractionsAPIResponse | AsyncIterator[InteractionsAPIStreamingResponse]: - unsupported: Final = tuple(name for name, value in unsupported_params.items() if value not in (None, False, (), {})) + unsupported: Final = tuple( + name + for name, value in unsupported_params.items() + if value not in (None, (), {}) and (name == "store" or value is not False) + ) if unsupported: raise ValueError(f"QCA interactions do not support {', '.join(sorted(unsupported))}") agent_id: Final = _require_identifier(agent, "agent_", "agent id") @@ -359,6 +405,10 @@ async def create_qca_interaction( observed: Final = _observed_stream(normalized, interaction_id, agent_id, logging_obj, start_time) return _closing_stream(observed, http_client) if owns_client else observed result: Final = await _collect_response(normalized, interaction_id, agent_id) + except httpx.HTTPStatusError as error: + if owns_client: + await http_client.aclose() + raise _provider_error(error, agent_id) from error except BaseException: if owns_client: await http_client.aclose() diff --git a/litellm/proxy/hooks/responses_id_security.py b/litellm/proxy/hooks/responses_id_security.py index 0269a784445..be5b897e42f 100644 --- a/litellm/proxy/hooks/responses_id_security.py +++ b/litellm/proxy/hooks/responses_id_security.py @@ -22,6 +22,7 @@ from litellm.proxy.common_utils.encrypt_decrypt_utils import ( encrypt_value_helper, ) from litellm.responses.utils import ResponsesAPIRequestUtils +from litellm.types.interactions import InteractionsAPIResponse, InteractionsAPIStreamingResponse from litellm.types.llms.openai import ( BaseLiteLLMOpenAIResponseObject, ResponsesAPIResponse, @@ -41,7 +42,7 @@ if TYPE_CHECKING: _RESPONSES_API_PROVIDER_PREFIX: Final = "/openai" -_RESPONSES_API_ROUTES: Final = frozenset({"/v1/responses", "/responses"}) +_RESPONSES_API_ROUTES: Final = frozenset({"/v1/responses", "/responses", "/v1beta/interactions", "/interactions"}) _ADDRESSED_RESPONSE_ID_FIELD_BY_CALL_TYPE: Final[Mapping[str, str]] = MappingProxyType( { "aresponses": "previous_response_id", @@ -50,6 +51,7 @@ _ADDRESSED_RESPONSE_ID_FIELD_BY_CALL_TYPE: Final[Mapping[str, str]] = MappingPro "adelete_responses": "response_id", "acancel_responses": "response_id", "alist_input_items": "response_id", + "acreate_interaction": "previous_interaction_id", } ) @@ -79,7 +81,9 @@ def _proxy_signing_key() -> str | None: _RESPONSE_PAYLOAD_ADAPTER: Final = TypeAdapter(Mapping[str, object]) _REQUEST_DATA_ADAPTER: Final = TypeAdapter(dict[str, object]) _ResponseObjectT = TypeVar("_ResponseObjectT", bound=BaseLiteLLMOpenAIResponseObject) -_AddressedResultT = TypeVar("_AddressedResultT", bound=ResponsesAPIResponse | DeleteResponseResult) +_AddressedResultT = TypeVar( + "_AddressedResultT", bound=ResponsesAPIResponse | InteractionsAPIResponse | DeleteResponseResult +) def _response_payload(response_obj: object) -> Mapping[str, object] | None: @@ -94,7 +98,9 @@ def _rewrite_advertised_id( rewrite: Callable[[str], str], ) -> _ResponseObjectT: event_id: Final = getattr(event, "id", None) - if isinstance(event_id, str) and event_id.startswith("resp_"): + if isinstance(event_id, str) and ( + event_id.startswith("resp_") or isinstance(event, InteractionsAPIResponse | InteractionsAPIStreamingResponse) + ): setattr(event, "id", rewrite(event_id)) return event @@ -151,7 +157,9 @@ def _held_id_naming(response_id: str | None, held_ids: Mapping[str, str]) -> str ) -def _id_sent_back(response: ResponsesAPIResponse | DeleteResponseResult, held_ids: Mapping[str, str]) -> str | None: +def _id_sent_back( + response: ResponsesAPIResponse | InteractionsAPIResponse | DeleteResponseResult, held_ids: Mapping[str, str] +) -> str | None: held_id: Final = _held_id_naming(response.id, held_ids) if held_id is not None: return held_id @@ -455,7 +463,7 @@ class ResponsesIDSecurity(CustomLogger): if general_settings.get("disable_responses_id_security", False): return response - if isinstance(response, ResponsesAPIResponse | DeleteResponseResult): + if isinstance(response, ResponsesAPIResponse | InteractionsAPIResponse | DeleteResponseResult): return self._advertised_response(response, user_api_key_dict, data) return response diff --git a/litellm/proxy/route_llm_request.py b/litellm/proxy/route_llm_request.py index 148d465099f..a0a94e1e45b 100644 --- a/litellm/proxy/route_llm_request.py +++ b/litellm/proxy/route_llm_request.py @@ -144,6 +144,25 @@ def _raise_if_model_fully_blocked(llm_router: LitellmRouter, model_name: object, ) +def _raise_if_qca_deployment_selectors_are_overridden( + llm_router: LitellmRouter, model_name: object, request_data: Mapping[str, object] +) -> None: + if not isinstance(model_name, str): + return + deployment: Final = llm_router.registry().deployments(model_name, source="first_in_group") + configured_model: Final = deployment.litellm_params.model if deployment is not None else model_name + if not configured_model.startswith("qca/"): + return + overridden: Final = tuple(name for name in ("agent", "environment") if request_data.get(name) is not None) + if overridden: + raise ProxyException( + message=f"Qoder Cloud Agents deployment selectors cannot be overridden: {', '.join(overridden)}", + type="auth_error", + param=overridden[0], + code=status.HTTP_403_FORBIDDEN, + ) + + ROUTE_ENDPOINT_MAPPING: Final = { "acompletion": "/chat/completions", "atext_completion": "/completions", @@ -816,6 +835,12 @@ async def _route_request_single_attempt( # noqa: ANN202 # returns unawaited pr router_model_names: Final = llm_router.registry().names(kind="group_set") if llm_router is not None else [] is_proxy_admin_without_team: Final = team_id is None and is_proxy_admin_request(data) + if route_type == "acreate_interaction" and llm_router is not None: + interaction_request: Final = JSON_OBJECT_ADAPTER.validate_python(data) + _raise_if_qca_deployment_selectors_are_overridden( + llm_router, interaction_request.get("model"), interaction_request + ) + # Preprocess Google GenAI generate content requests if route_type in ["agenerate_content", "agenerate_content_stream"]: # Map generationConfig to config parameter for Google GenAI compatibility diff --git a/litellm/types/utils.py b/litellm/types/utils.py index 0f672c5718e..f752f2b1f71 100644 --- a/litellm/types/utils.py +++ b/litellm/types/utils.py @@ -800,6 +800,7 @@ CallTypesLiteral = Literal[ "list_mcp_tools", "asend_message", "send_message", + "acreate_interaction", "aresponses", "responses", "acreate_skill", diff --git a/tests/unit/interactions/test_main.py b/tests/unit/interactions/test_main.py index ac3d4a8c199..5b7347ef52c 100644 --- a/tests/unit/interactions/test_main.py +++ b/tests/unit/interactions/test_main.py @@ -892,9 +892,7 @@ async def test_acreate_qca_runs_managed_session_flow(respx_mock: MockRouter) -> assert response.status == "completed" assert response.steps == [ { - "type": "message", - "id": "evt_answer", - "role": "assistant", + "type": "model_output", "content": [{"type": "text", "text": "ready"}], } ] diff --git a/tests/unit/llms/qca/interactions/test_handler.py b/tests/unit/llms/qca/interactions/test_handler.py index d58e0e775be..7575d486ee4 100644 --- a/tests/unit/llms/qca/interactions/test_handler.py +++ b/tests/unit/llms/qca/interactions/test_handler.py @@ -1,11 +1,15 @@ import json +from datetime import datetime from typing import Final -from unittest.mock import AsyncMock, MagicMock +from unittest.mock import MagicMock import httpx import pytest +from litellm.exceptions import RateLimitError from litellm.llms.qca.interactions.handler import create_qca_interaction +from litellm.integrations.custom_logger import CustomLogger +from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj from litellm.types.interactions import InteractionsAPIResponse @@ -13,6 +17,33 @@ def _sse_event(payload: dict[str, object]) -> bytes: return f"data: {json.dumps(payload)}\n\n".encode() +class _SuccessRecorder(CustomLogger): + def __init__(self) -> None: + self.responses: list[object] = [] + + async def async_log_success_event( + self, kwargs: dict[str, object], response_obj: object, start_time: datetime, end_time: datetime + ) -> None: + self.responses.append(response_obj) + + +def _stream_logging(recorder: _SuccessRecorder) -> LiteLLMLoggingObj: + logging_obj: Final = LiteLLMLoggingObj( + model="qca/agent_test", + messages=[], + stream=True, + call_type="acreate_interaction", + start_time=datetime.now(), + litellm_call_id="qca-test-call", + function_id="qca-test-function", + dynamic_async_success_callbacks=[recorder], + ) + logging_obj.update_environment_variables( + model="qca/agent_test", user="", optional_params={}, litellm_params={"api_base": "https://qca.test"} + ) + return logging_obj + + @pytest.mark.asyncio async def test_create_qca_interaction_maps_session_events_to_completed_response() -> None: requests: list[httpx.Request] = [] @@ -45,7 +76,7 @@ async def test_create_qca_interaction_maps_session_events_to_completed_response( return httpx.Response(200, content=body, headers={"content-type": "text/event-stream"}) async with httpx.AsyncClient(transport=httpx.MockTransport(route)) as client: - logging_obj: Final = MagicMock(dispatch_success_handlers=AsyncMock()) + logging_obj: Final = MagicMock() response: Final = await create_qca_interaction( agent="agent_test", input="finish the task", @@ -65,9 +96,7 @@ async def test_create_qca_interaction_maps_session_events_to_completed_response( assert response.status == "completed" assert response.steps == [ { - "type": "message", - "id": "evt_answer", - "role": "assistant", + "type": "model_output", "content": [{"type": "text", "text": "done"}], } ] @@ -98,11 +127,18 @@ async def test_create_qca_interaction_continues_session_and_streams_normalized_e return httpx.Response(202, json={"data": [{"id": "evt_next", "type": "user.message"}]}) body: Final = b"".join( ( + _sse_event( + { + "id": "evt_answer", + "type": "agent.message", + "content": [{"type": "text", "text": "continued"}], + } + ), _sse_event( { "id": "evt_action", "type": "session.status_idle", - "stop_reason": {"type": "requires_action"}, + "stop_reason": {"type": "end_turn"}, } ), ) @@ -110,7 +146,8 @@ async def test_create_qca_interaction_continues_session_and_streams_normalized_e return httpx.Response(200, content=body, headers={"content-type": "text/event-stream"}) async with httpx.AsyncClient(transport=httpx.MockTransport(route)) as client: - logging_obj: Final = MagicMock(dispatch_success_handlers=AsyncMock()) + recorder: Final = _SuccessRecorder() + logging_obj: Final = _stream_logging(recorder) result: Final = await create_qca_interaction( agent="agent_test", input="continue", @@ -127,15 +164,24 @@ async def test_create_qca_interaction_continues_session_and_streams_normalized_e assert not isinstance(result, InteractionsAPIResponse) chunks: Final = [chunk async for chunk in result] - assert [chunk.event_type for chunk in chunks] == ["interaction.created", "interaction.requires_action"] + assert [chunk.event_type for chunk in chunks] == [ + "interaction.created", + "step.delta", + "interaction.completed", + ] + assert chunks[1].delta == {"type": "text", "text": "continued"} assert chunks[-1].interaction_id == "qca1.sess_existing.evt_next" + assert len(recorder.responses) == 1 + logged: Final = recorder.responses[0] + assert isinstance(logged, InteractionsAPIResponse) + assert logged.steps == [{"type": "model_output", "content": [{"type": "text", "text": "continued"}]}] assert all(not request.url.path.endswith("/sessions") for request in requests) @pytest.mark.asyncio async def test_create_qca_interaction_rejects_unsupported_features_before_network_call() -> None: async with httpx.AsyncClient(transport=httpx.MockTransport(lambda _: httpx.Response(500))) as client: - logging_obj: Final = MagicMock(dispatch_success_handlers=AsyncMock()) + logging_obj: Final = MagicMock() with pytest.raises(ValueError, match="background"): await create_qca_interaction( agent="agent_test", @@ -152,6 +198,48 @@ async def test_create_qca_interaction_rejects_unsupported_features_before_networ ) +@pytest.mark.asyncio +async def test_create_qca_interaction_rejects_store_false_before_network_call() -> None: + async with httpx.AsyncClient(transport=httpx.MockTransport(lambda _: httpx.Response(500))) as client: + with pytest.raises(ValueError, match="store"): + await create_qca_interaction( + agent="agent_test", + input="hello", + environment="env_test", + previous_interaction_id=None, + stream=False, + api_key="secret", + api_base=None, + timeout=30, + unsupported_params={"store": False}, + logging_obj=MagicMock(), + client=client, + ) + + +@pytest.mark.asyncio +async def test_create_qca_interaction_preserves_rate_limit_status() -> None: + def route(request: httpx.Request) -> httpx.Response: + return httpx.Response(429, text="rate limited", request=request) + + async with httpx.AsyncClient(transport=httpx.MockTransport(route)) as client: + with pytest.raises(RateLimitError) as error: + await create_qca_interaction( + agent="agent_test", + input="hello", + environment="env_test", + previous_interaction_id=None, + stream=False, + api_key="secret", + api_base=None, + timeout=30, + unsupported_params={}, + logging_obj=MagicMock(), + client=client, + ) + assert error.value.status_code == 429 + + @pytest.mark.asyncio async def test_create_qca_interaction_rejects_continuation_for_another_agent() -> None: requests: list[httpx.Request] = [] @@ -164,7 +252,7 @@ async def test_create_qca_interaction_rejects_continuation_for_another_agent() - ) async with httpx.AsyncClient(transport=httpx.MockTransport(route)) as client: - logging_obj: Final = MagicMock(dispatch_success_handlers=AsyncMock()) + logging_obj: Final = MagicMock() with pytest.raises(ValueError, match="configured agent and environment"): await create_qca_interaction( agent="agent_test", diff --git a/tests/unit/proxy/test_route_llm_request.py b/tests/unit/proxy/test_route_llm_request.py index 71f816ea183..d64ae2be705 100644 --- a/tests/unit/proxy/test_route_llm_request.py +++ b/tests/unit/proxy/test_route_llm_request.py @@ -26,11 +26,12 @@ from fastapi import HTTPException from litellm.caching.caching import DualCache from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj -from litellm.proxy._types import UserAPIKeyAuth +from litellm.proxy._types import ProxyException, UserAPIKeyAuth from litellm.proxy.route_llm_request import ( JSON_OBJECT_ADAPTER, ProxyModelNotFoundError, UserConfigRouters, + _raise_if_qca_deployment_selectors_are_overridden, endpoint_unsupported_by_model, route_request, user_config_routers, @@ -43,6 +44,32 @@ from litellm.router_strategy.adaptive_router.config import ( from litellm.types.router import RouterErrors, RouterGeneralSettings, RouterRateLimitError +@pytest.mark.parametrize("selector", ("agent", "environment")) +def test_qca_alias_rejects_request_selector_overrides(selector: str) -> None: + router: Final = litellm.Router( + model_list=[ + { + "model_name": "qoder-agent", + "litellm_params": { + "model": "qca/agent_configured", + "environment": "env_configured", + "api_key": "unit-provider-key", + }, + } + ] + ) + + with pytest.raises(ProxyException) as error: + _raise_if_qca_deployment_selectors_are_overridden( + router, + "qoder-agent", + {"model": "qoder-agent", selector: f"{selector}_attacker"}, + ) + + assert error.value.code == "403" + + + @pytest.mark.parametrize( "route_type, required_body_params", [ diff --git a/tests/unit/test_responses_id_security.py b/tests/unit/test_responses_id_security.py index 789fdbd453e..47fed5b7f6e 100644 --- a/tests/unit/test_responses_id_security.py +++ b/tests/unit/test_responses_id_security.py @@ -23,6 +23,7 @@ from litellm.proxy.hooks.responses_id_security import ( ) from litellm.proxy.spend_tracking.spend_tracking_utils import get_logging_payload from litellm.responses.utils import ResponsesAPIRequestUtils +from litellm.types.interactions import InteractionsAPIResponse, InteractionsAPIStreamingResponse from litellm.types.llms.openai import ( GenericEvent, ResponseAPIUsage, @@ -803,6 +804,68 @@ def _comparison_options(data: object, nested: bool) -> Mapping[str, str]: return TypeAdapter(dict[str, str]).validate_python(container["prompt_cache_options"]) +@pytest.mark.asyncio +async def test_interaction_continuation_id_is_bound_to_its_user_and_team(salt_key_env: str) -> None: + hook: Final = _hook(signing_key=salt_key_env) + owner: Final = _auth() + issued: Final = InteractionsAPIResponse.model_validate( + await hook.async_post_call_success_hook( # pyright: ignore[reportArgumentType] # interactions share this proxy hook + data={}, + user_api_key_dict=owner, + response=InteractionsAPIResponse(id="qca1.sess_owner.evt_input", status="completed"), + ) + ) + assert issued.id is not None + assert issued.id != "qca1.sess_owner.evt_input" + authorized: Final = TypeAdapter(dict[str, object]).validate_python( + await hook.async_pre_call_hook( + user_api_key_dict=owner, + cache=DualCache(), + data={"previous_interaction_id": issued.id}, + call_type="acreate_interaction", + ) + ) + assert authorized["previous_interaction_id"] == "qca1.sess_owner.evt_input" + + with pytest.raises(HTTPException) as refusal: + await hook.async_pre_call_hook( + user_api_key_dict=_auth(user_id="other-user"), + cache=DualCache(), + data={"previous_interaction_id": issued.id}, + call_type="acreate_interaction", + ) + assert refusal.value.status_code == 403 + + +@pytest.mark.asyncio +async def test_streamed_interaction_ids_are_bound_to_the_caller(salt_key_env: str) -> None: + hook: Final = _hook(signing_key=salt_key_env) + caller: Final = _auth().model_copy(update={"request_route": "/interactions"}) + + async def chunks(): + yield InteractionsAPIStreamingResponse( + event_type="interaction.created", id="qca1.sess_owner.evt_input", status="in_progress" + ) + + protected: Final = tuple( + [ + chunk + async for chunk in hook.async_post_call_streaming_iterator_hook( + user_api_key_dict=caller, + response=chunks(), + request_data={}, + ) + ] + ) + protected_id: Final = protected[0].id + assert protected_id is not None + assert hook._decrypt_response_id(protected_id) == ( # pyright: ignore[reportPrivateUsage] # verifies ownership payload + "qca1.sess_owner.evt_input", + "owner-user", + "owner-team", + ) + + @pytest.mark.asyncio @pytest.mark.parametrize("nested", (False, True)) @pytest.mark.parametrize("stranger", ("user", "team")) From 8ff4d9eb17d4f8aa6dbf640fa5d979dbe8c627c3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=AE=89=E8=BE=B0?= Date: Thu, 5 Nov 2026 10:26:42 +0800 Subject: [PATCH 3/6] fix(interactions): close managed session security gaps --- litellm/proxy/hooks/responses_id_security.py | 21 ++++++ litellm/proxy/route_llm_request.py | 6 +- litellm/types/utils.py | 9 +++ tests/unit/proxy/test_route_llm_request.py | 25 +++++++ tests/unit/test_responses_id_security.py | 75 +++++++++++++++++++ ui/litellm-dashboard/src/lib/http/schema.d.ts | 2 +- 6 files changed, 135 insertions(+), 3 deletions(-) diff --git a/litellm/proxy/hooks/responses_id_security.py b/litellm/proxy/hooks/responses_id_security.py index be5b897e42f..1a8539ae8f8 100644 --- a/litellm/proxy/hooks/responses_id_security.py +++ b/litellm/proxy/hooks/responses_id_security.py @@ -52,6 +52,9 @@ _ADDRESSED_RESPONSE_ID_FIELD_BY_CALL_TYPE: Final[Mapping[str, str]] = MappingPro "acancel_responses": "response_id", "alist_input_items": "response_id", "acreate_interaction": "previous_interaction_id", + "aget_interaction": "interaction_id", + "adelete_interaction": "interaction_id", + "acancel_interaction": "interaction_id", } ) @@ -97,6 +100,24 @@ def _rewrite_advertised_id( event: _ResponseObjectT, rewrite: Callable[[str], str], ) -> _ResponseObjectT: + if isinstance(event, InteractionsAPIStreamingResponse): + interaction_id: Final = event.interaction_id + rewritten_interaction_id: Final = rewrite(interaction_id) if isinstance(interaction_id, str) else None + interaction: Final = _response_payload(event.interaction) + nested_id: Final = None if interaction is None else interaction.get("id") + rewritten_interaction: Final = ( + {**interaction, "id": rewrite(nested_id)} + if interaction is not None and isinstance(nested_id, str) + else None + ) + if rewritten_interaction_id is not None or rewritten_interaction is not None: + return event.model_copy( + update={ + **({"interaction_id": rewritten_interaction_id} if rewritten_interaction_id is not None else {}), + **({"interaction": rewritten_interaction} if rewritten_interaction is not None else {}), + } + ) + event_id: Final = getattr(event, "id", None) if isinstance(event_id, str) and ( event_id.startswith("resp_") or isinstance(event, InteractionsAPIResponse | InteractionsAPIStreamingResponse) diff --git a/litellm/proxy/route_llm_request.py b/litellm/proxy/route_llm_request.py index a0a94e1e45b..17f4eed2883 100644 --- a/litellm/proxy/route_llm_request.py +++ b/litellm/proxy/route_llm_request.py @@ -149,8 +149,10 @@ def _raise_if_qca_deployment_selectors_are_overridden( ) -> None: if not isinstance(model_name, str): return - deployment: Final = llm_router.registry().deployments(model_name, source="first_in_group") - configured_model: Final = deployment.litellm_params.model if deployment is not None else model_name + alias_target: Final = llm_router.registry().resolve(model_name, kind="alias") + resolved_model: Final = alias_target if isinstance(alias_target, str) else model_name + deployment: Final = llm_router.registry().deployments(resolved_model, source="first_in_group") + configured_model: Final = deployment.litellm_params.model if deployment is not None else resolved_model if not configured_model.startswith("qca/"): return overridden: Final = tuple(name for name in ("agent", "environment") if request_data.get(name) is not None) diff --git a/litellm/types/utils.py b/litellm/types/utils.py index f752f2b1f71..f29ac089495 100644 --- a/litellm/types/utils.py +++ b/litellm/types/utils.py @@ -660,6 +660,12 @@ class CallTypes(str, Enum): ######################################################### create_interaction = "create_interaction" acreate_interaction = "acreate_interaction" + get_interaction = "get_interaction" + aget_interaction = "aget_interaction" + delete_interaction = "delete_interaction" + adelete_interaction = "adelete_interaction" + cancel_interaction = "cancel_interaction" + acancel_interaction = "acancel_interaction" ######################################################### # Container Call Types @@ -801,6 +807,9 @@ CallTypesLiteral = Literal[ "asend_message", "send_message", "acreate_interaction", + "aget_interaction", + "adelete_interaction", + "acancel_interaction", "aresponses", "responses", "acreate_skill", diff --git a/tests/unit/proxy/test_route_llm_request.py b/tests/unit/proxy/test_route_llm_request.py index d64ae2be705..afbd238dcb7 100644 --- a/tests/unit/proxy/test_route_llm_request.py +++ b/tests/unit/proxy/test_route_llm_request.py @@ -69,6 +69,31 @@ def test_qca_alias_rejects_request_selector_overrides(selector: str) -> None: assert error.value.code == "403" +@pytest.mark.parametrize("selector", ("agent", "environment")) +def test_qca_model_group_alias_rejects_request_selector_overrides(selector: str) -> None: + router: Final = litellm.Router( + model_list=[ + { + "model_name": "qoder-agent", + "litellm_params": { + "model": "qca/agent_configured", + "environment": "env_configured", + "api_key": "unit-provider-key", + }, + } + ], + model_group_alias={"public-agent": "qoder-agent"}, + ) + + with pytest.raises(ProxyException) as error: + _raise_if_qca_deployment_selectors_are_overridden( + router, + "public-agent", + {"model": "public-agent", selector: f"{selector}_attacker"}, + ) + + assert error.value.code == "403" + @pytest.mark.parametrize( "route_type, required_body_params", diff --git a/tests/unit/test_responses_id_security.py b/tests/unit/test_responses_id_security.py index 47fed5b7f6e..cb09e5e4603 100644 --- a/tests/unit/test_responses_id_security.py +++ b/tests/unit/test_responses_id_security.py @@ -866,6 +866,81 @@ async def test_streamed_interaction_ids_are_bound_to_the_caller(salt_key_env: st ) +@pytest.mark.asyncio +@pytest.mark.parametrize("call_type", ("aget_interaction", "adelete_interaction", "acancel_interaction")) +async def test_interaction_lifecycle_ids_are_authorized_before_provider_calls( + salt_key_env: str, call_type: CallTypesLiteral +) -> None: + hook: Final = _hook(signing_key=salt_key_env) + owner: Final = _auth() + issued_response: Final = hook._encrypt_response_id( # pyright: ignore[reportPrivateUsage] # mints ids the way create does + InteractionsAPIResponse(id="interaction-provider-id", status="completed"), + owner, + ) + issued: Final = InteractionsAPIResponse.model_validate(issued_response).id + assert issued is not None + authorized: Final = TypeAdapter(dict[str, object]).validate_python( + await hook.async_pre_call_hook( + user_api_key_dict=owner, + cache=DualCache(), + data={"interaction_id": issued}, + call_type=call_type, + ) + ) + + assert authorized["interaction_id"] == "interaction-provider-id" + + with pytest.raises(HTTPException) as refusal: + await hook.async_pre_call_hook( + user_api_key_dict=_auth(user_id="other-user"), + cache=DualCache(), + data={"interaction_id": issued}, + call_type=call_type, + ) + assert refusal.value.status_code == 403 + + +@pytest.mark.asyncio +@pytest.mark.parametrize("id_location", ("interaction_id", "interaction")) +async def test_streamed_interaction_schema_ids_are_bound_to_the_caller(salt_key_env: str, id_location: str) -> None: + hook: Final = _hook(signing_key=salt_key_env) + caller: Final = _auth().model_copy(update={"request_route": "/interactions"}) + provider_id: Final = "qca1.sess_owner.evt_input" + + async def chunks(): + yield InteractionsAPIStreamingResponse( + event_type="interaction.created", + **( + {"interaction_id": provider_id} + if id_location == "interaction_id" + else {"interaction": {"id": provider_id}} + ), + ) + + protected: Final = tuple( + [ + chunk + async for chunk in hook.async_post_call_streaming_iterator_hook( + user_api_key_dict=caller, + response=chunks(), + request_data={}, + ) + ] + ) + protected_id: Final = ( + protected[0].interaction_id + if id_location == "interaction_id" + else TypeAdapter(dict[str, str]).validate_python(protected[0].interaction)["id"] + ) + + assert protected_id != provider_id + assert hook._decrypt_response_id(protected_id) == ( # pyright: ignore[reportPrivateUsage] # verifies ownership payload + provider_id, + "owner-user", + "owner-team", + ) + + @pytest.mark.asyncio @pytest.mark.parametrize("nested", (False, True)) @pytest.mark.parametrize("stranger", ("user", "team")) diff --git a/ui/litellm-dashboard/src/lib/http/schema.d.ts b/ui/litellm-dashboard/src/lib/http/schema.d.ts index e2e1210bc2a..9d049d976a0 100644 --- a/ui/litellm-dashboard/src/lib/http/schema.d.ts +++ b/ui/litellm-dashboard/src/lib/http/schema.d.ts @@ -28910,7 +28910,7 @@ export interface components { * CallTypes * @enum {string} */ - CallTypes: "embedding" | "aembedding" | "completion" | "acompletion" | "atext_completion" | "text_completion" | "image_generation" | "aimage_generation" | "image_edit" | "aimage_edit" | "moderation" | "amoderation" | "atranscription" | "transcription" | "aspeech" | "speech" | "rerank" | "arerank" | "search" | "asearch" | "decisions" | "adecisions" | "systemone" | "asystemone" | "extraction" | "aextraction" | "_arealtime" | "_aresponses_websocket" | "acreate_realtime_client_secret" | "arealtime_calls" | "acreate_realtime_transcription_session" | "acreate_realtime_translation_client_secret" | "arealtime_translation_calls" | "create_batch" | "acreate_batch" | "aretrieve_batch" | "retrieve_batch" | "acancel_batch" | "cancel_batch" | "pass_through_endpoint" | "anthropic_messages" | "aanthropic_messages" | "get_assistants" | "aget_assistants" | "create_assistants" | "acreate_assistants" | "delete_assistant" | "adelete_assistant" | "acreate_thread" | "create_thread" | "aget_thread" | "get_thread" | "a_add_message" | "add_message" | "aget_messages" | "get_messages" | "arun_thread" | "run_thread" | "arun_thread_stream" | "run_thread_stream" | "afile_retrieve" | "file_retrieve" | "afile_delete" | "file_delete" | "afile_list" | "file_list" | "acreate_file" | "create_file" | "afile_content" | "file_content" | "create_fine_tuning_job" | "acreate_fine_tuning_job" | "create_video" | "acreate_video" | "video_generation" | "avideo_generation" | "avideo_retrieve" | "video_retrieve" | "avideo_content" | "video_content" | "video_remix" | "avideo_remix" | "video_list" | "avideo_list" | "video_retrieve_job" | "avideo_retrieve_job" | "video_delete" | "avideo_delete" | "video_create_character" | "avideo_create_character" | "video_get_character" | "avideo_get_character" | "video_edit" | "avideo_edit" | "video_extension" | "avideo_extension" | "vector_store_file_create" | "avector_store_file_create" | "vector_store_file_list" | "avector_store_file_list" | "vector_store_file_retrieve" | "avector_store_file_retrieve" | "vector_store_file_content" | "avector_store_file_content" | "vector_store_file_update" | "avector_store_file_update" | "vector_store_file_delete" | "avector_store_file_delete" | "vector_store_create" | "avector_store_create" | "vector_store_search" | "avector_store_search" | "ingest" | "aingest" | "query" | "aquery" | "create_interaction" | "acreate_interaction" | "create_container" | "acreate_container" | "list_containers" | "alist_containers" | "retrieve_container" | "aretrieve_container" | "delete_container" | "adelete_container" | "list_container_files" | "alist_container_files" | "upload_container_file" | "aupload_container_file" | "create_sandbox" | "acreate_sandbox" | "delete_sandbox" | "adelete_sandbox" | "run_code" | "arun_code" | "code_interpreter_tool" | "acode_interpreter_tool" | "acancel_fine_tuning_job" | "cancel_fine_tuning_job" | "alist_fine_tuning_jobs" | "list_fine_tuning_jobs" | "aretrieve_fine_tuning_job" | "retrieve_fine_tuning_job" | "responses" | "aresponses" | "get_responses" | "aget_responses" | "delete_responses" | "adelete_responses" | "cancel_responses" | "acancel_responses" | "alist_input_items" | "llm_passthrough_route" | "allm_passthrough_route" | "generate_content" | "agenerate_content" | "generate_content_stream" | "agenerate_content_stream" | "ocr" | "aocr" | "call_mcp_tool" | "list_mcp_tools" | "asend_message" | "send_message" | "acreate_skill"; + CallTypes: "embedding" | "aembedding" | "completion" | "acompletion" | "atext_completion" | "text_completion" | "image_generation" | "aimage_generation" | "image_edit" | "aimage_edit" | "moderation" | "amoderation" | "atranscription" | "transcription" | "aspeech" | "speech" | "rerank" | "arerank" | "search" | "asearch" | "decisions" | "adecisions" | "systemone" | "asystemone" | "extraction" | "aextraction" | "_arealtime" | "_aresponses_websocket" | "acreate_realtime_client_secret" | "arealtime_calls" | "acreate_realtime_transcription_session" | "acreate_realtime_translation_client_secret" | "arealtime_translation_calls" | "create_batch" | "acreate_batch" | "aretrieve_batch" | "retrieve_batch" | "acancel_batch" | "cancel_batch" | "pass_through_endpoint" | "anthropic_messages" | "aanthropic_messages" | "get_assistants" | "aget_assistants" | "create_assistants" | "acreate_assistants" | "delete_assistant" | "adelete_assistant" | "acreate_thread" | "create_thread" | "aget_thread" | "get_thread" | "a_add_message" | "add_message" | "aget_messages" | "get_messages" | "arun_thread" | "run_thread" | "arun_thread_stream" | "run_thread_stream" | "afile_retrieve" | "file_retrieve" | "afile_delete" | "file_delete" | "afile_list" | "file_list" | "acreate_file" | "create_file" | "afile_content" | "file_content" | "create_fine_tuning_job" | "acreate_fine_tuning_job" | "create_video" | "acreate_video" | "video_generation" | "avideo_generation" | "avideo_retrieve" | "video_retrieve" | "avideo_content" | "video_content" | "video_remix" | "avideo_remix" | "video_list" | "avideo_list" | "video_retrieve_job" | "avideo_retrieve_job" | "video_delete" | "avideo_delete" | "video_create_character" | "avideo_create_character" | "video_get_character" | "avideo_get_character" | "video_edit" | "avideo_edit" | "video_extension" | "avideo_extension" | "vector_store_file_create" | "avector_store_file_create" | "vector_store_file_list" | "avector_store_file_list" | "vector_store_file_retrieve" | "avector_store_file_retrieve" | "vector_store_file_content" | "avector_store_file_content" | "vector_store_file_update" | "avector_store_file_update" | "vector_store_file_delete" | "avector_store_file_delete" | "vector_store_create" | "avector_store_create" | "vector_store_search" | "avector_store_search" | "ingest" | "aingest" | "query" | "aquery" | "create_interaction" | "acreate_interaction" | "get_interaction" | "aget_interaction" | "delete_interaction" | "adelete_interaction" | "cancel_interaction" | "acancel_interaction" | "create_container" | "acreate_container" | "list_containers" | "alist_containers" | "retrieve_container" | "aretrieve_container" | "delete_container" | "adelete_container" | "list_container_files" | "alist_container_files" | "upload_container_file" | "aupload_container_file" | "create_sandbox" | "acreate_sandbox" | "delete_sandbox" | "adelete_sandbox" | "run_code" | "arun_code" | "code_interpreter_tool" | "acode_interpreter_tool" | "acancel_fine_tuning_job" | "cancel_fine_tuning_job" | "alist_fine_tuning_jobs" | "list_fine_tuning_jobs" | "aretrieve_fine_tuning_job" | "retrieve_fine_tuning_job" | "responses" | "aresponses" | "get_responses" | "aget_responses" | "delete_responses" | "adelete_responses" | "cancel_responses" | "acancel_responses" | "alist_input_items" | "llm_passthrough_route" | "allm_passthrough_route" | "generate_content" | "agenerate_content" | "generate_content_stream" | "agenerate_content_stream" | "ocr" | "aocr" | "call_mcp_tool" | "list_mcp_tools" | "asend_message" | "send_message" | "acreate_skill"; /** CallbackDelete */ CallbackDelete: { /** Callback Name */ From 658f49b1cc1191c2163dfb03c821be02c59110af Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=AE=89=E8=BE=B0?= Date: Thu, 5 Nov 2026 11:10:42 +0800 Subject: [PATCH 4/6] fix(interactions): harden Qoder lifecycle responses --- litellm/llms/qca/interactions/handler.py | 77 ++++++++++++------- litellm/proxy/hooks/responses_id_security.py | 31 ++++++-- litellm/proxy/route_llm_request.py | 3 +- .../llms/qca/interactions/test_handler.py | 18 +++-- tests/unit/proxy/test_route_llm_request.py | 35 ++++++++- tests/unit/test_responses_id_security.py | 35 ++++++++- 6 files changed, 155 insertions(+), 44 deletions(-) diff --git a/litellm/llms/qca/interactions/handler.py b/litellm/llms/qca/interactions/handler.py index f6e6fb82916..dbd9f56278f 100644 --- a/litellm/llms/qca/interactions/handler.py +++ b/litellm/llms/qca/interactions/handler.py @@ -6,7 +6,7 @@ from urllib.parse import urlsplit import httpx from pydantic import BaseModel, ConfigDict, TypeAdapter, ValidationError -from litellm.exceptions import APIError, AuthenticationError, RateLimitError +from litellm.exceptions import APIError, AuthenticationError, BadRequestError, RateLimitError from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj from litellm.llms.custom_httpx.sse_lines import aiter_sse_lines from litellm.types.interactions import InteractionsAPIResponse, InteractionsAPIStreamingResponse @@ -68,9 +68,13 @@ class QCAEventsPage(BaseModel): data: tuple[QCAEvent, ...] +class QCAValidationError(ValueError): + pass + + def _require_identifier(value: str | None, prefix: str, name: str) -> str: if value is None or not value.startswith(prefix): - raise ValueError(f"QCA {name} must start with {prefix!r}") + raise QCAValidationError(f"QCA {name} must start with {prefix!r}") return value @@ -85,13 +89,13 @@ def _api_base(value: str | None) -> str: or bool(parsed.fragment) ) if invalid: - raise ValueError("QCA api_base must be an absolute HTTP URL without credentials, query, or fragment") + raise QCAValidationError("QCA api_base must be an absolute HTTP URL without credentials, query, or fragment") return base def _text_input(value: object) -> str: if not isinstance(value, str) or not value: - raise ValueError("QCA interactions currently require a non-empty text input") + raise QCAValidationError("QCA interactions currently require a non-empty text input") return value @@ -100,7 +104,7 @@ def _previous_session_id(previous_interaction_id: str | None) -> str | None: return None parts: Final = previous_interaction_id.split(".") if len(parts) != 3 or parts[0] != "qca1": - raise ValueError("QCA previous_interaction_id is invalid") + raise QCAValidationError("QCA previous_interaction_id is invalid") _require_identifier(parts[2], "evt_", "event id") return _require_identifier(parts[1], "sess_", "session id") @@ -205,11 +209,19 @@ async def _stream_events(response: httpx.Response) -> AsyncIterator[QCAEvent]: async def _normalized_stream( response: httpx.Response, interaction_id: str, agent: str ) -> AsyncIterator[InteractionsAPIStreamingResponse]: + created: Final = datetime.now(timezone.utc).isoformat() try: yield InteractionsAPIStreamingResponse( event_type="interaction.created", interaction_id=interaction_id, - interaction={"id": interaction_id, "agent": agent, "status": "in_progress", "steps": []}, + interaction={ + "id": interaction_id, + "agent": agent, + "status": "in_progress", + "created": created, + "updated": created, + "steps": [], + }, ) async for event in _stream_events(response): step, status = _normalized_event(event) @@ -228,6 +240,8 @@ async def _normalized_stream( "id": interaction_id, "agent": agent, "status": status, + "created": created, + "updated": datetime.now(timezone.utc).isoformat(), "usage": _usage(event), }, ) @@ -257,11 +271,15 @@ def _response_from_terminal( raise httpx.RemoteProtocolError("QCA terminal event is missing its interaction") status: Final = terminal.interaction.get("status") usage_value: Final = terminal.interaction.get("usage") + created: Final = terminal.interaction.get("created") + updated: Final = terminal.interaction.get("updated") usage: Final = _JSON_OBJECT.validate_python(usage_value) if isinstance(usage_value, dict) else None return InteractionsAPIResponse( id=interaction_id, agent=agent, status=status if isinstance(status, str) else None, + created=created if isinstance(created, str) else None, + updated=updated if isinstance(updated, str) else None, steps=[dict(step) for step in steps], usage=usage, ) @@ -327,7 +345,7 @@ async def _session_id( session: Final = QCASession.model_validate(existing_response.json()) session_agent: Final = session.agent.id if isinstance(session.agent, QCASessionAgent) else session.agent if session.id != previous_session_id or session_agent != agent_id or session.environment_id != environment_id: - raise ValueError("QCA previous_interaction_id does not match the configured agent and environment") + raise QCAValidationError("QCA previous_interaction_id does not match the configured agent and environment") return previous_session_id create_response: Final = await client.post( f"{base}/sessions", @@ -352,25 +370,28 @@ async def create_qca_interaction( logging_obj: LiteLLMLoggingObj, client: httpx.AsyncClient | None = None, ) -> InteractionsAPIResponse | AsyncIterator[InteractionsAPIStreamingResponse]: - unsupported: Final = tuple( - name - for name, value in unsupported_params.items() - if value not in (None, (), {}) and (name == "store" or value is not False) - ) - if unsupported: - raise ValueError(f"QCA interactions do not support {', '.join(sorted(unsupported))}") - agent_id: Final = _require_identifier(agent, "agent_", "agent id") - environment_id: Final = _require_identifier( - environment if isinstance(environment, str) else None, - "env_", - "environment id", - ) - token: Final = api_key - if not token: - raise ValueError("QCA api_key is required") - base: Final = _api_base(api_base) - text: Final = _text_input(input) - previous_session_id: Final = _previous_session_id(previous_interaction_id) + try: + unsupported: Final = tuple( + name + for name, value in unsupported_params.items() + if value not in (None, (), {}) and (name == "store" or value is not False) + ) + if unsupported: + raise QCAValidationError(f"QCA interactions do not support {', '.join(sorted(unsupported))}") + agent_id: Final = _require_identifier(agent, "agent_", "agent id") + environment_id: Final = _require_identifier( + environment if isinstance(environment, str) else None, + "env_", + "environment id", + ) + token: Final = api_key + if not token: + raise QCAValidationError("QCA api_key is required") + base: Final = _api_base(api_base) + text: Final = _text_input(input) + previous_session_id: Final = _previous_session_id(previous_interaction_id) + except QCAValidationError as error: + raise BadRequestError(message=str(error), model=agent or "", llm_provider="qca") from error owns_client: Final = client is None http_client: Final = client or httpx.AsyncClient(timeout=timeout) start_time: Final = datetime.now(timezone.utc) @@ -409,6 +430,10 @@ async def create_qca_interaction( if owns_client: await http_client.aclose() raise _provider_error(error, agent_id) from error + except QCAValidationError as error: + if owns_client: + await http_client.aclose() + raise BadRequestError(message=str(error), model=agent_id, llm_provider="qca") from error except BaseException: if owns_client: await http_client.aclose() diff --git a/litellm/proxy/hooks/responses_id_security.py b/litellm/proxy/hooks/responses_id_security.py index 1a8539ae8f8..4dbc9ece7cd 100644 --- a/litellm/proxy/hooks/responses_id_security.py +++ b/litellm/proxy/hooks/responses_id_security.py @@ -7,7 +7,7 @@ instead of writing immediately on each request. from collections.abc import AsyncGenerator, Callable, Mapping from types import MappingProxyType -from typing import TYPE_CHECKING, Any, Final, TypeVar +from typing import TYPE_CHECKING, Any, Final, TypeAlias, TypeVar from fastapi import HTTPException from pydantic import TypeAdapter, ValidationError @@ -22,7 +22,12 @@ from litellm.proxy.common_utils.encrypt_decrypt_utils import ( encrypt_value_helper, ) from litellm.responses.utils import ResponsesAPIRequestUtils -from litellm.types.interactions import InteractionsAPIResponse, InteractionsAPIStreamingResponse +from litellm.types.interactions import ( + CancelInteractionResult, + DeleteInteractionResult, + InteractionsAPIResponse, + InteractionsAPIStreamingResponse, +) from litellm.types.llms.openai import ( BaseLiteLLMOpenAIResponseObject, ResponsesAPIResponse, @@ -84,9 +89,14 @@ def _proxy_signing_key() -> str | None: _RESPONSE_PAYLOAD_ADAPTER: Final = TypeAdapter(Mapping[str, object]) _REQUEST_DATA_ADAPTER: Final = TypeAdapter(dict[str, object]) _ResponseObjectT = TypeVar("_ResponseObjectT", bound=BaseLiteLLMOpenAIResponseObject) -_AddressedResultT = TypeVar( - "_AddressedResultT", bound=ResponsesAPIResponse | InteractionsAPIResponse | DeleteResponseResult +_AddressedResult: TypeAlias = ( + ResponsesAPIResponse + | InteractionsAPIResponse + | DeleteResponseResult + | DeleteInteractionResult + | CancelInteractionResult ) +_AddressedResultT: Final = TypeVar("_AddressedResultT", bound=_AddressedResult) def _response_payload(response_obj: object) -> Mapping[str, object] | None: @@ -178,9 +188,7 @@ def _held_id_naming(response_id: str | None, held_ids: Mapping[str, str]) -> str ) -def _id_sent_back( - response: ResponsesAPIResponse | InteractionsAPIResponse | DeleteResponseResult, held_ids: Mapping[str, str] -) -> str | None: +def _id_sent_back(response: _AddressedResult, held_ids: Mapping[str, str]) -> str | None: held_id: Final = _held_id_naming(response.id, held_ids) if held_id is not None: return held_id @@ -484,7 +492,14 @@ class ResponsesIDSecurity(CustomLogger): if general_settings.get("disable_responses_id_security", False): return response - if isinstance(response, ResponsesAPIResponse | InteractionsAPIResponse | DeleteResponseResult): + if isinstance( + response, + ResponsesAPIResponse + | InteractionsAPIResponse + | DeleteResponseResult + | DeleteInteractionResult + | CancelInteractionResult, + ): return self._advertised_response(response, user_api_key_dict, data) return response diff --git a/litellm/proxy/route_llm_request.py b/litellm/proxy/route_llm_request.py index 17f4eed2883..9f40689272e 100644 --- a/litellm/proxy/route_llm_request.py +++ b/litellm/proxy/route_llm_request.py @@ -149,9 +149,10 @@ def _raise_if_qca_deployment_selectors_are_overridden( ) -> None: if not isinstance(model_name, str): return + deployment_by_id: Final = llm_router.registry().deployment(model_name) alias_target: Final = llm_router.registry().resolve(model_name, kind="alias") resolved_model: Final = alias_target if isinstance(alias_target, str) else model_name - deployment: Final = llm_router.registry().deployments(resolved_model, source="first_in_group") + deployment: Final = deployment_by_id or llm_router.registry().deployments(resolved_model, source="first_in_group") configured_model: Final = deployment.litellm_params.model if deployment is not None else resolved_model if not configured_model.startswith("qca/"): return diff --git a/tests/unit/llms/qca/interactions/test_handler.py b/tests/unit/llms/qca/interactions/test_handler.py index 7575d486ee4..38f0887a825 100644 --- a/tests/unit/llms/qca/interactions/test_handler.py +++ b/tests/unit/llms/qca/interactions/test_handler.py @@ -6,10 +6,10 @@ from unittest.mock import MagicMock import httpx import pytest -from litellm.exceptions import RateLimitError -from litellm.llms.qca.interactions.handler import create_qca_interaction +from litellm.exceptions import BadRequestError, RateLimitError from litellm.integrations.custom_logger import CustomLogger from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj +from litellm.llms.qca.interactions.handler import create_qca_interaction from litellm.types.interactions import InteractionsAPIResponse @@ -94,6 +94,9 @@ async def test_create_qca_interaction_maps_session_events_to_completed_response( assert isinstance(response, InteractionsAPIResponse) assert response.id == "qca1.sess_test.evt_input" assert response.status == "completed" + assert response.created is not None + assert response.updated is not None + assert datetime.fromisoformat(response.updated) >= datetime.fromisoformat(response.created) assert response.steps == [ { "type": "model_output", @@ -138,7 +141,7 @@ async def test_create_qca_interaction_continues_session_and_streams_normalized_e { "id": "evt_action", "type": "session.status_idle", - "stop_reason": {"type": "end_turn"}, + "stop_reason": {"type": "end_turn"}, } ), ) @@ -182,7 +185,7 @@ async def test_create_qca_interaction_continues_session_and_streams_normalized_e async def test_create_qca_interaction_rejects_unsupported_features_before_network_call() -> None: async with httpx.AsyncClient(transport=httpx.MockTransport(lambda _: httpx.Response(500))) as client: logging_obj: Final = MagicMock() - with pytest.raises(ValueError, match="background"): + with pytest.raises(BadRequestError, match="background") as error: await create_qca_interaction( agent="agent_test", input="hello", @@ -196,12 +199,13 @@ async def test_create_qca_interaction_rejects_unsupported_features_before_networ logging_obj=logging_obj, client=client, ) + assert error.value.status_code == 400 @pytest.mark.asyncio async def test_create_qca_interaction_rejects_store_false_before_network_call() -> None: async with httpx.AsyncClient(transport=httpx.MockTransport(lambda _: httpx.Response(500))) as client: - with pytest.raises(ValueError, match="store"): + with pytest.raises(BadRequestError, match="store") as error: await create_qca_interaction( agent="agent_test", input="hello", @@ -215,6 +219,7 @@ async def test_create_qca_interaction_rejects_store_false_before_network_call() logging_obj=MagicMock(), client=client, ) + assert error.value.status_code == 400 @pytest.mark.asyncio @@ -253,7 +258,7 @@ async def test_create_qca_interaction_rejects_continuation_for_another_agent() - async with httpx.AsyncClient(transport=httpx.MockTransport(route)) as client: logging_obj: Final = MagicMock() - with pytest.raises(ValueError, match="configured agent and environment"): + with pytest.raises(BadRequestError, match="configured agent and environment") as error: await create_qca_interaction( agent="agent_test", input="continue", @@ -267,5 +272,6 @@ async def test_create_qca_interaction_rejects_continuation_for_another_agent() - logging_obj=logging_obj, client=client, ) + assert error.value.status_code == 400 assert [request.method for request in requests] == ["GET"] diff --git a/tests/unit/proxy/test_route_llm_request.py b/tests/unit/proxy/test_route_llm_request.py index afbd238dcb7..4d3f49fbdf6 100644 --- a/tests/unit/proxy/test_route_llm_request.py +++ b/tests/unit/proxy/test_route_llm_request.py @@ -95,6 +95,33 @@ def test_qca_model_group_alias_rejects_request_selector_overrides(selector: str) assert error.value.code == "403" +@pytest.mark.parametrize("selector", ("agent", "environment")) +def test_qca_deployment_id_rejects_request_selector_overrides(selector: str) -> None: + deployment_id: Final = "qoder-deployment-id" + router: Final = litellm.Router( + model_list=[ + { + "model_name": "qoder-agent", + "litellm_params": { + "model": "qca/agent_configured", + "environment": "env_configured", + "api_key": "unit-provider-key", + }, + "model_info": {"id": deployment_id}, + } + ] + ) + + with pytest.raises(ProxyException) as error: + _raise_if_qca_deployment_selectors_are_overridden( + router, + deployment_id, + {"model": deployment_id, selector: f"{selector}_attacker"}, + ) + + assert error.value.code == "403" + + @pytest.mark.parametrize( "route_type, required_body_params", [ @@ -1451,7 +1478,9 @@ async def test_route_request_applies_override_fallbacks_when_request_carries_cre @pytest.mark.parametrize( "router_only_setting", [ - pytest.param({"model_group_retry_policy": {"gpt-4": {"TimeoutErrorRetries": 1}}}, id="model-group-retry-policy"), + pytest.param( + {"model_group_retry_policy": {"gpt-4": {"TimeoutErrorRetries": 1}}}, id="model-group-retry-policy" + ), pytest.param({"context_window_fallbacks": [{"gpt-4": ["gpt-3.5-turbo"]}]}, id="context-window-fallbacks"), pytest.param({"content_policy_fallbacks": [{"gpt-4": ["gpt-3.5-turbo"]}]}, id="content-policy-fallbacks"), ], @@ -2657,7 +2686,9 @@ async def test_a_user_config_adaptive_request_answers_with_the_model_it_picked() }, { **_mock_deployment("smart"), - "model_info": {"adaptive_router_preferences": {"quality_tier": 3, "strengths": ["code_generation"]}}, + "model_info": { + "adaptive_router_preferences": {"quality_tier": 3, "strengths": ["code_generation"]} + }, }, ] }, diff --git a/tests/unit/test_responses_id_security.py b/tests/unit/test_responses_id_security.py index cb09e5e4603..cc228861fe6 100644 --- a/tests/unit/test_responses_id_security.py +++ b/tests/unit/test_responses_id_security.py @@ -23,7 +23,11 @@ from litellm.proxy.hooks.responses_id_security import ( ) from litellm.proxy.spend_tracking.spend_tracking_utils import get_logging_payload from litellm.responses.utils import ResponsesAPIRequestUtils -from litellm.types.interactions import InteractionsAPIResponse, InteractionsAPIStreamingResponse +from litellm.types.interactions import ( + CancelInteractionResult, + InteractionsAPIResponse, + InteractionsAPIStreamingResponse, +) from litellm.types.llms.openai import ( GenericEvent, ResponseAPIUsage, @@ -900,6 +904,35 @@ async def test_interaction_lifecycle_ids_are_authorized_before_provider_calls( assert refusal.value.status_code == 403 +@pytest.mark.asyncio +async def test_cancel_interaction_returns_the_managed_id_held_by_the_caller(salt_key_env: str) -> None: + hook: Final = _hook(signing_key=salt_key_env) + owner: Final = _auth() + provider_id: Final = "interaction-provider-id" + issued_response: Final = hook._encrypt_response_id( # pyright: ignore[reportPrivateUsage] # mints ids the way create does + InteractionsAPIResponse(id=provider_id, status="in_progress"), + owner, + ) + issued: Final = InteractionsAPIResponse.model_validate(issued_response).id + assert issued is not None + authorized: Final = TypeAdapter(dict[str, object]).validate_python( + await hook.async_pre_call_hook( + user_api_key_dict=owner, + cache=DualCache(), + data={"interaction_id": issued}, + call_type="acancel_interaction", + ) + ) + + result: Final = await hook.async_post_call_success_hook( # pyright: ignore[reportArgumentType] # interactions share this proxy hook + data=authorized, + user_api_key_dict=owner, + response=CancelInteractionResult(id=provider_id, status="cancelled"), + ) + + assert CancelInteractionResult.model_validate(result).id == issued + + @pytest.mark.asyncio @pytest.mark.parametrize("id_location", ("interaction_id", "interaction")) async def test_streamed_interaction_schema_ids_are_bound_to_the_caller(salt_key_env: str, id_location: str) -> None: From c2960218291bfa6880584d9033f9cb83f4f98e88 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=AE=89=E8=BE=B0?= Date: Thu, 5 Nov 2026 14:43:59 +0800 Subject: [PATCH 5/6] fix(interactions): enforce Qoder budget safety --- docs/adrs/adr_003_qca_budget_enforcement.md | 32 ++++++ litellm/llms/qca/interactions/handler.py | 30 +++++- litellm/proxy/route_llm_request.py | 101 +++++++++++++++--- litellm/types/interactions/__init__.py | 2 + litellm/types/interactions/generated.py | 1 + .../llms/qca/interactions/test_handler.py | 56 +++++++++- tests/unit/proxy/test_route_llm_request.py | 101 +++++++++++++++--- 7 files changed, 288 insertions(+), 35 deletions(-) create mode 100644 docs/adrs/adr_003_qca_budget_enforcement.md diff --git a/docs/adrs/adr_003_qca_budget_enforcement.md b/docs/adrs/adr_003_qca_budget_enforcement.md new file mode 100644 index 00000000000..bd54d71052e --- /dev/null +++ b/docs/adrs/adr_003_qca_budget_enforcement.md @@ -0,0 +1,32 @@ +# Qoder Cloud Agents budget enforcement + +## Status + +Accepted + +## Context + +Qoder Cloud Agents reports cumulative provider credits for a session. LiteLLM monetary budgets require a cost in a +configured currency for each request. Qoder does not publish a stable currency conversion, and charging the cumulative +session credits on every continuation would bill earlier turns more than once + +## Decision + +The proxy rejects Qoder Cloud Agents requests from callers with an enforced key, user, team, organization, end-user, or +model monetary budget. The rejection happens before creating or continuing a provider session. Callers without monetary +budgets can use Qoder Cloud Agents, and the response keeps the cumulative Qoder credits as provider usage fields + +## Alternatives Considered + +Converting credits with a fixed rate was rejected because LiteLLM does not own a stable Qoder credit price + +Charging cumulative credits on every interaction was rejected because session continuations would repeatedly charge +earlier turns + +Tracking session deltas in proxy memory was rejected because it would not be reliable across workers or restarts + +## Consequences + +Monetary budgets cannot silently admit unpriced Qoder usage. Operators that need budgeted access must wait for a stable, +deployment-owned conversion and durable per-session delta accounting. Existing unbudgeted integrations continue to +receive the provider credit fields without a guessed monetary cost diff --git a/litellm/llms/qca/interactions/handler.py b/litellm/llms/qca/interactions/handler.py index dbd9f56278f..19074f7207c 100644 --- a/litellm/llms/qca/interactions/handler.py +++ b/litellm/llms/qca/interactions/handler.py @@ -9,7 +9,7 @@ from pydantic import BaseModel, ConfigDict, TypeAdapter, ValidationError from litellm.exceptions import APIError, AuthenticationError, BadRequestError, RateLimitError from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj from litellm.llms.custom_httpx.sse_lines import aiter_sse_lines -from litellm.types.interactions import InteractionsAPIResponse, InteractionsAPIStreamingResponse +from litellm.types.interactions import Error, InteractionsAPIResponse, InteractionsAPIStreamingResponse QCA_DEFAULT_API_BASE: Final = "https://api.qoder.com/api/v1/cloud" _JSON_OBJECT: Final = TypeAdapter(dict[str, object]) @@ -233,6 +233,16 @@ async def _normalized_stream( delta=dict(_text_delta(step)), ) if status is not None: + if status == "failed": + yield InteractionsAPIStreamingResponse( + event_type="error", + interaction_id=interaction_id, + error=Error( + code="qca_session_failed", + message=f"Qoder Cloud Agents session ended with {event.type}", + ), + ) + return yield InteractionsAPIStreamingResponse( event_type=f"interaction.{status}", interaction_id=interaction_id, @@ -306,12 +316,13 @@ async def _observed_stream( async for chunk in stream: if chunk.event_type == "step.delta" and chunk.delta is not None: steps = (*steps, _step_from_delta(chunk.delta)) - if chunk.event_type in {"interaction.completed", "interaction.requires_action", "interaction.failed"}: - await logging_obj.async_success_handler( + if chunk.event_type in {"interaction.completed", "interaction.requires_action"}: + await logging_obj.dispatch_success_handlers( # pyright: ignore[reportUnknownMemberType] # legacy callback API is untyped result=_terminal_for_logging(chunk, interaction_id, agent, steps), start_time=start_time, end_time=datetime.now(timezone.utc), cache_hit=None, + prefer_async_handlers=True, ) yield chunk @@ -324,7 +335,18 @@ async def _collect_response( async for chunk in stream: if chunk.event_type == "step.delta" and chunk.delta is not None: steps = (*steps, _step_from_delta(chunk.delta)) - if chunk.event_type in {"interaction.completed", "interaction.requires_action", "interaction.failed"}: + if chunk.event_type == "error": + raise APIError( + status_code=502, + message=( + chunk.error.message + if chunk.error is not None and chunk.error.message is not None + else "Qoder Cloud Agents session failed" + ), + llm_provider="qca", + model=agent, + ) + if chunk.event_type in {"interaction.completed", "interaction.requires_action"}: terminal = chunk if terminal is None: raise httpx.RemoteProtocolError("QCA event stream did not produce a terminal interaction") diff --git a/litellm/proxy/route_llm_request.py b/litellm/proxy/route_llm_request.py index 9f40689272e..6012720a8f6 100644 --- a/litellm/proxy/route_llm_request.py +++ b/litellm/proxy/route_llm_request.py @@ -11,7 +11,7 @@ from uuid import uuid4 import httpx from fastapi import HTTPException, status -from pydantic import BeforeValidator, TypeAdapter, ValidationError +from pydantic import BaseModel, BeforeValidator, ConfigDict, TypeAdapter, ValidationError import litellm from litellm.litellm_core_utils.core_helpers import normalize_drop_params @@ -61,6 +61,50 @@ _PER_REQUEST_ROUTER_SETTINGS: Final = ( _ROUTER_ONLY_SETTINGS: Final = frozenset( ("fallbacks", "context_window_fallbacks", "content_policy_fallbacks", "model_group_retry_policy") ) +_ROUTER_DEPLOYMENTS: Final = TypeAdapter(list[dict[str, object]]) + + +class _QCAMonetaryBudgetContext(BaseModel): + model_config = ConfigDict(frozen=True, from_attributes=True) + + max_budget: float | None = None + user_max_budget: float | None = None + team_max_budget: float | None = None + organization_max_budget: float | None = None + end_user_max_budget: float | None = None + budget_id: str | None = None + budget_reservation: Mapping[str, object] | None = None + model_max_budget: Mapping[str, object] | None = None + user_model_max_budget: Mapping[str, object] | None = None + team_model_max_budget: Mapping[str, object] | None = None + team_member_model_max_budget: Mapping[str, object] | None = None + end_user_model_max_budget: Mapping[str, object] | None = None + + def is_limited(self) -> bool: + values: Final = ( + self.max_budget, + self.user_max_budget, + self.team_max_budget, + self.organization_max_budget, + self.end_user_max_budget, + self.budget_id, + self.budget_reservation, + self.model_max_budget, + self.user_model_max_budget, + self.team_model_max_budget, + self.team_member_model_max_budget, + self.end_user_model_max_budget, + ) + return any(value is not None and value != {} for value in values) + + +def _provider_model(deployment: Mapping[str, object]) -> str | None: + unvalidated_litellm_params: Final = deployment.get("litellm_params") + if not isinstance(unvalidated_litellm_params, Mapping): + return None + litellm_params: Final = JSON_OBJECT_ADAPTER.validate_python(unvalidated_litellm_params) + model: Final = litellm_params.get("model") + return model if isinstance(model, str) else None class UserConfigRouters: @@ -145,17 +189,42 @@ def _raise_if_model_fully_blocked(llm_router: LitellmRouter, model_name: object, def _raise_if_qca_deployment_selectors_are_overridden( - llm_router: LitellmRouter, model_name: object, request_data: Mapping[str, object] + llm_router: LitellmRouter, + model_name: object, + request_data: Mapping[str, object], + user_api_key_dict: UserAPIKeyAuth | None = None, ) -> None: if not isinstance(model_name, str): return deployment_by_id: Final = llm_router.registry().deployment(model_name) - alias_target: Final = llm_router.registry().resolve(model_name, kind="alias") - resolved_model: Final = alias_target if isinstance(alias_target, str) else model_name - deployment: Final = deployment_by_id or llm_router.registry().deployments(resolved_model, source="first_in_group") - configured_model: Final = deployment.litellm_params.model if deployment is not None else resolved_model - if not configured_model.startswith("qca/"): + request_deployments: Final = _ROUTER_DEPLOYMENTS.validate_python( + llm_router.registry().deployments( + model_name, + source="for_request", + request=request_data, + call_type="acreate_interaction", + ) + ) + configured_models: Final = ( + (deployment_by_id.litellm_params.model,) + if deployment_by_id is not None + else tuple(model for deployment in request_deployments if (model := _provider_model(deployment)) is not None) + ) + if not model_name.startswith("qca/") and not any(model.startswith("qca/") for model in configured_models): return + budget_context: Final = ( + None if user_api_key_dict is None else _QCAMonetaryBudgetContext.model_validate(user_api_key_dict) + ) + if budget_context is not None and budget_context.is_limited(): + raise ProxyException( + message=( + "Qoder Cloud Agents cannot enforce monetary budgets because Qoder credits do not have a configured " + "currency conversion" + ), + type="invalid_request_error", + param="model", + code=status.HTTP_400_BAD_REQUEST, + ) overridden: Final = tuple(name for name in ("agent", "environment") if request_data.get(name) is not None) if overridden: raise ProxyException( @@ -473,13 +542,14 @@ def endpoint_unsupported_by_model( if not isinstance(model, str) or llm_router is None: return False metadata: Final = data.get("metadata") - request_kwargs: Final = { - **data, - "metadata": { - **(metadata if isinstance(metadata, Mapping) else {}), + validated_metadata: Final = JSON_OBJECT_ADAPTER.validate_python(metadata) if isinstance(metadata, Mapping) else {} + request_metadata: Final = MappingProxyType[str, object]( + { + **validated_metadata, "user_api_key_team_id": team_id, - }, - } + } + ) + request_kwargs: Final = MappingProxyType[str, object]({**data, "metadata": request_metadata}) registry: Final = llm_router.registry() return bool(registry.deployments(model, source="routable", request=request_kwargs)) and not registry.deployments( model, source="routable", request=request_kwargs, call_type=route_type @@ -841,7 +911,10 @@ async def _route_request_single_attempt( # noqa: ANN202 # returns unawaited pr if route_type == "acreate_interaction" and llm_router is not None: interaction_request: Final = JSON_OBJECT_ADAPTER.validate_python(data) _raise_if_qca_deployment_selectors_are_overridden( - llm_router, interaction_request.get("model"), interaction_request + llm_router, + interaction_request.get("model"), + interaction_request, + user_api_key_dict, ) # Preprocess Google GenAI generate content requests diff --git a/litellm/types/interactions/__init__.py b/litellm/types/interactions/__init__.py index 1dba2273267..381a58bff39 100644 --- a/litellm/types/interactions/__init__.py +++ b/litellm/types/interactions/__init__.py @@ -24,6 +24,7 @@ from litellm.types.interactions.generated import ( DeleteInteractionResult, DocumentContent, DynamicAgentConfig, + Error, ErrorEvent, FileSearch, FileSearchResultContent, @@ -92,6 +93,7 @@ __all__ = [ "DeleteInteractionResult", "DocumentContent", "DynamicAgentConfig", + "Error", "ErrorEvent", "FileSearch", "FileSearchResultContent", diff --git a/litellm/types/interactions/generated.py b/litellm/types/interactions/generated.py index c4f7a2d8a7b..7d4ec04526f 100644 --- a/litellm/types/interactions/generated.py +++ b/litellm/types/interactions/generated.py @@ -1121,6 +1121,7 @@ class InteractionsAPIStreamingResponse(BaseLiteLLMOpenAIResponseObject): steps: list[dict[str, Any]] | None = None usage: dict[str, Any] | None = None delta: dict[str, Any] | None = None + error: Error | None = None # New schema streaming fields index: int | None = None step: dict[str, Any] | None = None diff --git a/tests/unit/llms/qca/interactions/test_handler.py b/tests/unit/llms/qca/interactions/test_handler.py index 38f0887a825..0929c20f8d2 100644 --- a/tests/unit/llms/qca/interactions/test_handler.py +++ b/tests/unit/llms/qca/interactions/test_handler.py @@ -1,3 +1,4 @@ +import asyncio import json from datetime import datetime from typing import Final @@ -8,9 +9,10 @@ import pytest from litellm.exceptions import BadRequestError, RateLimitError from litellm.integrations.custom_logger import CustomLogger +from litellm.litellm_core_utils import litellm_logging from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj from litellm.llms.qca.interactions.handler import create_qca_interaction -from litellm.types.interactions import InteractionsAPIResponse +from litellm.types.interactions import InteractionsAPIResponse, InteractionSseEvent def _sse_event(payload: dict[str, object]) -> bytes: @@ -27,7 +29,7 @@ class _SuccessRecorder(CustomLogger): self.responses.append(response_obj) -def _stream_logging(recorder: _SuccessRecorder) -> LiteLLMLoggingObj: +def _stream_logging(recorder: _SuccessRecorder, include_sync_callback: bool = False) -> LiteLLMLoggingObj: logging_obj: Final = LiteLLMLoggingObj( model="qca/agent_test", messages=[], @@ -41,6 +43,8 @@ def _stream_logging(recorder: _SuccessRecorder) -> LiteLLMLoggingObj: logging_obj.update_environment_variables( model="qca/agent_test", user="", optional_params={}, litellm_params={"api_base": "https://qca.test"} ) + if include_sync_callback: + logging_obj.dynamic_success_callbacks = ["s3"] return logging_obj @@ -116,8 +120,12 @@ async def test_create_qca_interaction_maps_session_events_to_completed_response( @pytest.mark.asyncio -async def test_create_qca_interaction_continues_session_and_streams_normalized_events() -> None: +async def test_create_qca_interaction_continues_session_and_streams_normalized_events( + monkeypatch: pytest.MonkeyPatch, +) -> None: requests: list[httpx.Request] = [] + sync_logger: Final = MagicMock() + monkeypatch.setattr(litellm_logging, "s3Logger", sync_logger) def route(request: httpx.Request) -> httpx.Response: requests.append(request) @@ -150,7 +158,7 @@ async def test_create_qca_interaction_continues_session_and_streams_normalized_e async with httpx.AsyncClient(transport=httpx.MockTransport(route)) as client: recorder: Final = _SuccessRecorder() - logging_obj: Final = _stream_logging(recorder) + logging_obj: Final = _stream_logging(recorder, include_sync_callback=True) result: Final = await create_qca_interaction( agent="agent_test", input="continue", @@ -178,9 +186,49 @@ async def test_create_qca_interaction_continues_session_and_streams_normalized_e logged: Final = recorder.responses[0] assert isinstance(logged, InteractionsAPIResponse) assert logged.steps == [{"type": "model_output", "content": [{"type": "text", "text": "continued"}]}] + for _ in range(200): + if sync_logger.log_event.call_count == 1: + break + await asyncio.sleep(0.01) + sync_logger.log_event.assert_called_once() assert all(not request.url.path.endswith("/sessions") for request in requests) +@pytest.mark.asyncio +async def test_create_qca_interaction_stream_emits_schema_valid_error_event() -> None: + def route(request: httpx.Request) -> httpx.Response: + if request.url.path.endswith("/sessions"): + return httpx.Response(200, json={"id": "sess_test"}) + if request.url.path.endswith("/events"): + return httpx.Response(202, json={"data": [{"id": "evt_input", "type": "user.message"}]}) + return httpx.Response( + 200, + content=_sse_event({"id": "evt_error", "type": "session.error"}), + headers={"content-type": "text/event-stream"}, + ) + + async with httpx.AsyncClient(transport=httpx.MockTransport(route)) as client: + result: Final = await create_qca_interaction( + agent="agent_test", + input="fail", + environment="env_test", + previous_interaction_id=None, + stream=True, + api_key="secret", + api_base="https://qca.test/api/v1/cloud", + timeout=30, + unsupported_params={}, + logging_obj=_stream_logging(_SuccessRecorder()), + client=client, + ) + assert not isinstance(result, InteractionsAPIResponse) + chunks: Final = [chunk async for chunk in result] + + assert chunks[-1].event_type == "error" + validated: Final = InteractionSseEvent.model_validate(chunks[-1].model_dump(exclude_none=True)) + assert validated.root.event_type == "error" + + @pytest.mark.asyncio async def test_create_qca_interaction_rejects_unsupported_features_before_network_call() -> None: async with httpx.AsyncClient(transport=httpx.MockTransport(lambda _: httpx.Response(500))) as client: diff --git a/tests/unit/proxy/test_route_llm_request.py b/tests/unit/proxy/test_route_llm_request.py index 4d3f49fbdf6..6681d1b6d65 100644 --- a/tests/unit/proxy/test_route_llm_request.py +++ b/tests/unit/proxy/test_route_llm_request.py @@ -1,26 +1,21 @@ import asyncio import contextlib import gc +import json import threading import weakref +from collections.abc import Awaitable, Callable, Coroutine, Iterator, Mapping +from datetime import datetime +from types import SimpleNamespace +from typing import Final +from unittest.mock import MagicMock +from uuid import uuid4 + import httpx -import json import litellm import pytest import redis import respx - - -from collections.abc import Awaitable, Callable, Coroutine, Iterator, Mapping -from datetime import datetime -from types import SimpleNamespace -from uuid import uuid4 -from typing import Final -from unittest.mock import MagicMock - -from litellm.router_api import RouterSettings -from tests.unit.proxy.router_entry_point_stubs import route_entry_points_to_stubs - from fakeredis import TcpFakeServer from fastapi import HTTPException @@ -37,11 +32,13 @@ from litellm.proxy.route_llm_request import ( user_config_routers, ) from litellm.proxy.utils import ProxyLogging +from litellm.router_api import RouterSettings from litellm.router_strategy.adaptive_router.config import ( ADAPTIVE_ROUTER_CHOSEN_MODEL_KEY, ADAPTIVE_ROUTER_RESPONSE_HEADER, ) from litellm.types.router import RouterErrors, RouterGeneralSettings, RouterRateLimitError +from tests.unit.proxy.router_entry_point_stubs import route_entry_points_to_stubs @pytest.mark.parametrize("selector", ("agent", "environment")) @@ -122,6 +119,84 @@ def test_qca_deployment_id_rejects_request_selector_overrides(selector: str) -> assert error.value.code == "403" +@pytest.mark.parametrize("selector", ("agent", "environment")) +def test_qca_team_public_name_rejects_request_selector_overrides(selector: str) -> None: + team_id: Final = "qoder-team" + public_name: Final = "team-qoder-agent" + router: Final = litellm.Router( + model_list=[ + { + "model_name": "private-qoder-agent", + "litellm_params": { + "model": "qca/agent_configured", + "environment": "env_configured", + "api_key": "unit-provider-key", + }, + "model_info": {"team_id": team_id, "team_public_model_name": public_name}, + } + ] + ) + request: Final = { + "model": public_name, + selector: f"{selector}_attacker", + "metadata": {"user_api_key_team_id": team_id}, + } + + with pytest.raises(ProxyException) as error: + _raise_if_qca_deployment_selectors_are_overridden( + router, + public_name, + request, + UserAPIKeyAuth(team_id=team_id), + ) + + assert error.value.code == "403" + + +@pytest.mark.parametrize( + "budget_field,budget_value", + ( + ("max_budget", 10.0), + ("user_max_budget", 10.0), + ("team_max_budget", 10.0), + ("organization_max_budget", 10.0), + ("end_user_max_budget", 10.0), + ("budget_id", "budget-config-id"), + ("budget_reservation", {"entries": [{"counter_key": "spend:team:team-id"}]}), + ("model_max_budget", {"qoder-agent": {"max_budget": 10.0}}), + ("user_model_max_budget", {"qoder-agent": {"max_budget": 10.0}}), + ("team_model_max_budget", {"qoder-agent": {"max_budget": 10.0}}), + ("team_member_model_max_budget", {"qoder-agent": {"max_budget": 10.0}}), + ("end_user_model_max_budget", {"qoder-agent": {"max_budget": 10.0}}), + ), +) +def test_qca_rejects_callers_with_unenforceable_monetary_budgets(budget_field: str, budget_value: object) -> None: + router: Final = litellm.Router( + model_list=[ + { + "model_name": "qoder-agent", + "litellm_params": { + "model": "qca/agent_configured", + "environment": "env_configured", + "api_key": "unit-provider-key", + }, + } + ] + ) + caller: Final = UserAPIKeyAuth.model_validate({budget_field: budget_value}) + + with pytest.raises(ProxyException) as error: + _raise_if_qca_deployment_selectors_are_overridden( + router, + "qoder-agent", + {"model": "qoder-agent"}, + caller, + ) + + assert error.value.code == "400" + assert "cannot enforce monetary budgets" in error.value.message + + @pytest.mark.parametrize( "route_type, required_body_params", [ From 07a523f164384e546f7265673a5165abe63927c9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=AE=89=E8=BE=B0?= Date: Fri, 6 Nov 2026 14:21:58 +0800 Subject: [PATCH 6/6] fix(guardrails): reject unguarded interactions --- .../unified_guardrail/unified_guardrail.py | 23 +++++++++++++++++++ .../test_unified_guardrail.py | 22 ++++++++++++++++++ 2 files changed, 45 insertions(+) diff --git a/litellm/proxy/guardrails/guardrail_hooks/unified_guardrail/unified_guardrail.py b/litellm/proxy/guardrails/guardrail_hooks/unified_guardrail/unified_guardrail.py index 5ce2dfb3b87..55f0e7d6dd8 100644 --- a/litellm/proxy/guardrails/guardrail_hooks/unified_guardrail/unified_guardrail.py +++ b/litellm/proxy/guardrails/guardrail_hooks/unified_guardrail/unified_guardrail.py @@ -56,6 +56,24 @@ _DECODED_JSON: Final = TypeAdapter(object) _RequestData: TypeAlias = dict[str, object] +def _raise_for_unmapped_interactions_guardrail( + call_type: CallTypesLiteral | None, + mappings: Mapping[CallTypes, type["BaseTranslation"]], +) -> None: + if call_type is None: + return + try: + resolved_call_type: Final = CallTypes(call_type) + except ValueError: + return + if resolved_call_type != CallTypes.acreate_interaction or resolved_call_type in mappings: + return + raise HTTPException( + status_code=400, + detail="Interactions requests cannot enforce configured apply_guardrail policies without a translation handler", + ) + + @runtime_checkable class RequestAttachmentScanner(Protocol): async def async_scan_request_attachments( @@ -285,6 +303,8 @@ class UnifiedLLMGuardrails(CustomLogger): mappings: Final = load_guardrail_translation_mappings() + _raise_for_unmapped_interactions_guardrail(call_type, mappings) + try: if CallTypes(call_type) not in mappings: return data @@ -336,6 +356,7 @@ class UnifiedLLMGuardrails(CustomLogger): return data mappings: Final = load_guardrail_translation_mappings() + _raise_for_unmapped_interactions_guardrail(call_type, mappings) if call_type is not None and CallTypes(call_type) not in mappings: return data @@ -412,6 +433,8 @@ class UnifiedLLMGuardrails(CustomLogger): mappings: Final = load_guardrail_translation_mappings() + _raise_for_unmapped_interactions_guardrail(call_type, mappings) + if CallTypes(call_type) not in mappings: verbose_proxy_logger.warning( "Guardrail '%s' selected for route '%s' but call type '%s' has no guardrail translation handler; " diff --git a/tests/unit/proxy/guardrails/guardrail_hooks/unified_guardrails/test_unified_guardrail.py b/tests/unit/proxy/guardrails/guardrail_hooks/unified_guardrails/test_unified_guardrail.py index de3ced0c583..f560f5886ba 100644 --- a/tests/unit/proxy/guardrails/guardrail_hooks/unified_guardrails/test_unified_guardrail.py +++ b/tests/unit/proxy/guardrails/guardrail_hooks/unified_guardrails/test_unified_guardrail.py @@ -12,6 +12,7 @@ from typing import TYPE_CHECKING, Final, Literal import anyio import pytest +from fastapi import HTTPException import litellm from litellm.caching import DualCache @@ -337,6 +338,27 @@ class TestUnifiedLLMGuardrails: assert "tool" in roles class TestAsyncPreCallHook: + @pytest.mark.asyncio + async def test_interactions_rejects_apply_guardrail_without_translation(self) -> None: + guardrail: Final = RecordingGuardrail() + data: Final = { + "guardrail_to_apply": guardrail, + "model": "qca-agent-alias", + "input": "sensitive input", + } + + with pytest.raises(HTTPException) as error: + await UnifiedLLMGuardrails().async_pre_call_hook( + user_api_key_dict=UserAPIKeyAuth(api_key="test-key"), + cache=DualCache(), + data=data, + call_type=CallTypes.acreate_interaction.value, + ) + + assert error.value.status_code == 400 + assert "cannot enforce configured apply_guardrail policies" in error.value.detail + assert guardrail.apply_calls == [] + @pytest.mark.asyncio async def test_uses_mcp_event_type(self): """pre_call hook should swap to GuardrailEventHooks.pre_mcp_call for MCP calls."""