diff --git a/src/agents/memory/openai_responses_compaction_session.py b/src/agents/memory/openai_responses_compaction_session.py
index f09c3a6edd..af272d5ce0 100644
--- a/src/agents/memory/openai_responses_compaction_session.py
+++ b/src/agents/memory/openai_responses_compaction_session.py
@@ -27,10 +27,59 @@
 
 DEFAULT_COMPACTION_THRESHOLD = 10
 _ALL_SESSION_ITEMS_LIMIT = 2_147_483_647
+# Recorded response boundaries are pruned oldest first past this size so ids whose
+# compaction never runs (for example turns whose compaction was deferred) cannot
+# grow the map without bound.
+_MAX_RECORDED_RESPONSE_BOUNDARIES = 50
 
 OpenAIResponsesCompactionMode = Literal["previous_response_id", "input", "auto"]
 
 
+class _SessionOwnershipToken:
+    """One run's claim over the stored history it built its request from.
+
+    The runner captures a token under the mutation lock when it reads the
+    session to build a run's request input, before that read, and threads it
+    through the run's persists. ``count`` starts at the underlying item count
+    at capture time and advances by exactly the items the same run appends;
+    ``generation`` pins the destructive generation seen at capture. Any
+    history rewrite, including a compaction replacement this run itself
+    triggered, bumps the generation and retires the token for good: the
+    rewritten store may hold interleaved items the run's requests never saw,
+    so later persists of the run record nothing and their compactions skip.
+    The token lives only for the run that captured it and is never stored on
+    the session, so restored or resumed runs start without one.
+
+    A token can also be voided outright through ``invalidate``. The runner
+    calls it when the run's request input provably stops covering the whole
+    stored history: the request history read returned a different number of
+    items than the token captured, a session input callback rebuilt the
+    input, a call_model_input_filter rewrote the request, a handoff input
+    filter rewrote the accumulated history mid run, or handoff history
+    nesting collapsed it into a rendered transcript. Counts cannot describe
+    what the server saw in any of these cases, so a voided token never
+    records. Every persist that carries a token, in whatever state, marks the
+    session as boundary managed, which makes every compaction keyed on the
+    run's responses skip before the billed call whenever nothing was recorded
+    for them.
+    """
+
+    __slots__ = ("count", "generation", "invalidated")
+
+    def __init__(self, count: int, generation: int) -> None:
+        self.count = count
+        self.generation = generation
+        self.invalidated = False
+
+    def advance(self, item_count: int) -> None:
+        """Count a batch this run appended itself."""
+        self.count += item_count
+
+    def invalidate(self) -> None:
+        """Void the token: the run's request input skipped stored history."""
+        self.invalidated = True
+
+
 def select_compaction_candidate_items(
     items: list[TResponseInputItem],
 ) -> list[TResponseInputItem]:
@@ -137,11 +186,49 @@ def __init__(
         self._compaction_candidate_items: list[TResponseInputItem] | None = None
         self._session_items: list[TResponseInputItem] | None = None
         self._response_id: str | None = None
+        # Authoritative record of what previous_response_id compaction covers.
+        # Each response id maps to the exact local item count at the moment its
+        # batch was persisted, recorded under the mutation lock, and only when
+        # the persisting run's ownership token proves no other writer touched
+        # the store between the run's request input read and the persist. Such
+        # a compaction covers the server side history through that response
+        # only, so the replacement must preserve every local item past the
+        # recorded boundary, not past a snapshot taken later. clear_session and pop_item
+        # drop recorded boundaries outright, and a successful replacement
+        # translates boundaries at or past the rewritten prefix onto the new
+        # history and drops the rest, so a stale count can never be read from
+        # the map. An absent entry therefore means the boundary was dropped by
+        # one of those paths, evicted past the cap, or never recorded because
+        # the persisting run's request input skipped stored history or some
+        # other writer interleaved before the persist; once anything armed
+        # the gate, a compaction keyed on an absent id skips instead of
+        # guessing, because the snapshot fallback would claim turns the
+        # server side history through that response never contained. The
+        # fallback stays reserved for sessions no token carrying persist
+        # ever touched, whose direct callers persist before compacting and
+        # own that ordering.
+        self._response_boundaries: dict[str, int] = {}
+        # True once any response boundary has been recorded on this instance,
+        # and also once any persist arrived carrying an ownership token at
+        # all, whether the token was valid, voided, or stale: a token exists
+        # only inside a runner managed run, so its arrival alone proves a
+        # runner manages this session's boundaries and an absent entry must
+        # mean skip rather than snapshot fallback. Hookless persists never
+        # arm it, which keeps the fallback for direct callers who persist
+        # before compacting and own that ordering. It is never reset: a
+        # compaction keyed on a response recorded before a clear must still
+        # skip after the clear.
+        self._response_boundaries_ever_recorded = False
         self._deferred_response_id: str | None = None
         self._last_unstored_response_id: str | None = None
         # Serialize wrapper mutations against compaction snapshot/replace/restore so a
         # cancellation rollback cannot rewrite past a newer concurrent write.
         self._mutation_lock = asyncio.Lock()
+        # Bumped under the lock by every rewrite of stored history: clear_session,
+        # pop_item, and a successful compaction replacement. The prefix check in
+        # run_compaction cannot see a rewrite when the snapshot it compares against
+        # was empty, so the counter records those explicitly.
+        self._destructive_generation = 0
 
     @property
     def client(self) -> AsyncOpenAI:
@@ -201,37 +288,72 @@ async def run_compaction(
                 "when using previous_response_id compaction."
             )
 
-        compaction_candidate_items, session_items = await self._ensure_compaction_candidates()
+        # resolved_mode was derived from the response id as it is right now. A
+        # concurrent call can overwrite self._response_id while this one waits on
+        # the lock below, so everything after this line uses the paired local.
+        response_id = self._response_id
 
-        force = args.get("force", False) if args else False
-        should_compact = force or self.should_trigger_compaction(
-            {
-                "response_id": self._response_id,
-                "compaction_mode": resolved_mode,
-                "compaction_candidate_items": compaction_candidate_items,
-                "session_items": session_items,
-            }
-        )
-
-        if not should_compact:
-            logger.debug(
-                "skip: decision hook declined compaction for %s (mode=%s)",
-                self._response_id,
-                resolved_mode,
+        async with self._mutation_lock:
+            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": response_id,
+                    "compaction_mode": resolved_mode,
+                    "compaction_candidate_items": compaction_candidate_items,
+                    "session_items": session_items,
+                }
             )
-            return
 
-        self._deferred_response_id = None
+            if not should_compact:
+                logger.debug(
+                    "skip: decision hook declined compaction for %s (mode=%s)",
+                    response_id,
+                    resolved_mode,
+                )
+                return
+
+            # _response_boundaries is the authority on what previous_response_id
+            # compaction covers; the snapshot below only anchors the divergence
+            # check. Skip when the map cannot vouch for this response.
+            recorded_boundary: int | None = None
+            if resolved_mode == "previous_response_id" and response_id is not None:
+                if response_id in self._response_boundaries:
+                    recorded_boundary = self._response_boundaries[response_id]
+                elif self._response_boundaries_ever_recorded:
+                    logger.warning(
+                        "Skipped compaction for %s (mode=%s): this session manages response "
+                        "boundaries but has no entry for this response, because its boundary "
+                        "was invalidated by a history rewrite, evicted, or removed, the "
+                        "persisting run's request input skipped stored history, or another "
+                        "writer interleaved before the persist; the snapshot fallback would "
+                        "claim items the server side history through this response never "
+                        "contained.",
+                        response_id,
+                        resolved_mode,
+                    )
+                    return
+
+            # Capture the full stored history while the lock still excludes writers.
+            # It anchors the post-flight check that detects concurrent mutations.
+            snapshot_items = await self._get_all_underlying_session_items()
+            snapshot_generation = self._destructive_generation
+            # Clear the deferred id before releasing the lock: this call
+            # subsumes the deferral, and a concurrent run that defers after the
+            # release must not have its deferral wiped.
+            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
 
@@ -247,19 +369,96 @@ async def run_compaction(
 
         async with self._mutation_lock:
             previous_items = await self._get_all_underlying_session_items()
-            await self._replace_underlying_session_items(
-                output_items=output_items,
-                previous_items=previous_items,
-            )
-            self._compaction_candidate_items = select_compaction_candidate_items(output_items)
-            self._session_items = output_items
+            baseline_count = len(snapshot_items)
+            if (
+                self._destructive_generation != snapshot_generation
+                or previous_items[:baseline_count] != snapshot_items
+            ):
+                # The stored history no longer extends the snapshot, so
+                # replacing it could resurrect deleted items; keep the current
+                # items and drop the caches. The generation counter catches
+                # every rewrite made through this wrapper, including those the
+                # prefix comparison cannot see because the snapshot was empty
+                # or the rewrite restored identical items. The prefix
+                # comparison catches writers that bypass this wrapper and
+                # mutate the underlying session directly.
+                self._compaction_candidate_items = None
+                self._session_items = None
+                logger.warning(
+                    "Skipped compaction replacement for %s (mode=%s): session history "
+                    "diverged from the compaction snapshot while the request was in flight.",
+                    response_id,
+                    resolved_mode,
+                )
+                return
+            if recorded_boundary is not None and recorded_boundary > baseline_count:
+                # No healthy path records a boundary past the stored history:
+                # appends record the post append count, and rewrites drop or
+                # translate entries. Treat the boundary state as corrupt and
+                # skip instead of slicing past the end, which would classify
+                # the whole history as covered and replace it outright.
+                self._compaction_candidate_items = None
+                self._session_items = None
+                self._response_boundaries.clear()
+                logger.warning(
+                    "Skipped compaction replacement for %s (mode=%s): the recorded boundary "
+                    "exceeds the stored history, so the boundary state no longer describes "
+                    "this session.",
+                    response_id,
+                    resolved_mode,
+                )
+                return
+            # The recorded boundary, not the snapshot, decides what the
+            # replacement preserves; see _response_boundaries. The snapshot
+            # length substitutes only when nothing was recorded: direct
+            # run_compaction calls, and input mode, where the compaction input
+            # was captured under the same lock as the snapshot.
+            preserve_from = recorded_boundary if recorded_boundary is not None else baseline_count
+            concurrent_tail = previous_items[preserve_from:]
+            # Build the refreshed caches before the replacement so nothing that
+            # can raise sits between the durable write and the bookkeeping
+            # assignments below.
+            cached_items = output_items + _normalize_compaction_session_items(concurrent_tail)
+            cached_candidate_items = select_compaction_candidate_items(cached_items)
+            try:
+                await self._replace_underlying_session_items(
+                    output_items=output_items + concurrent_tail,
+                    previous_items=previous_items,
+                )
+            except (Exception, asyncio.CancelledError):
+                # The replacement transaction failed and its restore is best
+                # effort, so the store may hold anything. Count the attempt as
+                # a rewrite and drop every recorded boundary, matching pop_item
+                # and clear_session, so no later compaction trusts a count the
+                # store may no longer back.
+                self._destructive_generation += 1
+                self._response_boundaries.clear()
+                raise
+            self._compaction_candidate_items = cached_candidate_items
+            self._session_items = cached_items
+            # The replacement itself rewrote stored history, so a second in flight
+            # compaction that snapshotted before it must not treat this output as
+            # a concurrent tail. With a non empty snapshot the prefix check would
+            # catch that; when both snapshots were empty only the counter can.
+            self._destructive_generation += 1
+            # Translate the surviving boundaries onto the rewritten history: a
+            # boundary at or past preserve_from still counts the same persisted
+            # batch at its shifted position. A boundary inside the rewritten
+            # prefix has no counterpart in the new history, so drop it and let
+            # its compaction skip; see _response_boundaries.
+            self._response_boundaries = {
+                rid: boundary - preserve_from + len(output_items)
+                for rid, boundary in self._response_boundaries.items()
+                if boundary >= preserve_from
+            }
 
         logger.debug(
-            "compact: done for %s (mode=%s, output=%s, candidates=%s)",
-            self._response_id,
+            "compact: done for %s (mode=%s, output=%s, candidates=%s, concurrent_tail=%s)",
+            response_id,
             resolved_mode,
             len(output_items),
             len(self._compaction_candidate_items or []),
+            len(concurrent_tail),
         )
 
     async def get_items(self, limit: int | None = None) -> list[TResponseInputItem]:
@@ -384,24 +583,28 @@ async def _restore_underlying_session_items(
         )
 
     async def _defer_compaction(self, response_id: str, store: bool | None = None) -> None:
-        if self._deferred_response_id is not None:
-            return
-        compaction_candidate_items, session_items = await self._ensure_compaction_candidates()
-        resolved_mode = self._resolve_compaction_mode_for_response(
-            response_id=response_id,
-            store=store,
-            requested_mode=None,
-        )
-        should_compact = self.should_trigger_compaction(
-            {
-                "response_id": response_id,
-                "compaction_mode": resolved_mode,
-                "compaction_candidate_items": compaction_candidate_items,
-                "session_items": session_items,
-            }
-        )
-        if should_compact:
-            self._deferred_response_id = response_id
+        # The cache fill below must not interleave with a replacement that is
+        # installing fresh caches, so the whole decision runs under the
+        # mutation lock. The runner call site holds no lock here.
+        async with self._mutation_lock:
+            if self._deferred_response_id is not None:
+                return
+            compaction_candidate_items, session_items = await self._ensure_compaction_candidates()
+            resolved_mode = self._resolve_compaction_mode_for_response(
+                response_id=response_id,
+                store=store,
+                requested_mode=None,
+            )
+            should_compact = self.should_trigger_compaction(
+                {
+                    "response_id": response_id,
+                    "compaction_mode": resolved_mode,
+                    "compaction_candidate_items": compaction_candidate_items,
+                    "session_items": session_items,
+                }
+            )
+            if should_compact:
+                self._deferred_response_id = response_id
 
     def _get_deferred_compaction_response_id(self) -> str | None:
         return self._deferred_response_id
@@ -411,28 +614,146 @@ def _clear_deferred_compaction(self) -> None:
 
     async def add_items(self, items: list[TResponseInputItem]) -> None:
         async with self._mutation_lock:
-            try:
-                await self.underlying_session.add_items(items)
-            except (Exception, asyncio.CancelledError):
-                # The backend may have committed before acknowledgement failed. Re-read its
-                # authoritative history before compaction instead of retaining a stale cache.
-                self._compaction_candidate_items = None
-                self._session_items = None
-                raise
-            if self._compaction_candidate_items is not None:
-                new_items = _normalize_compaction_session_items(items)
-                new_candidates = select_compaction_candidate_items(new_items)
-                if new_candidates:
-                    self._compaction_candidate_items.extend(new_candidates)
-            if self._session_items is not None:
-                self._session_items.extend(_normalize_compaction_session_items(items))
+            await self._add_items_locked(items)
+
+    async def _capture_ownership_token(self) -> _SessionOwnershipToken:
+        """Capture the store state a run is about to build its request from.
+
+        The runner calls this immediately before it reads the session for a
+        run's request input. Capturing before the read keeps the token
+        conservative: a write that lands between the capture and the read can
+        only make the token stale, never let it claim an item the request
+        input missed.
+        """
+        async with self._mutation_lock:
+            count = len(await self._get_all_underlying_session_items())
+            return _SessionOwnershipToken(count=count, generation=self._destructive_generation)
+
+    async def _add_items_for_response(
+        self,
+        items: list[TResponseInputItem],
+        *,
+        response_id: str,
+        ownership_token: _SessionOwnershipToken | None = None,
+    ) -> None:
+        """Append a response's persisted batch and record its compaction boundary.
+
+        The runner calls this instead of add_items when the batch belongs to a
+        specific response. Appending and recording in one locked region keeps
+        the pairing exact: no other writer can slip items in between the batch
+        and the count recorded for it. The boundary records only when the
+        caller's ownership token still matches the store; without a token, or
+        with a voided or stale one, the batch is appended and nothing is
+        recorded. Any token at all, in whatever state, marks the session as
+        boundary managed, because a token exists only inside a runner managed
+        run; only hookless persists leave that mark alone. See
+        _response_boundaries for how the recorded boundary is consumed.
+
+        The count that seeds the boundary is read before the append. The lock
+        excludes every other writer for the whole region, so that count plus
+        the batch length equals the count a read after the append would return,
+        and once the append succeeds the recording below is pure assignment
+        that cannot fail. A history read that failed after a successful append
+        would surface an error for a batch the backend already holds, and
+        retrying the turn would persist it again. When the read itself fails,
+        the append has not started, so no boundary is recorded, the ever
+        recorded flag stays untouched, and a retry begins clean.
+
+        The ever recorded arm also precedes the append: the backend can
+        commit the batch and still raise before acknowledging, and a batch it
+        committed for a token carrying persist must leave the session
+        boundary managed, or a compaction keyed on one of the run's responses
+        would reach the snapshot fallback through the unarmed gate and claim
+        items the server side history never contained. Arming first is
+        strictly conservative; when the append never committed it costs at
+        most a skipped compaction.
+        """
+        async with self._mutation_lock:
+            count_before_append = len(await self._get_all_underlying_session_items())
+            # The token decides whether this response may own the whole prefix.
+            # Appends by any other writer leave the underlying count above the
+            # token's, because the token advances only for this run's own
+            # batches, and every destructive rewrite bumps the generation. Both
+            # matching therefore means the store holds exactly the history this
+            # run's request input was built from plus this run's own persisted
+            # batches, all of which the server side history through this
+            # response contains, so count plus the batch length is the true
+            # coverage boundary. On a mismatch some interleaved write sits
+            # below where this batch lands and the server side history cannot
+            # contain it; a recorded count would let the replacement delete it,
+            # so the batch is appended with nothing recorded and a compaction
+            # keyed on this response skips through the ever recorded gate
+            # before the billed call.
+            owns_prefix = (
+                ownership_token is not None
+                and not ownership_token.invalidated
+                and ownership_token.generation == self._destructive_generation
+                and ownership_token.count == count_before_append
+            )
+            if ownership_token is not None:
+                # The token, in whatever state it arrives, proves a runner is
+                # managing this session's boundaries, so arm the gate before
+                # the append and before the state checks below: a compaction
+                # keyed on any of this run's responses must skip through an
+                # absent entry instead of reaching the snapshot fallback.
+                # Arming after the append would leave the gate unarmed when
+                # the backend commits the batch and then raises before
+                # acknowledging, and the fallback would classify the
+                # committed items as covered and let the replacement delete
+                # them; the same unarmed gate would follow the first ever
+                # persist on a fresh wrapper arriving with a stale token.
+                # Only hookless persists, which carry no token, leave the
+                # gate alone, preserving the fallback for direct callers.
+                self._response_boundaries_ever_recorded = True
+            await self._add_items_locked(items)
+            if ownership_token is None:
+                return
+            if ownership_token.invalidated:
+                # The run's request input skipped stored history, so no count
+                # can describe what the server saw; nothing records.
+                return
+            if not owns_prefix:
+                # Another writer landed between the run's request input read
+                # and this persist; a recorded count would claim the
+                # interleaved items for this response, so nothing records.
+                return
+            ownership_token.advance(len(items))
+            boundary = count_before_append + len(items)
+            # Pop before reinserting so a recorded id moves to the newest slot;
+            # the eviction loop below removes oldest first by insertion order.
+            self._response_boundaries.pop(response_id, None)
+            self._response_boundaries[response_id] = boundary
+            while len(self._response_boundaries) > _MAX_RECORDED_RESPONSE_BOUNDARIES:
+                del self._response_boundaries[next(iter(self._response_boundaries))]
+
+    async def _add_items_locked(self, items: list[TResponseInputItem]) -> None:
+        try:
+            await self.underlying_session.add_items(items)
+        except (Exception, asyncio.CancelledError):
+            # The backend may have committed before acknowledgement failed. Read its
+            # authoritative history again before compaction instead of retaining a
+            # stale cache.
+            self._compaction_candidate_items = None
+            self._session_items = None
+            raise
+        if self._compaction_candidate_items is not None:
+            new_items = _normalize_compaction_session_items(items)
+            new_candidates = select_compaction_candidate_items(new_items)
+            if new_candidates:
+                self._compaction_candidate_items.extend(new_candidates)
+        if self._session_items is not None:
+            self._session_items.extend(_normalize_compaction_session_items(items))
 
     async def pop_item(self) -> TResponseInputItem | None:
         async with self._mutation_lock:
             popped = await self.underlying_session.pop_item()
-            if popped:
+            if popped is not None:
                 self._compaction_candidate_items = None
                 self._session_items = None
+                self._destructive_generation += 1
+                # Recorded boundaries are item counts, so removing an item makes
+                # them stale even when the pop only trimmed the tail.
+                self._response_boundaries.clear()
             return popped
 
     async def clear_session(self) -> None:
@@ -441,6 +762,8 @@ async def clear_session(self) -> None:
             self._compaction_candidate_items = []
             self._session_items = []
             self._deferred_response_id = None
+            self._destructive_generation += 1
+            self._response_boundaries.clear()
 
     async def _ensure_compaction_candidates(
         self,
diff --git a/src/agents/run.py b/src/agents/run.py
index c23b6363cf..788fc700f9 100644
--- a/src/agents/run.py
+++ b/src/agents/run.py
@@ -135,6 +135,7 @@
 from .run_internal.session_persistence import (
     _session_get_items,
     admit_pending_input,
+    capture_session_ownership_token,
     commit_server_pending_input,
     persist_session_items_for_guardrail_trip,
     prepare_input_with_session,
@@ -162,6 +163,9 @@
 from .tracing.span_data import AgentSpanData, TaskSpanData
 from .util import _error_tracing
 
+if TYPE_CHECKING:
+    from .memory.openai_responses_compaction_session import _SessionOwnershipToken
+
 DEFAULT_AGENT_RUNNER: AgentRunner = None  # type: ignore
 # the value is set at the end of the module
 
@@ -607,6 +611,18 @@ async def _run_impl(
         # Track the most recent input batch we persisted so conversation-lock retries can rewind
         # exactly those items (and not the full history).
         last_saved_input_snapshot_for_rewind: list[TResponseInputItem] | None = None
+        # Ownership of the session history this run builds its request input
+        # from; captured right before the history read and threaded through
+        # every persist so compaction boundaries record only when no other
+        # writer interleaved. Resumed runs never read the session for their
+        # request input, so they carry no token and record no boundaries.
+        # The token is voided wherever the request provably stops covering
+        # the stored history: a resolved limit truncating the history read,
+        # a session input callback rebuilding the input, a
+        # call_model_input_filter rewriting a request, or a handoff input
+        # filter rewriting the accumulated history. Such runs record no
+        # boundaries either.
+        session_ownership_token: _SessionOwnershipToken | None = None
 
         if is_resumed_state and run_state is not None:
             (
@@ -671,6 +687,9 @@ async def _run_impl(
                 original_input_for_state = raw_input
                 session_input_items_for_persistence = []
             else:
+                session_ownership_token = await capture_session_ownership_token(
+                    session, wrapper=context_wrapper
+                )
                 (
                     prepared_input,
                     session_input_items_for_persistence,
@@ -681,6 +700,7 @@ async def _run_impl(
                     run_config.session_settings,
                     reasoning_item_id_policy=resolved_reasoning_item_id_policy,
                     wrapper=context_wrapper,
+                    ownership_token=session_ownership_token,
                 )
                 original_input_for_state = prepared_input
 
@@ -952,6 +972,7 @@ def _mark_response_hooks_started() -> None:
                         run_state,
                         store=store_setting,
                         wrapper=context_wrapper,
+                        ownership_token=session_ownership_token,
                     )
                     session_input_items_for_persistence = []
             except BaseException:
@@ -1009,6 +1030,7 @@ def _mark_response_hooks_started() -> None:
                                     run_state,
                                     store=store_setting,
                                     wrapper=context_wrapper,
+                                    ownership_token=session_ownership_token,
                                 )
                             )
                             raise
@@ -1060,6 +1082,7 @@ def _mark_response_hooks_started() -> None:
                             run_state,
                             store=store_setting,
                             wrapper=context_wrapper,
+                            ownership_token=session_ownership_token,
                         )
                         session_input_items_for_persistence = []
                     if run_state is not None and run_state._current_step is not None:
@@ -1320,6 +1343,7 @@ def _mark_response_hooks_started() -> None:
                                             response_id=turn_result.model_response.response_id,
                                             store=store_setting,
                                             wrapper=context_wrapper,
+                                            ownership_token=session_ownership_token,
                                         )
                                     except BaseException as persistence_error:
                                         raise _safe_redacted_persistence_error(
@@ -1352,6 +1376,7 @@ def _mark_response_hooks_started() -> None:
                                             response_id=turn_result.model_response.response_id,
                                             store=store_setting,
                                             wrapper=context_wrapper,
+                                            ownership_token=session_ownership_token,
                                         )
                                     raise
 
@@ -1371,6 +1396,7 @@ def _mark_response_hooks_started() -> None:
                                     response_id=turn_result.model_response.response_id,
                                     store=store_setting,
                                     wrapper=context_wrapper,
+                                    ownership_token=session_ownership_token,
                                 )
                                 current_step = getattr(run_state, "_current_step", None)
                                 approvals_from_state = approvals_from_step(current_step)
@@ -1444,6 +1470,7 @@ def _mark_response_hooks_started() -> None:
                                 server_conversation_tracker=server_conversation_tracker,
                                 store=store_setting,
                                 wrapper=context_wrapper,
+                                ownership_token=session_ownership_token,
                             )
                             generated_items.extend(admission_items)
                             session_items.extend(admission_items)
