fix: normalize compacted Responses user inputs before session reuse (#2925)

Author: seratch

src/agents/memory/openai_responses_compaction_session.py

@@ -2,10 +2,11 @@
 
 import logging
 from collections.abc import Callable
-from typing import TYPE_CHECKING, Any, Literal
+from typing import TYPE_CHECKING, Any, Literal, cast
 
 from openai import AsyncOpenAI
 
+from ..items import TResponseInputItem
 from ..models._openai_shared import get_default_openai_client
 from ..run_internal.items import normalize_input_items_for_api
 from .openai_conversations_session import OpenAIConversationsSession
@@ -16,7 +17,6 @@
 )
 
 if TYPE_CHECKING:
-    from ..items import TResponseInputItem
     from .session import Session
 
 logger = logging.getLogger("openai-agents.openai.compaction")
@@ -213,19 +213,8 @@ async def run_compaction(self, args: OpenAIResponsesCompactionArgs | None = None
 
         compacted = await self.client.responses.compact(**compact_kwargs)
 
+        output_items = _normalize_compaction_output_items(compacted.output or [])
         await self.underlying_session.clear_session()
-        output_items: list[TResponseInputItem] = []
-        if compacted.output:
-            for item in compacted.output:
-                if isinstance(item, dict):
-                    output_items.append(item)
-                else:
-                    # Suppress Pydantic literal warnings: responses.compact can return
-                    # user-style input_text content inside ResponseOutputMessage.
-                    output_items.append(
-                        item.model_dump(exclude_unset=True, warnings=False)  # type: ignore
-                    )
-
         output_items = _strip_orphaned_assistant_ids(output_items)
 
         if output_items:
@@ -339,6 +328,101 @@ def _strip_orphaned_assistant_ids(
     return cleaned
 
 
+def _normalize_compaction_output_items(items: list[Any]) -> list[TResponseInputItem]:
+    """Normalize compacted output into replay-safe Responses input items."""
+    output_items: list[TResponseInputItem] = []
+    for item in items:
+        if isinstance(item, dict):
+            output_item = item
+        else:
+            # Suppress Pydantic literal warnings: responses.compact can return
+            # user-style input_text content inside ResponseOutputMessage.
+            output_item = item.model_dump(exclude_unset=True, warnings=False)
+
+        if (
+            isinstance(output_item, dict)
+            and output_item.get("type") == "message"
+            and output_item.get("role") == "user"
+        ):
+            output_items.append(_normalize_compaction_user_message(output_item))
+            continue
+
+        output_items.append(cast(TResponseInputItem, output_item))
+    return output_items
+
+
+def _normalize_compaction_user_message(item: dict[str, Any]) -> TResponseInputItem:
+    """Normalize compacted user message content before it is reused as input."""
+    content = item.get("content")
+    if not isinstance(content, list):
+        return cast(TResponseInputItem, item)
+
+    normalized_content: list[Any] = []
+    for content_item in content:
+        if not isinstance(content_item, dict):
+            normalized_content.append(content_item)
+            continue
+
+        content_type = content_item.get("type")
+        if content_type == "input_image":
+            normalized_content.append(_normalize_compaction_input_image(content_item))
+        elif content_type == "input_file":
+            normalized_content.append(_normalize_compaction_input_file(content_item))
+        else:
+            normalized_content.append(content_item)
+
+    normalized_item = dict(item)
+    normalized_item["content"] = normalized_content
+    return cast(TResponseInputItem, normalized_item)
+
+
+def _normalize_compaction_input_image(content_item: dict[str, Any]) -> dict[str, Any]:
+    """Return a valid replay shape for a compacted Responses image input."""
+    normalized = {"type": "input_image"}
+
+    image_url = content_item.get("image_url")
+    file_id = content_item.get("file_id")
+    if isinstance(image_url, str) and image_url:
+        normalized["image_url"] = image_url
+    elif isinstance(file_id, str) and file_id:
+        normalized["file_id"] = file_id
+    else:
+        raise ValueError("Compaction input_image item missing image_url or file_id.")
+
+    detail = content_item.get("detail")
+    if isinstance(detail, str) and detail:
+        normalized["detail"] = detail
+
+    return normalized
+
+
+def _normalize_compaction_input_file(content_item: dict[str, Any]) -> dict[str, Any]:
+    """Return a valid replay shape for a compacted Responses file input."""
+    normalized = {"type": "input_file"}
+
+    file_data = content_item.get("file_data")
+    file_url = content_item.get("file_url")
+    file_id = content_item.get("file_id")
+    if isinstance(file_data, str) and file_data:
+        normalized["file_data"] = file_data
+    elif isinstance(file_url, str) and file_url:
+        normalized["file_url"] = file_url
+    elif isinstance(file_id, str) and file_id:
+        normalized["file_id"] = file_id
+    else:
+        raise ValueError("Compaction input_file item missing file_data, file_url, or file_id.")
+
+    filename = content_item.get("filename")
+    if isinstance(filename, str) and filename:
+        normalized["filename"] = filename
+
+    detail = content_item.get("detail")
+    if isinstance(detail, str) and detail:
+        normalized["detail"] = detail
+
+    return normalized
+
+
 def _normalize_compaction_session_items(
     items: list[TResponseInputItem],
 ) -> list[TResponseInputItem]:

tests/memory/test_openai_responses_compaction_session.py

@@ -544,6 +544,206 @@ def model_dump(
             model="gpt-4.1",
         )
 
+    @pytest.mark.asyncio
+    async def test_run_compaction_normalizes_compacted_user_image_messages(self) -> None:
+        mock_session = self.create_mock_session()
+        mock_session.get_items.return_value = []
+
+        mock_compact_response = MagicMock()
+        mock_compact_response.output = [
+            {
+                "type": "message",
+                "role": "user",
+                "content": [
+                    {"type": "input_text", "text": "analyze this input"},
+                    {
+                        "type": "input_image",
+                        "image_url": "https://example.com/image.png",
+                        "file_id": None,
+                        "detail": "auto",
+                    },
+                ],
+            }
+        ]
+
+        mock_client = MagicMock()
+        mock_client.responses.compact = AsyncMock(return_value=mock_compact_response)
+
+        session = OpenAIResponsesCompactionSession(
+            session_id="test",
+            underlying_session=mock_session,
+            client=mock_client,
+            compaction_mode="input",
+        )
+
+        await session.run_compaction({"force": True, "compaction_mode": "input"})
+
+        stored_items = mock_session.add_items.call_args[0][0]
+        assert stored_items == [
+            {
+                "type": "message",
+                "role": "user",
+                "content": [
+                    {"type": "input_text", "text": "analyze this input"},
+                    {
+                        "type": "input_image",
+                        "image_url": "https://example.com/image.png",
+                        "detail": "auto",
+                    },
+                ],
+            }
+        ]
+
+    @pytest.mark.asyncio
+    async def test_run_compaction_normalizes_compacted_user_file_messages(self) -> None:
+        mock_session = self.create_mock_session()
+        mock_session.get_items.return_value = []
+
+        mock_compact_response = MagicMock()
+        mock_compact_response.output = [
+            {
+                "type": "message",
+                "role": "user",
+                "content": [
+                    {"type": "input_text", "text": "analyze this input"},
+                    {
+                        "type": "input_file",
+                        "file_url": "https://example.com/report.pdf",
+                        "file_id": None,
+                        "filename": "report.pdf",
+                        "detail": "high",
+                    },
+                ],
+            }
+        ]
+
+        mock_client = MagicMock()
+        mock_client.responses.compact = AsyncMock(return_value=mock_compact_response)
+
+        session = OpenAIResponsesCompactionSession(
+            session_id="test",
+            underlying_session=mock_session,
+            client=mock_client,
+            compaction_mode="input",
+        )
+
+        await session.run_compaction({"force": True, "compaction_mode": "input"})
+
+        stored_items = mock_session.add_items.call_args[0][0]
+        assert stored_items == [
+            {
+                "type": "message",
+                "role": "user",
+                "content": [
+                    {"type": "input_text", "text": "analyze this input"},
+                    {
+                        "type": "input_file",
+                        "file_url": "https://example.com/report.pdf",
+                        "filename": "report.pdf",
+                        "detail": "high",
+                    },
+                ],
+            }
+        ]
+
+    @pytest.mark.asyncio
+    async def test_run_compaction_normalizes_file_id_inputs_and_preserves_metadata(self) -> None:
+        mock_session = self.create_mock_session()
+        mock_session.get_items.return_value = []
+
+        mock_compact_response = MagicMock()
+        mock_compact_response.output = [
+            {
+                "type": "message",
+                "role": "user",
+                "content": [
+                    {"type": "input_text", "text": "analyze this input"},
+                    {
+                        "type": "input_file",
+                        "file_id": "file_123",
+                        "file_url": None,
+                        "filename": "report.pdf",
+                        "detail": "low",
+                    },
+                ],
+            }
+        ]
+
+        mock_client = MagicMock()
+        mock_client.responses.compact = AsyncMock(return_value=mock_compact_response)
+
+        session = OpenAIResponsesCompactionSession(
+            session_id="test",
+            underlying_session=mock_session,
+            client=mock_client,
+            compaction_mode="input",
+        )
+
+        await session.run_compaction({"force": True, "compaction_mode": "input"})
+
+        stored_items = mock_session.add_items.call_args[0][0]
+        assert stored_items == [
+            {
+                "type": "message",
+                "role": "user",
+                "content": [
+                    {"type": "input_text", "text": "analyze this input"},
+                    {
+                        "type": "input_file",
+                        "file_id": "file_123",
+                        "filename": "report.pdf",
+                        "detail": "low",
+                    },
+                ],
+            }
+        ]
+
+    @pytest.mark.asyncio
+    async def test_run_compaction_preserves_history_when_output_normalization_fails(self) -> None:
+        history = [
+            {
+                "type": "message",
+                "role": "user",
+                "content": [{"type": "input_text", "text": "hello"}],
+            },
+            {
+                "type": "message",
+                "role": "assistant",
+                "status": "completed",
+                "content": [{"type": "output_text", "text": "world"}],
+            },
+        ]
+        underlying = SimpleListSession(history=cast(list[TResponseInputItem], history))
+
+        mock_compact_response = MagicMock()
+        mock_compact_response.output = [
+            {
+                "type": "message",
+                "role": "user",
+                "content": [
+                    {"type": "input_text", "text": "hello"},
+                    {"type": "input_image", "detail": "auto"},
+                ],
+            }
+        ]
+
+        mock_client = MagicMock()
+        mock_client.responses.compact = AsyncMock(return_value=mock_compact_response)
+
+        session = OpenAIResponsesCompactionSession(
+            session_id="test",
+            underlying_session=underlying,
+            client=mock_client,
+            compaction_mode="input",
+        )
+
+        with pytest.raises(
+            ValueError, match="Compaction input_image item missing image_url or file_id."
+        ):
+            await session.run_compaction({"force": True, "compaction_mode": "input"})
+
+        assert await session.get_items() == history
+
     @pytest.mark.asyncio
     async def test_compaction_runs_during_runner_flow(self) -> None:
         """Ensure Runner triggers compaction when using a compaction-aware session."""