fix(memory): compact the full stored history of a limited session

src/agents/memory/openai_responses_compaction_session.py

@@ -442,7 +442,11 @@ async def _ensure_compaction_candidates(
         if self._compaction_candidate_items is not None and self._session_items is not None:
             return (self._compaction_candidate_items[:], self._session_items[:])
 
-        history = _normalize_compaction_session_items(await self.underlying_session.get_items())
+        # Bypass SessionSettings.limit so compaction sees stored history, not just the
+        # retrieval window. Replacement still writes over the full store.
+        history = _normalize_compaction_session_items(
+            await self._get_all_underlying_session_items()
+        )
         candidates = select_compaction_candidate_items(history)
         self._compaction_candidate_items = candidates
         self._session_items = history

tests/memory/test_openai_responses_compaction_session.py

@@ -1026,6 +1026,50 @@ async def clear_session(self) -> None:
         assert failing_session.clear_calls == 2
         assert failing_session.add_calls == 2
 
+    @pytest.mark.asyncio
+    async def test_run_compaction_input_uses_full_history_when_session_limit_applies(
+        self, tmp_path
+    ) -> None:
+        history: list[TResponseInputItem] = [
+            cast(TResponseInputItem, {"type": "message", "role": "user", "content": "oldest"}),
+            cast(
+                TResponseInputItem,
+                {"type": "message", "role": "assistant", "content": "middle"},
+            ),
+            cast(TResponseInputItem, {"type": "message", "role": "user", "content": "newest"}),
+        ]
+        compacted_items: list[TResponseInputItem] = [
+            cast(
+                TResponseInputItem,
+                {"type": "message", "role": "assistant", "content": "compacted"},
+            )
+        ]
+
+        underlying = SQLiteSession(
+            "limited-compact",
+            str(tmp_path / "limited_compact.db"),
+            session_settings=SessionSettings(limit=2),
+        )
+        await underlying.add_items(history)
+        assert len(await underlying.get_items()) == 2
+
+        mock_compact_response = MagicMock()
+        mock_compact_response.output = compacted_items
+        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",
+        )
+
+        await session.run_compaction({"force": True})
+
+        compact_input = mock_client.responses.compact.call_args.kwargs["input"]
+        assert compact_input == history
+
     @pytest.mark.asyncio
     async def test_run_compaction_does_not_restore_when_clear_fails_without_mutation(
         self,