@@ -1523,6 +1550,7 @@ async def _save_max_turns_handler_output(
                                     reasoning_item_id_policy=resolved_reasoning_item_id_policy,
                                     store=store_setting,
                                     wrapper=context_wrapper,
+                                    ownership_token=session_ownership_token,
                                 )
                             )
                             if not items:
@@ -1656,6 +1684,7 @@ async def _save_max_turns_handler_output(
                                         run_state,
                                         store=store_setting,
                                         wrapper=context_wrapper,
+                                        ownership_token=session_ownership_token,
                                     )
                                 )
                                 raise
@@ -1685,6 +1714,7 @@ async def _save_max_turns_handler_output(
                                     on_response_accepted=_commit_pending_server_response,
                                     on_response_hooks_started=_mark_response_hooks_started,
                                     run_state=run_state,
+                                    ownership_token=session_ownership_token,
                                 )
                             )
 
@@ -1717,6 +1747,7 @@ async def _save_max_turns_handler_output(
                                             run_state,
                                             store=store_setting,
                                             wrapper=context_wrapper,
+                                            ownership_token=session_ownership_token,
                                         )
                                     )
                                     raise
@@ -1760,6 +1791,7 @@ async def _save_max_turns_handler_output(
                                 on_response_accepted=_commit_pending_server_response,
                                 on_response_hooks_started=_mark_response_hooks_started,
                                 run_state=run_state,
+                                ownership_token=session_ownership_token,
                             )
                     finally:
                         if current_turn_span is not None:
@@ -1869,6 +1901,7 @@ async def _save_max_turns_handler_output(
                                         response_id=turn_result.model_response.response_id,
                                         store=store_setting,
                                         wrapper=context_wrapper,
+                                        ownership_token=session_ownership_token,
                                     )
 
                     # After the first resumed turn, treat subsequent turns as fresh
