LCORE-3582: fix compacted-mode 500s in the agent pipeline

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(

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

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,

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:

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.