fix: load full history when compacting a limited session

src/agents/memory/openai_responses_compaction_session.py

@@ -162,6 +162,8 @@ 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
+        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:
@@ -170,20 +172,36 @@ 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,
         )
 
+        # 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 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(
                 "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 +263,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,7 +353,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_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()
@@ -320,6 +367,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._input_covers_history(input_covered_full_history, None)
+        ):
+            resolved_mode = "input"
         should_compact = self.should_trigger_compaction(
             {
                 "response_id": response_id,
@@ -367,7 +422,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

@@ -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
@@ -1411,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)
@@ -1430,6 +1435,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
@@ -1491,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:

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,
@@ -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:

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
@@ -346,6 +359,24 @@ 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:
@@ -358,7 +389,11 @@ async def save_result_to_session(
         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_covered_full_history=input_covered_full_history,
+                )
                 if inspect.isawaitable(result):
                     await result
             logger.debug(
@@ -381,6 +416,7 @@ async def save_result_to_session(
         compaction_args: OpenAIResponsesCompactionArgs = {
             "response_id": response_id,
             "force": force_compaction,
+            "input_covered_full_history": input_covered_full_history,
         }
         if store is not None:
             compaction_args["store"] = store
@@ -397,6 +433,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 +446,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