@@ -1947,6 +1980,7 @@ async def _save_max_turns_handler_output(
                                         response_id=turn_result.model_response.response_id,
                                         store=store_setting,
                                         wrapper=context_wrapper,
+                                        ownership_token=session_ownership_token,
                                     )
                                 except BaseException as persistence_error:
                                     raise _safe_redacted_persistence_error(
@@ -1979,6 +2013,7 @@ async def _save_max_turns_handler_output(
                                         response_id=turn_result.model_response.response_id,
                                         store=store_setting,
                                         wrapper=context_wrapper,
+                                        ownership_token=session_ownership_token,
                                     )
                                 raise
 
@@ -1998,6 +2033,7 @@ async def _save_max_turns_handler_output(
                                 response_id=turn_result.model_response.response_id,
                                 store=store_setting,
                                 wrapper=context_wrapper,
+                                ownership_token=session_ownership_token,
                             )
 
                             # Ensure starting_input is not None and not RunState
@@ -2056,6 +2092,7 @@ async def _save_max_turns_handler_output(
                                         response_id=turn_result.model_response.response_id,
                                         store=store_setting,
                                         wrapper=context_wrapper,
+                                        ownership_token=session_ownership_token,
                                     )
                             append_model_response_if_new(
                                 model_responses, turn_result.model_response
@@ -2118,6 +2155,7 @@ async def _save_max_turns_handler_output(
                                 response_id=turn_result.model_response.response_id,
                                 store=store_setting,
                                 wrapper=context_wrapper,
+                                ownership_token=session_ownership_token,
                             )
                             continue
                         else:
diff --git a/src/agents/run_internal/agent_runner_helpers.py b/src/agents/run_internal/agent_runner_helpers.py
index be1d976724..17b78f6feb 100644
--- a/src/agents/run_internal/agent_runner_helpers.py
+++ b/src/agents/run_internal/agent_runner_helpers.py
@@ -3,7 +3,7 @@
 from __future__ import annotations
 
 from collections.abc import Mapping
-from typing import Any, cast
+from typing import TYPE_CHECKING, Any, cast
 
 from openai.types.responses.response_usage import OutputTokensDetails
 
@@ -45,6 +45,9 @@
 from .tool_use_tracker import AgentToolUseTracker, serialize_tool_use_tracker
 from .turn_preparation import get_model
 
+if TYPE_CHECKING:
+    from ..memory.openai_responses_compaction_session import _SessionOwnershipToken
+
 __all__ = [
     "apply_resumed_conversation_settings",
     "append_model_response_if_new",
@@ -535,6 +538,7 @@ async def save_turn_items_if_needed(
     response_id: str | None,
     store: bool | None = None,
     wrapper: RunContextWrapper[Any] | None = None,
+    ownership_token: _SessionOwnershipToken | None = None,
 ) -> None:
     """Persist turn items when persistence is enabled and guardrails allow it."""
     if not session_persistence_enabled:
@@ -551,6 +555,7 @@ async def save_turn_items_if_needed(
         response_id=response_id,
         store=store,
         wrapper=wrapper,
+        ownership_token=ownership_token,
     )
 
 
@@ -565,6 +570,7 @@ async def save_final_turn_items_after_guardrails(
     reasoning_item_id_policy: ReasoningItemIdPolicy | None = None,
     store: bool | None = None,
     wrapper: RunContextWrapper[Any] | None = None,
+    ownership_token: _SessionOwnershipToken | None = None,
 ) -> int:
     """Persist deferred final-turn items without skipping a partially persisted resumed turn."""
     if not session_persistence_enabled or not items:
@@ -572,6 +578,9 @@ async def save_final_turn_items_after_guardrails(
     if input_guardrails_triggered(input_guardrail_results):
         return 0
     if run_state is not None and run_state._current_turn_persisted_item_count > 0:
+        # The reconciling path may append behind an earlier partial persist, so
+        # no count read there can prove what this response covers; it carries
+        # no ownership token and records no boundary.
         run_state._current_turn_persisted_item_count = await save_resumed_turn_items(
             session=session,
             items=items,
@@ -591,6 +600,7 @@ async def save_final_turn_items_after_guardrails(
         reasoning_item_id_policy=reasoning_item_id_policy,
         store=store,
         wrapper=wrapper,
+        ownership_token=ownership_token,
     )
 
 
diff --git a/src/agents/run_internal/run_loop.py b/src/agents/run_internal/run_loop.py
index 3c293b6ee8..8bde970e31 100644
--- a/src/agents/run_internal/run_loop.py
+++ b/src/agents/run_internal/run_loop.py
@@ -10,7 +10,7 @@
 from collections.abc import Awaitable, Callable
 from contextlib import aclosing
 from functools import partial
-from typing import Any, TypeVar, cast
+from typing import TYPE_CHECKING, Any, TypeVar, cast
 from uuid import uuid4
 
 from openai.types.responses import (
@@ -173,6 +173,7 @@
 from .session_persistence import (
     _session_get_items,
     admit_pending_input,
+    capture_session_ownership_token,
     commit_server_pending_input,
     persist_session_items_for_guardrail_trip,
     prepare_input_with_session,
@@ -226,6 +227,9 @@
     run_final_output_hooks,
 )
 
+if TYPE_CHECKING:
+    from ..memory.openai_responses_compaction_session import _SessionOwnershipToken
+
 __all__ = [
     "extract_tool_call_id",
     "coerce_shell_call",
@@ -418,6 +422,7 @@ async def _save_stream_items(
     response_id: str | None,
     update_persisted_count: bool,
     store: bool | None = None,
+    ownership_token: _SessionOwnershipToken | None = None,
 ) -> None:
     if not await _should_persist_stream_items(
         session=session,
@@ -433,6 +438,7 @@ async def _save_stream_items(
         response_id=response_id,
         store=store,
         wrapper=streamed_result.context_wrapper,
+        ownership_token=ownership_token,
     )
     if update_persisted_count and streamed_result._state is not None:
         streamed_result._current_turn_persisted_item_count = (
@@ -794,6 +800,7 @@ async def _persist_stream_input_if_needed(
     session: Session | None,
     server_conversation_tracker: OpenAIServerConversationTracker | None,
     context_wrapper: RunContextWrapper[TContext],
+    ownership_token: _SessionOwnershipToken | None = None,
 ) -> None:
     if (
         streamed_result._stream_input_persisted
@@ -817,6 +824,7 @@ async def _persist_stream_input_if_needed(
             [],
             streamed_result._state,
             wrapper=context_wrapper,
+            ownership_token=ownership_token,
         )
     streamed_result._stream_input_persisted = True
 
@@ -1075,6 +1083,18 @@ def _mark_response_hooks_started() -> None:
         streamed_result._event_queue.put_nowait(AgentUpdatedStreamEvent(new_agent=current_agent))
 
         prepared_input: str | list[TResponseInputItem]
+        # Ownership of the session history this run builds its request input
+        # from; captured right before the history read and threaded through
+        # every persist so compaction boundaries record only when no other
+        # writer interleaved. Resumed runs never read the session for their
+        # request input, so they carry no token and record no boundaries.
+        # The token is voided wherever the request provably stops covering
+        # the stored history: a resolved limit truncating the history read,
+        # a session input callback rebuilding the input, a
+        # call_model_input_filter rewriting a request, or a handoff input
+        # filter rewriting the accumulated history. Such runs record no
+        # boundaries either.
+        session_ownership_token: _SessionOwnershipToken | None = None
         if is_resumed_state and run_state is not None:
             prepared_input = normalize_resumed_input(starting_input)
             (
@@ -1094,6 +1114,10 @@ def _mark_response_hooks_started() -> None:
             streamed_result._stream_input_persisted = True
         else:
             server_manages_conversation = server_conversation_tracker is not None
+            if not server_manages_conversation:
+                session_ownership_token = await capture_session_ownership_token(
+                    session, wrapper=context_wrapper
+                )
             prepared_input, session_items_snapshot = await prepare_input_with_session(
                 starting_input,
                 session,
@@ -1103,6 +1127,7 @@ def _mark_response_hooks_started() -> None:
                 preserve_dropped_new_items=True,
                 reasoning_item_id_policy=resolved_reasoning_item_id_policy,
                 wrapper=context_wrapper,
+                ownership_token=session_ownership_token,
             )
             streamed_result.input = prepared_input
             streamed_result._original_input = copy_input_items(prepared_input)
@@ -1138,6 +1163,7 @@ async def _save_stream_items_with_count(
                 response_id=response_id,
                 update_persisted_count=True,
                 store=store_setting,
+                ownership_token=session_ownership_token,
             )
 
         async def _save_stream_items_without_count(
@@ -1152,6 +1178,7 @@ async def _save_stream_items_without_count(
                 response_id=response_id,
                 update_persisted_count=False,
                 store=store_setting,
+                ownership_token=session_ownership_token,
             )
 
         async def _save_max_turns_items(
@@ -1172,6 +1199,7 @@ async def _save_max_turns_items(
                 reasoning_item_id_policy=streamed_result._reasoning_item_id_policy,
                 store=store_setting,
                 wrapper=streamed_result.context_wrapper,
+                ownership_token=session_ownership_token,
             )
             streamed_result._current_turn_persisted_item_count += saved_count
     except BaseException:
@@ -1235,6 +1263,7 @@ async def _save_max_turns_items(
                                     run_config.model_settings
                                 ).store,
                                 wrapper=context_wrapper,
+                                ownership_token=session_ownership_token,
                             )
                         )
                         raise InputGuardrailTripwireTriggered(result)
@@ -1519,6 +1548,7 @@ async def _save_max_turns_items(
                     server_conversation_tracker=server_conversation_tracker,
                     store=store_setting,
                     wrapper=context_wrapper,
+                    ownership_token=session_ownership_token,
                 )
                 streamed_result._model_input_items.extend(admission_items)
                 streamed_result.new_items.extend(admission_items)
@@ -1595,6 +1625,7 @@ async def _save_max_turns_items(
                     session=session,
                     server_conversation_tracker=server_conversation_tracker,
                     context_wrapper=context_wrapper,
+                    ownership_token=session_ownership_token,
                 )
 
                 validated_output = validate_handler_final_output(
@@ -1696,6 +1727,7 @@ def _record_max_turns_handler_output(
                                         run_config.model_settings
                                     ).store,
                                     wrapper=context_wrapper,
+                                    ownership_token=session_ownership_token,
                                 )
                             )
                             raise InputGuardrailTripwireTriggered(result)
@@ -1791,6 +1823,7 @@ def _record_max_turns_handler_output(
                         on_response_accepted=_commit_pending_server_response,
                         on_response_hooks_started=_mark_response_hooks_started,
                         run_state=run_state,
+                        ownership_token=session_ownership_token,
                     )
                 finally:
                     if current_turn_span is not None:
@@ -2062,6 +2095,7 @@ async def run_single_turn_streamed(
     on_response_accepted: Callable[[ModelResponse, ProcessedResponse | None], bool] | None = None,
     on_response_hooks_started: Callable[[], None] | None = None,
     run_state: RunState[Any] | None = None,
+    ownership_token: _SessionOwnershipToken | None = None,
 ) -> SingleStepResult:
     """Run a single streamed turn and emit events as results arrive."""
     public_agent = bindings.public_agent
@@ -2164,6 +2198,7 @@ async def raise_if_input_guardrail_tripwire_known() -> None:
         context_wrapper=context_wrapper,
         input_items=input,
         system_instructions=system_prompt,
+        ownership_token=ownership_token,
     )
     if isinstance(filtered.input, list):
         filtered.input = deduplicate_input_items_preferring_latest(filtered.input)
@@ -2197,6 +2232,7 @@ async def raise_if_input_guardrail_tripwire_known() -> None:
         session=session,
         server_conversation_tracker=server_conversation_tracker,
         context_wrapper=context_wrapper,
+        ownership_token=ownership_token,
     )
 
     previous_response_id = (
@@ -2376,6 +2412,7 @@ async def check_input_guardrails_before_side_effects() -> None:
         after_invocation_validation=after_invocation_validation,
         before_side_effects=check_input_guardrails_before_side_effects,
         run_state=run_state,
+        ownership_token=ownership_token,
     )
 
     items_to_filter = session_items_for_turn(single_step_result)
@@ -2412,6 +2449,7 @@ async def run_single_turn(
     on_response_accepted: Callable[[ModelResponse, ProcessedResponse | None], bool] | None = None,
     on_response_hooks_started: Callable[[], None] | None = None,
     run_state: RunState[Any] | None = None,
+    ownership_token: _SessionOwnershipToken | None = None,
 ) -> SingleStepResult:
     """Run a single non-streaming turn of the agent loop."""
     public_agent = bindings.public_agent
@@ -2480,6 +2518,7 @@ async def run_single_turn(
         session_items_to_rewind=session_items_to_rewind,
         prompt_cache_key_resolver=prompt_cache_key_resolver,
         defer_llm_end_hooks=True,
+        ownership_token=ownership_token,
     )
 
     response_accepted = False
@@ -2519,6 +2558,7 @@ async def after_invocation_validation(
         server_manages_conversation=server_conversation_tracker is not None,
         after_invocation_validation=after_invocation_validation,
         run_state=run_state,
+        ownership_token=ownership_token,
     )
 
 
@@ -2539,6 +2579,7 @@ async def get_new_response(
     session_items_to_rewind: list[TResponseInputItem] | None = None,
     prompt_cache_key_resolver: PromptCacheKeyResolver | None = None,
     defer_llm_end_hooks: bool = False,
+    ownership_token: _SessionOwnershipToken | None = None,
 ) -> ModelResponse:
     """Call the model and return the raw response, handling retries and hooks."""
     public_agent = bindings.public_agent
@@ -2549,6 +2590,7 @@ async def get_new_response(
         context_wrapper=context_wrapper,
         input_items=input,
         system_instructions=system_prompt,
+        ownership_token=ownership_token,
     )
     if isinstance(filtered.input, list):
         filtered.input = deduplicate_input_items_preferring_latest(filtered.input)
diff --git a/src/agents/run_internal/session_persistence.py b/src/agents/run_internal/session_persistence.py
index bfe500b544..44315fcd58 100644
--- a/src/agents/run_internal/session_persistence.py
+++ b/src/agents/run_internal/session_persistence.py
@@ -12,7 +12,7 @@
 import json
 from collections import deque
 from collections.abc import Sequence
-from typing import Any, cast
+from typing import TYPE_CHECKING, Any, cast
 
 from .. import _debug
 from ..exceptions import UserError
@@ -63,8 +63,12 @@
 from .oai_conversation import OpenAIServerConversationTracker
 from .run_steps import NextStepInterruption, NextStepRunAgain, ProcessedResponse, SingleStepResult
 
+if TYPE_CHECKING:
+    from ..memory.openai_responses_compaction_session import _SessionOwnershipToken
+
 __all__ = [
     "admit_pending_input",
+    "capture_session_ownership_token",
     "commit_server_pending_input",
     "prepare_input_with_session",
     "persist_session_items_for_guardrail_trip",
@@ -92,6 +96,7 @@ async def admit_pending_input(
     server_conversation_tracker: OpenAIServerConversationTracker | None,
     store: bool | None,
     wrapper: RunContextWrapper[Any],
+    ownership_token: _SessionOwnershipToken | None = None,
 ) -> list[RunItem]:
     """Admit staged RunState input into the active conversation ownership boundary.
 
@@ -115,6 +120,7 @@ async def admit_pending_input(
             None,
             store=store,
             wrapper=wrapper,
+            ownership_token=ownership_token,
         )
     if server_conversation_tracker is None:
         run_state.clear_pending_input()
@@ -188,6 +194,33 @@ def retain_accepted_admissions(items: list[RunItem]) -> None:
     return True
 
 
+async def capture_session_ownership_token(
+    session: Session | None,
+    *,
+    wrapper: RunContextWrapper[Any] | None = None,
+) -> _SessionOwnershipToken | None:
+    """Capture a run's ownership of the session history it is about to read.
+
+    The runner calls this immediately before reading the session to build a
+    run's request input, and threads the token through that run's persists so
+    the session can tell whether any other writer landed in between. The
+    ordering matters: capturing before the read keeps the token conservative,
+    because a write that slips in between makes the token stale instead of
+    letting it claim items the request input never saw. Plain sessions record
+    no boundaries, so they get no token and keep their behavior unchanged.
+    """
+    if session is None or not is_openai_responses_compaction_aware_session(session):
+        return None
+    capture = getattr(session, "_capture_ownership_token", None)
+    if not callable(capture):
+        return None
+    wrapper = _get_session_wrapper(session, wrapper)
+    return cast(
+        "_SessionOwnershipToken | None",
+        await _call_session_method(capture, wrapper=wrapper),
+    )
+
+
 async def _session_get_items(
     session: Session,
     limit: int | None | object = _SESSION_LIMIT_UNSET,
@@ -207,11 +240,39 @@ async def _session_add_items(
     session: Session,
     items: list[TResponseInputItem],
     *,
+    response_id: str | None = None,
+    ownership_token: _SessionOwnershipToken | None = None,
     wrapper: RunContextWrapper[Any] | None = None,
 ) -> None:
-    """Append session items while preserving the legacy method call shape."""
+    """Append session items while preserving the legacy method call shape.
+
+    When a response id is supplied and the session is compaction aware, the
+    batch is routed through the session's boundary hook so the response id is
+    paired with its exact item count in one locked region. The hook records
+    that pairing only when the run's ownership token still matches the store,
+    proving nothing interleaved since the run read the session for its
+    request input.
+    """
     wrapper = _get_session_wrapper(session, wrapper)
+    add_items_for_response = (
+        getattr(session, "_add_items_for_response", None)
+        if response_id and is_openai_responses_compaction_aware_session(session)
+        else None
+    )
+    if callable(add_items_for_response):
+        await _call_session_method(
+            add_items_for_response,
+            items,
+            response_id=response_id,
+            ownership_token=ownership_token,
+            wrapper=wrapper,
+        )
+        return
     await _call_session_method(session.add_items, items, wrapper=wrapper)
+    if ownership_token is not None:
+        # This run appended the batch itself, so its token keeps pace with the
+        # store; only a write from outside the run may leave the counts apart.
+        ownership_token.advance(len(items))
 
 
 async def _session_pop_item(
@@ -326,6 +387,7 @@ async def prepare_input_with_session(
     preserve_dropped_new_items: bool = False,
     reasoning_item_id_policy: ReasoningItemIdPolicy | None = None,
     wrapper: RunContextWrapper[Any] | None = None,
+    ownership_token: _SessionOwnershipToken | None = None,
 ) -> tuple[str | list[TResponseInputItem], list[TResponseInputItem]]:
     """Prepare model input from session history plus the new turn input.
 
@@ -340,6 +402,15 @@ async def prepare_input_with_session(
     against deep-copied history and new-input lists, first by object identity and then by
     content frequency, so retries and custom merge strategies do not accidentally re-persist
     old history as fresh input.
+
+    ``ownership_token`` is the token the runner captured for this read. The
+    preparation below is where the run's request input can silently stop
+    covering the stored history: the history read can return a different
+    number of items than the token counted, and a ``session_input_callback``
+    rebuilds the input outright. Both cases void the token here, at the site
+    where the coverage is lost, so the run's persists record no compaction
+    boundary and its previous_response_id compactions skip instead of letting
+    a replacement delete prefix items the server never saw.
     """
 
     if session is None:
@@ -357,6 +428,17 @@ async def prepare_input_with_session(
         )
     else:
         history = await _session_get_items(session, wrapper=wrapper)
+    if ownership_token is not None and len(history) != ownership_token.count:
+        # The read that feeds the request returned a different number of items
+        # than the token counted in the store, so the request either leaves
+        # stored items out or was built over an interleaved write, and no
+        # recorded count could describe what the server saw. Asking instead
+        # whether a limit resolved here would miss one configured on a wrapped
+        # session, whose settings this decorator never exposes, and any other
+        # backend that windows its own reads. The raw read is compared, before
+        # normalization and dedupe legitimately merge items; a window that
+        # admits the whole store still matches it and still records.
+        ownership_token.invalidate()
     is_openai_conversation_session = isinstance(session, OpenAIConversationsSession)
     converted_history = [
         strip_internal_input_item_metadata(ensure_input_item_format(item)) for item in history
@@ -393,6 +475,12 @@ async def prepare_input_with_session(
                 f"Invalid `session_input_callback` value: {session_input_callback}. "
                 "Choose between `None` or a custom callable function."
             )
+        if ownership_token is not None:
+            # The callback below rebuilds the request input and may drop,
+            # reorder, or rewrite stored history, so item counts can no longer
+            # prove what the server saw. Void the token before invoking it so
+            # this run records no boundaries and its compactions skip.
+            ownership_token.invalidate()
         history_for_callback = copy.deepcopy(converted_history)
         new_items_for_callback = copy.deepcopy(new_input_list)
         # Keep the original history objects alive so their identities remain valid even if the
@@ -487,6 +575,7 @@ async def persist_session_items_for_guardrail_trip(
     run_state: RunState | None,
     store: bool | None = None,
     wrapper: RunContextWrapper[Any] | None = None,
+    ownership_token: _SessionOwnershipToken | None = None,
 ) -> list[TResponseInputItem] | None:
     """
     Persist input items when a guardrail tripwire is triggered.
@@ -508,6 +597,7 @@ async def persist_session_items_for_guardrail_trip(
         run_state,
         store=store,
         wrapper=wrapper,
+        ownership_token=ownership_token,
     )
     return updated_session_input_items
 
@@ -555,6 +645,7 @@ async def save_result_to_session(
     store: bool | None = None,
     wrapper: RunContextWrapper[Any] | None = None,
     resumed_write_state: RunState | None = None,
+    ownership_token: _SessionOwnershipToken | None = None,
 ) -> int:
     """
     Persist a turn to the session store, keeping track of what was already saved so retries
@@ -662,9 +753,24 @@ async def save_result_to_session(
                 resumed_write_state._current_turn_persisted_item_count + saved_run_items_count
             ),
         }
-        await resume_pending_session_write(resumed_write_state, session, wrapper=wrapper)
+        await resume_pending_session_write(
+            resumed_write_state, session, response_id=response_id, wrapper=wrapper
+        )
     else:
-        await _session_add_items(session, items_to_save, wrapper=wrapper)
+        # Persisting this response's batch is the only moment its response id
+        # and its exact local item boundary coincide, so pass the id along with
+        # the run's ownership token and let the boundary hook record the
+        # pairing when nothing interleaved since the run read the session for
+        # its request input. A later previous_response_id compaction preserves
+        # everything past the recorded boundary, including turns other runs
+        # append before that compaction takes its own snapshot.
+        await _session_add_items(
+            session,
+            items_to_save,
+            response_id=response_id,
+            ownership_token=ownership_token,
+            wrapper=wrapper,
+        )
 
     if run_state is not None:
         run_state._current_turn_persisted_item_count = already_persisted + saved_run_items_count
@@ -751,6 +857,7 @@ async def resume_pending_session_write(
     run_state: RunState,
     session: Session | None,
     *,
+    response_id: str | None = None,
     wrapper: RunContextWrapper[Any] | None = None,
 ) -> None:
     """Settle a resumed output batch before allowing further model work.
@@ -758,6 +865,12 @@ async def resume_pending_session_write(
     The application must supply the original backend and serialize access to its history,
     including independently restored RunState copies. Session has no distributed compare-and-swap
     or backend identity contract. A changed tail is not repaired or searched for similar items.
+
+    When a response_id is supplied and this call performs the append itself, the batch is
+    persisted through the compaction boundary hook. Resumed writes carry no ownership
+    token, because the resumed run never read the session to build its request input, so
+    the hook appends the batch without recording a boundary and a compaction keyed on
+    the response skips instead of guessing at its coverage.
     """
     pending = run_state._pending_session_write
     if pending is None:
@@ -801,7 +914,15 @@ def digests(items: Sequence[TResponseInputItem]) -> list[str]:
             append = unchanged
         if append:
             # Backends may retain or transform their input; the durable checkpoint stays detached.
-            await _session_add_items(session, copy.deepcopy(pending["items"]), wrapper=wrapper)
+            items_to_append = copy.deepcopy(pending["items"])
+            # The append goes through the boundary hook, but with no ownership
+            # token it records nothing: a resumed run never read the session to
+            # build its request input, so no count read here can prove what the
+            # response covers, and other writers may have appended since an
+            # earlier attempt.
+            await _session_add_items(
+                session, items_to_append, response_id=response_id, wrapper=wrapper
+            )
         run_state._current_turn_persisted_item_count = pending["persisted_count"]
         run_state._pending_session_write = None
     finally:
diff --git a/src/agents/run_internal/turn_preparation.py b/src/agents/run_internal/turn_preparation.py
index da531afb6a..56c131cc68 100644
--- a/src/agents/run_internal/turn_preparation.py
+++ b/src/agents/run_internal/turn_preparation.py
@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import inspect
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 from ..agent import Agent
 from ..agent_output import AgentOutputSchema, AgentOutputSchemaBase
@@ -19,6 +19,9 @@
 from ..util import _error_tracing
 from ..util._asyncio_tasks import gather_with_cancel
 
+if TYPE_CHECKING:
+    from ..memory.openai_responses_compaction_session import _SessionOwnershipToken
+
 __all__ = [
     "validate_run_hooks",
     "maybe_filter_model_input",
@@ -55,14 +58,29 @@ async def maybe_filter_model_input(
     context_wrapper: RunContextWrapper[TContext],
     input_items: list[TResponseInputItem],
     system_instructions: str | None,
+    ownership_token: _SessionOwnershipToken | None = None,
 ) -> ModelInputData:
-    """Apply optional call_model_input_filter to modify model input."""
+    """Apply optional call_model_input_filter to modify model input.
+
+    ``ownership_token`` is the token the runner captured when it read the
+    session for this run's request input. The filter below can drop stored
+    history from the request it returns, so once it is invoked no item count
+    can prove what the server saw. The token is voided here, at the site
+    where the coverage is lost, unconditionally and with no comparison of
+    the filter's output, mirroring the session input callback: the run then
+    records no compaction boundaries and its previous_response_id
+    compactions skip instead of letting a replacement delete prefix items
+    the server never saw.
+    """
     effective_instructions = system_instructions
     effective_input: list[TResponseInputItem] = input_items
 
     if run_config.call_model_input_filter is None:
         return ModelInputData(input=effective_input, instructions=effective_instructions)
 
+    if ownership_token is not None:
+        ownership_token.invalidate()
+
     try:
         model_input = ModelInputData(
             input=effective_input.copy(),
diff --git a/src/agents/run_internal/turn_resolution.py b/src/agents/run_internal/turn_resolution.py
index ae9fec7619..e79d9c1859 100644
--- a/src/agents/run_internal/turn_resolution.py
+++ b/src/agents/run_internal/turn_resolution.py
@@ -4,7 +4,7 @@
 from collections.abc import Awaitable, Callable, Container, Mapping, Sequence
 from copy import deepcopy
 from dataclasses import replace
-from typing import Any, Literal, cast
+from typing import TYPE_CHECKING, Any, Literal, cast
 
 from openai.types.responses import (
     ResponseCompactionItem,
@@ -187,6 +187,9 @@
 )
 from .turn_preparation import get_handoffs, get_output_schema
 
+if TYPE_CHECKING:
+    from ..memory.openai_responses_compaction_session import _SessionOwnershipToken
+
 _DEFAULT_NEST_HANDOFF_HISTORY = nest_handoff_history
 
 __all__ = [
@@ -540,8 +543,18 @@ async def execute_handoffs(
     tool_input_guardrail_results: list[ToolInputGuardrailResult] | None = None,
     tool_output_guardrail_results: list[ToolOutputGuardrailResult] | None = None,
     handoff_output_committer: Callable[[HandoffOutputItem, Agent[Any]], None] | None = None,
+    ownership_token: _SessionOwnershipToken | None = None,
 ) -> SingleStepResult:
-    """Execute a handoff and prepare the next turn for the new agent."""
+    """Execute a handoff and prepare the next turn for the new agent.
+
+    ``ownership_token`` is the running run's claim over the session history
+    its request input was built from. A handoff input filter and nested
+    handoff history both rewrite the accumulated history the next requests
+    are built from while the session keeps every persisted turn, so applying
+    either voids the token: later persists then record no compaction
+    boundaries and their compactions skip instead of letting a replacement
+    delete stored items the rewritten requests never carried.
+    """
 
     def nest_history(
         data: HandoffInputData,
@@ -661,6 +674,14 @@ def nest_history(
             )
 
         if input_filter is not None and handoff_input_data is not None:
+            if ownership_token is not None:
+                # The filter below may drop or rewrite stored turns in the
+                # history the next requests carry, so from here on no item
+                # count can prove what the server saw. Void the token before
+                # invoking it, unconditionally and with no comparison of the
+                # filter's output, mirroring the session input callback and
+                # call_model_input_filter precedents.
+                ownership_token.invalidate()
             filter_name = getattr(input_filter, "__qualname__", repr(input_filter))
             from_agent = getattr(public_agent, "name", public_agent.__class__.__name__)
             to_agent = getattr(new_agent, "name", new_agent.__class__.__name__)
@@ -710,6 +731,16 @@ def nest_history(
                 _get_nested_history_owned_items(filtered, source_data=handoff_input_data)
             )
         elif should_nest_history and handoff_input_data is not None:
+            if ownership_token is not None:
+                # Nesting folds the accumulated history the next requests are
+                # built from into a rendered transcript while the session
+                # keeps the original turns, so from here on no item count can
+                # prove what the server saw. Void the token before nesting,
+                # unconditionally and with no comparison of the nested
+                # output, mirroring the input filter branch above: the
+                # default nesting moves every prior item into wrapper text,
+                # and a custom history mapper may drop anything outright.
+                ownership_token.invalidate()
             nested, nested_history_owned_items = nest_history(
                 handoff_input_data,
                 run_config.handoff_history_mapper,
@@ -796,6 +827,7 @@ async def execute_tools_and_side_effects(
     server_manages_conversation: bool = False,
     precomputed_skipped_raw_item_ids: set[int] | None = None,
     run_state: RunState[Any] | None = None,
+    ownership_token: _SessionOwnershipToken | None = None,
 ) -> SingleStepResult:
     """Run one turn of the loop, coordinating tools, approvals, guardrails, and handoffs."""
     public_agent = bindings.public_agent
@@ -932,6 +964,7 @@ def _commit_accepted_response_tool_output(item: RunItem) -> None:
             server_manages_conversation=server_manages_conversation,
             tool_input_guardrail_results=tool_input_guardrail_results,
             tool_output_guardrail_results=tool_output_guardrail_results,
+            ownership_token=ownership_token,
         )
 
     tool_final_output = await _maybe_finalize_from_tool_results(
@@ -3553,6 +3586,7 @@ async def get_single_step_result_from_response(
     | None = None,
     before_side_effects: Callable[[], Awaitable[None]] | None = None,
     run_state: RunState[Any] | None = None,
+    ownership_token: _SessionOwnershipToken | None = None,
 ) -> SingleStepResult:
     item_agent = bindings.public_agent
     try:
@@ -3615,4 +3649,5 @@ async def get_single_step_result_from_response(
         server_manages_conversation=server_manages_conversation,
         precomputed_skipped_raw_item_ids=skipped_raw_item_ids,
         run_state=run_state,
+        ownership_token=ownership_token,
     )
diff --git a/tests/memory/test_openai_responses_compaction_session.py b/tests/memory/test_openai_responses_compaction_session.py
index 5519228ea6..002513d8fd 100644
--- a/tests/memory/test_openai_responses_compaction_session.py
+++ b/tests/memory/test_openai_responses_compaction_session.py
@@ -15,9 +15,10 @@
 )
 
 import agents._debug as _debug
-from agents import Agent, Runner
-from agents.items import TResponseInputItem
+from agents import Agent, HandoffInputData, RunConfig, Runner, handoff
+from agents.items import RunItem, TResponseInputItem
 from agents.memory import (
+    OpenAIResponsesCompactionArgs,
     OpenAIResponsesCompactionSession,
     Session,
     SessionSettings,
@@ -30,15 +31,38 @@
     is_openai_model_name,
     select_compaction_candidate_items,
 )
+from agents.run import CallModelData, ModelInputData
 from agents.run_internal.items import (
     TOOL_CALL_SESSION_DESCRIPTION_KEY,
     TOOL_CALL_SESSION_TITLE_KEY,
 )
-from agents.testing import ScriptedModel
-from tests.test_responses import get_function_tool, get_function_tool_call, get_text_message
+from agents.run_internal.session_persistence import save_result_to_session
+from agents.testing import ModelStep, ScriptedModel
+from tests.test_responses import (
+    get_function_tool,
+    get_function_tool_call,
+    get_handoff_tool_call,
+    get_text_message,
+)
 from tests.utils.simple_session import SimpleListSession
 
 
+async def persist_response_batch(
+    session: OpenAIResponsesCompactionSession,
+    items: list[TResponseInputItem],
+    response_id: str,
+) -> None:
+    """Persist a batch the way the runner does on a clean turn.
+
+    Ownership is captured at the request input read and nothing interleaves
+    before the persist, so the boundary hook records the response's coverage.
+    """
+    ownership_token = await session._capture_ownership_token()
+    await session._add_items_for_response(
+        items, response_id=response_id, ownership_token=ownership_token
+    )
+
+
 class TestIsOpenAIModelName:
     def test_gpt_models(self) -> None:
         assert is_openai_model_name("gpt-4o") is True
@@ -1495,6 +1519,45 @@ async def test_compaction_runs_during_runner_flow(self) -> None:
         items = await session.get_items()
         assert any(isinstance(item, dict) and item.get("type") == "compaction" for item in items)
 
+    @pytest.mark.asyncio
+    async def test_streamed_run_records_boundary_and_compacts(self) -> None:
+        """The streamed path threads ownership so boundaries record and compaction lands."""
+        underlying = SimpleListSession()
+        compacted = SimpleNamespace(
+            output=[{"type": "compaction", "summary": "compacted"}],
+        )
+        mock_client = MagicMock()
+        mock_client.responses.compact = AsyncMock(return_value=compacted)
+
+        session = OpenAIResponsesCompactionSession(
+            session_id="stream-clean",
+            underlying_session=underlying,
+            client=mock_client,
+            compaction_mode="previous_response_id",
+            should_trigger_compaction=lambda ctx: True,
+        )
+
+        model = ScriptedModel(
+            steps=[ModelStep(output=[get_text_message("ok")], response_id="resp_stream")]
+        )
+        worker = Agent(name="stream_worker", model=model)
+
+        result = Runner.run_streamed(worker, "hello", session=session)
+        async for _ in result.stream_events():
+            pass
+
+        mock_client.responses.compact.assert_awaited_once()
+        assert mock_client.responses.compact.call_args.kwargs["previous_response_id"] == (
+            "resp_stream"
+        )
+        # The boundary recorded through the streamed persists and was
+        # translated onto the replaced history; the snapshot fallback never
+        # records, so this proves the ownership token reached the hook.
+        assert session._response_boundaries_ever_recorded is True
+        assert session._response_boundaries == {"resp_stream": 1}
+        items = await session.get_items()
+        assert any(isinstance(item, dict) and item.get("type") == "compaction" for item in items)
+
     @pytest.mark.asyncio
     async def test_compaction_skips_when_tool_outputs_present(self) -> None:
         underlying = SimpleListSession()
@@ -1780,6 +1843,2118 @@ async def test_run_compaction_keeps_ids_when_reasoning_present(self) -> None:
         assert assistant_items[0]["id"] == "msg_aaa"
 
 
+class TestCompactionConcurrentMutations:
+    """Wrapper mutations racing the in-flight responses.compact call."""
+
+    def create_gated_compact_client(
+        self,
+        output_items: list[TResponseInputItem],
+        compact_entered: asyncio.Event,
+        release_compact: asyncio.Event,
+    ) -> MagicMock:
+        """Build a client whose compact call blocks until the test releases it."""
+        mock_compact_response = MagicMock()
+        mock_compact_response.output = output_items
+
+        async def gated_compact(**kwargs: Any) -> MagicMock:
+            compact_entered.set()
+            await release_compact.wait()
+            return mock_compact_response
+
+        mock_client = MagicMock()
+        mock_client.responses.compact = AsyncMock(side_effect=gated_compact)
+        return mock_client
+
+    @pytest.mark.asyncio
+    async def test_concurrent_add_items_during_forced_compaction_survives(self) -> None:
+        """Items added while responses.compact is in flight must survive replacement."""
+        history: list[TResponseInputItem] = [
+            cast(TResponseInputItem, {"type": "message", "role": "assistant", "content": f"msg{i}"})
+            for i in range(3)
+        ]
+        compacted_item = cast(TResponseInputItem, {"type": "compaction", "summary": "compacted"})
+        concurrent_user_item = cast(
+            TResponseInputItem,
+            {"type": "message", "role": "user", "content": "written mid-flight"},
+        )
+        concurrent_assistant_item = cast(
+            TResponseInputItem,
+            {"type": "message", "role": "assistant", "content": "reply mid-flight"},
+        )
+
+        underlying = SimpleListSession(history=history)
+        compact_entered = asyncio.Event()
+        release_compact = asyncio.Event()
+        session = OpenAIResponsesCompactionSession(
+            session_id="concurrent-add",
+            underlying_session=underlying,
+            client=self.create_gated_compact_client(
+                [compacted_item], compact_entered, release_compact
+            ),
+            compaction_mode="input",
+        )
+
+        compaction_task = asyncio.create_task(session.run_compaction({"force": True}))
+        try:
+            await compact_entered.wait()
+            await session.add_items([concurrent_user_item, concurrent_assistant_item])
+            assert await underlying.get_items() == [
+                *history,
+                concurrent_user_item,
+                concurrent_assistant_item,
+            ]
+        finally:
+            release_compact.set()
+        await compaction_task
+
+        expected = [compacted_item, concurrent_user_item, concurrent_assistant_item]
+        assert await session.get_items() == expected
+        assert session._session_items == expected
+        assert session._compaction_candidate_items == [concurrent_assistant_item]
+
+    @pytest.mark.asyncio
+    async def test_concurrent_add_items_during_threshold_compaction_survives(self) -> None:
+        """The default threshold trigger path must also preserve mid-flight writes."""
+        history: list[TResponseInputItem] = [
+            cast(TResponseInputItem, {"type": "message", "role": "assistant", "content": f"msg{i}"})
+            for i in range(DEFAULT_COMPACTION_THRESHOLD)
+        ]
+        compacted_item = cast(TResponseInputItem, {"type": "compaction", "summary": "compacted"})
+        concurrent_item = cast(
+            TResponseInputItem,
+            {"type": "message", "role": "user", "content": "written mid-flight"},
+        )
+
+        underlying = SimpleListSession(history=history)
+        compact_entered = asyncio.Event()
+        release_compact = asyncio.Event()
+        mock_client = self.create_gated_compact_client(
+            [compacted_item], compact_entered, release_compact
+        )
+        session = OpenAIResponsesCompactionSession(
+            session_id="concurrent-threshold",
+            underlying_session=underlying,
+            client=mock_client,
+            compaction_mode="input",
+        )
+
+        compaction_task = asyncio.create_task(session.run_compaction())
+        try:
+            await compact_entered.wait()
+            await session.add_items([concurrent_item])
+        finally:
+            release_compact.set()
+        await compaction_task
+
+        mock_client.responses.compact.assert_called_once()
+        assert await session.get_items() == [compacted_item, concurrent_item]
+        assert session._session_items == [compacted_item, concurrent_item]
+
+    @pytest.mark.asyncio
+    async def test_clear_session_during_compaction_is_not_resurrected(
+        self, caplog: pytest.LogCaptureFixture
+    ) -> None:
+        """A clear_session issued mid-flight must win over the stale snapshot."""
+        history: list[TResponseInputItem] = [
+            cast(TResponseInputItem, {"type": "message", "role": "assistant", "content": f"msg{i}"})
+            for i in range(3)
+        ]
+        compacted_item = cast(TResponseInputItem, {"type": "compaction", "summary": "compacted"})
+
+        underlying = SimpleListSession(history=history)
+        compact_entered = asyncio.Event()
+        release_compact = asyncio.Event()
+        session = OpenAIResponsesCompactionSession(
+            session_id="concurrent-clear",
+            underlying_session=underlying,
+            client=self.create_gated_compact_client(
+                [compacted_item], compact_entered, release_compact
+            ),
+            compaction_mode="input",
+        )
+
+        with caplog.at_level(logging.WARNING, logger="openai-agents.openai.compaction"):
+            compaction_task = asyncio.create_task(session.run_compaction({"force": True}))
+            try:
+                await compact_entered.wait()
+                await session.clear_session()
+                assert await underlying.get_items() == []
+            finally:
+                release_compact.set()
+            await compaction_task
+
+        assert await session.get_items() == []
+        candidates, session_items = await session._ensure_compaction_candidates()
+        assert candidates == []
+        assert session_items == []
+        assert "Skipped compaction replacement" in caplog.text
+
+    @pytest.mark.asyncio
+    async def test_replacement_preserves_metadata_tail_across_storage_round_trip(
+        self, tmp_path
+    ) -> None:
+        """The snapshot prefix check must match items that round-tripped through storage."""
+        history: list[TResponseInputItem] = [
+            cast(TResponseInputItem, {"type": "message", "role": "assistant", "content": "reply"}),
+            cast(
+                TResponseInputItem,
+                {
+                    "type": "function_call",
+                    "call_id": "call_history",
+                    "name": "lookup",
+                    "arguments": "{}",
+                    TOOL_CALL_SESSION_DESCRIPTION_KEY: "Lookup private records.",
+                },
+            ),
+        ]
+        concurrent_item = cast(
+            TResponseInputItem,
+            {
+                "type": "function_call",
+                "call_id": "call_tail",
+                "name": "lookup",
+                "arguments": "{}",
+                TOOL_CALL_SESSION_TITLE_KEY: "Lookup",
+            },
+        )
+        compacted_item = cast(TResponseInputItem, {"type": "compaction", "summary": "compacted"})
+
+        underlying = SQLiteSession("round-trip", str(tmp_path / "compaction_round_trip.db"))
+        compact_entered = asyncio.Event()
+        release_compact = asyncio.Event()
+        session = OpenAIResponsesCompactionSession(
+            session_id="round-trip",
+            underlying_session=underlying,
+            client=self.create_gated_compact_client(
+                [compacted_item], compact_entered, release_compact
+            ),
+            compaction_mode="input",
+        )
+
+        # Warm the caches, then add through the wrapper so the cached items hold the
+        # normalized shapes while the underlying store keeps the raw metadata keys.
+        await session._ensure_compaction_candidates()
+        await session.add_items(history)
+        stored_history = [cast(dict, item) for item in await underlying.get_items()]
+        assert any(TOOL_CALL_SESSION_DESCRIPTION_KEY in item for item in stored_history)
+        assert all(
+            TOOL_CALL_SESSION_DESCRIPTION_KEY not in cast(dict, item)
+            for item in session._session_items or []
+        )
+
+        compaction_task = asyncio.create_task(session.run_compaction({"force": True}))
+        try:
+            await compact_entered.wait()
+            await session.add_items([concurrent_item])
+        finally:
+            release_compact.set()
+        await compaction_task
+
+        assert await underlying.get_items() == [compacted_item, concurrent_item]
+        assert session._session_items is not None
+        cached_tail = cast(dict, session._session_items[1])
+        assert cached_tail.get("call_id") == "call_tail"
+        assert TOOL_CALL_SESSION_TITLE_KEY not in cached_tail
+
+    @pytest.mark.asyncio
+    async def test_clear_session_with_empty_snapshot_is_not_repopulated(self) -> None:
+        """A clear during flight must hold even when the snapshot itself was empty.
+
+        With previous_response_id compaction the local session can be empty when
+        the request starts, so the post flight prefix check compares two empty
+        lists and cannot see the clear. The destructive generation counter can.
+        """
+        compacted_item = cast(TResponseInputItem, {"type": "compaction", "summary": "compacted"})
+        underlying = SimpleListSession(history=[])
+        compact_entered = asyncio.Event()
+        release_compact = asyncio.Event()
+        session = OpenAIResponsesCompactionSession(
+            session_id="clear-empty-baseline",
+            underlying_session=underlying,
+            client=self.create_gated_compact_client(
+                [compacted_item], compact_entered, release_compact
+            ),
+            compaction_mode="previous_response_id",
+        )
+
+        compaction_task = asyncio.create_task(
+            session.run_compaction({"force": True, "response_id": "resp_baseline"})
+        )
+        try:
+            await compact_entered.wait()
+            await session.clear_session()
+        finally:
+            release_compact.set()
+        await compaction_task
+
+        assert await underlying.get_items() == []
+        assert await session.get_items() == []
+
+    @pytest.mark.asyncio
+    async def test_compact_uses_response_id_captured_before_lock_wait(self) -> None:
+        """The compact call must use the response id its mode was resolved from.
+
+        While one run_compaction waits on the mutation lock, a second call can
+        overwrite the shared response id at entry. The waiter must not pick up
+        the newer id after resuming.
+        """
+        compacted_item = cast(TResponseInputItem, {"type": "compaction", "summary": "compacted"})
+        underlying = SimpleListSession(history=[])
+        mock_compact_response = MagicMock()
+        mock_compact_response.output = [compacted_item]
+        mock_client = MagicMock()
+        mock_client.responses.compact = AsyncMock(return_value=mock_compact_response)
+        session = OpenAIResponsesCompactionSession(
+            session_id="captured-response-id",
+            underlying_session=underlying,
+            client=mock_client,
+            compaction_mode="previous_response_id",
+        )
+
+        await session._mutation_lock.acquire()
+        try:
+            first = asyncio.create_task(
+                session.run_compaction({"force": True, "response_id": "resp_first"})
+            )
+            await asyncio.sleep(0)
+            await asyncio.sleep(0)
+            second = asyncio.create_task(
+                session.run_compaction({"force": True, "response_id": "resp_second"})
+            )
+            await asyncio.sleep(0)
+            await asyncio.sleep(0)
+            # The second call already overwrote the shared id while the first waits.
+            assert session._response_id == "resp_second"
+        finally:
+            session._mutation_lock.release()
+        await first
+        await second
+
+        sent_ids = [
+            call.kwargs["previous_response_id"]
+            for call in mock_client.responses.compact.call_args_list
+        ]
+        assert sent_ids == ["resp_first", "resp_second"]
+
+    @pytest.mark.asyncio
+    async def test_overlapping_compactions_with_empty_snapshots_keep_one_output(self) -> None:
+        """The second of two overlapping compactions must not absorb the first.
+
+        Both calls snapshot an empty session, so the prefix check alone cannot
+        tell the first replacement from a concurrent append. Without the
+        generation bump on replacement, the second call persists both outputs
+        concatenated.
+        """
+        first_output = cast(TResponseInputItem, {"type": "compaction", "summary": "first"})
+        second_output = cast(TResponseInputItem, {"type": "compaction", "summary": "second"})
+        underlying = SimpleListSession(history=[])
+
+        entered: list[asyncio.Event] = [asyncio.Event(), asyncio.Event()]
+        release: list[asyncio.Event] = [asyncio.Event(), asyncio.Event()]
+        outputs = [[first_output], [second_output]]
+        call_index = 0
+
+        async def gated_compact(**kwargs: Any) -> MagicMock:
+            nonlocal call_index
+            index = call_index
+            call_index += 1
+            entered[index].set()
+            await release[index].wait()
+            response = MagicMock()
+            response.output = outputs[index]
+            return response
+
+        mock_client = MagicMock()
+        mock_client.responses.compact = AsyncMock(side_effect=gated_compact)
+        session = OpenAIResponsesCompactionSession(
+            session_id="overlapping-empty",
+            underlying_session=underlying,
+            client=mock_client,
+            compaction_mode="previous_response_id",
+        )
+
+        first = asyncio.create_task(
+            session.run_compaction({"force": True, "response_id": "resp_overlap"})
+        )
+        await entered[0].wait()
+        second = asyncio.create_task(
+            session.run_compaction({"force": True, "response_id": "resp_overlap"})
+        )
+        await entered[1].wait()
+
+        release[0].set()
+        await first
+        release[1].set()
+        await second
+
+        # The first replacement landed; the second detected the rewrite and skipped.
+        assert await underlying.get_items() == [first_output]
+        assert await session.get_items() == [first_output]
+
+    @pytest.mark.asyncio
+    async def test_turn_persisted_before_late_snapshot_survives_replacement(self) -> None:
+        """A turn landing between a response's persisted batch and its snapshot survives.
+
+        Ordering under test: run A's response batch is persisted, run B persists
+        its own turn, and only then does run A snapshot and compact. The compact
+        call with previous_response_id covers history through A's batch only, so
+        the replacement must preserve B's turn even though A's late snapshot
+        already contains it and the prefix and generation checks see no rewrite.
+        """
+        compacted_item = cast(TResponseInputItem, {"type": "compaction", "summary": "compacted"})
+        a_turn = cast(
+            TResponseInputItem, {"type": "message", "role": "assistant", "content": "turn a"}
+        )
+        b_turn = cast(
+            TResponseInputItem, {"type": "message", "role": "assistant", "content": "turn b"}
+        )
+
+        underlying = SimpleListSession(history=[])
+        compact_entered = asyncio.Event()
+        release_compact = asyncio.Event()
+        mock_client = self.create_gated_compact_client(
+            [compacted_item], compact_entered, release_compact
+        )
+        session = OpenAIResponsesCompactionSession(
+            session_id="persist-boundary",
+            underlying_session=underlying,
+            client=mock_client,
+            compaction_mode="previous_response_id",
+            should_trigger_compaction=lambda context: context["response_id"] == "resp_a",
+        )
+
+        class StubMessageRunItem:
+            def __init__(self, payload: TResponseInputItem) -> None:
+                self.raw_item = payload
+                self.type = "message_output_item"
+
+            def to_input_item(self) -> TResponseInputItem:
+                return self.raw_item
+
+        real_run_compaction = session.run_compaction
+        run_a_compaction_requested = asyncio.Event()
+        release_run_a_compaction = asyncio.Event()
+
+        async def paused_run_compaction(args: OpenAIResponsesCompactionArgs | None = None) -> None:
+            # Hold back only the first call, which belongs to run A. Run B's call
+            # passes straight through and is declined by the decision hook.
+            if not run_a_compaction_requested.is_set():
+                run_a_compaction_requested.set()
+                await release_run_a_compaction.wait()
+            await real_run_compaction(args)
+
+        session.run_compaction = paused_run_compaction  # type: ignore[method-assign]
+
+        token_a = await session._capture_ownership_token()
+        save_a = asyncio.create_task(
+            save_result_to_session(
+                session,
+                [],
+                [cast(RunItem, StubMessageRunItem(a_turn))],
+                None,
+                response_id="resp_a",
+                ownership_token=token_a,
+            )
+        )
+        await run_a_compaction_requested.wait()
+        # Run A's batch is persisted and paired with its boundary, but its
+        # compaction has not snapshotted yet. Run B's turn lands now; B read
+        # the session after A's batch was stored, so B's ownership holds and
+        # its boundary records too.
+        token_b = await session._capture_ownership_token()
+        await save_result_to_session(
+            session,
+            [],
+            [cast(RunItem, StubMessageRunItem(b_turn))],
+            None,
+            response_id="resp_b",
+            ownership_token=token_b,
+        )
+        assert await underlying.get_items() == [a_turn, b_turn]
+
+        release_run_a_compaction.set()
+        await compact_entered.wait()
+        release_compact.set()
+        await save_a
+
+        mock_client.responses.compact.assert_awaited_once()
+        assert mock_client.responses.compact.call_args.kwargs["previous_response_id"] == "resp_a"
+        assert await underlying.get_items() == [compacted_item, b_turn]
+        assert await session.get_items() == [compacted_item, b_turn]
+
+    @pytest.mark.asyncio
+    async def test_turn_interleaved_during_request_survives_compaction(self) -> None:
+        """A turn persisted while another run's request is in flight must survive.
+
+        Ordering under test: run A reads the session and its request goes
+        out, run B completes a full turn against the same session while A
+        waits, and A's batch persists after B's. The stored history now
+        holds B's turn below the point where A's batch lands, but the
+        server side history through A's response cannot contain a turn
+        written after A's request was sent, so a boundary taken from the
+        count at persist time would let A's replacement delete B's turn.
+        A must record nothing, and its compaction must skip before the
+        billed call, leaving both turns stored.
+        """
+        underlying = SimpleListSession()
+        compacted = SimpleNamespace(
+            output=[{"type": "compaction", "summary": "through a"}],
+        )
+        mock_client = MagicMock()
+        mock_client.responses.compact = AsyncMock(return_value=compacted)
+        session = OpenAIResponsesCompactionSession(
+            session_id="interleaved-runs",
+            underlying_session=underlying,
+            client=mock_client,
+            compaction_mode="previous_response_id",
+            should_trigger_compaction=lambda context: context["response_id"] == "resp_a",
+        )
+
+        request_a_in_flight = asyncio.Event()
+        release_request_a = asyncio.Event()
+
+        async def hold_request_a(call: Any) -> ModelStep:
+            request_a_in_flight.set()
+            await release_request_a.wait()
+            return ModelStep(output=[get_text_message("reply a")], response_id="resp_a")
+
+        worker_a = Agent(
+            name="worker_a",
+            model=ScriptedModel(steps=[ModelStep.respond(hold_request_a)]),
+        )
+        worker_b = Agent(
+            name="worker_b",
+            model=ScriptedModel(
+                steps=[ModelStep(output=[get_text_message("reply b")], response_id="resp_b")]
+            ),
+        )
+
+        run_a = asyncio.create_task(Runner.run(worker_a, "hello a", session=session))
+        await request_a_in_flight.wait()
+        # B's full turn, input and reply, lands while A's request is in flight.
+        await Runner.run(worker_b, "hello b", session=session)
+        release_request_a.set()
+        await run_a
+
+        stored_text = str(await underlying.get_items())
+        # B's turn must still be stored; deleting it is the data loss.
+        assert "hello b" in stored_text
+        assert "reply b" in stored_text
+        assert "hello a" in stored_text
+        assert "reply a" in stored_text
+        # A's compaction skipped before the billed call instead of replacing
+        # history it does not cover.
+        mock_client.responses.compact.assert_not_called()
+        assert "resp_a" not in session._response_boundaries
+        # B read the session after A's input was stored and nothing
+        # interleaved before B's persist, so B's boundary recorded: A's
+        # input, B's input, and B's reply.
+        assert session._response_boundaries == {"resp_b": 3}
+
+    @pytest.mark.asyncio
+    async def test_streamed_turn_interleaved_during_request_survives_compaction(self) -> None:
+        """A turn persisted while a streamed run's request is in flight must survive.
+
+        The streamed variant of the ordering above: run A streams, run B
+        completes a full turn against the same session while A's request
+        is held open, and A's batch persists after B's. A's ownership went
+        stale, so A records nothing and its compaction skips before the
+        billed call, leaving both turns stored. This pins the ownership
+        threading through the streaming loop's own persist helpers.
+        """
+        underlying = SimpleListSession()
+        compacted = SimpleNamespace(
+            output=[{"type": "compaction", "summary": "through a"}],
+        )
+        mock_client = MagicMock()
+        mock_client.responses.compact = AsyncMock(return_value=compacted)
+        session = OpenAIResponsesCompactionSession(
+            session_id="interleaved-streamed-runs",
+            underlying_session=underlying,
+            client=mock_client,
+            compaction_mode="previous_response_id",
+            should_trigger_compaction=lambda context: context["response_id"] == "resp_a",
+        )
+
+        request_a_in_flight = asyncio.Event()
+        release_request_a = asyncio.Event()
+
+        async def hold_request_a(call: Any) -> ModelStep:
+            request_a_in_flight.set()
+            await release_request_a.wait()
+            return ModelStep(output=[get_text_message("reply a")], response_id="resp_a")
+
+        worker_a = Agent(
+            name="worker_a",
+            model=ScriptedModel(steps=[ModelStep.respond(hold_request_a)]),
+        )
+        worker_b = Agent(
+            name="worker_b",
+            model=ScriptedModel(
+                steps=[ModelStep(output=[get_text_message("reply b")], response_id="resp_b")]
+            ),
+        )
+
+        async def stream_run_a() -> None:
+            result = Runner.run_streamed(worker_a, "hello a", session=session)
+            async for _ in result.stream_events():
+                pass
+
+        run_a = asyncio.create_task(stream_run_a())
+        await request_a_in_flight.wait()
+        # B's full turn, input and reply, lands while A's stream is held open.
+        await Runner.run(worker_b, "hello b", session=session)
+        release_request_a.set()
+        await run_a
+
+        stored_text = str(await underlying.get_items())
+        # B's turn must still be stored; deleting it is the data loss.
+        assert "hello b" in stored_text
+        assert "reply b" in stored_text
+        assert "hello a" in stored_text
+        assert "reply a" in stored_text
+        # A's compaction skipped before the billed call instead of replacing
+        # history it does not cover.
+        mock_client.responses.compact.assert_not_called()
+        assert "resp_a" not in session._response_boundaries
+        # B read the session after A's input was stored and nothing
+        # interleaved before B's persist, so B's boundary recorded: A's
+        # input, B's input, and B's reply.
+        assert session._response_boundaries == {"resp_b": 3}
+
+    @pytest.mark.asyncio
+    async def test_first_interleaved_persist_arms_skip_gate(
+        self, caplog: pytest.LogCaptureFixture
+    ) -> None:
+        """A first ever persist with stale ownership must arm the skip gate.
+
+        Ordering under test: a fresh wrapper over an empty store, run A's
+        request goes out, and a plain add_items call lands another
+        writer's item while A waits. A's persist records nothing because
+        its ownership went stale, and nothing was ever recorded on this
+        wrapper before, so without the stale persist arming the gate the
+        compaction keyed on A's response would fall through to the
+        snapshot fallback, classify the interleaved item as covered, and
+        delete it in the replacement. The stale token must arm the gate
+        by itself: A's compaction skips before the billed call and every
+        stored item survives. The interleaved write goes through plain
+        add_items with no token, so it neither records nor arms anything
+        on its own.
+        """
+        interleaved_item = cast(
+            TResponseInputItem, {"type": "message", "role": "assistant", "content": "turn b"}
+        )
+        underlying = SimpleListSession()
+        compacted = SimpleNamespace(
+            output=[{"type": "compaction", "summary": "through a"}],
+        )
+        mock_client = MagicMock()
+        mock_client.responses.compact = AsyncMock(return_value=compacted)
+        session = OpenAIResponsesCompactionSession(
+            session_id="first-interleave",
+            underlying_session=underlying,
+            client=mock_client,
+            compaction_mode="previous_response_id",
+            should_trigger_compaction=lambda context: context["response_id"] == "resp_a",
+        )
+
+        request_a_in_flight = asyncio.Event()
+        release_request_a = asyncio.Event()
+
+        async def hold_request_a(call: Any) -> ModelStep:
+            request_a_in_flight.set()
+            await release_request_a.wait()
+            return ModelStep(output=[get_text_message("reply a")], response_id="resp_a")
+
+        worker_a = Agent(
+            name="worker_a",
+            model=ScriptedModel(steps=[ModelStep.respond(hold_request_a)]),
+        )
+
+        run_a = asyncio.create_task(Runner.run(worker_a, "hello a", session=session))
+        await request_a_in_flight.wait()
+        await session.add_items([interleaved_item])
+        with caplog.at_level(logging.WARNING, logger="openai-agents.openai.compaction"):
+            release_request_a.set()
+            await run_a
+
+        # The interleaved item must still be stored; deleting it is the loss.
+        stored_text = str(await underlying.get_items())
+        assert "turn b" in stored_text
+        assert "hello a" in stored_text
+        assert "reply a" in stored_text
+        # The stale persist recorded nothing but armed the gate, so the
+        # compaction skipped before the billed call instead of reaching the
+        # snapshot fallback.
+        mock_client.responses.compact.assert_not_called()
+        assert session._response_boundaries == {}
+        assert session._response_boundaries_ever_recorded is True
+        assert "Skipped compaction for resp_a" in caplog.text
+
+    @pytest.mark.asyncio
+    async def test_interleaved_write_before_persist_records_no_boundary(
+        self, caplog: pytest.LogCaptureFixture
+    ) -> None:
+        """A batch persisted after an interleaved write must record no boundary.
+
+        Ordering under test: run A captures ownership when it reads the
+        session for its request input, run B's turn persists through the
+        boundary hook while A's request is in flight, and A's batch then
+        persists. A's ownership no longer matches the store, so the hook
+        appends A's batch without recording, and A's compaction skips
+        through the ever recorded gate before the billed call instead of
+        deleting B's turn.
+        """
+        a_turn = cast(
+            TResponseInputItem, {"type": "message", "role": "assistant", "content": "turn a"}
+        )
+        b_turn = cast(
+            TResponseInputItem, {"type": "message", "role": "assistant", "content": "turn b"}
+        )
+
+        underlying = SimpleListSession(history=[])
+        mock_client = MagicMock()
+        mock_client.responses.compact = AsyncMock()
+        session = OpenAIResponsesCompactionSession(
+            session_id="interleaved-persist",
+            underlying_session=underlying,
+            client=mock_client,
+            compaction_mode="previous_response_id",
+        )
+
+        token_a = await session._capture_ownership_token()
+        # B's turn lands through the hook while A's request is in flight.
+        await persist_response_batch(session, [b_turn], "resp_b")
+        await session._add_items_for_response(
+            [a_turn], response_id="resp_a", ownership_token=token_a
+        )
+
+        assert "resp_a" not in session._response_boundaries
+        assert session._response_boundaries == {"resp_b": 1}
+
+        with caplog.at_level(logging.WARNING, logger="openai-agents.openai.compaction"):
+            await session.run_compaction({"force": True, "response_id": "resp_a"})
+
+        assert "Skipped compaction for resp_a" in caplog.text
+        mock_client.responses.compact.assert_not_awaited()
+        assert await underlying.get_items() == [b_turn, a_turn]
+        assert await session.get_items() == [b_turn, a_turn]
+
+    @pytest.mark.asyncio
+    async def test_destructive_interleave_invalidates_ownership(
+        self, caplog: pytest.LogCaptureFixture
+    ) -> None:
+        """A destructive rewrite between capture and persist must void ownership.
+
+        Ordering under test: a first response records cleanly so the ever
+        recorded gate is armed, run A captures ownership, then a pop and a
+        fresh append rewrite the store back to the captured count before
+        A's batch persists. The count alone cannot see that rewrite, but
+        the generation can, so nothing records and A's compaction skips
+        before the billed call.
+        """
+        seed_one = cast(
+            TResponseInputItem, {"type": "message", "role": "assistant", "content": "seed 1"}
+        )
+        seed_two = cast(
+            TResponseInputItem, {"type": "message", "role": "assistant", "content": "seed 2"}
+        )
+        replacement_turn = cast(
+            TResponseInputItem, {"type": "message", "role": "assistant", "content": "replacement"}
+        )
+        a_turn = cast(
+            TResponseInputItem, {"type": "message", "role": "assistant", "content": "turn a"}
+        )
+
+        underlying = SimpleListSession(history=[])
+        mock_client = MagicMock()
+        mock_client.responses.compact = AsyncMock()
+        session = OpenAIResponsesCompactionSession(
+            session_id="destructive-interleave",
+            underlying_session=underlying,
+            client=mock_client,
+            compaction_mode="previous_response_id",
+        )
+
+        await persist_response_batch(session, [seed_one, seed_two], "resp_seed")
+        token_a = await session._capture_ownership_token()
+        # The pop bumps the generation and drops every recorded boundary;
+        # the append restores the captured count without restoring the
+        # captured history.
+        await session.pop_item()
+        await session.add_items([replacement_turn])
+        await session._add_items_for_response(
+            [a_turn], response_id="resp_a", ownership_token=token_a
+        )
+
+        assert session._response_boundaries == {}
+        assert session._response_boundaries_ever_recorded is True
+
+        with caplog.at_level(logging.WARNING, logger="openai-agents.openai.compaction"):
+            await session.run_compaction({"force": True, "response_id": "resp_a"})
+
+        assert "Skipped compaction for resp_a" in caplog.text
+        mock_client.responses.compact.assert_not_awaited()
+        assert await underlying.get_items() == [seed_one, replacement_turn, a_turn]
+
+    @pytest.mark.asyncio
+    async def test_clean_persist_with_token_records_and_compacts(self) -> None:
+        """With ownership intact the boundary records and the replacement lands.
+
+        Ordering under test: ownership is captured, the batch persists with
+        no interleaved write, and a later turn lands past the recorded
+        boundary. The compaction keyed on the response replaces exactly the
+        covered prefix and preserves the later turn as the tail.
+        """
+        turn = cast(
+            TResponseInputItem, {"type": "message", "role": "assistant", "content": "turn a"}
+        )
+        later_turn = cast(
+            TResponseInputItem, {"type": "message", "role": "assistant", "content": "turn later"}
+        )
+        summary = cast(TResponseInputItem, {"type": "compaction", "summary": "through a"})
+
+        underlying = SimpleListSession(history=[])
+        mock_compact_response = MagicMock()
+        mock_compact_response.output = [summary]
+        mock_client = MagicMock()
+        mock_client.responses.compact = AsyncMock(return_value=mock_compact_response)
+        session = OpenAIResponsesCompactionSession(
+            session_id="clean-token",
+            underlying_session=underlying,
+            client=mock_client,
+            compaction_mode="previous_response_id",
+        )
+
+        ownership_token = await session._capture_ownership_token()
+        await session._add_items_for_response(
+            [turn], response_id="resp_a", ownership_token=ownership_token
+        )
+
+        assert session._response_boundaries == {"resp_a": 1}
+        assert ownership_token.count == 1
+
+        await session.add_items([later_turn])
+        await session.run_compaction({"force": True, "response_id": "resp_a"})
+
+        mock_client.responses.compact.assert_awaited_once()
+        assert mock_client.responses.compact.call_args.kwargs["previous_response_id"] == "resp_a"
+        assert await underlying.get_items() == [summary, later_turn]
+        assert await session.get_items() == [summary, later_turn]
+
+    @pytest.mark.asyncio
+    async def test_overlapping_compaction_lands_on_translated_boundary(self) -> None:
+        """A second compaction must preserve turns past its translated boundary.
+
+        Ordering under test: response A's batch is persisted, response B's batch
+        follows, and A's replacement lands while run C's turn is appended mid
+        flight. A's replacement rewrites the prefix that B's recorded boundary
+        counted into A's summary, so the boundary must be translated onto the
+        rewritten history. B's later replacement then preserves C's turn.
+        Dropping the boundary instead would make B skip its compaction, and
+        keeping the raw count would misplace the boundary and drop the turn.
+        """
+        a_turn = cast(
+            TResponseInputItem, {"type": "message", "role": "assistant", "content": "turn a"}
+        )
+        b_turn = cast(
+            TResponseInputItem, {"type": "message", "role": "assistant", "content": "turn b"}
+        )
+        c_turn = cast(
+            TResponseInputItem, {"type": "message", "role": "assistant", "content": "turn c"}
+        )
+        a_summary = cast(TResponseInputItem, {"type": "compaction", "summary": "through a"})
+        b_summary = cast(TResponseInputItem, {"type": "compaction", "summary": "through b"})
+
+        underlying = SimpleListSession(history=[])
+        first_compact_entered = asyncio.Event()
+        release_first_compact = asyncio.Event()
+        outputs = [[a_summary], [b_summary]]
+        call_index = 0
+
+        async def gated_compact(**kwargs: Any) -> MagicMock:
+            nonlocal call_index
+            index = call_index
+            call_index += 1
+            if index == 0:
+                first_compact_entered.set()
+                await release_first_compact.wait()
+            response = MagicMock()
+            response.output = outputs[index]
+            return response
+
+        mock_client = MagicMock()
+        mock_client.responses.compact = AsyncMock(side_effect=gated_compact)
+        session = OpenAIResponsesCompactionSession(
+            session_id="translated-boundary",
+            underlying_session=underlying,
+            client=mock_client,
+            compaction_mode="previous_response_id",
+        )
+
+        await persist_response_batch(session, [a_turn], "resp_a")
+        await persist_response_batch(session, [b_turn], "resp_b")
+
+        compaction_a = asyncio.create_task(
+            session.run_compaction({"force": True, "response_id": "resp_a"})
+        )
+        try:
+            await first_compact_entered.wait()
+            # A's request is in flight with its snapshot taken; C's turn lands now.
+            await session.add_items([c_turn])
+        finally:
+            release_first_compact.set()
+        await compaction_a
+        assert await underlying.get_items() == [a_summary, b_turn, c_turn]
+
+        await session.run_compaction({"force": True, "response_id": "resp_b"})
+
+        sent_ids = [
+            call.kwargs["previous_response_id"]
+            for call in mock_client.responses.compact.call_args_list
+        ]
+        assert sent_ids == ["resp_a", "resp_b"]
+        # B's boundary now sits past A's summary and B's own batch, so C's turn
+        # is the only tail B's replacement must keep.
+        assert await underlying.get_items() == [b_summary, c_turn]
+        assert await session.get_items() == [b_summary, c_turn]
+
+    @pytest.mark.asyncio
+    async def test_compaction_skips_when_replacement_rewrote_its_recorded_baseline(
+        self, caplog: pytest.LogCaptureFixture
+    ) -> None:
+        """A compaction whose recorded prefix was rewritten away must skip.
+
+        Ordering under test: response B's batch is persisted with a boundary,
+        response A's batch follows, run C's turn lands while A's compaction is
+        in flight, and A's replacement rewrites both batches into A's summary
+        before B takes its snapshot. B's recorded prefix ended inside the
+        rewritten region, so the replacement drops B's entry: no boundary on
+        the new history describes what B's compaction covers. B must skip
+        through the absent entry path without a billed call; falling back to
+        its snapshot would classify C's turn and A's summary into B's baseline
+        and drop both.
+        """
+        b_turn = cast(
+            TResponseInputItem, {"type": "message", "role": "assistant", "content": "turn b"}
+        )
+        a_turn = cast(
+            TResponseInputItem, {"type": "message", "role": "assistant", "content": "turn a"}
+        )
+        c_turn = cast(
+            TResponseInputItem, {"type": "message", "role": "assistant", "content": "turn c"}
+        )
+        a_summary = cast(TResponseInputItem, {"type": "compaction", "summary": "through a"})
+        stale_b_summary = cast(TResponseInputItem, {"type": "compaction", "summary": "through b"})
+
+        underlying = SimpleListSession(history=[])
+        compact_entered = asyncio.Event()
+        release_compact = asyncio.Event()
+        outputs = [[a_summary], [stale_b_summary]]
+        call_index = 0
+
+        async def gated_compact(**kwargs: Any) -> MagicMock:
+            nonlocal call_index
+            index = call_index
+            call_index += 1
+            if index == 0:
+                compact_entered.set()
+                await release_compact.wait()
+            response = MagicMock()
+            response.output = outputs[index]
+            return response
+
+        mock_client = MagicMock()
+        mock_client.responses.compact = AsyncMock(side_effect=gated_compact)
+        session = OpenAIResponsesCompactionSession(
+            session_id="rewritten-baseline",
+            underlying_session=underlying,
+            client=mock_client,
+            compaction_mode="previous_response_id",
+        )
+
+        await persist_response_batch(session, [b_turn], "resp_b")
+        await persist_response_batch(session, [a_turn], "resp_a")
+
+        compaction_a = asyncio.create_task(
+            session.run_compaction({"force": True, "response_id": "resp_a"})
+        )
+        try:
+            await compact_entered.wait()
+            # A's request is in flight; C's turn lands before A's replacement.
+            await session.add_items([c_turn])
+        finally:
+            release_compact.set()
+        await compaction_a
+        # A's replacement covered both batches and preserved C's turn, and it
+        # dropped B's entry because B's recorded prefix was rewritten.
+        assert await underlying.get_items() == [a_summary, c_turn]
+        assert "resp_b" not in session._response_boundaries
+
+        with caplog.at_level(logging.WARNING, logger="openai-agents.openai.compaction"):
+            await session.run_compaction({"force": True, "response_id": "resp_b"})
+
+        # B's compaction skipped without calling the API and replaced nothing,
+        # so C's turn and A's summary both survive.
+        assert await underlying.get_items() == [a_summary, c_turn]
+        assert await session.get_items() == [a_summary, c_turn]
+        assert mock_client.responses.compact.await_count == 1
+        assert "Skipped compaction for resp_b" in caplog.text
+
+    @pytest.mark.asyncio
+    async def test_compaction_skips_when_recorded_boundary_was_evicted(
+        self, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+    ) -> None:
+        """A delayed compaction whose recorded boundary was evicted must skip.
+
+        Ordering under test: an old response's batch is persisted with a
+        boundary, enough newer response batches follow to push that entry past
+        the recording cap, and only then does the compaction keyed on the old
+        response run. Its entry is gone, so no boundary on the current history
+        describes what the compaction covers. Falling back to the snapshot
+        would classify every newer turn into the old response's baseline and
+        replace them all with a summary that covers the old response alone.
+        """
+        monkeypatch.setattr(
+            "agents.memory.openai_responses_compaction_session._MAX_RECORDED_RESPONSE_BOUNDARIES",
+            2,
+        )
+        old_turn = cast(
+            TResponseInputItem, {"type": "message", "role": "assistant", "content": "turn old"}
+        )
+        newer_turns = [
+            cast(
+                TResponseInputItem,
+                {"type": "message", "role": "assistant", "content": f"turn new {index}"},
+            )
+            for index in range(2)
+        ]
+        stale_summary = cast(TResponseInputItem, {"type": "compaction", "summary": "through old"})
+
+        underlying = SimpleListSession(history=[])
+        mock_compact_response = MagicMock()
+        mock_compact_response.output = [stale_summary]
+        mock_client = MagicMock()
+        mock_client.responses.compact = AsyncMock(return_value=mock_compact_response)
+        session = OpenAIResponsesCompactionSession(
+            session_id="evicted-boundary",
+            underlying_session=underlying,
+            client=mock_client,
+            compaction_mode="previous_response_id",
+        )
+
+        await persist_response_batch(session, [old_turn], "resp_old")
+        for index, turn in enumerate(newer_turns):
+            await persist_response_batch(session, [turn], f"resp_new_{index}")
+        assert "resp_old" not in session._response_boundaries
+
+        with caplog.at_level(logging.WARNING, logger="openai-agents.openai.compaction"):
+            await session.run_compaction({"force": True, "response_id": "resp_old"})
+
+        # The compaction skipped before the billed call and replaced nothing,
+        # so every newer turn survives.
+        assert await underlying.get_items() == [old_turn, *newer_turns]
+        assert await session.get_items() == [old_turn, *newer_turns]
+        mock_client.responses.compact.assert_not_awaited()
+        assert "Skipped compaction for resp_old" in caplog.text
+
+    @pytest.mark.asyncio
+    async def test_compaction_skips_for_recorded_response_after_clear_session(
+        self, caplog: pytest.LogCaptureFixture
+    ) -> None:
+        """A compaction keyed on a response recorded before clear_session must skip.
+
+        Ordering under test: a response's batch is persisted with a boundary,
+        clear_session wipes the history and every recorded boundary, and a
+        fresh turn lands afterwards. The delayed compaction keyed on the
+        cleared response then runs. Its entry is gone, so falling back to the
+        snapshot would replace the fresh turn with a summary of history the
+        clear already discarded.
+        """
+        recorded_turn = cast(
+            TResponseInputItem, {"type": "message", "role": "assistant", "content": "recorded"}
+        )
+        later_turn = cast(
+            TResponseInputItem, {"type": "message", "role": "assistant", "content": "later"}
+        )
+        stale_summary = cast(
+            TResponseInputItem, {"type": "compaction", "summary": "cleared history"}
+        )
+
+        underlying = SimpleListSession(history=[])
+        mock_compact_response = MagicMock()
+        mock_compact_response.output = [stale_summary]
+        mock_client = MagicMock()
+        mock_client.responses.compact = AsyncMock(return_value=mock_compact_response)
+        session = OpenAIResponsesCompactionSession(
+            session_id="cleared-boundary",
+            underlying_session=underlying,
+            client=mock_client,
+            compaction_mode="previous_response_id",
+        )
+
+        await persist_response_batch(session, [recorded_turn], "resp_recorded")
+        await session.clear_session()
+        await session.add_items([later_turn])
+        assert "resp_recorded" not in session._response_boundaries
+
+        with caplog.at_level(logging.WARNING, logger="openai-agents.openai.compaction"):
+            await session.run_compaction({"force": True, "response_id": "resp_recorded"})
+
+        # The compaction skipped instead of repopulating cleared history, so
+        # the turn persisted after the clear survives untouched.
+        assert await underlying.get_items() == [later_turn]
+        assert await session.get_items() == [later_turn]
+        mock_client.responses.compact.assert_not_awaited()
+        assert "Skipped compaction for resp_recorded" in caplog.text
+
+    @pytest.mark.asyncio
+    async def test_persisted_batch_survives_read_failure_after_append(self) -> None:
+        """A failing history read must not fail a batch the backend already holds.
+
+        Ordering under test: a response's batch is persisted through the
+        boundary hook while the underlying store fails every read issued
+        after an append. The count that seeds the boundary is read before
+        the append, so the hook records the boundary without touching the
+        failing read and the caller never sees an error for a turn that is
+        already stored; a read after the append would raise here, the run
+        would treat the persisted turn as failed, and a retry would append
+        it again. The compaction keyed on the response then proves the
+        recorded boundary equals the count a read after the append would
+        have produced: the turn persisted after the batch is preserved as
+        the tail.
+        """
+        turn = cast(
+            TResponseInputItem, {"type": "message", "role": "assistant", "content": "turn a"}
+        )
+        later_turn = cast(
+            TResponseInputItem, {"type": "message", "role": "assistant", "content": "turn later"}
+        )
+        summary = cast(TResponseInputItem, {"type": "compaction", "summary": "through a"})
+
+        class ReadFailsAfterAppendSession(SimpleListSession):
+            def __init__(self) -> None:
+                super().__init__()
+                self.add_calls = 0
+                self.failing = True
+
+            async def add_items(self, items: list[TResponseInputItem]) -> None:
+                self.add_calls += 1
+                await super().add_items(items)
+
+            async def get_items(self, limit: int | None = None) -> list[TResponseInputItem]:
+                if self.failing and self.add_calls:
+                    raise RuntimeError("history read failed after the batch was appended")
+                return await super().get_items(limit)
+
+        underlying = ReadFailsAfterAppendSession()
+        mock_compact_response = MagicMock()
+        mock_compact_response.output = [summary]
+        mock_client = MagicMock()
+        mock_client.responses.compact = AsyncMock(return_value=mock_compact_response)
+        session = OpenAIResponsesCompactionSession(
+            session_id="read-failure-boundary",
+            underlying_session=underlying,
+            client=mock_client,
+            compaction_mode="previous_response_id",
+        )
+
+        await persist_response_batch(session, [turn], "resp_a")
+
+        # The append landed exactly once and the boundary bookkeeping finished
+        # even though every read after the append would have failed.
+        assert underlying.add_calls == 1
+        assert session._response_boundaries["resp_a"] == 1
+        assert session._response_boundaries_ever_recorded is True
+
+        underlying.failing = False
+        assert await underlying.get_items() == [turn]
+
+        await session.add_items([later_turn])
+        await session.run_compaction({"force": True, "response_id": "resp_a"})
+
+        # The recorded boundary matches the count a read after the append
+        # would have produced, so the turn persisted after the batch is the
+        # tail the replacement preserves.
+        assert await underlying.get_items() == [summary, later_turn]
+        assert await session.get_items() == [summary, later_turn]
+
+    @pytest.mark.asyncio
+    async def test_commit_then_failed_append_still_arms_skip_gate(
+        self, caplog: pytest.LogCaptureFixture
+    ) -> None:
+        """An append that commits and then raises must still arm the skip gate.
+
+        Ordering under test: a fresh wrapper over an empty store, a run
+        captures ownership, and its persist reaches a backend that stores
+        the batch but raises before acknowledging. The batch is durable
+        while the persist surfaced an error, so nothing was recorded for
+        the response. With the ever recorded arm sitting after the append,
+        the raise skipped it, and the compaction keyed on the response fell
+        through to the snapshot fallback, classified the committed turn as
+        covered, and replaced it with a summary the server side history
+        through that response cannot vouch for. The arm now precedes the
+        append, so every token carrying persist marks the session boundary
+        managed even when the backend commits and then fails: the compaction
+        skips before the billed call and the committed turn survives. Arming
+        first is strictly conservative, costing at most a skip when the
+        append never committed at all.
+        """
+        turn = cast(
+            TResponseInputItem, {"type": "message", "role": "assistant", "content": "turn a"}
+        )
+
+        class CommitThenRaiseSession(SimpleListSession):
+            """Backend that stores the first batch, then fails its acknowledgement."""
+
+            def __init__(self) -> None:
+                super().__init__()
+                self.failed_once = False
+
+            async def add_items(self, items: list[TResponseInputItem]) -> None:
+                await super().add_items(items)
+                if not self.failed_once:
+                    self.failed_once = True
+                    raise RuntimeError("acknowledgement lost after commit")
+
+        underlying = CommitThenRaiseSession()
+        compacted = SimpleNamespace(output=[{"type": "compaction", "summary": "through a"}])
+        mock_client = MagicMock()
+        mock_client.responses.compact = AsyncMock(return_value=compacted)
+        session = OpenAIResponsesCompactionSession(
+            session_id="commit-then-raise",
+            underlying_session=underlying,
+            client=mock_client,
+            compaction_mode="previous_response_id",
+        )
+
+        ownership_token = await session._capture_ownership_token()
+        with pytest.raises(RuntimeError, match="acknowledgement lost after commit"):
+            await session._add_items_for_response(
+                [turn], response_id="resp_a", ownership_token=ownership_token
+            )
+
+        # The backend committed the batch even though the persist errored,
+        # and nothing was recorded for the response; only the armed gate
+        # stands between that absent entry and the snapshot fallback.
+        assert await underlying.get_items() == [turn]
+        assert session._response_boundaries == {}
+        assert session._response_boundaries_ever_recorded is True
+
+        with caplog.at_level(logging.WARNING, logger="openai-agents.openai.compaction"):
+            await session.run_compaction({"force": True, "response_id": "resp_a"})
+
+        # The compaction skipped before the billed call and the committed
+        # turn survived instead of being replaced through the fallback.
+        mock_client.responses.compact.assert_not_called()
+        assert await underlying.get_items() == [turn]
+        assert "Skipped compaction for resp_a" in caplog.text
+
+    @pytest.mark.asyncio
+    async def test_replacement_translates_boundaries_at_nonzero_shift(self) -> None:
+        """Boundary translation must apply the exact shift of the rewrite.
+
+        Ordering under test: response A persists a two item batch, response B
+        persists one item, and A's replacement rewrites A's batch into a one
+        item summary. The rewrite shrinks the prefix by one, so B's recorded
+        boundary must shift from three to two. B's compaction then preserves
+        exactly the turn appended past the translated boundary. Keeping B's
+        raw count instead would place the boundary past that turn and drop
+        it from the replaced history.
+        """
+        a_turn_one = cast(
+            TResponseInputItem, {"type": "message", "role": "assistant", "content": "turn a1"}
+        )
+        a_turn_two = cast(
+            TResponseInputItem, {"type": "message", "role": "assistant", "content": "turn a2"}
+        )
+        b_turn = cast(
+            TResponseInputItem, {"type": "message", "role": "assistant", "content": "turn b"}
+        )
+        later_turn = cast(
+            TResponseInputItem, {"type": "message", "role": "assistant", "content": "turn later"}
+        )
+        a_summary = cast(TResponseInputItem, {"type": "compaction", "summary": "through a"})
+        b_summary = cast(TResponseInputItem, {"type": "compaction", "summary": "through b"})
+
+        underlying = SimpleListSession(history=[])
+        outputs = [[a_summary], [b_summary]]
+        call_index = 0
+
+        async def sequenced_compact(**kwargs: Any) -> MagicMock:
+            nonlocal call_index
+            response = MagicMock()
+            response.output = outputs[call_index]
+            call_index += 1
+            return response
+
+        mock_client = MagicMock()
+        mock_client.responses.compact = AsyncMock(side_effect=sequenced_compact)
+        session = OpenAIResponsesCompactionSession(
+            session_id="translated-shift",
+            underlying_session=underlying,
+            client=mock_client,
+            compaction_mode="previous_response_id",
+        )
+
+        await persist_response_batch(session, [a_turn_one, a_turn_two], "resp_a")
+        await persist_response_batch(session, [b_turn], "resp_b")
+        assert session._response_boundaries == {"resp_a": 2, "resp_b": 3}
+
+        await session.run_compaction({"force": True, "response_id": "resp_a"})
+
+        # The replacement rewrote two items into one, so every surviving
+        # boundary shifts back by one.
+        assert await underlying.get_items() == [a_summary, b_turn]
+        assert session._response_boundaries["resp_b"] == 2
+
+        await session.add_items([later_turn])
+        await session.run_compaction({"force": True, "response_id": "resp_b"})
+
+        # B's replacement preserved exactly the turn past the translated
+        # boundary.
+        assert await underlying.get_items() == [b_summary, later_turn]
+        assert await session.get_items() == [b_summary, later_turn]
+
+    @pytest.mark.asyncio
+    async def test_defer_compaction_waits_for_mutation_lock(self) -> None:
+        """The deferral decision must not run while another task holds the lock.
+
+        Ordering under test: a writer holds the mutation lock, the way the
+        replacement phase does, while _defer_compaction is called with cold
+        caches. The deferral must park at the lock instead of reading the
+        store and installing caches mid replacement; an unserialized fill
+        here could overwrite the caches a replacement just installed with a
+        torn read of the store.
+        """
+        turn = cast(TResponseInputItem, {"type": "message", "role": "assistant", "content": "turn"})
+        underlying = SimpleListSession(history=[turn])
+        session = OpenAIResponsesCompactionSession(
+            session_id="defer-lock",
+            underlying_session=underlying,
+            client=MagicMock(),
+            should_trigger_compaction=lambda context: True,
+        )
+
+        async with session._mutation_lock:
+            defer_task = asyncio.create_task(session._defer_compaction("resp_a"))
+            for _ in range(5):
+                await asyncio.sleep(0)
+            # Parked at the lock: no cache fill and no deferred id yet.
+            assert not defer_task.done()
+            assert session._compaction_candidate_items is None
+            assert session._session_items is None
+            assert session._get_deferred_compaction_response_id() is None
+        await defer_task
+
+        assert session._get_deferred_compaction_response_id() == "resp_a"
+        assert session._session_items == [turn]
+
+    @pytest.mark.asyncio
+    async def test_deferred_id_survives_concurrent_run_compaction(self) -> None:
+        """A deferral landing after the decision phase must not be wiped.
+
+        Ordering under test: run A passes the compaction decision, and run B
+        defers its own response while A still holds the decision phase lock.
+        A subsumes only deferrals made before its decision, so B's deferral
+        must survive A and keep the forced follow up compaction for B's
+        response armed.
+        """
+        history: list[TResponseInputItem] = [
+            cast(TResponseInputItem, {"type": "message", "role": "assistant", "content": f"m{i}"})
+            for i in range(3)
+        ]
+        summary = cast(TResponseInputItem, {"type": "compaction", "summary": "compacted"})
+
+        class SnapshotPausingSession(SimpleListSession):
+            def __init__(self, history: list[TResponseInputItem]) -> None:
+                super().__init__(history=history)
+                self.snapshot_entered = asyncio.Event()
+                self.release_snapshot = asyncio.Event()
+                self.full_reads = 0
+
+            async def get_items(self, limit: int | None = None) -> list[TResponseInputItem]:
+                if limit is not None and limit > 1_000_000:
+                    self.full_reads += 1
+                    if self.full_reads == 1:
+                        self.snapshot_entered.set()
+                        await self.release_snapshot.wait()
+                return await super().get_items(limit)
+
+        underlying = SnapshotPausingSession(history=history)
+        mock_compact_response = MagicMock()
+        mock_compact_response.output = [summary]
+        mock_client = MagicMock()
+        mock_client.responses.compact = AsyncMock(return_value=mock_compact_response)
+        session = OpenAIResponsesCompactionSession(
+            session_id="deferral-survives",
+            underlying_session=underlying,
+            client=mock_client,
+            compaction_mode="input",
+            should_trigger_compaction=lambda context: True,
+        )
+
+        compaction_task = asyncio.create_task(session.run_compaction({"force": True}))
+        try:
+            await underlying.snapshot_entered.wait()
+            # A holds the decision phase lock at its snapshot read; B's
+            # deferral arrives now and parks at the lock.
+            defer_task = asyncio.create_task(session._defer_compaction("resp_b"))
+            for _ in range(5):
+                await asyncio.sleep(0)
+            assert not defer_task.done()
+        finally:
+            underlying.release_snapshot.set()
+        await compaction_task
+        await defer_task
+
+        # B's deferral survived A's compaction instead of being wiped.
+        assert session._get_deferred_compaction_response_id() == "resp_b"
+        assert await underlying.get_items() == [summary]
+
+    @pytest.mark.asyncio
+    async def test_failed_replacement_drops_boundary_state(
+        self, caplog: pytest.LogCaptureFixture
+    ) -> None:
+        """A failed replacement transaction must invalidate the boundary records.
+
+        Ordering under test: a response's batch is persisted with a boundary,
+        its replacement fails, and the restore fails too, leaving the store
+        empty while the boundary still counts the old history. The failure
+        must drop every recorded boundary and count as a rewrite, matching
+        pop_item and clear_session; a retried compaction keyed on the same
+        response then skips. Keeping the boundary instead would let the retry
+        slice a shorter store past its end and silently delete a batch
+        persisted while the retry was in flight.
+        """
+        turns: list[TResponseInputItem] = [
+            cast(TResponseInputItem, {"type": "message", "role": "assistant", "content": f"t{i}"})
+            for i in range(3)
+        ]
+        batch_turns: list[TResponseInputItem] = [
+            cast(TResponseInputItem, {"type": "message", "role": "assistant", "content": f"b{i}"})
+            for i in range(2)
+        ]
+        stale_summary = cast(TResponseInputItem, {"type": "compaction", "summary": "stale"})
+
+        class DoubleFailureSession(SimpleListSession):
+            def __init__(self) -> None:
+                super().__init__()
+                self.failing_adds = 0
+
+            async def add_items(self, items: list[TResponseInputItem]) -> None:
+                if self.failing_adds > 0:
+                    self.failing_adds -= 1
+                    raise RuntimeError("backend write failed")
+                await super().add_items(items)
+
+        underlying = DoubleFailureSession()
+        mock_compact_response = MagicMock()
+        mock_compact_response.output = [stale_summary]
+        mock_client = MagicMock()
+        mock_client.responses.compact = AsyncMock(return_value=mock_compact_response)
+        session = OpenAIResponsesCompactionSession(
+            session_id="failed-replacement",
+            underlying_session=underlying,
+            client=mock_client,
+            compaction_mode="previous_response_id",
+        )
+
+        await persist_response_batch(session, turns, "resp_a")
+        generation_before = session._destructive_generation
+
+        # The replacement add and the restore add both fail in one outage.
+        underlying.failing_adds = 2
+        with pytest.raises(RuntimeError, match="backend write failed"):
+            await session.run_compaction({"force": True, "response_id": "resp_a"})
+
+        assert await underlying.get_items() == []
+        assert session._response_boundaries == {}
+        assert session._destructive_generation == generation_before + 1
+        assert session._response_boundaries_ever_recorded is True
+
+        # A batch persisted after the failure records its boundary cleanly.
+        await persist_response_batch(session, batch_turns, "resp_b")
+
+        with caplog.at_level(logging.WARNING, logger="openai-agents.openai.compaction"):
+            await session.run_compaction({"force": True, "response_id": "resp_a"})
+
+        # The retry skipped before the billed call instead of trusting a
+        # boundary the store no longer backs, so the newer batch survives.
+        assert "Skipped compaction for resp_a" in caplog.text
+        assert await underlying.get_items() == batch_turns
+        assert mock_client.responses.compact.await_count == 1
+
+    @pytest.mark.asyncio
+    async def test_compaction_skips_when_recorded_boundary_exceeds_history(
+        self, caplog: pytest.LogCaptureFixture
+    ) -> None:
+        """A boundary past the stored history must skip the replacement.
+
+        No healthy path records such a boundary: appends record the post
+        append count and every rewrite drops or translates entries, so the
+        state here is planted directly to stand in for corrupted
+        bookkeeping. The replacement must treat it as invalid instead of
+        slicing past the end, which would classify the whole history as
+        covered and replace it outright.
+        """
+        turn = cast(TResponseInputItem, {"type": "message", "role": "assistant", "content": "turn"})
+        stale_summary = cast(TResponseInputItem, {"type": "compaction", "summary": "stale"})
+
+        underlying = SimpleListSession(history=[])
+        mock_compact_response = MagicMock()
+        mock_compact_response.output = [stale_summary]
+        mock_client = MagicMock()
+        mock_client.responses.compact = AsyncMock(return_value=mock_compact_response)
+        session = OpenAIResponsesCompactionSession(
+            session_id="oversized-boundary",
+            underlying_session=underlying,
+            client=mock_client,
+            compaction_mode="previous_response_id",
+        )
+
+        await persist_response_batch(session, [turn], "resp_a")
+        session._response_boundaries["resp_a"] = 99
+
+        with caplog.at_level(logging.WARNING, logger="openai-agents.openai.compaction"):
+            await session.run_compaction({"force": True, "response_id": "resp_a"})
+
+        # The replacement was skipped and the corrupt records were dropped.
+        assert "Skipped compaction replacement for resp_a" in caplog.text
+        assert await underlying.get_items() == [turn]
+        assert session._response_boundaries == {}
+        assert mock_client.responses.compact.await_count == 1
+
+    @pytest.mark.asyncio
+    async def test_failed_tail_normalization_keeps_store_untouched(self) -> None:
+        """A cache build failure must surface before the store is rewritten.
+
+        Ordering under test: a response's batch is persisted with a boundary,
+        and while the compaction request is in flight a writer sharing the
+        underlying store appends an item whose serialization hook raises.
+        Building the refreshed caches from that tail fails; the failure must
+        surface before the replacement, leaving the store, the generation
+        counter, and the recorded boundaries all describing the same history.
+        A replacement that lands before the failure would strand a rewritten
+        store behind bookkeeping that still describes the old one.
+        """
+
+        class UnserializableItem:
+            @property
+            def model_dump(self) -> Any:
+                raise RuntimeError("serialization failed")
+
+        a_turn_one = cast(
+            TResponseInputItem, {"type": "message", "role": "assistant", "content": "turn a1"}
+        )
+        a_turn_two = cast(
+            TResponseInputItem, {"type": "message", "role": "assistant", "content": "turn a2"}
+        )
+        poisoned_item = cast(TResponseInputItem, UnserializableItem())
+        summary = cast(TResponseInputItem, {"type": "compaction", "summary": "through a"})
+
+        underlying = SimpleListSession(history=[])
+        compact_entered = asyncio.Event()
+        release_compact = asyncio.Event()
+        session = OpenAIResponsesCompactionSession(
+            session_id="poisoned-tail",
+            underlying_session=underlying,
+            client=self.create_gated_compact_client([summary], compact_entered, release_compact),
+            compaction_mode="previous_response_id",
+        )
+
+        await persist_response_batch(session, [a_turn_one, a_turn_two], "resp_a")
+        generation_before = session._destructive_generation
+
+        compaction_task = asyncio.create_task(
+            session.run_compaction({"force": True, "response_id": "resp_a"})
+        )
+        try:
+            await compact_entered.wait()
+            # A writer sharing the store appends the item that cannot be
+            # serialized while the request is in flight.
+            await underlying.add_items([poisoned_item])
+        finally:
+            release_compact.set()
+        with pytest.raises(RuntimeError, match="serialization failed"):
+            await compaction_task
+
+        # The store was not rewritten, and the bookkeeping still describes it.
+        assert await underlying.get_items() == [a_turn_one, a_turn_two, poisoned_item]
+        assert session._response_boundaries == {"resp_a": 2}
+        assert session._destructive_generation == generation_before
+
+
+class TestPartialRequestInputCompaction:
+    """Runs whose request input skips stored history must not record boundaries.
+
+    A session limit windows the history read, whether it resolves at the
+    wrapper or only inside the wrapped backend, a session input callback
+    rebuilds the prepared input outright, a call_model_input_filter
+    rewrites any request before it goes out, a handoff input filter rewrites
+    the accumulated history mid run, and nested handoff history folds it
+    into a rendered transcript. In every case the request carries less than
+    the stored history while the count at persist time spans
+    all of it, so a recorded boundary would let a previous_response_id
+    replacement delete prefix items the server never saw, with no concurrency
+    involved. Such runs must record nothing and their compactions must skip
+    before the billed call, even on a fresh wrapper instance over a restored
+    store where no boundary was ever recorded.
+    """
+
+    @staticmethod
+    def _prior_turns() -> list[TResponseInputItem]:
+        return [
+            cast(TResponseInputItem, {"role": "user", "content": "prefix question"}),
+            cast(
+                TResponseInputItem,
+                {"type": "message", "role": "assistant", "content": "prefix answer"},
+            ),
+            cast(TResponseInputItem, {"role": "user", "content": "recent question"}),
+            cast(
+                TResponseInputItem,
+                {"type": "message", "role": "assistant", "content": "recent answer"},
+            ),
+        ]
+
+    @staticmethod
+    def _make_session(
+        underlying: SimpleListSession, session_id: str
+    ) -> tuple[OpenAIResponsesCompactionSession, MagicMock]:
+        compacted = SimpleNamespace(
+            output=[{"type": "compaction", "summary": "window only"}],
+        )
+        mock_client = MagicMock()
+        mock_client.responses.compact = AsyncMock(return_value=compacted)
+        session = OpenAIResponsesCompactionSession(
+            session_id=session_id,
+            underlying_session=underlying,
+            client=mock_client,
+            compaction_mode="previous_response_id",
+            should_trigger_compaction=lambda context: True,
+        )
+        return session, mock_client
+
+    @pytest.mark.asyncio
+    async def test_limit_windowed_run_records_nothing_and_compaction_skips(
+        self, caplog: pytest.LogCaptureFixture
+    ) -> None:
+        """A run under a truncating session limit must leave the full history intact.
+
+        The store holds four turns and the run's limit admits only the last
+        two, so the request input never carries the oldest turns. On the
+        prior behavior the persist recorded a boundary spanning the whole
+        store and the compaction keyed on the response replaced the prefix,
+        deleting the turns the server never saw. Now the voided ownership
+        token records nothing, arms the skip gate, and the compaction skips
+        before the billed call with every stored item preserved.
+        """
+        prior_turns = self._prior_turns()
+        underlying = SimpleListSession(history=list(prior_turns))
+        session, mock_client = self._make_session(underlying, "windowed-limit")
+
+        model = ScriptedModel(
+            steps=[ModelStep(output=[get_text_message("windowed reply")], response_id="resp_win")]
+        )
+        worker = Agent(name="windowed_worker", model=model)
+
+        with caplog.at_level(logging.WARNING, logger="openai-agents.openai.compaction"):
+            await Runner.run(
+                worker,
+                "hello there",
+                session=session,
+                run_config=RunConfig(session_settings=SessionSettings(limit=2)),
+            )
+
+        # The request really was windowed: the oldest turns were left out of
+        # it, so the server side history cannot contain them.
+        assert model.last_call is not None
+        request_text = str(model.last_call.input)
+        assert "prefix question" not in request_text
+        assert "recent question" in request_text
+        assert "hello there" in request_text
+
+        # The full history survived, including the prefix the server never
+        # saw; deleting it is the data loss under test.
+        stored = await underlying.get_items()
+        assert stored[: len(prior_turns)] == prior_turns
+        stored_text = str(stored)
+        assert "hello there" in stored_text
+        assert "windowed reply" in stored_text
+
+        # Nothing recorded, and the voided token armed the gate so the
+        # compaction skipped before the billed call even though this wrapper
+        # instance never recorded any boundary.
+        assert session._response_boundaries == {}
+        assert session._response_boundaries_ever_recorded is True
+        mock_client.responses.compact.assert_not_called()
+        assert "Skipped compaction for resp_win" in caplog.text
+
+    @pytest.mark.asyncio
+    async def test_session_level_limit_records_nothing_and_compaction_skips(
+        self, caplog: pytest.LogCaptureFixture
+    ) -> None:
+        """A truncating limit set on the session itself must fire the guard too.
+
+        The windowed test above configures the limit through RunConfig.
+        Here the session's own session_settings carry it and the run
+        supplies only an empty override, so the resolved settings inherit
+        the session's window through the same resolve() path. The guard
+        keys on the resolved value, not on where it came from: the persist
+        records nothing, the gate arms, and the compaction skips with the
+        full store preserved.
+        """
+        prior_turns = self._prior_turns()
+        underlying = SimpleListSession(history=list(prior_turns))
+        session, mock_client = self._make_session(underlying, "session-level-limit")
+        session.session_settings = SessionSettings(limit=2)
+
+        model = ScriptedModel(
+            steps=[ModelStep(output=[get_text_message("session reply")], response_id="resp_sess")]
+        )
+        worker = Agent(name="session_limit_worker", model=model)
+
+        with caplog.at_level(logging.WARNING, logger="openai-agents.openai.compaction"):
+            await Runner.run(
+                worker,
+                "hello again",
+                session=session,
+                run_config=RunConfig(session_settings=SessionSettings()),
+            )
+
+        # The session level limit really windowed the request: the oldest
+        # turns were left out, so the server side history cannot contain
+        # them.
+        assert model.last_call is not None
+        request_text = str(model.last_call.input)
+        assert "prefix question" not in request_text
+        assert "recent question" in request_text
+        assert "hello again" in request_text
+
+        stored = await underlying.get_items()
+        assert stored[: len(prior_turns)] == prior_turns
+        stored_text = str(stored)
+        assert "hello again" in stored_text
+        assert "session reply" in stored_text
+
+        assert session._response_boundaries == {}
+        assert session._response_boundaries_ever_recorded is True
+        mock_client.responses.compact.assert_not_called()
+        assert "Skipped compaction for resp_sess" in caplog.text
+
+    @pytest.mark.asyncio
+    async def test_wrapped_session_limit_records_nothing_and_compaction_skips(
+        self, caplog: pytest.LogCaptureFixture
+    ) -> None:
+        """A limit carried by the wrapped backend must fire the guard too.
+
+        Nothing above the backend knows about this window. The decorator
+        inherits ``session_settings = None`` rather than proxying the wrapped
+        session's setting, and the run sets no limit either, so the resolved
+        settings look unlimited while the backend still applies its own limit
+        to the unlimited read and returns only the newest turns. The request
+        input therefore omits the oldest stored turns with no limit visible
+        anywhere and no concurrency involved, and a boundary recorded over
+        the full store would let the previous_response_id replacement delete
+        the omitted prefix. The token is voided because the read came back
+        shorter than the count it captured, so nothing records, the gate
+        arms, the compaction skips before the billed call, and every stored
+        item survives.
+        """
+
+        class WrappedLimitSession(SimpleListSession):
+            """Mirror a backend whose own settings window an unlimited read."""
+
+            def __init__(self, history: list[TResponseInputItem]) -> None:
+                super().__init__(history=history)
+                self.session_settings = SessionSettings(limit=2)
+
+            async def get_items(self, limit: int | None = None) -> list[TResponseInputItem]:
+                if limit is None and self.session_settings is not None:
+                    limit = self.session_settings.limit
+                return await super().get_items(limit)
+
+        prior_turns = self._prior_turns()
+        underlying = WrappedLimitSession(history=list(prior_turns))
+        session, mock_client = self._make_session(underlying, "wrapped-limit")
+        # The window is invisible from the wrapper, which is what made the
+        # earlier check on the resolved limit miss this case.
+        assert session.session_settings is None
+
+        model = ScriptedModel(
+            steps=[ModelStep(output=[get_text_message("wrapped reply")], response_id="resp_wrap")]
+        )
+        worker = Agent(name="wrapped_limit_worker", model=model)
+
+        with caplog.at_level(logging.WARNING, logger="openai-agents.openai.compaction"):
+            await Runner.run(worker, "wrapped question", session=session)
+
+        # The backend really windowed the request: the oldest turns were left
+        # out of it, so the server side history cannot contain them.
+        assert model.last_call is not None
+        request_text = str(model.last_call.input)
+        assert "prefix question" not in request_text
+        assert "recent question" in request_text
+        assert "wrapped question" in request_text
+
+        # The full history survived, including the prefix the server never
+        # saw; deleting it is the data loss under test. Read past the
+        # backend's own window so the whole store is inspected.
+        stored = await underlying.get_items(limit=100)
+        assert stored[: len(prior_turns)] == prior_turns
+        stored_text = str(stored)
+        assert "wrapped question" in stored_text
+        assert "wrapped reply" in stored_text
+
+        assert session._response_boundaries == {}
+        assert session._response_boundaries_ever_recorded is True
+        mock_client.responses.compact.assert_not_called()
+        assert "Skipped compaction for resp_wrap" in caplog.text
+
+    @pytest.mark.asyncio
+    async def test_filtering_input_callback_records_nothing_and_compaction_skips(
+        self, caplog: pytest.LogCaptureFixture
+    ) -> None:
+        """A streamed run whose callback drops history must leave it intact.
+
+        The callback keeps only the new turn input, so the request carries
+        none of the four stored turns. On the prior behavior the persist
+        still recorded a boundary spanning the whole store and the
+        compaction replaced it with a summary of the filtered request. Now
+        the token is voided where the callback runs, nothing records, and
+        the compaction skips with the stored history preserved. Running
+        streamed exercises the token threading on the streaming path.
+        """
+        prior_turns = self._prior_turns()
+        underlying = SimpleListSession(history=list(prior_turns))
+        session, mock_client = self._make_session(underlying, "filtered-callback")
+
+        model = ScriptedModel(
+            steps=[ModelStep(output=[get_text_message("filtered reply")], response_id="resp_filt")]
+        )
+        worker = Agent(name="filtered_worker", model=model)
+
+        def keep_only_new_items(
+            history: list[TResponseInputItem], new_items: list[TResponseInputItem]
+        ) -> list[TResponseInputItem]:
+            return list(new_items)
+
+        with caplog.at_level(logging.WARNING, logger="openai-agents.openai.compaction"):
+            result = Runner.run_streamed(
+                worker,
+                "fresh question",
+                session=session,
+                run_config=RunConfig(session_input_callback=keep_only_new_items),
+            )
+            async for _ in result.stream_events():
+                pass
+
+        assert model.last_call is not None
+        request_text = str(model.last_call.input)
+        assert "prefix question" not in request_text
+        assert "recent question" not in request_text
+        assert "fresh question" in request_text
+
+        # The stored history survived even though none of it went out with
+        # the request; deleting it is the data loss under test.
+        stored = await underlying.get_items()
+        assert stored[: len(prior_turns)] == prior_turns
+        stored_text = str(stored)
+        assert "fresh question" in stored_text
+        assert "filtered reply" in stored_text
+
+        assert session._response_boundaries == {}
+        assert session._response_boundaries_ever_recorded is True
+        mock_client.responses.compact.assert_not_called()
+        assert "Skipped compaction for resp_filt" in caplog.text
+
+    @pytest.mark.asyncio
+    async def test_call_model_input_filter_records_nothing_and_compaction_skips(
+        self, caplog: pytest.LogCaptureFixture
+    ) -> None:
+        """A run whose call_model_input_filter drops history must leave it intact.
+
+        The filter runs on every request, and here it drops the oldest
+        stored turns from the request input, so the server side history
+        through the response never contains them while the count at
+        persist time spans the whole store. On the prior behavior the
+        persist recorded that count and the compaction keyed on the
+        response replaced the prefix, deleting the dropped turns. Invoking
+        the filter now voids the ownership token, unconditionally like the
+        session input callback, so nothing records and the compaction
+        skips with every stored item preserved.
+        """
+        prior_turns = self._prior_turns()
+        underlying = SimpleListSession(history=list(prior_turns))
+        session, mock_client = self._make_session(underlying, "request-filter")
+
+        model = ScriptedModel(
+            steps=[ModelStep(output=[get_text_message("kept reply")], response_id="resp_drop")]
+        )
+        worker = Agent(name="request_filter_worker", model=model)
+
+        def drop_prefix_turns(data: CallModelData[Any]) -> ModelInputData:
+            kept = [
+                item
+                for item in data.model_data.input
+                if "prefix" not in str(item.get("content") if isinstance(item, dict) else item)
+            ]
+            return ModelInputData(input=kept, instructions=data.model_data.instructions)
+
+        with caplog.at_level(logging.WARNING, logger="openai-agents.openai.compaction"):
+            await Runner.run(
+                worker,
+                "fresh question",
+                session=session,
+                run_config=RunConfig(call_model_input_filter=drop_prefix_turns),
+            )
+
+        # The filter really ran: the oldest turns were left out of the
+        # request, so the server side history cannot contain them.
+        assert model.last_call is not None
+        request_text = str(model.last_call.input)
+        assert "prefix question" not in request_text
+        assert "recent question" in request_text
+        assert "fresh question" in request_text
+
+        # The stored history survived, including the turns the filter
+        # dropped from the request; deleting them is the data loss under
+        # test.
+        stored = await underlying.get_items()
+        assert stored[: len(prior_turns)] == prior_turns
+        stored_text = str(stored)
+        assert "fresh question" in stored_text
+        assert "kept reply" in stored_text
+
+        assert session._response_boundaries == {}
+        assert session._response_boundaries_ever_recorded is True
+        mock_client.responses.compact.assert_not_called()
+        assert "Skipped compaction for resp_drop" in caplog.text
+
+    @pytest.mark.asyncio
+    async def test_handoff_input_filter_records_nothing_and_compaction_skips(
+        self, caplog: pytest.LogCaptureFixture
+    ) -> None:
+        """A handoff filter that rewrites history must void the run's ownership.
+
+        The triage response hands off through a filter that drops the
+        oldest stored turns from the accumulated history, so every request
+        after the handoff omits them while the store keeps all of them.
+        The post handoff persist would otherwise record a count spanning
+        the whole store, and the compaction keyed on that response, forced
+        after the deferral for the handoff output, would replace the
+        prefix and delete the dropped turns. Applying the filter now voids
+        the ownership token, so nothing records and the compaction skips
+        with the store intact.
+        """
+        prior_turns = self._prior_turns()
+        underlying = SimpleListSession(history=list(prior_turns))
+        session, mock_client = self._make_session(underlying, "handoff-filter")
+
+        def drop_prefix_history(data: HandoffInputData) -> HandoffInputData:
+            history = data.input_history
+            if not isinstance(history, str):
+                history = tuple(
+                    item
+                    for item in history
+                    if "prefix" not in str(item.get("content") if isinstance(item, dict) else item)
+                )
+            return HandoffInputData(
+                input_history=history,
+                pre_handoff_items=data.pre_handoff_items,
+                new_items=data.new_items,
+                run_context=data.run_context,
+            )
+
+        model = ScriptedModel()
+        delegate = Agent(name="delegate", model=model)
+        triage = Agent(
+            name="triage",
+            model=model,
+            handoffs=[handoff(delegate, input_filter=drop_prefix_history)],
+        )
+        model.extend(
+            [
+                ModelStep(output=[get_handoff_tool_call(delegate)], response_id="resp_hand"),
+                ModelStep(output=[get_text_message("delegate reply")], response_id="resp_post"),
+            ]
+        )
+
+        with caplog.at_level(logging.WARNING, logger="openai-agents.openai.compaction"):
+            await Runner.run(triage, "route this", session=session)
+
+        # The desync is real: the request before the handoff carried the
+        # oldest turns, the request after it did not, and the store kept
+        # them the whole time, so counts at persist time can no longer
+        # describe what the server saw.
+        assert len(model.calls) == 2
+        assert "prefix question" in str(model.calls[0].input)
+        post_handoff_request_text = str(model.calls[1].input)
+        assert "prefix question" not in post_handoff_request_text
+        assert "recent question" in post_handoff_request_text
+
+        # The stored history survived, including the turns the filter
+        # dropped from the post handoff requests; deleting them is the
+        # data loss under test.
+        stored = await underlying.get_items()
+        assert stored[: len(prior_turns)] == prior_turns
+        stored_text = str(stored)
+        assert "route this" in stored_text
+        assert "delegate reply" in stored_text
+
+        assert session._response_boundaries == {}
+        assert session._response_boundaries_ever_recorded is True
+        mock_client.responses.compact.assert_not_called()
+        assert "Skipped compaction for resp_post" in caplog.text
+
+    @pytest.mark.asyncio
+    async def test_nested_handoff_history_records_nothing_and_compaction_skips(
+        self, caplog: pytest.LogCaptureFixture
+    ) -> None:
+        """Nested handoff history must void the run's ownership like a filter.
+
+        The triage response hands off with nest_handoff_history enabled, so
+        the accumulated history is folded into one synthesized message that
+        renders the stored turns as wrapped transcript text. The traced
+        desync: the pre handoff request carried the stored turns as items,
+        the post handoff request carried none of them as items, only the
+        rendering, while the session kept every original turn because
+        session_step_items deliberately preserves the lossless history. The
+        count at persist time still spans the whole store, so on the prior
+        behavior the persist recorded a boundary for the post handoff
+        response and the compaction keyed on it, forced after the deferral
+        for the handoff output, replaced the lossless stored items with a
+        compaction of the rendering; summarized tool items survive only as
+        text and a custom handoff_history_mapper may drop anything outright.
+        Applying the nesting now voids the ownership token, unconditionally
+        like the input filter branch, so nothing records and the compaction
+        skips with the store intact.
+        """
+        prior_turns = self._prior_turns()
+        underlying = SimpleListSession(history=list(prior_turns))
+        session, mock_client = self._make_session(underlying, "nested-handoff")
+
+        model = ScriptedModel()
+        delegate = Agent(name="delegate", model=model)
+        triage = Agent(name="triage", model=model, handoffs=[handoff(delegate)])
+        model.extend(
+            [
+                ModelStep(output=[get_handoff_tool_call(delegate)], response_id="resp_hand"),
+                ModelStep(output=[get_text_message("nested reply")], response_id="resp_nest"),
+            ]
+        )
+
+        with caplog.at_level(logging.WARNING, logger="openai-agents.openai.compaction"):
+            await Runner.run(
+                triage,
+                "route this",
+                session=session,
+                run_config=RunConfig(nest_handoff_history=True),
+            )
+
+        # The desync is real: the request before the handoff carried the
+        # stored turns as items, the request after it carried exactly one
+        # synthesized message rendering them as wrapped transcript text, so
+        # counts at persist time can no longer describe item for item what
+        # the server saw.
+        assert len(model.calls) == 2
+        assert "prefix question" in str(model.calls[0].input)
+        post_request_items = model.calls[1].input
+        assert isinstance(post_request_items, list)
+        assert len(post_request_items) == 1
+        rendered_history = str(post_request_items[0])
+        assert "<CONVERSATION HISTORY>" in rendered_history
+        assert "prefix question" in rendered_history
+
+        # The stored history survived as real items, including everything
+        # the post handoff request only rendered as text; replacing it with
+        # a compaction of that rendering is the data loss under test.
+        stored = await underlying.get_items()
+        assert stored[: len(prior_turns)] == prior_turns
+        stored_text = str(stored)
+        assert "route this" in stored_text
+        assert "nested reply" in stored_text
+
+        assert session._response_boundaries == {}
+        assert session._response_boundaries_ever_recorded is True
+        mock_client.responses.compact.assert_not_called()
+        assert "Skipped compaction for resp_nest" in caplog.text
+
+    @pytest.mark.asyncio
+    async def test_unwindowed_run_still_records_and_replaces(self) -> None:
+        """Without a limit or callback the boundary records and compaction lands.
+
+        This pins the other side of the guard: the same seeded store, the
+        same runner flow, no windowing and no filtering, so the request
+        carried the entire stored history, the persist recorded the exact
+        boundary, and the replacement swapped the covered prefix for the
+        compacted output just as it did before the guard existed.
+        """
+        prior_turns = self._prior_turns()
+        underlying = SimpleListSession(history=list(prior_turns))
+        session, mock_client = self._make_session(underlying, "unwindowed-control")
+
+        model = ScriptedModel(
+            steps=[ModelStep(output=[get_text_message("plain reply")], response_id="resp_plain")]
+        )
+        worker = Agent(name="plain_worker", model=model)
+
+        await Runner.run(worker, "plain question", session=session)
+
+        assert model.last_call is not None
+        request_text = str(model.last_call.input)
+        assert "prefix question" in request_text
+        assert "plain question" in request_text
+
+        mock_client.responses.compact.assert_awaited_once()
+        assert (
+            mock_client.responses.compact.call_args.kwargs["previous_response_id"] == "resp_plain"
+        )
+        # Four prior items plus the persisted input and reply were covered, so
+        # the replacement left exactly the compacted output, and the boundary
+        # was translated onto the rewritten history.
+        assert await underlying.get_items() == [{"type": "compaction", "summary": "window only"}]
+        assert session._response_boundaries == {"resp_plain": 1}
+
+    @pytest.mark.asyncio
+    async def test_limit_covering_whole_store_still_records_and_replaces(self) -> None:
+        """A limit that admits the entire store keeps recording boundaries.
+
+        The window and the captured count match, so the request input covered
+        every stored item and the boundary stays provable. The persist then
+        records it exactly as an unwindowed run would, and the compaction
+        replaces the covered prefix. Only a limit that actually leaves items
+        out voids the token.
+        """
+        prior_turns = self._prior_turns()
+        underlying = SimpleListSession(history=list(prior_turns))
+        session, mock_client = self._make_session(underlying, "generous-limit")
+
+        model = ScriptedModel(
+            steps=[ModelStep(output=[get_text_message("roomy reply")], response_id="resp_roomy")]
+        )
+        worker = Agent(name="roomy_worker", model=model)
+
+        await Runner.run(
+            worker,
+            "roomy question",
+            session=session,
+            run_config=RunConfig(session_settings=SessionSettings(limit=50)),
+        )
+
+        assert model.last_call is not None
+        request_text = str(model.last_call.input)
+        assert "prefix question" in request_text
+        assert "roomy question" in request_text
+
+        mock_client.responses.compact.assert_awaited_once()
+        assert (
+            mock_client.responses.compact.call_args.kwargs["previous_response_id"] == "resp_roomy"
+        )
+        assert await underlying.get_items() == [{"type": "compaction", "summary": "window only"}]
+        assert session._response_boundaries == {"resp_roomy": 1}
+
+
 class TestTypeGuard:
     def test_is_compaction_aware_session_true(self) -> None:
         mock_underlying = MagicMock(spec=Session)