From d6c84248840252204cf6f55c1e5803832d1e2be5 Mon Sep 17 00:00:00 2001
From: winklemad <winklemad@outlook.com>
Date: Tue, 14 Jul 2026 00:03:43 +0530
Subject: [PATCH 1/9] fix: load full history when compacting a limited session
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

OpenAIResponsesCompactionSession._ensure_compaction_candidates loaded the
underlying history with a bare underlying_session.get_items(), which resolves
to session_settings.limit and returns only the latest N items. In input/auto
mode those truncated items are what gets sent to responses.compact, and
run_compaction then clears the session and replaces it with the compacted
summary — so any history older than the limit window is silently and
permanently lost.

The restore/replace paths already read the full history through
_get_all_underlying_session_items() (limit=_ALL_SESSION_ITEMS_LIMIT), added in
#3117; the candidate-loading path was missed. Use the same helper there so
compaction always operates on the complete stored history.
---
 .../openai_responses_compaction_session.py    |  4 +-
 ...est_openai_responses_compaction_session.py | 44 +++++++++++++++++++
 2 files changed, 47 insertions(+), 1 deletion(-)

diff --git a/src/agents/memory/openai_responses_compaction_session.py b/src/agents/memory/openai_responses_compaction_session.py
index 2ec40663db..f8c01c8c96 100644
--- a/src/agents/memory/openai_responses_compaction_session.py
+++ b/src/agents/memory/openai_responses_compaction_session.py
@@ -367,7 +367,9 @@ 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())
+        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
diff --git a/tests/memory/test_openai_responses_compaction_session.py b/tests/memory/test_openai_responses_compaction_session.py
index fe893cf88a..f5251ec41b 100644
--- a/tests/memory/test_openai_responses_compaction_session.py
+++ b/tests/memory/test_openai_responses_compaction_session.py
@@ -2,6 +2,7 @@
 
 import logging
 import warnings as warnings_module
+from pathlib import Path
 from types import SimpleNamespace
 from typing import Any, cast
 from unittest.mock import AsyncMock, MagicMock
@@ -22,6 +23,7 @@
     is_openai_model_name,
     select_compaction_candidate_items,
 )
+from agents.memory.sqlite_session import SQLiteSession
 from agents.run_internal.items import (
     TOOL_CALL_SESSION_DESCRIPTION_KEY,
     TOOL_CALL_SESSION_TITLE_KEY,
@@ -194,6 +196,48 @@ async def test_run_compaction_input_mode_without_response_id(self) -> None:
         assert "previous_response_id" not in call_kwargs
         assert call_kwargs.get("input") == items
 
+    @pytest.mark.asyncio
+    async def test_run_compaction_uses_full_history_when_underlying_session_has_limit(
+        self, tmp_path: Path
+    ) -> None:
+        # A limited underlying session must not cause compaction to lose older history:
+        # the compact call has to receive the full stored history, not just the latest
+        # `limit` items, otherwise everything before the window is silently dropped when
+        # the session is cleared and replaced with the compacted summary.
+        underlying = SQLiteSession(
+            "test",
+            db_path=str(tmp_path / "history.db"),
+            session_settings=SessionSettings(limit=3),
+        )
+        items: list[TResponseInputItem] = [
+            cast(
+                TResponseInputItem,
+                {"type": "message", "role": "assistant", "content": f"m{i}"},
+            )
+            for i in range(8)
+        ]
+        await underlying.add_items(items)
+
+        mock_compact_response = MagicMock()
+        mock_compact_response.output = [
+            {"type": "message", "role": "assistant", "content": "compacted"}
+        ]
+        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})
+
+        mock_client.responses.compact.assert_called_once()
+        sent = mock_client.responses.compact.call_args.kwargs.get("input")
+        assert [item["content"] for item in sent] == [f"m{i}" for i in range(8)]
+
     @pytest.mark.asyncio
     async def test_run_compaction_auto_without_response_id_uses_input(self) -> None:
         mock_session = self.create_mock_session()

From 5469581c1cb1ea4d710ad7dd1bc82c5666baf004 Mon Sep 17 00:00:00 2001
From: winklemad <winklemad@outlook.com>
Date: Tue, 14 Jul 2026 04:37:16 +0530
Subject: [PATCH 2/9] Fall back to input compaction in auto mode when the
 session view is limited

In auto mode with a stored response_id, compaction resolves to
previous_response_id, which does not send the local history. When the
underlying session's default view is limit-bounded and does not cover the
full stored history, clearing and replacing the session with the
server-derived summary can still drop the unrepresented items. Auto now
falls back to full-history input compaction whenever the limited view does
not cover the stored history; an explicitly requested previous_response_id
compaction stays opt-in.

Add a no-force regression test for the auto path, and guard the test
assertions against an Optional input value for pyright.
---
 .../openai_responses_compaction_session.py    | 29 ++++++++++-
 ...est_openai_responses_compaction_session.py | 48 +++++++++++++++++++
 2 files changed, 75 insertions(+), 2 deletions(-)

