fix: load full history when compacting a limited session

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,35 +158,61 @@ 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"]
+        # 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
         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
+
+        # 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=self._response_id,
+            response_id=response_id,
             store=store,
             requested_mode=requested_mode,
         )
 
-        if resolved_mode == "previous_response_id" and not self._response_id:
+        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.
+        if (
+            (requested_mode or self.compaction_mode) == "auto"
+            and resolved_mode == "previous_response_id"
+            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
+            # compaction reuse it once the session shrinks back within the limit.
+            if response_id is not None:
+                self._last_unstored_response_id = 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."
             )
 
-        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(
             {
-                "response_id": self._response_id,
+                "response_id": response_id,
                 "compaction_mode": resolved_mode,
                 "compaction_candidate_items": compaction_candidate_items,
                 "session_items": session_items,
@@ -197,22 +222,22 @@ 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
 
         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
 
@@ -233,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),
@@ -245,6 +270,30 @@ 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, input_history_limit: int | None) -> bool:
+        """Whether a response's prepared input window held every stored item.
+
+        ``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=input_history_limit)
+        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,
         *,
@@ -311,15 +360,30 @@ 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_covered_full_history: bool | None = None,
+    ) -> 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 (
+            self.compaction_mode == "auto"
+            and resolved_mode == "previous_response_id"
+            and not await self._input_covers_history(input_covered_full_history, None)
+        ):
+            resolved_mode = "input"
         should_compact = self.should_trigger_compaction(
             {
                 "response_id": response_id,
@@ -367,7 +431,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

src/agents/memory/session.py

@@ -127,6 +127,22 @@ 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.
+    """
+
+    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):

src/agents/run.py

@@ -936,6 +936,7 @@ def _finalize_result(result: RunResult) -> RunResult:
                                             run_state._reasoning_item_id_policy
                                         ),
                                         store=store_setting,
+                                        session_settings=run_config.session_settings,
                                     )
                                 )
 
@@ -1048,6 +1049,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)
@@ -1414,6 +1416,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:
@@ -1424,6 +1427,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
@@ -1476,6 +1480,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)
@@ -1495,6 +1500,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
@@ -1556,6 +1562,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:

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,
     )
 
 

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,
@@ -323,6 +323,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,
@@ -337,6 +338,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 = (
@@ -354,6 +356,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,
@@ -368,6 +371,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 = (
@@ -655,6 +659,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(
@@ -669,6 +674,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(
@@ -683,6 +689,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: