@@ -52,14 +52,16 @@ class _SessionOwnershipToken:
 
     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: a windowed read left the oldest items out, 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.
+    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")
 
@@ -405,12 +405,12 @@ async def prepare_input_with_session(
 
     ``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: a resolved ``limit`` windows the history read,
-    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.
+    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:
@@ -426,19 +426,19 @@ async def prepare_input_with_session(
             limit=resolved_settings.limit,
             wrapper=wrapper,
         )
-        if ownership_token is not None and len(history) != ownership_token.count:
-            # The windowed read returned something other than the exact items
-            # the token was captured over, so the request input either leaves
-            # out the oldest stored items or was built over an interleaved
-            # write; either way no recorded count could describe what the
-            # server saw. Void the token so this run records no boundaries.
-            # When the counts match, the window held the entire store at
-            # capture time, and the token's generation and count checks at
-            # persist still prove nothing changed in between, so recording
-            # stays exactly as sound as an unwindowed read.
-            ownership_token.invalidate()
     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
 
@@ -3388,8 +3388,9 @@ def model_dump(self) -> Any:
 class TestPartialRequestInputCompaction:
     """Runs whose request input skips stored history must not record boundaries.
 
-    A resolved session limit windows the history read, a session input
-    callback rebuilds the prepared input outright, a call_model_input_filter
+    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
@@ -3541,6 +3542,75 @@ async def test_session_level_limit_records_nothing_and_compaction_skips(
         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