From d6c84248840252204cf6f55c1e5803832d1e2be5 Mon Sep 17 00:00:00 2001 From: winklemad Date: Tue, 14 Jul 2026 00:03:43 +0530 Subject: [PATCH 01/11] 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 Date: Tue, 14 Jul 2026 04:37:16 +0530 Subject: [PATCH 02/11] 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 Date: Tue, 14 Jul 2026 05:02:58 +0530 Subject: [PATCH 03/11] 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 Date: Tue, 14 Jul 2026 05:19:38 +0530 Subject: [PATCH 04/11] 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 Date: Wed, 15 Jul 2026 06:04:07 +0530 Subject: [PATCH 05/11] 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 Date: Wed, 15 Jul 2026 06:54:42 +0530 Subject: [PATCH 06/11] 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 Date: Wed, 15 Jul 2026 07:32:05 +0530 Subject: [PATCH 07/11] 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 Date: Wed, 15 Jul 2026 10:45:38 +0530 Subject: [PATCH 08/11] 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 Date: Wed, 15 Jul 2026 11:14:43 +0530 Subject: [PATCH 09/11] 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; From 8a43af3ff3a52f6cab846d2815b55ae6cdfb46bc Mon Sep 17 00:00:00 2001 From: winklemad Date: Thu, 16 Jul 2026 05:16:22 +0530 Subject: [PATCH 10/11] Keep run_compaction response id attempt-local across awaits run_compaction wrote the call's response id into the shared self._response_id and then, after awaiting candidate loading, read that shared field back for the mode decision, the previous_response_id it sends, and the unstored marking. An interleaved run mutating the field in that gap could make this call compact under another call's id and coverage. Capture the response id in a local at the top (coverage and mode are already local) and use it for every post-await decision; self._response_id is still updated for external state but no longer read within the call. Replace the sequential interleaved test with an event-controlled two-call regression that parks one call at its first await while the other overwrites the shared field, asserting each call compacts under its own response id and full stored history survives. --- .../openai_responses_compaction_session.py | 35 ++++++----- ...est_openai_responses_compaction_session.py | 62 ++++++++++++++----- 2 files changed, 66 insertions(+), 31 deletions(-) diff --git a/src/agents/memory/openai_responses_compaction_session.py b/src/agents/memory/openai_responses_compaction_session.py index a5867d9435..861d6802ef 100644 --- a/src/agents/memory/openai_responses_compaction_session.py +++ b/src/agents/memory/openai_responses_compaction_session.py @@ -159,23 +159,30 @@ def _resolve_compaction_mode_for_response( async def run_compaction(self, args: OpenAIResponsesCompactionArgs | None = None) -> None: """Run compaction using responses.compact API.""" - if args and args.get("response_id"): - self._response_id = args["response_id"] + # Capture this call's response id locally (like ``input_covered_full_history`` and + # the mode below) so the decision stays attempt-local across the awaits that follow. + # ``self._response_id`` is shared, so an interleaved run could otherwise overwrite it + # and make this call compact under another call's id and coverage. + response_id = args.get("response_id") if args else None + if response_id is not None: + self._response_id = response_id + else: + response_id = self._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: - self._last_unstored_response_id = self._response_id - elif store is True and self._response_id == self._last_unstored_response_id: + if store is False and response_id: + self._last_unstored_response_id = response_id + elif store is True and response_id == self._last_unstored_response_id: 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, + response_id=response_id, store=store, requested_mode=requested_mode, ) @@ -193,10 +200,10 @@ async def run_compaction(self, args: OpenAIResponsesCompactionArgs | None = None 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 response_id is not None: + self._last_unstored_response_id = response_id - if resolved_mode == "previous_response_id" and not self._response_id: + if resolved_mode == "previous_response_id" and not response_id: raise ValueError( "OpenAIResponsesCompactionSession.run_compaction requires a response_id " "when using previous_response_id compaction." @@ -205,7 +212,7 @@ async def run_compaction(self, args: OpenAIResponsesCompactionArgs | None = None force = args.get("force", False) if args else False should_compact = force or self.should_trigger_compaction( { - "response_id": self._response_id, + "response_id": response_id, "compaction_mode": resolved_mode, "compaction_candidate_items": compaction_candidate_items, "session_items": session_items, @@ -215,7 +222,7 @@ async def run_compaction(self, args: OpenAIResponsesCompactionArgs | None = None if not should_compact: logger.debug( "skip: decision hook declined compaction for %s (mode=%s)", - self._response_id, + response_id, resolved_mode, ) return @@ -223,14 +230,14 @@ async def run_compaction(self, args: OpenAIResponsesCompactionArgs | None = None self._deferred_response_id = None logger.debug( "compact: start for %s using %s (mode=%s)", - self._response_id, + response_id, self.model, resolved_mode, ) compact_kwargs: dict[str, Any] = {"model": self.model} if resolved_mode == "previous_response_id": - compact_kwargs["previous_response_id"] = self._response_id + compact_kwargs["previous_response_id"] = response_id else: compact_kwargs["input"] = session_items @@ -251,7 +258,7 @@ async def run_compaction(self, args: OpenAIResponsesCompactionArgs | None = None logger.debug( "compact: done for %s (mode=%s, output=%s, candidates=%s)", - self._response_id, + response_id, resolved_mode, len(output_items), len(self._compaction_candidate_items), diff --git a/tests/memory/test_openai_responses_compaction_session.py b/tests/memory/test_openai_responses_compaction_session.py index b24038a3f6..ff676a4835 100644 --- a/tests/memory/test_openai_responses_compaction_session.py +++ b/tests/memory/test_openai_responses_compaction_session.py @@ -1,5 +1,6 @@ from __future__ import annotations +import asyncio import logging import warnings as warnings_module from pathlib import Path @@ -387,15 +388,15 @@ async def test_run_compaction_auto_falls_back_when_run_limit_truncates_input( 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( + async def test_run_compaction_keeps_response_id_and_coverage_attempt_local( 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. + # One session instance can be shared by concurrent runs. Each run_compaction call + # must compact under its OWN response id and coverage, held locally across the awaits + # it performs. Regression: run_compaction read the shared self._response_id after + # awaiting candidate loading, so a call that suspended could resume and compact under + # a response id another run left in the shared field -- e.g. a full-coverage call + # inheriting a limited call's id and dropping the older history on replace. underlying = SQLiteSession("test", db_path=str(tmp_path / "history.db")) items: list[TResponseInputItem] = [ cast( @@ -419,19 +420,46 @@ async def test_run_compaction_uses_per_response_limit_under_interleaved_get_item 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) + # Park the first run_compaction (call B) inside its first await so call A runs to + # completion in between and overwrites the shared self._response_id. + b_parked = asyncio.Event() + a_done = asyncio.Event() + real_ensure = session._ensure_compaction_candidates + gated: dict[str, bool] = {"seen": False} - # 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} + async def gated_ensure() -> Any: + if not gated["seen"]: + gated["seen"] = True + b_parked.set() + await a_done.wait() + return await real_ensure() + + session._ensure_compaction_candidates = gated_ensure # type: ignore[method-assign] + + # B saw the full history -> previous_response_id with resp_B. + b_task = asyncio.create_task( + session.run_compaction( + {"response_id": "resp_B", "force": True, "input_covered_full_history": True} + ) ) + await b_parked.wait() - 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) + # A saw a limited window -> must compact via full-history input, and sets the shared + # self._response_id to resp_A while B is suspended. + await session.run_compaction( + {"response_id": "resp_A", "force": True, "input_covered_full_history": False} + ) + a_done.set() + await b_task + + calls = mock_client.responses.compact.call_args_list + sent_previous_ids = {c.kwargs.get("previous_response_id") for c in calls} + # B compacted under its own id, not the resp_A that A left in the shared field. + assert "resp_B" in sent_previous_ids + assert "resp_A" not in sent_previous_ids + # A used full-history input compaction and kept every stored item. + a_call = next(c for c in calls if "input" in c.kwargs) + assert len(a_call.kwargs["input"]) == len(items) @pytest.mark.asyncio async def test_save_result_to_session_carries_run_limit_to_fresh_wrapper_on_resume( From 80afea60958e9bf1932c5ce6b0d390a57de3ef54 Mon Sep 17 00:00:00 2001 From: winklemad Date: Sun, 19 Jul 2026 18:17:30 +0530 Subject: [PATCH 11/11] Keep compaction attempt state local across awaits run_compaction stored each call's response id in a shared self._response_id and, when a later call had none, read it back -- so a concurrent no-id call could compact under another run's previous_response_id and drop older history on replace. It also resolved the mode after awaiting candidate loading, where a concurrent run could shift the shared unstored-id hint. Drop the shared response-id field so each call decides from its own id, and resolve the mode before the candidate-loading await in both run_compaction and _defer_compaction. Replace the interleaving test with an event-controlled two-call regression proving a no-id call falls back to full-history input instead of inheriting a concurrent run's response id. --- .../openai_responses_compaction_session.py | 24 ++++--- ...est_openai_responses_compaction_session.py | 70 ++++++++++--------- 2 files changed, 49 insertions(+), 45 deletions(-) diff --git a/src/agents/memory/openai_responses_compaction_session.py b/src/agents/memory/openai_responses_compaction_session.py index 861d6802ef..dd3db660c2 100644 --- a/src/agents/memory/openai_responses_compaction_session.py +++ b/src/agents/memory/openai_responses_compaction_session.py @@ -130,7 +130,6 @@ def __init__( # cache for incremental candidate tracking self._compaction_candidate_items: list[TResponseInputItem] | None = None self._session_items: list[TResponseInputItem] | None = None - self._response_id: str | None = None self._deferred_response_id: str | None = None self._last_unstored_response_id: str | None = None @@ -159,15 +158,12 @@ def _resolve_compaction_mode_for_response( async def run_compaction(self, args: OpenAIResponsesCompactionArgs | None = None) -> None: """Run compaction using responses.compact API.""" - # Capture this call's response id locally (like ``input_covered_full_history`` and - # the mode below) so the decision stays attempt-local across the awaits that follow. - # ``self._response_id`` is shared, so an interleaved run could otherwise overwrite it - # and make this call compact under another call's id and coverage. + # Everything this attempt decides on -- its response id, coverage, store flag, and the + # resulting mode -- is captured in locals before the first await. run_compaction can run + # concurrently on a shared session, so reading any of these back from the instance after + # an await would let a different run's values leak in (e.g. compacting under another + # call's response id and dropping older history when the session is replaced). response_id = args.get("response_id") if args else None - if response_id is not None: - self._response_id = response_id - else: - response_id = self._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 @@ -179,14 +175,18 @@ 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() + # Resolve the mode from this attempt's own values before awaiting candidate loading, so a + # concurrent run cannot change the shared unstored-id hint between the check and the + # decision. resolved_mode = self._resolve_compaction_mode_for_response( response_id=response_id, store=store, requested_mode=requested_mode, ) + compaction_candidate_items, session_items = await self._ensure_compaction_candidates() + # 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. @@ -368,12 +368,14 @@ async def _defer_compaction( ) -> None: if self._deferred_response_id is not None: return - compaction_candidate_items, session_items = await self._ensure_compaction_candidates() + # Match run_compaction: resolve the mode from this attempt's own values before the + # candidate-loading await, so a concurrent run cannot shift the decision. resolved_mode = self._resolve_compaction_mode_for_response( response_id=response_id, store=store, requested_mode=None, ) + compaction_candidate_items, session_items = await self._ensure_compaction_candidates() # 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 ( diff --git a/tests/memory/test_openai_responses_compaction_session.py b/tests/memory/test_openai_responses_compaction_session.py index ff676a4835..548644fb6f 100644 --- a/tests/memory/test_openai_responses_compaction_session.py +++ b/tests/memory/test_openai_responses_compaction_session.py @@ -388,15 +388,16 @@ async def test_run_compaction_auto_falls_back_when_run_limit_truncates_input( assert "previous_response_id" not in mock_client.responses.compact.call_args.kwargs @pytest.mark.asyncio - async def test_run_compaction_keeps_response_id_and_coverage_attempt_local( + async def test_run_compaction_no_response_id_does_not_inherit_concurrent_response_id( self, tmp_path: Path ) -> None: - # One session instance can be shared by concurrent runs. Each run_compaction call - # must compact under its OWN response id and coverage, held locally across the awaits - # it performs. Regression: run_compaction read the shared self._response_id after - # awaiting candidate loading, so a call that suspended could resume and compact under - # a response id another run left in the shared field -- e.g. a full-coverage call - # inheriting a limited call's id and dropping the older history on replace. + # One session instance can be shared by concurrent runs, and each run_compaction call + # must decide from its OWN attempt state. A call that carries no response id must never + # pick up an id another concurrent run left on the session. Regression: run_compaction + # published each call's id to a shared self._response_id and, when a later call arrived + # without one, read that shared field back -- so a no-id call could compact under a + # concurrent call's previous_response_id and drop the older history that only a + # full-history input compaction preserves. underlying = SQLiteSession("test", db_path=str(tmp_path / "history.db")) items: list[TResponseInputItem] = [ cast( @@ -420,46 +421,47 @@ async def test_run_compaction_keeps_response_id_and_coverage_attempt_local( client=mock_client, ) - # Park the first run_compaction (call B) inside its first await so call A runs to - # completion in between and overwrites the shared self._response_id. - b_parked = asyncio.Event() - a_done = asyncio.Event() + # Park call A inside candidate loading -- past the point where the old code published + # its id to the session -- so the no-id call B runs to completion in between. + a_parked = asyncio.Event() + release_a = asyncio.Event() real_ensure = session._ensure_compaction_candidates - gated: dict[str, bool] = {"seen": False} + gate: dict[str, bool] = {"parked": False} async def gated_ensure() -> Any: - if not gated["seen"]: - gated["seen"] = True - b_parked.set() - await a_done.wait() + if not gate["parked"]: + gate["parked"] = True + a_parked.set() + await release_a.wait() return await real_ensure() session._ensure_compaction_candidates = gated_ensure # type: ignore[method-assign] - # B saw the full history -> previous_response_id with resp_B. - b_task = asyncio.create_task( + # A carries its own id and saw the full history -> previous_response_id with resp_A. + a_task = asyncio.create_task( session.run_compaction( - {"response_id": "resp_B", "force": True, "input_covered_full_history": True} + {"response_id": "resp_A", "force": True, "input_covered_full_history": True} ) ) - await b_parked.wait() + await a_parked.wait() - # A saw a limited window -> must compact via full-history input, and sets the shared - # self._response_id to resp_A while B is suspended. - await session.run_compaction( - {"response_id": "resp_A", "force": True, "input_covered_full_history": False} - ) - a_done.set() - await b_task + # B has no id of its own. It must fall back to full-history input compaction, never to + # the resp_A that A left on the shared session while suspended. + await session.run_compaction({"force": True}) + release_a.set() + await a_task calls = mock_client.responses.compact.call_args_list - sent_previous_ids = {c.kwargs.get("previous_response_id") for c in calls} - # B compacted under its own id, not the resp_A that A left in the shared field. - assert "resp_B" in sent_previous_ids - assert "resp_A" not in sent_previous_ids - # A used full-history input compaction and kept every stored item. - a_call = next(c for c in calls if "input" in c.kwargs) - assert len(a_call.kwargs["input"]) == len(items) + # The no-id call compacted via input over the full stored history -- it did not inherit + # A's response id. + input_calls = [c for c in calls if "input" in c.kwargs] + assert len(input_calls) == 1 + assert len(input_calls[0].kwargs["input"]) == len(items) + # Only A ever compacted under a previous_response_id, and only under its own. + previous_ids = [ + c.kwargs["previous_response_id"] for c in calls if "previous_response_id" in c.kwargs + ] + assert previous_ids == ["resp_A"] @pytest.mark.asyncio async def test_save_result_to_session_carries_run_limit_to_fresh_wrapper_on_resume(