diff --git a/src/pydantic_ai_lightspeed/llamastack/_model.py b/src/pydantic_ai_lightspeed/llamastack/_model.py index 330077e6a..80a608a3f 100644 --- a/src/pydantic_ai_lightspeed/llamastack/_model.py +++ b/src/pydantic_ai_lightspeed/llamastack/_model.py @@ -74,6 +74,16 @@ def _model_settings_from_responses_params( """Map ``ResponsesApiParams`` into Pydantic AI OpenAI Responses model settings.""" payload = responses_params.model_dump(exclude_none=True) extra_body = {k: v for k, v in payload.items() if k in _LLS_RESPONSES_EXTRA_FIELDS} + if responses_params.omit_conversation and not isinstance( + responses_params.input, str + ): + # Compacted mode (LCORE-3582): the request must carry the explicit item + # list (summaries + recent turns + new query), but pydantic-ai builds + # the wire ``input`` from the prompt alone. Overriding via extra_body + # replaces it with the explicit list, exactly as the non-agent + # /v1/responses path sends it. Dropped again on tool-loop + # continuations — see ``_prepare_compacted_input``. + extra_body["input"] = payload["input"] settings_dict: dict[str, Any] = {} if extra_body: settings_dict["extra_body"] = extra_body @@ -291,6 +301,7 @@ async def request( # pylint: disable=unused-argument messages, model_settings = self._prepare_conversation_continuation( messages, model_settings ) + model_settings = self._prepare_compacted_input(messages, model_settings) return await super().request(messages, model_settings, model_request_parameters) def _prepare_conversation_continuation( @@ -334,6 +345,34 @@ def _prepare_conversation_continuation( new_settings.pop("openai_previous_response_id", None) return trimmed_messages, cast(ModelSettings, new_settings) + def _prepare_compacted_input( + self, + messages: list[ModelMessage], + model_settings: Optional[ModelSettings], + ) -> Optional[ModelSettings]: + """Drop the compacted ``input`` override on tool-loop continuations. + + In compacted mode (LCORE-3582) the request body ``input`` is overridden + via ``extra_body`` with the explicit item list. That override is only + valid for the first request of an agent run: on client-side tool-loop + iterations pydantic-ai's mapped messages carry the tool results and + must win, so the override is removed once a ``ModelResponse`` exists in + the message history. + """ + if not model_settings or not isinstance(model_settings, dict): + return model_settings + extra_body = model_settings.get("extra_body") + if not isinstance(extra_body, dict) or "input" not in extra_body: + return model_settings + if not any(isinstance(message, ModelResponse) for message in messages): + return model_settings + + new_extra_body = dict(extra_body) + new_extra_body.pop("input") + new_settings = dict(model_settings) + new_settings["extra_body"] = new_extra_body + return cast(ModelSettings, new_settings) + @asynccontextmanager async def request_stream( # pylint: disable=unused-argument self, @@ -360,6 +399,7 @@ async def request_stream( # pylint: disable=unused-argument messages, model_settings = self._prepare_conversation_continuation( messages, model_settings ) + model_settings = self._prepare_compacted_input(messages, model_settings) model_settings_cast = cast(OpenAIResponsesModelSettings, model_settings or {}) response = await self._responses_create( diff --git a/src/utils/agents/error_handler.py b/src/utils/agents/error_handler.py index 15c28ba03..8b16efb70 100644 --- a/src/utils/agents/error_handler.py +++ b/src/utils/agents/error_handler.py @@ -1,5 +1,6 @@ """Error mapping for agent inference failures to structured API error responses.""" +from fastapi import status from ogx_client import APIConnectionError, APIStatusError from pydantic_ai.exceptions import ( AgentRunError, @@ -51,18 +52,35 @@ def map_agent_inference_error( """ match exc: case AgentRunError() as agent_exc: - return map_pydantic_agent_run_error(agent_exc, model_id) + response = map_pydantic_agent_run_error(agent_exc, model_id) case APIStatusError() as status_exc: - return handle_known_apistatus_errors(status_exc, model_id) + response = handle_known_apistatus_errors(status_exc, model_id) case APIConnectionError() as connection_exc: - return ServiceUnavailableResponse( + response = ServiceUnavailableResponse( backend_name="OGX", cause=str(connection_exc), ) case RuntimeError() as runtime_exc if is_context_length_error(str(runtime_exc)): - return PromptTooLongResponse(model=model_id) + response = PromptTooLongResponse(model=model_id) case _: - return InternalServerErrorResponse.generic() + response = InternalServerErrorResponse.generic() + + # The mapped HTTPException loses the original exception, and callers raise + # it without logging — log here so failures are diagnosable (LCORE-3582). + # Only unclassified failures are genuine service faults worth a traceback: + # a mapped 4xx is caller-caused (413 over-long prompt, 429 rate limit) and + # a 503 is an upstream outage, so logging those at error level with a stack + # trace would bury the 500s this logging exists to surface. + status_code = getattr(response, "status_code", None) + caller_or_upstream = isinstance(status_code, int) and ( + status_code < status.HTTP_500_INTERNAL_SERVER_ERROR + or status_code == status.HTTP_503_SERVICE_UNAVAILABLE + ) + if caller_or_upstream: + logger.warning("Agent inference returned %s: %s", status_code, exc) + else: + logger.error("Agent inference failed: %s", exc, exc_info=exc) + return response def map_pydantic_agent_run_error( # pylint: disable=too-many-return-statements diff --git a/src/utils/agents/query.py b/src/utils/agents/query.py index f0660faf1..1c3d1a80d 100644 --- a/src/utils/agents/query.py +++ b/src/utils/agents/query.py @@ -3,7 +3,7 @@ from __future__ import annotations from enum import Enum -from typing import Optional, cast +from typing import Optional from fastapi import HTTPException from ogx_client import APIConnectionError, APIStatusError, AsyncOgxClient @@ -36,6 +36,10 @@ process_native_tool_call, process_native_tool_result, ) +from utils.conversation_compaction import ( + agent_prompt_text, + reject_image_attachments_in_compacted_mode, +) from utils.conversations import append_turn_items_to_conversation from utils.otel_tracing import ( SpanAttributes, @@ -276,12 +280,13 @@ async def retrieve_agent_response( ) if moderation_result.decision == "blocked": - await append_turn_items_to_conversation( - client, - responses_params.conversation, - responses_params.input, - [moderation_result.refusal_response], - ) + if not responses_params.omit_conversation: + await append_turn_items_to_conversation( + client, + responses_params.conversation, + responses_params.input, + [moderation_result.refusal_response], + ) return TurnSummary( id=moderation_result.moderation_id, llm_response=moderation_result.message, @@ -299,13 +304,16 @@ async def retrieve_agent_response( no_tools=no_tools, ) logger.debug("Starting agent non-streaming response processing") + reject_image_attachments_in_compacted_mode( + responses_params, image_attachments + ) if image_attachments: prompt = build_multimodal_input( - cast(str, responses_params.input), + agent_prompt_text(responses_params), image_attachments, ) else: - prompt = cast(str, responses_params.input) + prompt = agent_prompt_text(responses_params) run_result = await agent.run(prompt) except ( AgentRunError, diff --git a/src/utils/agents/streaming.py b/src/utils/agents/streaming.py index 8bbc19cce..7ecf2d603 100644 --- a/src/utils/agents/streaming.py +++ b/src/utils/agents/streaming.py @@ -8,7 +8,7 @@ import datetime from collections.abc import AsyncIterator from functools import singledispatch -from typing import Any, Final, Optional, cast +from typing import Any, Final, Optional from fastapi import HTTPException from ogx_client import APIConnectionError, APIStatusError @@ -60,6 +60,10 @@ process_native_tool_call, process_native_tool_result, ) +from utils.conversation_compaction import ( + agent_prompt_text, + reject_image_attachments_in_compacted_mode, +) from utils.conversations import append_turn_items_to_conversation from utils.otel_tracing import ( SpanAttributes, @@ -386,13 +390,14 @@ async def agent_response_generator( rag_id_mapping=context.rag_id_mapping, turn_summary=turn_summary, ) + reject_image_attachments_in_compacted_mode(responses_params, image_attachments) if image_attachments: prompt = build_multimodal_input( - cast(str, responses_params.input), + agent_prompt_text(responses_params), image_attachments, ) else: - prompt = cast(str, responses_params.input) + prompt = agent_prompt_text(responses_params) logger.debug("Starting agent streaming response processing") async with agent.run_stream_events(prompt) as stream: diff --git a/src/utils/conversation_compaction.py b/src/utils/conversation_compaction.py index 479ed8263..f826bab84 100644 --- a/src/utils/conversation_compaction.py +++ b/src/utils/conversation_compaction.py @@ -47,6 +47,7 @@ from dataclasses import dataclass from typing import Any, Optional, cast +from fastapi import HTTPException from ogx_api.openai_responses import OpenAIResponseMessage from ogx_client import AsyncOgxClient from ogx_client.types.conversations.item_create_params import Item @@ -55,6 +56,7 @@ from cache.cache_error import CacheError from configuration import configuration from log import get_logger +from models.api.responses.error import UnprocessableEntityResponse from models.common.responses.responses_api_params import ResponsesApiParams from models.common.responses.types import ResponseInput from models.common.turn_summary import ContextStatus @@ -249,6 +251,85 @@ def _verbatim_input_message(item: Any) -> Optional[OpenAIResponseMessage]: return OpenAIResponseMessage(role=cast(Any, role), content=text) +def agent_prompt_text(params: ResponsesApiParams) -> str: + """Return the textual user prompt for a pydantic-ai agent run. + + In compacted mode ``params.input`` is the explicit item list built by + :func:`_build_explicit_input` (summaries + recent turns + new query), so + the new user query is the trailing message item. The agent pipeline still + needs a plain string prompt (capabilities and multimodal input operate on + it); the full explicit list reaches the request body separately via the + ``extra_body`` input override (LCORE-3582). + + Args: + params: Prepared (possibly compaction-rewritten) request parameters. + + Returns: + ``params.input`` unchanged when it is a string; otherwise the text of + the last message item in the explicit list, or ``""`` when there is + none. + """ + if isinstance(params.input, str): + return params.input + for item in reversed(list(params.input)): + if is_message_item(item): + text = extract_message_text(item) + if text: + return text + # The wire input still carries the explicit list via the extra_body + # override, so the request itself is well-formed — but capabilities and + # multimodal construction operate on this prompt, and an explicit input + # with no textual message item means compaction produced something + # unexpected. Surface it rather than degrading silently. + logger.warning( + "Explicit compacted input carries no textual message item; " + "agent prompt falls back to an empty string" + ) + return "" + + +def reject_image_attachments_in_compacted_mode( + params: ResponsesApiParams, + image_attachments: Optional[Sequence[Any]], +) -> None: + """Reject a compacted turn that carries image attachments (LCORE-3582). + + In compacted mode the wire ``input`` is overridden with the explicit item + list built by :func:`_build_explicit_input`, which is text-only: image + attachments are converted to pydantic-ai ``ImageUrl`` parts on the prompt, + and the override replaces the prompt-derived input wholesale, so those + parts never reach the request body. + + Answering anyway would return a confident response that never saw the + image, with nothing to tell the caller their attachment was ignored. Fail + explicitly instead until the explicit input can carry ``input_image`` + content parts of its own (LCORE-3789). + + Args: + params: Prepared (possibly compaction-rewritten) request parameters. + image_attachments: Image attachments for this turn, if any. + + Raises: + HTTPException: 422 when the turn is compacted and carries images. + """ + if not image_attachments or not params.omit_conversation: + return + logger.warning( + "Rejecting compacted turn with %d image attachment(s): the explicit " + "input override cannot carry image content parts (LCORE-3789)", + len(image_attachments), + ) + response = UnprocessableEntityResponse( + response="Image attachments are not supported on this conversation", + cause=( + "This conversation has been compacted to fit the model's context " + "window, and compacted turns cannot carry image attachments yet. " + "Send the image in a new conversation, or retry without it." + ), + ) + raise HTTPException(**response.model_dump()) + + def _query_input_message(original_input: ResponseInput) -> list[Any]: """Render the new user query as explicit input items. diff --git a/tests/unit/pydantic_ai_lightspeed/llamastack/test_model.py b/tests/unit/pydantic_ai_lightspeed/llamastack/test_model.py index 9a059cd72..3773c37d9 100644 --- a/tests/unit/pydantic_ai_lightspeed/llamastack/test_model.py +++ b/tests/unit/pydantic_ai_lightspeed/llamastack/test_model.py @@ -6,6 +6,7 @@ from typing import Any import pytest +from ogx_api.openai_responses import OpenAIResponseMessage from openai.types import responses from pydantic_ai import ModelMessage, UnexpectedModelBehavior from pydantic_ai.messages import ModelResponse @@ -120,6 +121,86 @@ def test_none_fields_excluded(self) -> None: assert "extra_headers" not in settings assert "openai_previous_response_id" not in settings + def test_compacted_input_overrides_via_extra_body(self) -> None: + """Test compacted params carry the explicit input list in extra_body.""" + items = [ + OpenAIResponseMessage( + role="user", content="Summary of earlier conversation:\nS1" + ), + OpenAIResponseMessage(role="user", content="new question"), + ] + params = _make_params(input=items, omit_conversation=True) + settings = _model_settings_from_responses_params(params) + extra_body = settings["extra_body"] + assert "conversation" not in extra_body + assert extra_body["input"] == [ + { + "role": "user", + "content": "Summary of earlier conversation:\nS1", + "type": "message", + }, + {"role": "user", "content": "new question", "type": "message"}, + ] + + def test_string_input_never_lands_in_extra_body(self) -> None: + """Test that a plain string input is not duplicated into extra_body.""" + params = _make_params(input="hello", omit_conversation=True) + settings = _model_settings_from_responses_params(params) + assert "input" not in settings.get("extra_body", {}) + + def test_non_compacted_list_input_not_in_extra_body(self) -> None: + """Test that without omit_conversation the input stays out of extra_body.""" + items = [OpenAIResponseMessage(role="user", content="q")] + params = _make_params(input=items, omit_conversation=False) + settings = _model_settings_from_responses_params(params) + assert "input" not in settings.get("extra_body", {}) + + +class TestPrepareCompactedInput: + """Tests for the compacted-input tool-loop guard.""" + + @pytest.fixture(name="model") + def model_fixture(self, mocker: MockerFixture) -> OgxResponsesModel: + """Create a OgxResponsesModel with mocked __init__.""" + mocker.patch.object(OgxResponsesModel, "__init__", return_value=None) + return OgxResponsesModel("test-model") + + def test_no_input_override_returns_unchanged( + self, model: OgxResponsesModel, mocker: MockerFixture + ) -> None: + """Test settings without an input override pass through untouched.""" + messages = [mocker.Mock()] + settings: ModelSettings = {"extra_body": {"max_infer_iters": 5}} + assert model._prepare_compacted_input(messages, settings) is settings + + def test_first_request_keeps_input_override( + self, model: OgxResponsesModel, mocker: MockerFixture + ) -> None: + """Test the override survives when no ModelResponse is in messages.""" + messages = [mocker.Mock(spec=[])] + settings: ModelSettings = { + "extra_body": {"input": [{"role": "user", "content": "q"}]} + } + assert model._prepare_compacted_input(messages, settings) is settings + + def test_tool_loop_continuation_drops_input_override( + self, model: OgxResponsesModel + ) -> None: + """Test the override is dropped once a ModelResponse exists.""" + messages: list[ModelMessage] = [ModelResponse(parts=[])] + settings: ModelSettings = { + "extra_body": { + "input": [{"role": "user", "content": "q"}], + "max_infer_iters": 5, + } + } + result = model._prepare_compacted_input(messages, settings) + assert result is not settings + assert "input" not in result["extra_body"] + assert result["extra_body"]["max_infer_iters"] == 5 + # original settings untouched + assert "input" in settings["extra_body"] + class TestFromOgxClient: """Tests for OgxResponsesModel.from_ogx_client factory.""" diff --git a/tests/unit/utils/agents/test_query.py b/tests/unit/utils/agents/test_query.py index ea630159b..b31253aac 100644 --- a/tests/unit/utils/agents/test_query.py +++ b/tests/unit/utils/agents/test_query.py @@ -6,6 +6,7 @@ import pytest from fastapi import HTTPException +from ogx_api.openai_responses import OpenAIResponseMessage from ogx_client import APIConnectionError, APIStatusError from pydantic_ai.messages import ( FinishReason, @@ -427,6 +428,141 @@ async def test_success_returns_turn_summary( assert summary.llm_response == "Hello!" assert summary.id == "resp-success" + @pytest.mark.asyncio + async def test_compacted_input_runs_agent_with_prompt_text( + self, + mocker: MockerFixture, + make_agent_run_result: Callable[..., Any], + make_responses_params: Callable[..., ResponsesApiParams], + patch_recording_metrics: None, + ) -> None: + """Test compacted explicit input is reduced to the query text for agent.run.""" + explicit = [ + OpenAIResponseMessage( + role="user", content="Summary of earlier conversation:\nS1" + ), + OpenAIResponseMessage(role="user", content="new question"), + ] + params = make_responses_params(input_text="ignored").model_copy( + update={"input": explicit, "omit_conversation": True} + ) + run_result = make_agent_run_result(content="Answer") + mock_agent = mocker.AsyncMock() + mock_agent.run = mocker.AsyncMock(return_value=run_result) + mocker.patch("utils.agents.query.build_agent", return_value=mock_agent) + + summary = await retrieve_agent_response( + client=mocker.AsyncMock(), + responses_params=params, + moderation_result=ShieldModerationPassed(), + endpoint_path=ENDPOINT_PATH_QUERY, + ) + + mock_agent.run.assert_awaited_once_with("new question") + assert summary.llm_response == "Answer" + + @pytest.mark.asyncio + async def test_blocked_moderation_compacted_skips_append( + self, + mocker: MockerFixture, + make_responses_params: Callable[..., ResponsesApiParams], + blocked_moderation: ShieldModerationBlocked, + ) -> None: + """Test blocked moderation does not append explicit input in compacted mode.""" + params = make_responses_params().model_copy( + update={ + "input": [OpenAIResponseMessage(role="user", content="q")], + "omit_conversation": True, + } + ) + mock_append = mocker.patch( + "utils.agents.query.append_turn_items_to_conversation", + new=mocker.AsyncMock(), + ) + + summary = await retrieve_agent_response( + client=mocker.AsyncMock(), + responses_params=params, + moderation_result=blocked_moderation, + endpoint_path=ENDPOINT_PATH_QUERY, + ) + + mock_append.assert_not_awaited() + assert summary.llm_response == "Content blocked by shield." + + @pytest.mark.asyncio + async def test_inference_error_is_logged( + self, + mocker: MockerFixture, + make_responses_params: Callable[..., ResponsesApiParams], + patch_recording_metrics: None, + caplog: pytest.LogCaptureFixture, + ) -> None: + """Test an upstream connection failure logs a warning, not an error. + + APIConnectionError maps to 503: the backend is down, which is not a + service fault of ours, so it must not be logged at error level with a + traceback — that noise would bury the genuine 500s this logging exists + to surface (LCORE-3582). + """ + mock_agent = mocker.AsyncMock() + mock_agent.run = mocker.AsyncMock( + side_effect=APIConnectionError(request=mocker.Mock()) + ) + mocker.patch("utils.agents.query.build_agent", return_value=mock_agent) + + with caplog.at_level("WARNING"): + with pytest.raises(HTTPException): + await retrieve_agent_response( + client=mocker.AsyncMock(), + responses_params=make_responses_params(), + moderation_result=ShieldModerationPassed(), + endpoint_path=ENDPOINT_PATH_QUERY, + ) + + assert any( + record.levelname == "WARNING" + and "Agent inference returned 503" in record.message + for record in caplog.records + ) + assert not any(record.levelname == "ERROR" for record in caplog.records) + + @pytest.mark.asyncio + async def test_unclassified_inference_error_is_logged_with_traceback( + self, + mocker: MockerFixture, + make_responses_params: Callable[..., ResponsesApiParams], + patch_recording_metrics: None, + caplog: pytest.LogCaptureFixture, + ) -> None: + """Test an unclassified failure is logged at error level with the traceback. + + This is the case the logging was added for: it maps to a generic 500, + the mapped HTTPException discards the original exception, and callers + raise without logging, so without this the failure left nothing behind. + """ + mock_agent = mocker.AsyncMock() + mock_agent.run = mocker.AsyncMock(side_effect=RuntimeError("kaboom")) + mocker.patch("utils.agents.query.build_agent", return_value=mock_agent) + + with caplog.at_level("ERROR"): + with pytest.raises(HTTPException): + await retrieve_agent_response( + client=mocker.AsyncMock(), + responses_params=make_responses_params(), + moderation_result=ShieldModerationPassed(), + endpoint_path=ENDPOINT_PATH_QUERY, + ) + + matching = [ + record + for record in caplog.records + if record.levelname == "ERROR" + and "Agent inference failed" in record.message + ] + assert matching, "unclassified failure was not logged at error level" + assert matching[0].exc_info is not None, "traceback was not attached" + @pytest.mark.asyncio async def test_success_with_image_attachments_sends_multimodal_prompt( self, diff --git a/tests/unit/utils/agents/test_streaming.py b/tests/unit/utils/agents/test_streaming.py index ff7f58ffc..9d2c2d172 100644 --- a/tests/unit/utils/agents/test_streaming.py +++ b/tests/unit/utils/agents/test_streaming.py @@ -10,6 +10,7 @@ import pytest from fastapi import HTTPException +from ogx_api.openai_responses import OpenAIResponseMessage from ogx_client import APIStatusError from opentelemetry.sdk.trace.export.in_memory_span_exporter import ( InMemorySpanExporter, @@ -1284,6 +1285,57 @@ async def test_streams_token_events_and_updates_summary( assert turn_summary.token_usage.input_tokens == 4 assert turn_summary.token_usage.output_tokens == 2 + @pytest.mark.asyncio + async def test_compacted_input_streams_with_prompt_text( + self, + mocker: MockerFixture, + make_generator_context: Callable[..., ResponseGeneratorContext], + make_responses_params: Callable[..., ResponsesApiParams], + make_agent_run_result: Callable[..., Any], + patch_recording_metrics: None, + ) -> None: + """Test compacted explicit input is reduced to the query text for streaming.""" + context = make_generator_context() + turn_summary = TurnSummary() + run_result = make_agent_run_result(content="Answer", response_id="resp-c1") + events = [ + PartStartEvent(index=0, part=TextPart(content="Answer")), + AgentRunResultEvent(result=run_result), + ] + mock_agent = mocker.Mock() + mock_agent.run_stream_events.return_value = _mock_run_stream(events) + mocker.patch( + "utils.agents.streaming.get_agent_finish_reason", + return_value=AgentFinishReason.SUCCESS, + ) + mocker.patch( + "utils.agents.streaming.deduplicate_referenced_documents", + side_effect=lambda docs: docs, + ) + + explicit = [ + OpenAIResponseMessage( + role="user", content="Summary of earlier conversation:\nS1" + ), + OpenAIResponseMessage(role="user", content="new question"), + ] + params = make_responses_params(input_text="ignored").model_copy( + update={"input": explicit, "omit_conversation": True} + ) + + _ = [ + event + async for event in agent_response_generator( + mock_agent, + params, + context, + turn_summary, + ENDPOINT_PATH_STREAMING_QUERY, + ) + ] + + assert mock_agent.run_stream_events.call_args[0][0] == "new question" + @pytest.mark.asyncio async def test_streams_with_image_attachments_passes_multimodal_prompt( self, diff --git a/tests/unit/utils/test_conversation_compaction.py b/tests/unit/utils/test_conversation_compaction.py index 4499a4a16..f7b3a4af2 100644 --- a/tests/unit/utils/test_conversation_compaction.py +++ b/tests/unit/utils/test_conversation_compaction.py @@ -8,6 +8,7 @@ from typing import Any, Optional, cast import pytest +from fastapi import HTTPException from ogx_api.openai_responses import OpenAIResponseMessage from pytest_mock import MockerFixture @@ -128,6 +129,59 @@ def test_compaction_result_context_status() -> None: assert cc.CompactionResult(params, compacted=True).context_status == "summarized" +def test_agent_prompt_text_string_input() -> None: + """A plain string input is returned unchanged.""" + assert cc.agent_prompt_text(_params("what is a pod?")) == "what is a pod?" + + +def test_agent_prompt_text_explicit_list_returns_last_message_text() -> None: + """For compacted explicit input, the trailing user query text is returned.""" + params = _params() + explicit = cc._build_explicit_input( + ["earlier summary"], [_msg("assistant", "prior answer")], "new question" + ) + compacted = params.model_copy(update={"input": explicit, "omit_conversation": True}) + assert cc.agent_prompt_text(compacted) == "new question" + + +@pytest.mark.parametrize( + ("omit_conversation", "has_images"), + [(True, False), (False, True), (False, False)], +) +def test_reject_image_attachments_allows_supported_combinations( + omit_conversation: bool, has_images: bool +) -> None: + """Only a compacted turn carrying images is rejected; the rest pass through.""" + params = _params().model_copy(update={"omit_conversation": omit_conversation}) + attachments = [object()] if has_images else None + cc.reject_image_attachments_in_compacted_mode(params, attachments) + + +def test_reject_image_attachments_in_compacted_mode_raises_422() -> None: + """A compacted turn with images fails explicitly instead of ignoring them. + + The explicit-input override replaces the prompt-derived request input, and + the explicit list is text-only, so the images would never reach the model. + Answering anyway would return a confident response that never saw the + image, with nothing telling the caller it was dropped (LCORE-3582). + """ + params = _params().model_copy(update={"omit_conversation": True}) + + with pytest.raises(HTTPException) as exc_info: + cc.reject_image_attachments_in_compacted_mode(params, [object()]) + + assert exc_info.value.status_code == 422 + detail = exc_info.value.detail + assert "Image attachments are not supported" in detail["response"] + assert "compacted" in detail["cause"] + + +def test_agent_prompt_text_empty_list_returns_empty() -> None: + """An empty explicit list yields an empty prompt rather than crashing.""" + params = _params().model_copy(update={"input": [], "omit_conversation": True}) + assert cc.agent_prompt_text(params) == "" + + # --- apply_compaction ---