diff --git a/src/agents/memory/openai_responses_compaction_session.py b/src/agents/memory/openai_responses_compaction_session.py
index f8c01c8c96..c8b3a5b55a 100644
--- a/src/agents/memory/openai_responses_compaction_session.py
+++ b/src/agents/memory/openai_responses_compaction_session.py
@@ -170,20 +170,34 @@ async def run_compaction(self, args: OpenAIResponsesCompactionArgs | None = None
                 self._last_unstored_response_id = None
         else:
             store = None
+        compaction_candidate_items, session_items = await self._ensure_compaction_candidates()
+
         resolved_mode = self._resolve_compaction_mode_for_response(
             response_id=self._response_id,
             store=store,
             requested_mode=requested_mode,
         )
 
+        # In auto mode, previous_response_id compaction relies on the server-side
+        # history for `response_id`. If the underlying session holds more history
+        # than its default (limit-bounded) view exposes, that server view may not
+        # represent everything; clearing and replacing the session with the
+        # server-derived summary would then silently drop the unrepresented items.
+        # Fall back to full-history input compaction in that case. An explicitly
+        # requested previous_response_id compaction is left opt-in and unchanged.
+        if (
+            (requested_mode or self.compaction_mode) == "auto"
+            and resolved_mode == "previous_response_id"
+            and not await self._underlying_view_covers_history()
+        ):
+            resolved_mode = "input"
+
         if resolved_mode == "previous_response_id" and not self._response_id:
             raise ValueError(
                 "OpenAIResponsesCompactionSession.run_compaction requires a response_id "
                 "when using previous_response_id compaction."
             )
 
-        compaction_candidate_items, session_items = await self._ensure_compaction_candidates()
-
         force = args.get("force", False) if args else False
         should_compact = force or self.should_trigger_compaction(
             {
@@ -245,6 +259,17 @@ async def get_items(self, limit: int | None = None) -> list[TResponseInputItem]:
     async def _get_all_underlying_session_items(self) -> list[TResponseInputItem]:
         return await self.underlying_session.get_items(limit=_ALL_SESSION_ITEMS_LIMIT)
 
+    async def _underlying_view_covers_history(self) -> bool:
+        """Whether the underlying session's default view contains the full history.
+
+        Returns False when the session applies a limit that hides older items, in
+        which case previous_response_id compaction could drop the unrepresented
+        history when the session is cleared and replaced.
+        """
+        default_view = await self.underlying_session.get_items()
+        full_view = await self._get_all_underlying_session_items()
+        return len(default_view) >= len(full_view)
+
     async def _replace_underlying_session_items(
         self,
         *,
diff --git a/tests/memory/test_openai_responses_compaction_session.py b/tests/memory/test_openai_responses_compaction_session.py
index f5251ec41b..2a59644b27 100644
--- a/tests/memory/test_openai_responses_compaction_session.py
+++ b/tests/memory/test_openai_responses_compaction_session.py
@@ -236,8 +236,56 @@ async def test_run_compaction_uses_full_history_when_underlying_session_has_limi
 
         mock_client.responses.compact.assert_called_once()
         sent = mock_client.responses.compact.call_args.kwargs.get("input")
+        assert sent is not None
         assert [item["content"] for item in sent] == [f"m{i}" for i in range(8)]
 
+    @pytest.mark.asyncio
+    async def test_run_compaction_auto_falls_back_to_input_when_history_truncated(
+        self, tmp_path: Path
+    ) -> None:
+        # Auto mode with a stored response_id normally resolves to
+        # previous_response_id, which does not send the local history. When the
+        # underlying session is limited so its default view does not cover the full
+        # stored history, that path would clear+replace the session and drop the
+        # unrepresented items. Auto must fall back to full-history input compaction
+        # instead. This exercises the no-force threshold path before replacement.
+        underlying = SQLiteSession(
+            "test",
+            db_path=str(tmp_path / "history.db"),
+            session_settings=SessionSettings(limit=3),
+        )
+        items: list[TResponseInputItem] = [
+            cast(
+                TResponseInputItem,
+                {"type": "message", "role": "assistant", "content": f"m{i}"},
+            )
+            for i in range(12)
+        ]
+        await underlying.add_items(items)
+
+        mock_compact_response = MagicMock()
+        mock_compact_response.output = [
+            {"type": "message", "role": "assistant", "content": "compacted"}
+        ]
+        mock_client = MagicMock()
+        mock_client.responses.compact = AsyncMock(return_value=mock_compact_response)
+
+        session = OpenAIResponsesCompactionSession(
+            session_id="test",
+            underlying_session=underlying,
+            client=mock_client,
+        )
+
+        # No force: 12 candidates exceed the default threshold and trigger compaction.
+        await session.run_compaction({"response_id": "resp_123"})
+
+        mock_client.responses.compact.assert_called_once()
+        call_kwargs = mock_client.responses.compact.call_args.kwargs
+        assert "previous_response_id" not in call_kwargs
+        sent = call_kwargs.get("input")
+        assert sent is not None
+        assert [item["content"] for item in sent] == [f"m{i}" for i in range(12)]
+
     @pytest.mark.asyncio
     async def test_run_compaction_auto_without_response_id_uses_input(self) -> None:
         mock_session = self.create_mock_session()

From 10471fd7f5f35b348f1a9931eb3c60ba3e2854e1 Mon Sep 17 00:00:00 2001
From: winklemad <winklemad@outlook.com>
Date: Tue, 14 Jul 2026 05:02:58 +0530
Subject: [PATCH 3/9] Do not reuse an unsafe response_id after auto falls back
 to input

When auto compaction falls back to input because the limited session view did
not cover the full history, the stored response_id still points at a server
response that omits the hidden items. Once the session later shrinks back
within the limit, a forced compaction would otherwise switch back to
previous_response_id and replace the full-history summary with a server-derived
one that drops those items. Mark the response_id unsafe on fallback, mirroring
the store=False path, and cover it with a regression test.
---
 .../openai_responses_compaction_session.py    | 20 ++++----
 ...est_openai_responses_compaction_session.py | 46 +++++++++++++++++++
 2 files changed, 55 insertions(+), 11 deletions(-)

diff --git a/src/agents/memory/openai_responses_compaction_session.py b/src/agents/memory/openai_responses_compaction_session.py
index c8b3a5b55a..f868a9fbdb 100644
--- a/src/agents/memory/openai_responses_compaction_session.py
+++ b/src/agents/memory/openai_responses_compaction_session.py
@@ -178,19 +178,19 @@ async def run_compaction(self, args: OpenAIResponsesCompactionArgs | None = None
             requested_mode=requested_mode,
         )
 
-        # In auto mode, previous_response_id compaction relies on the server-side
-        # history for `response_id`. If the underlying session holds more history
-        # than its default (limit-bounded) view exposes, that server view may not
-        # represent everything; clearing and replacing the session with the
-        # server-derived summary would then silently drop the unrepresented items.
-        # Fall back to full-history input compaction in that case. An explicitly
-        # requested previous_response_id compaction is left opt-in and unchanged.
+        # A limit-bounded session view hides older items from previous_response_id
+        # compaction, which drops them when the session is cleared and replaced. Fall
+        # back to full-history input; an explicit previous_response_id stays opt-in.
         if (
             (requested_mode or self.compaction_mode) == "auto"
             and resolved_mode == "previous_response_id"
             and not await self._underlying_view_covers_history()
         ):
             resolved_mode = "input"
+            # This response_id never covered the full history, so don't let a later
+            # compaction reuse it once the session shrinks back within the limit.
+            if self._response_id is not None:
+                self._last_unstored_response_id = self._response_id
 
         if resolved_mode == "previous_response_id" and not self._response_id:
             raise ValueError(
@@ -260,11 +260,9 @@ async def _get_all_underlying_session_items(self) -> list[TResponseInputItem]:
         return await self.underlying_session.get_items(limit=_ALL_SESSION_ITEMS_LIMIT)
 
     async def _underlying_view_covers_history(self) -> bool:
-        """Whether the underlying session's default view contains the full history.
+        """Whether the default session view already holds every stored item.
 
-        Returns False when the session applies a limit that hides older items, in
-        which case previous_response_id compaction could drop the unrepresented
-        history when the session is cleared and replaced.
+        False when a session limit hides older items from the default view.
         """
         default_view = await self.underlying_session.get_items()
         full_view = await self._get_all_underlying_session_items()
diff --git a/tests/memory/test_openai_responses_compaction_session.py b/tests/memory/test_openai_responses_compaction_session.py
index 2a59644b27..eae1b7a85c 100644
--- a/tests/memory/test_openai_responses_compaction_session.py
+++ b/tests/memory/test_openai_responses_compaction_session.py
@@ -286,6 +286,52 @@ async def test_run_compaction_auto_falls_back_to_input_when_history_truncated(
         assert sent is not None
         assert [item["content"] for item in sent] == [f"m{i}" for i in range(12)]
 
+    @pytest.mark.asyncio
+    async def test_run_compaction_auto_does_not_reuse_unsafe_response_id_after_input_fallback(
+        self, tmp_path: Path
+    ) -> None:
+        # After auto falls back to input because the stored response_id did not
+        # represent the hidden history, that response_id must not be reused later.
+        # Once the session shrinks to fit the limit, a second compaction would
+        # otherwise switch back to previous_response_id and drop the older items
+        # that only the full-history summary preserved.
+        underlying = SQLiteSession(
+            "test",
+            db_path=str(tmp_path / "history.db"),
+            session_settings=SessionSettings(limit=3),
+        )
+        items: list[TResponseInputItem] = [
+            cast(
+                TResponseInputItem,
+                {"type": "message", "role": "assistant", "content": f"m{i}"},
+            )
+            for i in range(12)
+        ]
+        await underlying.add_items(items)
+
+        mock_compact_response = MagicMock()
+        mock_compact_response.output = [
+            {"type": "message", "role": "assistant", "content": "summary"}
+        ]
+        mock_client = MagicMock()
+        mock_client.responses.compact = AsyncMock(return_value=mock_compact_response)
+
+        session = OpenAIResponsesCompactionSession(
+            session_id="test",
+            underlying_session=underlying,
+            client=mock_client,
+        )
+
+        # First compaction: truncated view -> falls back to input.
+        await session.run_compaction({"response_id": "resp_A"})
+        assert "previous_response_id" not in mock_client.responses.compact.call_args.kwargs
+
+        # The session now holds only the small summary (fits within limit=3). A second
+        # compaction must still use input rather than reuse the unsafe response_id.
+        mock_client.responses.compact.reset_mock()
+        await session.run_compaction({"force": True})
+        assert "previous_response_id" not in mock_client.responses.compact.call_args.kwargs
+
     @pytest.mark.asyncio
     async def test_run_compaction_auto_without_response_id_uses_input(self) -> None:
         mock_session = self.create_mock_session()

From 1ff8d282f96265f9ddf3abcf5ebc220ade581d56 Mon Sep 17 00:00:00 2001
From: winklemad <winklemad@outlook.com>
Date: Tue, 14 Jul 2026 05:19:38 +0530
Subject: [PATCH 4/9] Honor per-run session limits when choosing the compaction
 mode

The auto->input fallback checked the underlying session's default view, which
misses a limit supplied per run (e.g. RunConfig session settings) on an
otherwise-unlimited store: the response only saw the limited window, but the
check reported full coverage and kept using previous_response_id. Remember the
limit the Runner uses to prepare each turn's input via get_items and compare
against that window instead, so the fallback triggers whenever the response did
not see the full history.
---
 .../openai_responses_compaction_session.py    | 18 +++++++--
 ...est_openai_responses_compaction_session.py | 40 +++++++++++++++++++
 2 files changed, 54 insertions(+), 4 deletions(-)

diff --git a/src/agents/memory/openai_responses_compaction_session.py b/src/agents/memory/openai_responses_compaction_session.py
index f868a9fbdb..20933aeab1 100644
--- a/src/agents/memory/openai_responses_compaction_session.py
+++ b/src/agents/memory/openai_responses_compaction_session.py
@@ -133,6 +133,7 @@ def __init__(
         self._response_id: str | None = None
         self._deferred_response_id: str | None = None
         self._last_unstored_response_id: str | None = None
+        self._last_prepared_input_limit: int | None = None
 
     @property
     def client(self) -> AsyncOpenAI:
@@ -254,19 +255,28 @@ async def run_compaction(self, args: OpenAIResponsesCompactionArgs | None = None
         )
 
     async def get_items(self, limit: int | None = None) -> list[TResponseInputItem]:
+        # The Runner reads history through get_items(limit=...) to build each turn's
+        # model input, so this is the window the resulting response actually saw. Record
+        # it so compaction can tell whether the response covered the full history or only
+        # a limited slice (the limit can come from a per-run RunConfig, not just the
+        # underlying session's own settings).
+        self._last_prepared_input_limit = limit
         return await self.underlying_session.get_items(limit)
 
     async def _get_all_underlying_session_items(self) -> list[TResponseInputItem]:
         return await self.underlying_session.get_items(limit=_ALL_SESSION_ITEMS_LIMIT)
 
     async def _underlying_view_covers_history(self) -> bool:
-        """Whether the default session view already holds every stored item.
+        """Whether the last prepared input window held every stored item.
 
-        False when a session limit hides older items from the default view.
+        False when a session or per-run limit hid older items from the window the model
+        saw, so its response_id does not represent the full history.
         """
-        default_view = await self.underlying_session.get_items()
+        prepared_view = await self.underlying_session.get_items(
+            limit=self._last_prepared_input_limit
+        )
         full_view = await self._get_all_underlying_session_items()
-        return len(default_view) >= len(full_view)
+        return len(prepared_view) >= len(full_view)
 
     async def _replace_underlying_session_items(
         self,
diff --git a/tests/memory/test_openai_responses_compaction_session.py b/tests/memory/test_openai_responses_compaction_session.py
index eae1b7a85c..29e2f3200c 100644
--- a/tests/memory/test_openai_responses_compaction_session.py
+++ b/tests/memory/test_openai_responses_compaction_session.py
@@ -332,6 +332,46 @@ async def test_run_compaction_auto_does_not_reuse_unsafe_response_id_after_input
         await session.run_compaction({"force": True})
         assert "previous_response_id" not in mock_client.responses.compact.call_args.kwargs
 
+    @pytest.mark.asyncio
+    async def test_run_compaction_auto_falls_back_when_run_limit_truncates_input(
+        self, tmp_path: Path
+    ) -> None:
+        # The limit can come from the run (e.g. RunConfig session settings) rather than
+        # the underlying session's own settings: the Runner reads history through
+        # get_items(limit=N) to build the model input. Even with an unlimited store, if
+        # that window did not cover the full history the response_id is not
+        # representative, so auto must use input rather than previous_response_id.
+        underlying = SQLiteSession("test", db_path=str(tmp_path / "history.db"))
+        items: list[TResponseInputItem] = [
+            cast(
+                TResponseInputItem,
+                {"type": "message", "role": "assistant", "content": f"m{i}"},
+            )
+            for i in range(12)
+        ]
+        await underlying.add_items(items)
+
+        mock_compact_response = MagicMock()
+        mock_compact_response.output = [
+            {"type": "message", "role": "assistant", "content": "summary"}
+        ]
+        mock_client = MagicMock()
+        mock_client.responses.compact = AsyncMock(return_value=mock_compact_response)
+
+        session = OpenAIResponsesCompactionSession(
+            session_id="test",
+            underlying_session=underlying,
+            client=mock_client,
+        )
+
+        # Simulate the Runner building model input with a per-run limit of 3.
+        await session.get_items(limit=3)
+
+        await session.run_compaction({"response_id": "resp_A", "force": True})
+
+        mock_client.responses.compact.assert_called_once()
+        assert "previous_response_id" not in mock_client.responses.compact.call_args.kwargs
+
     @pytest.mark.asyncio
     async def test_run_compaction_auto_without_response_id_uses_input(self) -> None:
         mock_session = self.create_mock_session()

From d79002827f0f48f629a1475082a109ba879ae85f Mon Sep 17 00:00:00 2001
From: winklemad <winklemad@outlook.com>
Date: Wed, 15 Jul 2026 06:04:07 +0530
Subject: [PATCH 5/9] Carry per-response history window into compaction instead
 of shared session state

---
 .../openai_responses_compaction_session.py    |  39 +++---
 src/agents/memory/session.py                  |   8 ++
 src/agents/run.py                             |   5 +
 .../run_internal/session_persistence.py       |  37 ++++-
 ...est_openai_responses_compaction_session.py | 126 +++++++++++++++++-
 5 files changed, 187 insertions(+), 28 deletions(-)

diff --git a/src/agents/memory/openai_responses_compaction_session.py b/src/agents/memory/openai_responses_compaction_session.py
index 20933aeab1..af3bd6b79f 100644
--- a/src/agents/memory/openai_responses_compaction_session.py
+++ b/src/agents/memory/openai_responses_compaction_session.py
@@ -133,7 +133,6 @@ def __init__(
         self._response_id: str | None = None
         self._deferred_response_id: str | None = None
         self._last_unstored_response_id: str | None = None
-        self._last_prepared_input_limit: int | None = None
 
     @property
     def client(self) -> AsyncOpenAI:
@@ -163,6 +162,7 @@ async def run_compaction(self, args: OpenAIResponsesCompactionArgs | None = None
         if args and args.get("response_id"):
             self._response_id = args["response_id"]
         requested_mode = args.get("compaction_mode") if args else None
+        input_history_limit = args.get("input_history_limit") if args else None
         if args and "store" in args:
             store = args["store"]
             if store is False and self._response_id:
@@ -185,7 +185,7 @@ async def run_compaction(self, args: OpenAIResponsesCompactionArgs | None = None
         if (
             (requested_mode or self.compaction_mode) == "auto"
             and resolved_mode == "previous_response_id"
-            and not await self._underlying_view_covers_history()
+            and not await self._underlying_view_covers_history(input_history_limit)
         ):
             resolved_mode = "input"
             # This response_id never covered the full history, so don't let a later
@@ -255,26 +255,20 @@ async def run_compaction(self, args: OpenAIResponsesCompactionArgs | None = None
         )
 
     async def get_items(self, limit: int | None = None) -> list[TResponseInputItem]:
-        # The Runner reads history through get_items(limit=...) to build each turn's
-        # model input, so this is the window the resulting response actually saw. Record
-        # it so compaction can tell whether the response covered the full history or only
-        # a limited slice (the limit can come from a per-run RunConfig, not just the
-        # underlying session's own settings).
-        self._last_prepared_input_limit = limit
         return await self.underlying_session.get_items(limit)
 
     async def _get_all_underlying_session_items(self) -> list[TResponseInputItem]:
         return await self.underlying_session.get_items(limit=_ALL_SESSION_ITEMS_LIMIT)
 
-    async def _underlying_view_covers_history(self) -> bool:
-        """Whether the last prepared input window held every stored item.
+    async def _underlying_view_covers_history(self, input_history_limit: int | None) -> bool:
+        """Whether a response's prepared input window held every stored item.
 
-        False when a session or per-run limit hid older items from the window the model
-        saw, so its response_id does not represent the full history.
+        ``input_history_limit`` is the window used to prepare that specific response's input,
+        carried alongside the response instead of inferred from shared session state. Returns
+        False when a session or per-run limit hid older items from the window the model saw, so
+        its response_id does not represent the full history.
         """
-        prepared_view = await self.underlying_session.get_items(
-            limit=self._last_prepared_input_limit
-        )
+        prepared_view = await self.underlying_session.get_items(limit=input_history_limit)
         full_view = await self._get_all_underlying_session_items()
         return len(prepared_view) >= len(full_view)
 
@@ -344,7 +338,12 @@ async def _restore_underlying_session_items(
             replacement_error,
         )
 
-    async def _defer_compaction(self, response_id: str, store: bool | None = None) -> None:
+    async def _defer_compaction(
+        self,
+        response_id: str,
+        store: bool | None = None,
+        input_history_limit: int | None = None,
+    ) -> None:
         if self._deferred_response_id is not None:
             return
         compaction_candidate_items, session_items = await self._ensure_compaction_candidates()
@@ -353,6 +352,14 @@ async def _defer_compaction(self, response_id: str, store: bool | None = None) -
             store=store,
             requested_mode=None,
         )
+        # Mirror run_compaction: a limited view means previous_response_id would drop older
+        # items on replace, so the deferred turn also compacts from full-history input.
+        if (
+            self.compaction_mode == "auto"
+            and resolved_mode == "previous_response_id"
+            and not await self._underlying_view_covers_history(input_history_limit)
+        ):
+            resolved_mode = "input"
         should_compact = self.should_trigger_compaction(
             {
                 "response_id": response_id,
diff --git a/src/agents/memory/session.py b/src/agents/memory/session.py
index 1781b7ac9f..a132405778 100644
--- a/src/agents/memory/session.py
+++ b/src/agents/memory/session.py
@@ -127,6 +127,14 @@ class OpenAIResponsesCompactionArgs(TypedDict, total=False):
     force: bool
     """Whether to force compaction even if the threshold is not met."""
 
+    input_history_limit: int | None
+    """The history-window limit used to prepare this response's model input.
+
+    Carries the effective per-run limit (session or RunConfig) alongside the specific
+    response so compaction can tell whether that response saw the full stored history,
+    instead of inferring it from shared session state. ``None`` means no limit was applied.
+    """
+
 
 @runtime_checkable
 class OpenAIResponsesCompactionAwareSession(Session, Protocol):
diff --git a/src/agents/run.py b/src/agents/run.py
index f28462d461..f72e9a3aa8 100644
--- a/src/agents/run.py
+++ b/src/agents/run.py
@@ -894,6 +894,7 @@ def _finalize_result(result: RunResult) -> RunResult:
                                             run_state._reasoning_item_id_policy
                                         ),
                                         store=store_setting,
+                                        session_settings=run_config.session_settings,
                                     )
                                 )
 
@@ -1006,6 +1007,7 @@ def _finalize_result(result: RunResult) -> RunResult:
                                         run_state,
                                         response_id=turn_result.model_response.response_id,
                                         store=store_setting,
+                                        session_settings=run_config.session_settings,
                                     )
                                 result._original_input = copy_input_items(original_input)
                                 return _finalize_result(result)
@@ -1349,6 +1351,7 @@ def _finalize_result(result: RunResult) -> RunResult:
                                             run_state._reasoning_item_id_policy
                                         ),
                                         store=store_setting,
+                                        session_settings=run_config.session_settings,
                                     )
                                     run_state._current_turn_persisted_item_count += saved_count
                                 else:
@@ -1359,6 +1362,7 @@ def _finalize_result(result: RunResult) -> RunResult:
                                         run_state,
                                         response_id=turn_result.model_response.response_id,
                                         store=store_setting,
+                                        session_settings=run_config.session_settings,
                                     )
 
                     # After the first resumed turn, treat subsequent turns as fresh
@@ -1430,6 +1434,7 @@ def _finalize_result(result: RunResult) -> RunResult:
                                         run_state,
                                         response_id=turn_result.model_response.response_id,
                                         store=store_setting,
+                                        session_settings=run_config.session_settings,
                                     )
                             append_model_response_if_new(
                                 model_responses, turn_result.model_response
diff --git a/src/agents/run_internal/session_persistence.py b/src/agents/run_internal/session_persistence.py
index f483da13a3..2f5c528bed 100644
--- a/src/agents/run_internal/session_persistence.py
+++ b/src/agents/run_internal/session_persistence.py
@@ -51,6 +51,21 @@
 ]
 
 
+def resolve_session_history_limit(
+    session: Session, session_settings: SessionSettings | None
+) -> int | None:
+    """Return the effective history-window limit for a turn.
+
+    Combines the session's own settings with any per-run ``RunConfig`` override, matching
+    how ``prepare_input_with_session`` builds the model input. ``None`` means no limit is
+    applied and the full stored history is used.
+    """
+    resolved_settings = getattr(session, "session_settings", None) or SessionSettings()
+    if session_settings is not None:
+        resolved_settings = resolved_settings.resolve(session_settings)
+    return resolved_settings.limit
+
+
 async def prepare_input_with_session(
     input: str | list[TResponseInputItem],
     session: Session | None,
@@ -78,12 +93,9 @@ async def prepare_input_with_session(
     if session is None:
         return input, []
 
-    resolved_settings = getattr(session, "session_settings", None) or SessionSettings()
-    if session_settings is not None:
-        resolved_settings = resolved_settings.resolve(session_settings)
-
-    if resolved_settings.limit is not None:
-        history = await session.get_items(limit=resolved_settings.limit)
+    resolved_limit = resolve_session_history_limit(session, session_settings)
+    if resolved_limit is not None:
+        history = await session.get_items(limit=resolved_limit)
     else:
         history = await session.get_items()
     is_openai_conversation_session = isinstance(session, OpenAIConversationsSession)
@@ -253,6 +265,7 @@ async def save_result_to_session(
     response_id: str | None = None,
     reasoning_item_id_policy: ReasoningItemIdPolicy | None = None,
     store: bool | None = None,
+    session_settings: SessionSettings | None = None,
 ) -> int:
     """
     Persist a turn to the session store, keeping track of what was already saved so retries
@@ -352,13 +365,20 @@ async def save_result_to_session(
         run_state._current_turn_persisted_item_count = already_persisted + saved_run_items_count
 
     if response_id and is_openai_responses_compaction_aware_session(session):
+        # Carry the window used to prepare this response's input alongside the response
+        # itself, so compaction can tell whether it saw the full stored history instead of
+        # inferring it from shared session state (which breaks under interleaved runs and
+        # fresh wrappers on resume that skip input preparation).
+        input_history_limit = resolve_session_history_limit(session, session_settings)
         has_local_tool_outputs = any(
             isinstance(item, ToolCallOutputItem | HandoffOutputItem) for item in new_items
         )
         if has_local_tool_outputs:
             defer_compaction = getattr(session, "_defer_compaction", None)
             if callable(defer_compaction):
-                result = defer_compaction(response_id, store=store)
+                result = defer_compaction(
+                    response_id, store=store, input_history_limit=input_history_limit
+                )
                 if inspect.isawaitable(result):
                     await result
             logger.debug(
@@ -381,6 +401,7 @@ async def save_result_to_session(
         compaction_args: OpenAIResponsesCompactionArgs = {
             "response_id": response_id,
             "force": force_compaction,
+            "input_history_limit": input_history_limit,
         }
         if store is not None:
             compaction_args["store"] = store
@@ -397,6 +418,7 @@ async def save_resumed_turn_items(
     response_id: str | None,
     reasoning_item_id_policy: ReasoningItemIdPolicy | None = None,
     store: bool | None = None,
+    session_settings: SessionSettings | None = None,
 ) -> int:
     """Persist resumed turn items and return the updated persisted count."""
     if session is None or not items:
@@ -409,6 +431,7 @@ async def save_resumed_turn_items(
         response_id=response_id,
         reasoning_item_id_policy=reasoning_item_id_policy,
         store=store,
+        session_settings=session_settings,
     )
     return persisted_count + saved_count
 
diff --git a/tests/memory/test_openai_responses_compaction_session.py b/tests/memory/test_openai_responses_compaction_session.py
index 29e2f3200c..58db06d6a3 100644
--- a/tests/memory/test_openai_responses_compaction_session.py
+++ b/tests/memory/test_openai_responses_compaction_session.py
@@ -28,11 +28,24 @@
     TOOL_CALL_SESSION_DESCRIPTION_KEY,
     TOOL_CALL_SESSION_TITLE_KEY,
 )
+from agents.run_internal.session_persistence import save_result_to_session
 from tests.fake_model import FakeModel
 from tests.test_responses import get_function_tool, get_function_tool_call, get_text_message
 from tests.utils.simple_session import SimpleListSession
 
 
+class _DummyMessageRunItem:
+    """Minimal RunItem stand-in that persists a single message input item."""
+
+    type = "message_output_item"
+
+    def __init__(self, payload: dict[str, Any]) -> None:
+        self._payload = payload
+
+    def to_input_item(self) -> dict[str, Any]:
+        return self._payload
+
+
 class TestIsOpenAIModelName:
     def test_gpt_models(self) -> None:
         assert is_openai_model_name("gpt-4o") is True
@@ -337,8 +350,8 @@ async def test_run_compaction_auto_falls_back_when_run_limit_truncates_input(
         self, tmp_path: Path
     ) -> None:
         # The limit can come from the run (e.g. RunConfig session settings) rather than
-        # the underlying session's own settings: the Runner reads history through
-        # get_items(limit=N) to build the model input. Even with an unlimited store, if
+        # the underlying session's own settings. The Runner carries that window alongside
+        # the specific response as input_history_limit. Even with an unlimited store, if
         # that window did not cover the full history the response_id is not
         # representative, so auto must use input rather than previous_response_id.
         underlying = SQLiteSession("test", db_path=str(tmp_path / "history.db"))
@@ -364,13 +377,116 @@ async def test_run_compaction_auto_falls_back_when_run_limit_truncates_input(
             client=mock_client,
         )
 
-        # Simulate the Runner building model input with a per-run limit of 3.
+        # The response only saw the latest 3 of the 12 stored items, carried alongside it.
+        await session.run_compaction(
+            {"response_id": "resp_A", "force": True, "input_history_limit": 3}
+        )
+
+        mock_client.responses.compact.assert_called_once()
+        assert "previous_response_id" not in mock_client.responses.compact.call_args.kwargs
+
+    @pytest.mark.asyncio
+    async def test_run_compaction_uses_per_response_limit_under_interleaved_get_items(
+        self, tmp_path: Path
+    ) -> None:
+        # One session instance can be shared by concurrent runs. The window a response saw
+        # travels with that response as input_history_limit, so an interleaved get_items()
+        # from another run cannot make this response look like it covered the full history.
+        # Regression: tracking the last get_items() limit on the session let a concurrent
+        # get_items(limit=None) flip this response to previous_response_id and drop the older
+        # items on replace.
+        underlying = SQLiteSession("test", db_path=str(tmp_path / "history.db"))
+        items: list[TResponseInputItem] = [
+            cast(
+                TResponseInputItem,
+                {"type": "message", "role": "assistant", "content": f"m{i}"},
+            )
+            for i in range(12)
+        ]
+        await underlying.add_items(items)
+
+        mock_compact_response = MagicMock()
+        mock_compact_response.output = [
+            {"type": "message", "role": "assistant", "content": "summary"}
+        ]
+        mock_client = MagicMock()
+        mock_client.responses.compact = AsyncMock(return_value=mock_compact_response)
+
+        session = OpenAIResponsesCompactionSession(
+            session_id="test",
+            underlying_session=underlying,
+            client=mock_client,
+        )
+
+        # Response A saw only the latest 3 items; a second run then reads the full history.
         await session.get_items(limit=3)
+        await session.get_items(limit=None)
 
-        await session.run_compaction({"response_id": "resp_A", "force": True})
+        # Compacting A must still use full-history input, not previous_response_id.
+        await session.run_compaction(
+            {"response_id": "resp_A", "force": True, "input_history_limit": 3}
+        )
 
         mock_client.responses.compact.assert_called_once()
-        assert "previous_response_id" not in mock_client.responses.compact.call_args.kwargs
+        compact_kwargs = mock_client.responses.compact.call_args.kwargs
+        assert "previous_response_id" not in compact_kwargs
+        assert len(compact_kwargs["input"]) == len(items)
+
+    @pytest.mark.asyncio
+    async def test_save_result_to_session_carries_run_limit_to_fresh_wrapper_on_resume(
+        self, tmp_path: Path
+    ) -> None:
+        # On RunState resume the compaction wrapper is recreated fresh and resume skips input
+        # preparation, so it never observes the run's get_items(limit=N). The Runner instead
+        # carries the window with the response through save_result_to_session, so a fresh
+        # wrapper still compacts from full-history input rather than a previous_response_id
+        # that only covered the truncated per-run window.
+        underlying = SQLiteSession("resume", db_path=str(tmp_path / "history.db"))
+        items: list[TResponseInputItem] = [
+            cast(
+                TResponseInputItem,
+                {"type": "message", "role": "assistant", "content": f"m{i}"},
+            )
+            for i in range(12)
+        ]
+        await underlying.add_items(items)
+
+        mock_compact_response = MagicMock()
+        mock_compact_response.output = [
+            {"type": "message", "role": "assistant", "content": "summary"}
+        ]
+        mock_client = MagicMock()
+        mock_client.responses.compact = AsyncMock(return_value=mock_compact_response)
+
+        # Fresh wrapper, as recreated on resume: get_items is never called to prepare input.
+        session = OpenAIResponsesCompactionSession(
+            session_id="resume",
+            underlying_session=underlying,
+            client=mock_client,
+        )
+
+        new_items = [
+            cast(
+                Any,
+                _DummyMessageRunItem({"type": "message", "role": "assistant", "content": "latest"}),
+            )
+        ]
+        # A RunConfig applied a per-run limit of 3 that the store itself does not carry.
+        await save_result_to_session(
+            session,
+            [],
+            new_items,
+            None,
+            response_id="resp_A",
+            store=True,
+            session_settings=SessionSettings(limit=3),
+        )
+
+        mock_client.responses.compact.assert_called_once()
+        compact_kwargs = mock_client.responses.compact.call_args.kwargs
+        assert "previous_response_id" not in compact_kwargs
+        # The full stored history is compacted, not just the truncated per-run window.
+        assert len(compact_kwargs["input"]) >= len(items)
 
     @pytest.mark.asyncio
     async def test_run_compaction_auto_without_response_id_uses_input(self) -> None:

From e2ace6c9ec7fec34a78fedffc9684d2e26a132dd Mon Sep 17 00:00:00 2001
From: winklemad <winklemad@outlook.com>
Date: Wed, 15 Jul 2026 06:54:42 +0530
Subject: [PATCH 6/9] Carry the per-run history window through the streaming
 and final-output save paths

---
 src/agents/run.py                             |   2 +
 .../run_internal/agent_runner_helpers.py      |   4 +-
 src/agents/run_internal/run_loop.py           |   9 +-
 ...est_openai_responses_compaction_session.py | 179 +++++++++++++++++-
 tests/test_agent_runner_streamed.py           |   2 +
 5 files changed, 193 insertions(+), 3 deletions(-)

diff --git a/src/agents/run.py b/src/agents/run.py
index f72e9a3aa8..d57a2bf201 100644
--- a/src/agents/run.py
+++ b/src/agents/run.py
@@ -1415,6 +1415,7 @@ def _finalize_result(result: RunResult) -> RunResult:
                                 items=session_items_for_turn(turn_result),
                                 response_id=turn_result.model_response.response_id,
                                 store=store_setting,
+                                session_settings=run_config.session_settings,
                             )
                             result._original_input = copy_input_items(original_input)
                             return _finalize_result(result)
@@ -1496,6 +1497,7 @@ def _finalize_result(result: RunResult) -> RunResult:
                                 items=session_items_for_turn(turn_result),
                                 response_id=turn_result.model_response.response_id,
                                 store=store_setting,
+                                session_settings=run_config.session_settings,
                             )
                             continue
                         else:
diff --git a/src/agents/run_internal/agent_runner_helpers.py b/src/agents/run_internal/agent_runner_helpers.py
index 2e212e597f..c186da8ca2 100644
--- a/src/agents/run_internal/agent_runner_helpers.py
+++ b/src/agents/run_internal/agent_runner_helpers.py
@@ -12,7 +12,7 @@
 from ..exceptions import UserError
 from ..guardrail import InputGuardrailResult
 from ..items import ModelResponse, RunItem, ToolApprovalItem, TResponseInputItem
-from ..memory import Session
+from ..memory import Session, SessionSettings
 from ..models.openai_agent_registration import add_openai_harness_id_to_metadata
 from ..result import RunResult
 from ..run_config import RunConfig
@@ -473,6 +473,7 @@ async def save_turn_items_if_needed(
     items: list[RunItem],
     response_id: str | None,
     store: bool | None = None,
+    session_settings: SessionSettings | None = None,
 ) -> None:
     """Persist turn items when persistence is enabled and guardrails allow it."""
     if not session_persistence_enabled:
@@ -488,6 +489,7 @@ async def save_turn_items_if_needed(
         run_state,
         response_id=response_id,
         store=store,
+        session_settings=session_settings,
     )
 
 
diff --git a/src/agents/run_internal/run_loop.py b/src/agents/run_internal/run_loop.py
index b85347f6bf..7a978edac2 100644
--- a/src/agents/run_internal/run_loop.py
+++ b/src/agents/run_internal/run_loop.py
@@ -57,7 +57,7 @@
 )
 from ..lifecycle import RunHooks
 from ..logger import logger
-from ..memory import Session
+from ..memory import Session, SessionSettings
 from ..models._response_terminal import (
     response_error_event_failure_error,
     response_terminal_failure_error,
@@ -319,6 +319,7 @@ async def _save_resumed_stream_items(
     items: list[RunItem],
     response_id: str | None,
     store: bool | None = None,
+    session_settings: SessionSettings | None = None,
 ) -> None:
     if not await _should_persist_stream_items(
         session=session,
@@ -333,6 +334,7 @@ async def _save_resumed_stream_items(
         response_id=response_id,
         reasoning_item_id_policy=streamed_result._reasoning_item_id_policy,
         store=store,
+        session_settings=session_settings,
     )
     if run_state is not None:
         run_state._current_turn_persisted_item_count = (
@@ -350,6 +352,7 @@ async def _save_stream_items(
     response_id: str | None,
     update_persisted_count: bool,
     store: bool | None = None,
+    session_settings: SessionSettings | None = None,
 ) -> None:
     if not await _should_persist_stream_items(
         session=session,
@@ -364,6 +367,7 @@ async def _save_stream_items(
         run_state,
         response_id=response_id,
         store=store,
+        session_settings=session_settings,
     )
     if update_persisted_count and streamed_result._state is not None:
         streamed_result._current_turn_persisted_item_count = (
@@ -635,6 +639,7 @@ async def _save_resumed_items(
                 items=items,
                 response_id=response_id,
                 store=store_setting,
+                session_settings=run_config.session_settings,
             )
 
         async def _save_stream_items_with_count(
@@ -649,6 +654,7 @@ async def _save_stream_items_with_count(
                 response_id=response_id,
                 update_persisted_count=True,
                 store=store_setting,
+                session_settings=run_config.session_settings,
             )
 
         async def _save_stream_items_without_count(
@@ -663,6 +669,7 @@ async def _save_stream_items_without_count(
                 response_id=response_id,
                 update_persisted_count=False,
                 store=store_setting,
+                session_settings=run_config.session_settings,
             )
     except BaseException:
         if current_task_span:
diff --git a/tests/memory/test_openai_responses_compaction_session.py b/tests/memory/test_openai_responses_compaction_session.py
index 58db06d6a3..88b46c5929 100644
--- a/tests/memory/test_openai_responses_compaction_session.py
+++ b/tests/memory/test_openai_responses_compaction_session.py
@@ -10,7 +10,7 @@
 import pytest
 
 from agents import Agent, Runner
-from agents.items import TResponseInputItem
+from agents.items import ToolCallOutputItem, TResponseInputItem
 from agents.memory import (
     OpenAIResponsesCompactionSession,
     Session,
@@ -24,6 +24,7 @@
     select_compaction_candidate_items,
 )
 from agents.memory.sqlite_session import SQLiteSession
+from agents.run_internal.agent_runner_helpers import save_turn_items_if_needed
 from agents.run_internal.items import (
     TOOL_CALL_SESSION_DESCRIPTION_KEY,
     TOOL_CALL_SESSION_TITLE_KEY,
@@ -488,6 +489,182 @@ async def test_save_result_to_session_carries_run_limit_to_fresh_wrapper_on_resu
         # The full stored history is compacted, not just the truncated per-run window.
         assert len(compact_kwargs["input"]) >= len(items)
 
+    @pytest.mark.asyncio
+    async def test_save_turn_items_if_needed_carries_run_limit(self, tmp_path: Path) -> None:
+        # The final-output and run-again paths persist through save_turn_items_if_needed,
+        # which must also carry the per-run window so a limited run does not compact via
+        # previous_response_id and drop the older history.
+        underlying = SQLiteSession("turn", db_path=str(tmp_path / "history.db"))
+        items: list[TResponseInputItem] = [
+            cast(
+                TResponseInputItem,
+                {"type": "message", "role": "assistant", "content": f"m{i}"},
+            )
+            for i in range(12)
+        ]
+        await underlying.add_items(items)
+
+        mock_compact_response = MagicMock()
+        mock_compact_response.output = [
+            {"type": "message", "role": "assistant", "content": "summary"}
+        ]
+        mock_client = MagicMock()
+        mock_client.responses.compact = AsyncMock(return_value=mock_compact_response)
+
+        session = OpenAIResponsesCompactionSession(
+            session_id="turn",
+            underlying_session=underlying,
+            client=mock_client,
+        )
+
+        new_items = [
+            cast(
+                Any,
+                _DummyMessageRunItem({"type": "message", "role": "assistant", "content": "latest"}),
+            )
+        ]
+        await save_turn_items_if_needed(
+            session=session,
+            run_state=None,
+            session_persistence_enabled=True,
+            input_guardrail_results=[],
+            items=new_items,
+            response_id="resp_A",
+            store=True,
+            session_settings=SessionSettings(limit=3),
+        )
+
+        mock_client.responses.compact.assert_called_once()
+        compact_kwargs = mock_client.responses.compact.call_args.kwargs
+        assert "previous_response_id" not in compact_kwargs
+        assert len(compact_kwargs["input"]) >= len(items)
+
+    @pytest.mark.asyncio
+    async def test_deferred_compaction_under_limit_uses_full_history(self, tmp_path: Path) -> None:
+        # A turn whose output includes a local tool result DEFERS compaction. When the run's
+        # window is limited, the deferred compaction — fired on the next turn — must still
+        # compact the full stored history via input, not a previous_response_id that only saw
+        # the limited window.
+        underlying = SQLiteSession("deferred", db_path=str(tmp_path / "history.db"))
+        items: list[TResponseInputItem] = [
+            cast(
+                TResponseInputItem,
+                {"type": "message", "role": "assistant", "content": f"m{i}"},
+            )
+            for i in range(12)
+        ]
+        await underlying.add_items(items)
+
+        mock_compact_response = MagicMock()
+        mock_compact_response.output = [
+            {"type": "message", "role": "assistant", "content": "summary"}
+        ]
+        mock_client = MagicMock()
+        mock_client.responses.compact = AsyncMock(return_value=mock_compact_response)
+
+        session = OpenAIResponsesCompactionSession(
+            session_id="deferred",
+            underlying_session=underlying,
+            client=mock_client,
+        )
+
+        agent = Agent(name="assistant", model=FakeModel())
+        tool_output = cast(
+            Any,
+            ToolCallOutputItem(
+                agent=agent,
+                raw_item={"type": "function_call_output", "call_id": "c1", "output": "done"},
+                output="done",
+            ),
+        )
+        # Turn 1: the local tool output defers compaction rather than running it now.
+        await save_result_to_session(
+            session,
+            [],
+            [tool_output],
+            None,
+            response_id="resp_1",
+            store=True,
+            session_settings=SessionSettings(limit=3),
+        )
+        assert not mock_client.responses.compact.called
+
+        # Turn 2: the deferred compaction fires and must not drop the older history.
+        new_msg = [
+            cast(
+                Any,
+                _DummyMessageRunItem({"type": "message", "role": "assistant", "content": "latest"}),
+            )
+        ]
+        await save_result_to_session(
+            session,
+            [],
+            new_msg,
+            None,
+            response_id="resp_2",
+            store=True,
+            session_settings=SessionSettings(limit=3),
+        )
+
+        mock_client.responses.compact.assert_called_once()
+        compact_kwargs = mock_client.responses.compact.call_args.kwargs
+        assert "previous_response_id" not in compact_kwargs
+        assert len(compact_kwargs["input"]) >= len(items)
+
+    @pytest.mark.asyncio
+    async def test_save_stream_items_carries_run_limit(self, tmp_path: Path) -> None:
+        # The streaming persistence path carries the per-run window too, so a limited
+        # streamed run compacts the full stored history rather than a previous_response_id
+        # that only saw the window.
+        from agents.run_internal.run_loop import _save_stream_items
+
+        underlying = SQLiteSession("stream", db_path=str(tmp_path / "history.db"))
+        items: list[TResponseInputItem] = [
+            cast(
+                TResponseInputItem,
+                {"type": "message", "role": "assistant", "content": f"m{i}"},
+            )
+            for i in range(12)
+        ]
+        await underlying.add_items(items)
+
+        mock_compact_response = MagicMock()
+        mock_compact_response.output = [
+            {"type": "message", "role": "assistant", "content": "summary"}
+        ]
+        mock_client = MagicMock()
+        mock_client.responses.compact = AsyncMock(return_value=mock_compact_response)
+
+        session = OpenAIResponsesCompactionSession(
+            session_id="stream",
+            underlying_session=underlying,
+            client=mock_client,
+        )
+
+        streamed_result = SimpleNamespace(_input_guardrails_task=None, _state=None)
+        new_items = [
+            cast(
+                Any,
+                _DummyMessageRunItem({"type": "message", "role": "assistant", "content": "latest"}),
+            )
+        ]
+        await _save_stream_items(
+            session=session,
+            server_conversation_tracker=None,
+            streamed_result=cast(Any, streamed_result),
+            run_state=None,
+            items=new_items,
+            response_id="resp_A",
+            update_persisted_count=False,
+            store=True,
+            session_settings=SessionSettings(limit=3),
+        )
+
+        mock_client.responses.compact.assert_called_once()
+        compact_kwargs = mock_client.responses.compact.call_args.kwargs
+        assert "previous_response_id" not in compact_kwargs
+        assert len(compact_kwargs["input"]) >= len(items)
+
     @pytest.mark.asyncio
     async def test_run_compaction_auto_without_response_id_uses_input(self) -> None:
         mock_session = self.create_mock_session()
diff --git a/tests/test_agent_runner_streamed.py b/tests/test_agent_runner_streamed.py
index 8ee3a55db4..14715ff06b 100644
--- a/tests/test_agent_runner_streamed.py
+++ b/tests/test_agent_runner_streamed.py
@@ -1916,6 +1916,7 @@ async def save_wrapper(
         response_id: str | None,
         reasoning_item_id_policy: str | None = None,
         store: bool | None = None,
+        session_settings: Any = None,
     ) -> int:
         observed_counts.append(persisted_count)
         result = await real_save_resumed(
@@ -1925,6 +1926,7 @@ async def save_wrapper(
             response_id=response_id,
             reasoning_item_id_policy=reasoning_item_id_policy,
             store=store,
+            session_settings=session_settings,
         )
         return int(result)
 

From 2788e93e1c26e6a2e037b151e947671480e3bdb1 Mon Sep 17 00:00:00 2001
From: winklemad <winklemad@outlook.com>
Date: Wed, 15 Jul 2026 07:32:05 +0530
Subject: [PATCH 7/9] Measure compaction history coverage against the pre-save
 store

---
 .../openai_responses_compaction_session.py    | 21 ++++++-
 src/agents/memory/session.py                  |  8 +++
 .../run_internal/session_persistence.py       | 29 +++++++---
 ...est_openai_responses_compaction_session.py | 55 +++++++++++++++++++
 4 files changed, 103 insertions(+), 10 deletions(-)

diff --git a/src/agents/memory/openai_responses_compaction_session.py b/src/agents/memory/openai_responses_compaction_session.py
index af3bd6b79f..a5867d9435 100644
--- a/src/agents/memory/openai_responses_compaction_session.py
+++ b/src/agents/memory/openai_responses_compaction_session.py
@@ -163,6 +163,7 @@ async def run_compaction(self, args: OpenAIResponsesCompactionArgs | None = None
             self._response_id = args["response_id"]
         requested_mode = args.get("compaction_mode") if args else None
         input_history_limit = args.get("input_history_limit") if args else None
+        input_covered_full_history = args.get("input_covered_full_history") if args else None
         if args and "store" in args:
             store = args["store"]
             if store is False and self._response_id:
@@ -185,7 +186,9 @@ async def run_compaction(self, args: OpenAIResponsesCompactionArgs | None = None
         if (
             (requested_mode or self.compaction_mode) == "auto"
             and resolved_mode == "previous_response_id"
-            and not await self._underlying_view_covers_history(input_history_limit)
+            and not await self._input_covers_history(
+                input_covered_full_history, input_history_limit
+            )
         ):
             resolved_mode = "input"
             # This response_id never covered the full history, so don't let a later
@@ -272,6 +275,18 @@ async def _underlying_view_covers_history(self, input_history_limit: int | None)
         full_view = await self._get_all_underlying_session_items()
         return len(prepared_view) >= len(full_view)
 
+    async def _input_covers_history(
+        self, input_covered_full_history: bool | None, input_history_limit: int | None
+    ) -> bool:
+        """Whether the response's input saw the full stored history.
+
+        Prefers the pre-save answer the Runner computed for this specific response; falls back
+        to a live comparison against the limit for callers that don't carry it.
+        """
+        if input_covered_full_history is not None:
+            return input_covered_full_history
+        return await self._underlying_view_covers_history(input_history_limit)
+
     async def _replace_underlying_session_items(
         self,
         *,
@@ -342,7 +357,7 @@ async def _defer_compaction(
         self,
         response_id: str,
         store: bool | None = None,
-        input_history_limit: int | None = None,
+        input_covered_full_history: bool | None = None,
     ) -> None:
         if self._deferred_response_id is not None:
             return
@@ -357,7 +372,7 @@ async def _defer_compaction(
         if (
             self.compaction_mode == "auto"
             and resolved_mode == "previous_response_id"
-            and not await self._underlying_view_covers_history(input_history_limit)
+            and not await self._input_covers_history(input_covered_full_history, None)
         ):
             resolved_mode = "input"
         should_compact = self.should_trigger_compaction(
diff --git a/src/agents/memory/session.py b/src/agents/memory/session.py
index a132405778..3a2487253c 100644
--- a/src/agents/memory/session.py
+++ b/src/agents/memory/session.py
@@ -135,6 +135,14 @@ class OpenAIResponsesCompactionArgs(TypedDict, total=False):
     instead of inferring it from shared session state. ``None`` means no limit was applied.
     """
 
+    input_covered_full_history: bool | None
+    """Whether this response's input window held the full stored history.
+
+    Resolved by the Runner against the pre-save store (the state the response was actually
+    prepared from), so a turn that grows the session past its limit doesn't look truncated.
+    When present it is authoritative; ``None`` falls back to ``input_history_limit``.
+    """
+
 
 @runtime_checkable
 class OpenAIResponsesCompactionAwareSession(Session, Protocol):
diff --git a/src/agents/run_internal/session_persistence.py b/src/agents/run_internal/session_persistence.py
index 2f5c528bed..fd422e56ab 100644
--- a/src/agents/run_internal/session_persistence.py
+++ b/src/agents/run_internal/session_persistence.py
@@ -359,17 +359,30 @@ async def save_result_to_session(
             run_state._current_turn_persisted_item_count = already_persisted + saved_run_items_count
         return saved_run_items_count
 
+    # Whether this response's input window held the full stored history, measured against the
+    # PRE-save store -- the state the response was actually prepared from. Computed before
+    # add_items so this turn's freshly-saved items don't make an untruncated response look
+    # truncated (which would needlessly force full-history input compaction instead of the
+    # cheaper server-side previous_response_id path). Carried with the response so it is never
+    # inferred from shared session state, which breaks under interleaved runs and fresh
+    # wrappers on resume that skip input preparation.
+    input_covered_full_history: bool | None = None
+    if response_id and is_openai_responses_compaction_aware_session(session):
+        history_limit = resolve_session_history_limit(session, session_settings)
+        if history_limit is None:
+            input_covered_full_history = True
+        else:
+            # Fewer stored items than the window means the window held them all; a full
+            # window may still hide older items, so treat that as truncated.
+            prepared_view = await session.get_items(limit=history_limit)
+            input_covered_full_history = len(prepared_view) < history_limit
+
     await session.add_items(items_to_save)
 
     if run_state:
         run_state._current_turn_persisted_item_count = already_persisted + saved_run_items_count
 
     if response_id and is_openai_responses_compaction_aware_session(session):
-        # Carry the window used to prepare this response's input alongside the response
-        # itself, so compaction can tell whether it saw the full stored history instead of
-        # inferring it from shared session state (which breaks under interleaved runs and
-        # fresh wrappers on resume that skip input preparation).
-        input_history_limit = resolve_session_history_limit(session, session_settings)
         has_local_tool_outputs = any(
             isinstance(item, ToolCallOutputItem | HandoffOutputItem) for item in new_items
         )
@@ -377,7 +390,9 @@ async def save_result_to_session(
             defer_compaction = getattr(session, "_defer_compaction", None)
             if callable(defer_compaction):
                 result = defer_compaction(
-                    response_id, store=store, input_history_limit=input_history_limit
+                    response_id,
+                    store=store,
+                    input_covered_full_history=input_covered_full_history,
                 )
                 if inspect.isawaitable(result):
                     await result
@@ -401,7 +416,7 @@ async def save_result_to_session(
         compaction_args: OpenAIResponsesCompactionArgs = {
             "response_id": response_id,
             "force": force_compaction,
-            "input_history_limit": input_history_limit,
+            "input_covered_full_history": input_covered_full_history,
         }
         if store is not None:
             compaction_args["store"] = store
diff --git a/tests/memory/test_openai_responses_compaction_session.py b/tests/memory/test_openai_responses_compaction_session.py
index 88b46c5929..6cb7dfbbca 100644
--- a/tests/memory/test_openai_responses_compaction_session.py
+++ b/tests/memory/test_openai_responses_compaction_session.py
@@ -665,6 +665,61 @@ async def test_save_stream_items_carries_run_limit(self, tmp_path: Path) -> None
         assert "previous_response_id" not in compact_kwargs
         assert len(compact_kwargs["input"]) >= len(items)
 
+    @pytest.mark.asyncio
+    async def test_compaction_reuses_previous_response_id_when_pre_save_history_fits(
+        self, tmp_path: Path
+    ) -> None:
+        # If the history the response was prepared from already fit within the limit, the
+        # response saw everything, so compaction can reuse previous_response_id even when this
+        # turn's saved items push the stored session over the limit -- no need to force a
+        # full-history input request. Coverage is measured against the pre-save store.
+        underlying = SQLiteSession("fits", db_path=str(tmp_path / "history.db"))
+        items: list[TResponseInputItem] = [
+            cast(
+                TResponseInputItem,
+                {"type": "message", "role": "assistant", "content": f"m{i}"},
+            )
+            for i in range(4)
+        ]
+        await underlying.add_items(items)
+
+        mock_compact_response = MagicMock()
+        mock_compact_response.output = [
+            {"type": "message", "role": "assistant", "content": "summary"}
+        ]
+        mock_client = MagicMock()
+        mock_client.responses.compact = AsyncMock(return_value=mock_compact_response)
+
+        session = OpenAIResponsesCompactionSession(
+            session_id="fits",
+            underlying_session=underlying,
+            client=mock_client,
+            should_trigger_compaction=lambda _context: True,
+        )
+
+        new_items = [
+            cast(
+                Any,
+                _DummyMessageRunItem({"type": "message", "role": "assistant", "content": f"n{i}"}),
+            )
+            for i in range(4)
+        ]
+        # Pre-save history (4 items) fits within limit=5; this turn pushes the store to 8.
+        await save_result_to_session(
+            session,
+            [],
+            new_items,
+            None,
+            response_id="resp_A",
+            store=True,
+            session_settings=SessionSettings(limit=5),
+        )
+
+        mock_client.responses.compact.assert_called_once()
+        compact_kwargs = mock_client.responses.compact.call_args.kwargs
+        assert compact_kwargs.get("previous_response_id") == "resp_A"
+        assert "input" not in compact_kwargs
+
     @pytest.mark.asyncio
     async def test_run_compaction_auto_without_response_id_uses_input(self) -> None:
         mock_session = self.create_mock_session()

From 5570d8091aa900a06975cac359f487ad88ddce08 Mon Sep 17 00:00:00 2001
From: winklemad <winklemad@outlook.com>
Date: Wed, 15 Jul 2026 10:45:38 +0530
Subject: [PATCH 8/9] Honor an underlying session's own limit when resolving
 the history window

---
 .../run_internal/session_persistence.py       | 20 ++++++--
 ...est_openai_responses_compaction_session.py | 48 +++++++++++++++++++
 2 files changed, 64 insertions(+), 4 deletions(-)

diff --git a/src/agents/run_internal/session_persistence.py b/src/agents/run_internal/session_persistence.py
index fd422e56ab..4bc451b3a5 100644
--- a/src/agents/run_internal/session_persistence.py
+++ b/src/agents/run_internal/session_persistence.py
@@ -56,14 +56,26 @@ def resolve_session_history_limit(
 ) -> int | None:
     """Return the effective history-window limit for a turn.
 
-    Combines the session's own settings with any per-run ``RunConfig`` override, matching
-    how ``prepare_input_with_session`` builds the model input. ``None`` means no limit is
-    applied and the full stored history is used.
+    Combines the session's own settings with any per-run ``RunConfig`` override, matching how
+    ``prepare_input_with_session`` builds the model input. When no explicit limit applies, a
+    wrapping session (e.g. the compaction session) can still be truncated by an underlying
+    session's own limit -- ``get_items(None)`` delegates to it -- so the delegate chain is
+    consulted. ``None`` means the full stored history is used.
     """
     resolved_settings = getattr(session, "session_settings", None) or SessionSettings()
     if session_settings is not None:
         resolved_settings = resolved_settings.resolve(session_settings)
-    return resolved_settings.limit
+    if resolved_settings.limit is not None:
+        return resolved_settings.limit
+
+    delegate = getattr(session, "underlying_session", None)
+    while delegate is not None:
+        delegate_settings: SessionSettings | None = getattr(delegate, "session_settings", None)
+        delegate_limit = delegate_settings.limit if delegate_settings is not None else None
+        if delegate_limit is not None:
+            return delegate_limit
+        delegate = getattr(delegate, "underlying_session", None)
+    return None
 
 
 async def prepare_input_with_session(
diff --git a/tests/memory/test_openai_responses_compaction_session.py b/tests/memory/test_openai_responses_compaction_session.py
index 6cb7dfbbca..0cfb474680 100644
--- a/tests/memory/test_openai_responses_compaction_session.py
+++ b/tests/memory/test_openai_responses_compaction_session.py
@@ -720,6 +720,54 @@ async def test_compaction_reuses_previous_response_id_when_pre_save_history_fits
         assert compact_kwargs.get("previous_response_id") == "resp_A"
         assert "input" not in compact_kwargs
 
+    @pytest.mark.asyncio
+    async def test_compaction_honors_underlying_session_limit(self, tmp_path: Path) -> None:
+        # The limit can live on the underlying store rather than the wrapper or the RunConfig;
+        # get_items(None) delegates to the underlying, whose own limit truncates the input. A
+        # normal Runner save must still detect that truncation and compact from full history
+        # instead of a previous_response_id that never saw the hidden older items.
+        underlying = SQLiteSession(
+            "wrapped",
+            db_path=str(tmp_path / "history.db"),
+            session_settings=SessionSettings(limit=3),
+        )
+        items: list[TResponseInputItem] = [
+            cast(
+                TResponseInputItem,
+                {"type": "message", "role": "assistant", "content": f"m{i}"},
+            )
+            for i in range(12)
+        ]
+        await underlying.add_items(items)
+
+        mock_compact_response = MagicMock()
+        mock_compact_response.output = [
+            {"type": "message", "role": "assistant", "content": "summary"}
+        ]
+        mock_client = MagicMock()
+        mock_client.responses.compact = AsyncMock(return_value=mock_compact_response)
+
+        session = OpenAIResponsesCompactionSession(
+            session_id="wrapped",
+            underlying_session=underlying,
+            client=mock_client,
+            should_trigger_compaction=lambda _context: True,
+        )
+
+        new_items = [
+            cast(
+                Any,
+                _DummyMessageRunItem({"type": "message", "role": "assistant", "content": "latest"}),
+            )
+        ]
+        # No RunConfig limit and no wrapper limit -- only the underlying store is limited.
+        await save_result_to_session(session, [], new_items, None, response_id="resp_A", store=True)
+
+        mock_client.responses.compact.assert_called_once()
+        compact_kwargs = mock_client.responses.compact.call_args.kwargs
+        assert "previous_response_id" not in compact_kwargs
+        assert len(compact_kwargs["input"]) >= len(items)
+
     @pytest.mark.asyncio
     async def test_run_compaction_auto_without_response_id_uses_input(self) -> None:
         mock_session = self.create_mock_session()

From 8c417413942a929c8a42c9e9609d6bd621af744c Mon Sep 17 00:00:00 2001
From: winklemad <winklemad@outlook.com>
Date: Wed, 15 Jul 2026 11:14:43 +0530
Subject: [PATCH 9/9] Treat an exact-limit history window as covered, not
 truncated

---
 .../run_internal/session_persistence.py       |  9 ++--
 ...est_openai_responses_compaction_session.py | 54 +++++++++++++++++++
 2 files changed, 59 insertions(+), 4 deletions(-)

diff --git a/src/agents/run_internal/session_persistence.py b/src/agents/run_internal/session_persistence.py
index 4bc451b3a5..afe622b541 100644
--- a/src/agents/run_internal/session_persistence.py
+++ b/src/agents/run_internal/session_persistence.py
@@ -384,10 +384,11 @@ async def save_result_to_session(
         if history_limit is None:
             input_covered_full_history = True
         else:
-            # Fewer stored items than the window means the window held them all; a full
-            # window may still hide older items, so treat that as truncated.
-            prepared_view = await session.get_items(limit=history_limit)
-            input_covered_full_history = len(prepared_view) < history_limit
+            # The window held everything only if the store has no item beyond it. Read one
+            # extra item so an exact fill (covered) is distinguished from a real overflow
+            # (truncated), rather than treating a full window as always truncated.
+            prepared_view = await session.get_items(limit=history_limit + 1)
+            input_covered_full_history = len(prepared_view) <= history_limit
 
     await session.add_items(items_to_save)
 
diff --git a/tests/memory/test_openai_responses_compaction_session.py b/tests/memory/test_openai_responses_compaction_session.py
index 0cfb474680..b24038a3f6 100644
--- a/tests/memory/test_openai_responses_compaction_session.py
+++ b/tests/memory/test_openai_responses_compaction_session.py
@@ -720,6 +720,60 @@ async def test_compaction_reuses_previous_response_id_when_pre_save_history_fits
         assert compact_kwargs.get("previous_response_id") == "resp_A"
         assert "input" not in compact_kwargs
 
+    @pytest.mark.asyncio
+    async def test_compaction_reuses_previous_response_id_at_exact_limit(
+        self, tmp_path: Path
+    ) -> None:
+        # Pre-save history that exactly fills the limit was fully seen by the response, so an
+        # exact-fill window counts as covered (not truncated) and compaction can reuse
+        # previous_response_id even when this turn grows the store past the limit.
+        underlying = SQLiteSession("exact", db_path=str(tmp_path / "history.db"))
+        items: list[TResponseInputItem] = [
+            cast(
+                TResponseInputItem,
+                {"type": "message", "role": "assistant", "content": f"m{i}"},
+            )
+            for i in range(5)
+        ]
+        await underlying.add_items(items)
+
+        mock_compact_response = MagicMock()
+        mock_compact_response.output = [
+            {"type": "message", "role": "assistant", "content": "summary"}
+        ]
+        mock_client = MagicMock()
+        mock_client.responses.compact = AsyncMock(return_value=mock_compact_response)
+
+        session = OpenAIResponsesCompactionSession(
+            session_id="exact",
+            underlying_session=underlying,
+            client=mock_client,
+            should_trigger_compaction=lambda _context: True,
+        )
+
+        new_items = [
+            cast(
+                Any,
+                _DummyMessageRunItem({"type": "message", "role": "assistant", "content": f"n{i}"}),
+            )
+            for i in range(3)
+        ]
+        # Pre-save history (5 items) exactly fills limit=5; this turn pushes the store to 8.
+        await save_result_to_session(
+            session,
+            [],
+            new_items,
+            None,
+            response_id="resp_A",
+            store=True,
+            session_settings=SessionSettings(limit=5),
+        )
+
+        mock_client.responses.compact.assert_called_once()
+        compact_kwargs = mock_client.responses.compact.call_args.kwargs
+        assert compact_kwargs.get("previous_response_id") == "resp_A"
+        assert "input" not in compact_kwargs
+
     @pytest.mark.asyncio
     async def test_compaction_honors_underlying_session_limit(self, tmp_path: Path) -> None:
         # The limit can live on the underlying store rather than the wrapper or the RunConfig;