Anchor the thread fetch at the cursor, so a long thread stops going silent

The thread-seeding cursor had a limit nobody had named: the fetch is capped at
THREAD_CONTEXT_MESSAGE_LIMIT and returns the *newest* messages in the thread, so
once the cursor was older than that window it was no longer in the page, there
was no tail to find inside it, and every later turn answered without ever seeing
what the bot had said. The code degraded deliberately — deliver nothing rather
than re-seed and duplicate — which made it a permanent mute, not a degradation,
of exactly the feature that exists to keep the agent's own output reachable.

Anchoring fixes it rather than widening the window. Read against the Chatto
server's own handler: `after` walks forward from the cursor's stream sequence,
takes entries in ascending order and truncates to `limit`, so an anchored page is
the *oldest* messages past the cursor — the window and the thing being delivered
become the same window. No silent skipping, which an unstated assumption about
truncation direction could have introduced.

It also removes a constraint instead of adding a rule. A cursor no longer has to
be a position inside a window, only a boundary the server can walk from, so
`_seeded_through` no longer has to return "maybe unknowable" — the trigger is
always a valid cursor, which deleted `_messages_after`, the None case, the
unreachable-cursor fallback and the whole degradation branch with them.

And it saves a call: the server sets IncludeRoot false for an anchored page, and
the root is not needed anyway behind a cursor, so the separate root fetch is
skipped. One RPC per turn instead of two.

What is left is honest rather than silent: a backlog larger than the window drains
over turns, each turn taking the next slice after the cursor, and stopping there.

Tests: _thread_page models the server's three page shapes instead of returning a
fixed page, because a fixture that answers an anchored fetch with the thread's
start is precisely the bug being removed — six tests were passing against that
lie. History in a thread test now grows before the dispatch that posts into it, as
it does on a server. The empty-fetch test asserted the opposite of the new
invariant and was rewritten around it.

Four mutations, all caught: dropping the anchor, fetching the root anyway,
ignoring the limit, and the old unanchored slicing.

AGENTS.md

@@ -243,8 +243,11 @@ changes must keep them, or consciously renegotiate the docs.**
   thread root to the last message ID the session reached; the cap evicts the
   least recently touched thread. The cursor passes the trigger as well as the
   delivered block — the trigger is in the session, so a cursor left behind it
-  would hand the next turn the previous question a second time — and never
-  points past the fetched window, or the next fetch cannot find it again. The
+  would hand the next turn the previous question a second time. The
+  fetch is **anchored at the cursor** (`after=`), so a cursor is a boundary the
+  server walks forward from rather than a position inside a window, and a thread
+  that outgrew `THREAD_CONTEXT_MESSAGE_LIMIT` keeps answering instead of going
+  quiet for good. The
   cursor is **stored**, in memory and on the session
   (`chatto.thread_seed_cursor`), because a session's frontier is not derivable
   from the message list: a person who merely talked is in the session, the
@@ -258,9 +261,9 @@ changes must keep them, or consciously renegotiate the docs.**
   turn seeded nothing, no session yet, restart before the second turn — and that
   pass re-delivers the bot's own messages, leaves the people's messages to their
   own turns, and records the frontier; prefer that bounded duplicate to a
-  permanent loss. A cursor that has fallen out of the fetched window means the
-  thread outgrew `THREAD_CONTEXT_MESSAGE_LIMIT`: deliver nothing rather than
-  re-seed from the top, which would duplicate what the session holds. A failed
+  permanent loss. A backlog larger than the window drains over turns rather than
+  in one: each turn takes the next slice after the cursor and stops there, so
+  nothing is skipped and nothing repeats. A failed
   lookup leaves the thread unseeded so its next turn retries. Not discarded on
   reconnect, unlike the roster: thread history is immutable, so a refetch could
   only duplicate what the session already has. Note there

CONCEPTS.md

@@ -42,10 +42,21 @@ facts about it drive everything else:
    holds it; a cursor left behind it hands the next turn the previous question a
    second time, as an `[Earlier in this thread]` line beside a conversation that
    already contains it.
-2. **It cannot point past the fetched window.** The fetch is capped at
-   `THREAD_CONTEXT_MESSAGE_LIMIT`, and a cursor the next fetch does not contain
-   cannot be found again. When the trigger is missing from the window, the cursor
-   records what was actually delivered instead.
+2. **The fetch is anchored at the cursor**, which is what makes a cursor usable
+   rather than merely informative. An unanchored page returns the *newest*
+   messages in the thread; an anchored one (`after=<cursor>`) returns the
+   *oldest* messages past the cursor, walking forward. So the window and the
+   thing being delivered are the same window, and a thread that outgrew
+   `THREAD_CONTEXT_MESSAGE_LIMIT` keeps answering — where an unanchored page put
+   the cursor outside the window, left no tail to find inside it, and muted the
+   thread for good.
+
+   Two consequences ride along. The root is not part of an anchored page (the
+   server sets `IncludeRoot` false whenever `after` is set), and it is not needed
+   either — everything up to the cursor is already in the session — so the extra
+   root fetch is skipped: one RPC fewer per turn. And a backlog bigger than the
+   window drains over turns instead of arriving at once: each turn takes the next
+   slice and stops there, so nothing is skipped and nothing repeats.
 
 ### Why the frontier is stored, not derived
 
@@ -114,7 +125,9 @@ multiplexed gateway into `agent:main:` and names a session nobody is running.
 ### What this costs
 
 - One extra RPC per turn that delivers something: `get_thread_events`, capped at
-  20 messages. Quiet threads skip it — the cursor means there is nothing to fetch.
+  20 messages. Quiet threads skip it — the cursor means there is nothing to
+  fetch. It used to be two: the unanchored page omits the thread root, so the
+  root was fetched separately on top.
 - One session-metadata write per turn that delivered own output, plus one retry
   after a thread's first turn. Not per turn.
 - `[Earlier in this thread]` is reference text, not trusted input: only the

adapter.py

@@ -312,20 +312,6 @@ the thread's content is mirrored anywhere, and the value dies with the session.
 _T = TypeVar("_T")
 
 
-def _messages_after(history: list[Message], cursor_id: str) -> list[Message] | None:
-    """The messages ordered after ``cursor_id``, or ``None`` if it is not there.
-
-    ``None`` means the cursor fell out of the fetched window — the thread grew
-    past the limit this lookup fetches. The caller must not then re-seed from
-    the top: that repeats context the session already holds, which is exactly
-    what the cursor exists to prevent.
-    """
-    for index, message in enumerate(history):
-        if message.id == cursor_id:
-            return history[index + 1 :]
-    return None
-
-
 def hermes_adapter_factory(config: PlatformConfig):
     """Construct a ChattoAdapter from a PlatformConfig."""
     return ChattoAdapter(config)
@@ -1283,8 +1269,29 @@ class ChattoAdapter(BasePlatformAdapter):
         if not thread_root_event_id:
             return ""
 
+        session_key = self._session_key_for(source)
+
+        # The cursor is resolved *before* the fetch, because it anchors it. That
+        # is what makes a long thread work: the server returns the oldest messages
+        # past the anchor rather than the newest ones in the thread, so a thread
+        # that grew past the fetch limit still answers with what the session is
+        # missing instead of leaving the cursor outside the window.
+        cursor = self._seeded_threads.get(thread_root_event_id, _UNSEEDED)
+        if cursor is _UNSEEDED:
+            resumed = self._persisted_cursor(session_key, _UNSEEDED)
+            if isinstance(resumed, str):
+                # A cursor an earlier process left behind: it outlived that
+                # process with the session, so the thread goes on delivering its
+                # tail.
+                self._record_thread_seeded(thread_root_event_id, resumed)
+                cursor = resumed
+        anchored = cursor if isinstance(cursor, str) else ""
+
         history = await self._thread_history(
-            client, room_id=room_id, thread_root_event_id=thread_root_event_id
+            client,
+            room_id=room_id,
+            thread_root_event_id=thread_root_event_id,
+            after=anchored,
         )
         if history is None:
             logger.debug(
@@ -1293,26 +1300,12 @@ class ChattoAdapter(BasePlatformAdapter):
             )
             return ""
 
-        session_key = self._session_key_for(source)
-
-        if thread_root_event_id in self._seeded_threads:
+        if anchored:
+            # Everything the anchored page holds landed after the cursor, so it is
+            # the tail — no cutting point to find inside the window.
             return self._seed_thread_tail(
                 thread_root_event_id,
                 history,
-                self._seeded_threads[thread_root_event_id],
-                current_message_id,
-                session_key=session_key,
-            )
-
-        persisted = self._persisted_cursor(session_key, _UNSEEDED)
-        if persisted is not _UNSEEDED and isinstance(persisted, str):
-            # A cursor an earlier process left behind: it outlived that process
-            # with the session, so the thread goes on delivering its tail.
-            self._record_thread_seeded(thread_root_event_id, persisted)
-            return self._seed_thread_tail(
-                thread_root_event_id,
-                history,
-                persisted,
                 current_message_id,
                 session_key=session_key,
             )
@@ -1337,121 +1330,114 @@ class ChattoAdapter(BasePlatformAdapter):
                 thread_root_event_id,
                 len(own),
             )
-            cursor = self._seeded_through(history, own, current_message_id)
-            if cursor is not None:
-                self._record_thread_seeded(thread_root_event_id, cursor)
-                self._persist_thread_cursor(session_key, thread_root_event_id, cursor)
+            established = self._seeded_through(history, own, current_message_id)
+            if established is not None:
+                self._record_thread_seeded(thread_root_event_id, established)
+                self._persist_thread_cursor(
+                    session_key, thread_root_event_id, established
+                )
             return self._thread_context_block(own)
 
         # First turn into this thread: the session starts empty, so hand it the
         # whole thread before the reply.
         earlier = [m for m in history if m.id != current_message_id]
-        self._advance_thread_cursor(
-            session_key, thread_root_event_id, history, earlier, current_message_id
+        self._remember_thread_cursor(
+            session_key,
+            thread_root_event_id,
+            self._seeded_through(history, earlier, current_message_id),
+            delivered=earlier,
         )
         return self._thread_context_block(earlier)
 
     @staticmethod
     def _seeded_through(
         history: list[Message], delivered: list[Message], current_message_id: str
-    ) -> str | None:
-        """How far this turn's session now reaches, or None if the window cannot say.
+    ) -> str:
+        """How far this turn's session now reaches.
 
         The trigger is excluded from the *block* — it is the message text — but
         the session holds it, so the cursor passes it too. Leaving it behind
         would hand the next turn the previous question a second time, as an
         `[Earlier in this thread]` line beside the conversation that already
-        contains it. Only when the fetched window does not list the trigger: a
-        cursor beyond the window would be unreachable, so fall back to what was
-        actually delivered.
-
-        None when neither is in the window — an empty fetch. The trigger is *not*
-        a usable fallback there: recording it leaves a cursor the next fetch
-        cannot find, which then reads as a thread that grew past its cursor and
-        mutes it for good.
+        contains it.
+
+        The trigger is also the fallback when the fetched window does not list
+        it: the session holds this turn's own message either way. That is safe
+        because the fetch is anchored *at* the cursor rather than sliced around
+        it — a cursor names a boundary, not a position inside the window, so one
+        the window does not contain is still a usable cursor. That is what lets
+        this be a plain ``str`` instead of "maybe unknowable".
         """
         if any(m.id == current_message_id for m in history):
             return current_message_id
-        return delivered[-1].id if delivered else None
-
-    def _advance_thread_cursor(
-        self,
-        session_key: str,
-        thread_root_event_id: str,
-        history: list[Message],
-        delivered: list[Message],
-        current_message_id: str,
-    ) -> None:
-        """Record the frontier this turn established, if the window holds one.
-
-        A turn that cannot name a position leaves the thread unrecorded, which is
-        the state its next turn retries from — strictly better than recording a
-        cursor that can never be found again.
-        """
-        cursor = self._seeded_through(history, delivered, current_message_id)
-        if cursor is None:
-            logger.debug(
-                "Chatto: thread %s delivered nothing the fetched window can "
-                "name a cursor from; left unrecorded for the next turn",
-                thread_root_event_id,
-            )
-            return
-        self._remember_thread_cursor(
-            session_key, thread_root_event_id, cursor, delivered=delivered
-        )
+        return delivered[-1].id if delivered else current_message_id
 
     def _seed_thread_tail(
         self,
         thread_root_event_id: str,
         history: list[Message],
-        delivered_to: str,
         current_message_id: str,
         *,
         session_key: str,
     ) -> str:
-        """Deliver only what landed in an already-seeded thread since the cursor.
+        """Deliver what landed in an already-seeded thread since the cursor.
 
         This is the agent's own output the session never saw: a message it posted
         into the thread through a tool, whose realtime echo is filtered as a
         self-event. Without it the next human message in that thread is answered
         by an agent that does not know what it said.
+
+        ``history`` is anchored at the cursor, so every message in it is by
+        definition one the session does not hold — there is no cut point to find
+        inside the window, and no case where the window cannot answer.
         """
-        tail = _messages_after(history, delivered_to)
-        if tail is None:
-            # The cursor is no longer in the fetched window, so the thread grew
-            # past it and this process cannot tell what the session already has.
-            # Delivering nothing repeats the old behaviour; re-seeding from the
-            # top would duplicate history the session holds.
-            logger.debug(
-                "Chatto: thread %s grew past the seeded cursor %s; no tail delivered",
-                thread_root_event_id,
-                delivered_to,
-            )
-            return ""
-        unseen = [m for m in tail if m.id != current_message_id]
+        unseen = [m for m in history if m.id != current_message_id]
         # The cursor moves even when the trigger was all that arrived: it is in
         # the session, and a cursor left behind it would hand the next turn the
         # previous question a second time.
-        self._advance_thread_cursor(
-            session_key, thread_root_event_id, history, unseen, current_message_id
+        self._remember_thread_cursor(
+            session_key,
+            thread_root_event_id,
+            self._seeded_through(history, unseen, current_message_id),
+            delivered=unseen,
         )
         return self._thread_context_block(unseen)
 
     async def _thread_history(
-        self, client: ChattoClient, *, room_id: str, thread_root_event_id: str
+        self,
+        client: ChattoClient,
+        *,
+        room_id: str,
+        thread_root_event_id: str,
+        after: str = "",
     ) -> list[Message] | None:
         """Fetch a thread's messages oldest-first, or ``None`` if the lookup failed.
 
-        The thread root is fetched explicitly when the timeline omits it: a
-        cron message is a plain room message until someone replies to it, and
-        whether the server lists it back as the thread's first event is not
-        something the client signature promises.
+        ``after`` anchors the page at a cursor instead of at the thread's start,
+        and that is the whole point: the server then returns the *oldest* messages
+        past that cursor rather than the newest ones in the thread. A thread that
+        has grown past the fetch limit therefore still answers with exactly what
+        the session is missing, where an unanchored page would have left the
+        cursor outside the window and the turn with nothing to deliver.
+
+        Two things follow from the anchor, both from the server's own handler: the
+        root is not part of an anchored page (``IncludeRoot`` is false whenever
+        ``after`` is set), and it is not needed either — a cursor means
+        everything up to and including it is already in the session. So the extra
+        root fetch is skipped, which is one RPC fewer per turn than the unanchored
+        path.
+
+        When the gap is larger than the limit the page holds only its oldest
+        slice and the cursor stops at the last message delivered; the next turn
+        continues from there. A backlog drains over turns rather than in one, and
+        nothing is skipped either way.
         """
         try:
             page = await client.get_thread_events(
                 room_id,
                 thread_root_event_id,
                 limit=ChattoConstants.THREAD_CONTEXT_MESSAGE_LIMIT,
+                after=after or None,
             )
         except Exception:
             logger.debug(
@@ -1463,7 +1449,11 @@ class ChattoAdapter(BasePlatformAdapter):
             return None
 
         messages = [ev.message for ev in page.events if ev.message is not None]
-        if not any(m.id == thread_root_event_id for m in messages):
+        if not after and not any(m.id == thread_root_event_id for m in messages):
+            # The thread root is fetched explicitly when the unanchored timeline
+            # omits it: a cron message is a plain room message until someone
+            # replies to it, and whether the server lists it back as the thread's
+            # first event is not something the client signature promises.
             messages = (
                 await self._thread_root_message(
                     client, room_id=room_id, thread_root_event_id=thread_root_event_id

test_adapter.py

@@ -2164,6 +2164,56 @@ class TestSessionKey:
         assert adapter._session_key_for(source) == adapter._source_session_key(source)
 
 
+def _thread_page(
+    messages, thread_root_event_id, *, limit=None, before=None, after=None
+):
+    """``get_thread_events`` modelled on the server's own handler, not a fixture.
+
+    Three shapes, and the adapter's cursor logic depends on all three:
+
+    - unanchored: the root plus the **newest** ``limit`` replies,
+    - ``after=cursor``: the **oldest** ``limit`` replies past the cursor, exclusive,
+      and without the root — the server sets ``IncludeRoot`` false for an anchor,
+    - ``before``: the newest ``limit`` replies ahead of it.
+
+    A fixed ``TimelinePage`` fixture cannot stand in for this: it would answer an
+    anchored fetch with the thread's start, which is exactly the bug the anchor
+    exists to remove.
+    """
+    ordered = list(messages)
+    if after is not None:
+        index = next((i for i, m in enumerate(ordered) if m.id == after), None)
+        ahead = ordered[index + 1 :] if index is not None else []
+        selected = ahead[:limit] if limit else ahead
+    elif before is not None:
+        index = next((i for i, m in enumerate(ordered) if m.id == before), None)
+        head = ordered[:index] if index is not None else ordered
+        selected = head[-limit:] if limit else head
+    else:
+        root = next((m for m in ordered if m.id == thread_root_event_id), None)
+        newest = ordered[-limit:] if limit else ordered
+        selected = ([root] if root is not None else []) + [
+            m for m in newest if m.id != thread_root_event_id
+        ]
+    return TimelinePage(events=[_thread_event(m) for m in selected])
+
+
+def _serve_thread(adapter, messages, limit=None):
+    """Point the adapter's thread fetch at a message list the anchor can walk.
+
+    ``messages`` is the thread's whole history, oldest first, root included.
+    """
+    adapter._chatto_client.get_thread_events = AsyncMock(
+        side_effect=lambda room_id, thread_root_event_id, **kwargs: _thread_page(
+            messages,
+            thread_root_event_id,
+            limit=kwargs.get("limit") or limit,
+            before=kwargs.get("before"),
+            after=kwargs.get("after"),
+        )
+    )
+
+
 class _FakeSessionStore:
     """An in-memory session store with the real store's shape.
 
@@ -2262,9 +2312,10 @@ class TestThreadContext:
 
     async def test_second_turn_in_the_thread_repeats_no_context(self):
         """The session already holds what the first turn seeded — repeating it
-        would duplicate context the agent has. Seeding a later turn refetches to
-        deliver whatever landed since the cursor, so the invariant to hold is the
-        absence of duplication, not the absence of the lookup."""
+        would duplicate context the agent has. Seeding a later turn refetches,
+        anchored at the cursor, to deliver whatever landed since; so the
+        invariant to hold is the absence of duplication, not the absence of the
+        lookup."""
         adapter = self._adapter()
         self._seed_channel(adapter)
         cron = _make_message(
@@ -2273,12 +2324,16 @@ class TestThreadContext:
             actor_id="bot-user-id",
             created_at=datetime(2026, 10, 11, 8, 0, tzinfo=UTC),
         )
-        adapter._chatto_client.get_thread_events = AsyncMock(
-            return_value=TimelinePage(events=[_thread_event(cron)])
+        first = _make_message(
+            body="@hermes_bot first",
+            message_id="msg-1",
+            created_at=datetime(2026, 10, 11, 8, 1, tzinfo=UTC),
         )
+        history = [cron, first]
+        _serve_thread(adapter, history)
 
         await self._dispatch(
-            adapter, "@hermes_bot first", message_id="msg-1", thread_root="cron-1"
+            adapter, first.body, message_id="msg-1", thread_root="cron-1"
         )
         await self._dispatch(
             adapter, "@hermes_bot second", message_id="msg-2", thread_root="cron-1"
@@ -2294,6 +2349,10 @@ class TestThreadContext:
         on delivering its tail from exactly where it stopped — our own tool
         output arrives, and the human messages the session already holds do not.
         """
+
+        def at(minute: int):
+            return datetime(2026, 10, 11, 8, minute, tzinfo=UTC)
+
         store = _FakeSessionStore()
         before = self._adapter()
         self._seed_channel(before)
@@ -2302,24 +2361,16 @@ class TestThreadContext:
             body="Nightly build finished",
             message_id="cron-1",
             actor_id="bot-user-id",
-            created_at=datetime(2026, 10, 11, 8, 0, tzinfo=UTC),
-        )
-        human = _make_message(
-            body="@hermes_bot and the deploy?",
-            message_id="msg-1",
-            created_at=datetime(2026, 10, 11, 8, 5, tzinfo=UTC),
+            created_at=at(0),
         )
-        from_bot = _make_message(
-            body="Deployed, all green",
-            message_id="msg-tool",
-            actor_id="bot-user-id",
-            created_at=datetime(2026, 10, 11, 8, 6, tzinfo=UTC),
-        )
-        before._chatto_client.get_thread_events = AsyncMock(
-            return_value=TimelinePage(events=[_thread_event(cron)])
+        first = _make_message(
+            body="@hermes_bot and the deploy?", message_id="msg-1", created_at=at(1)
         )
+        history = [cron, first]
+        _serve_thread(before, history)
+
         await self._dispatch(
-            before, "@hermes_bot first", message_id="msg-1", thread_root="cron-1"
+            before, first.body, message_id="msg-1", thread_root="cron-1"
         )
         session_key = before._session_key_for(
             before.handle_message.await_args.args[0].source
@@ -2327,15 +2378,14 @@ class TestThreadContext:
         # The gateway creates the session once it has run a turn, which is what
         # lets the *next* turn's cursor land on it.
         store.create_session(session_key)
-        before._chatto_client.get_thread_events = AsyncMock(
-            return_value=TimelinePage(
-                events=[
-                    _thread_event(cron),
-                    _thread_event(human),
-                    _thread_event(from_bot),
-                ]
-            )
+
+        from_bot = _make_message(
+            body="Deployed, all green",
+            message_id="msg-tool",
+            actor_id="bot-user-id",
+            created_at=at(2),
         )
+        history.append(from_bot)
         await self._dispatch(
             before, "@hermes_bot thanks", message_id="msg-2", thread_root="cron-1"
         )
@@ -2345,41 +2395,22 @@ class TestThreadContext:
             "the pending cursor from turn one rides along on it rather than "
             "costing a second routing-table rewrite"
         )
-        later_tool_post = _make_message(
+
+        rolled_back = _make_message(
             body="Rolled back",
             message_id="msg-tool2",
             actor_id="bot-user-id",
-            created_at=datetime(2026, 10, 11, 8, 7, tzinfo=UTC),
-        )
-        before._chatto_client.get_thread_events = AsyncMock(
-            return_value=TimelinePage(
-                events=[
-                    _thread_event(cron),
-                    _thread_event(human),
-                    _thread_event(from_bot),
-                    _thread_event(later_tool_post),
-                ]
-            )
+            created_at=at(3),
         )
+        history.append(rolled_back)
         await self._dispatch(
             before, "@hermes_bot next", message_id="msg-3", thread_root="cron-1"
         )
+
         quiet = _make_message(
-            body="@hermes_bot and now?",
-            message_id="msg-4",
-            created_at=datetime(2026, 10, 11, 8, 8, tzinfo=UTC),
-        )
-        before._chatto_client.get_thread_events = AsyncMock(
-            return_value=TimelinePage(
-                events=[
-                    _thread_event(cron),
-                    _thread_event(human),
-                    _thread_event(from_bot),
-                    _thread_event(later_tool_post),
-                    _thread_event(quiet),
-                ]
-            )
+            body="@hermes_bot and now?", message_id="msg-4", created_at=at(4)
         )
+        history.append(quiet)
         await self._dispatch(
             before, quiet.body, message_id="msg-4", thread_root="cron-1"
         )
@@ -2396,28 +2427,15 @@ class TestThreadContext:
         assert dict(after._seeded_threads) == {}
         after_me = _make_message(
             body="Standing by",
-            message_id="msg-tool2",
+            message_id="msg-tool3",
             actor_id="bot-user-id",
-            created_at=datetime(2026, 10, 11, 8, 9, tzinfo=UTC),
+            created_at=at(5),
         )
         newest = _make_message(
-            body="@hermes_bot status please",
-            message_id="msg-5",
-            created_at=datetime(2026, 10, 11, 8, 10, tzinfo=UTC),
-        )
-        after._chatto_client.get_thread_events = AsyncMock(
-            return_value=TimelinePage(
-                events=[
-                    _thread_event(cron),
-                    _thread_event(human),
-                    _thread_event(from_bot),
-                    _thread_event(later_tool_post),
-                    _thread_event(quiet),
-                    _thread_event(after_me),
-                    _thread_event(newest),
-                ]
-            )
+            body="@hermes_bot status please", message_id="msg-5", created_at=at(6)
         )
+        history.extend([after_me, newest])
+        _serve_thread(after, history)
 
         await self._dispatch(
             after, newest.body, message_id="msg-5", thread_root="cron-1"
@@ -2471,12 +2489,16 @@ class TestThreadContext:
             actor_id="bot-user-id",
             created_at=datetime(2026, 10, 11, 8, 0, tzinfo=UTC),
         )
-        adapter._chatto_client.get_thread_events = AsyncMock(
-            return_value=TimelinePage(events=[_thread_event(cron)])
+        first = _make_message(
+            body="@hermes_bot first",
+            message_id="msg-1",
+            created_at=datetime(2026, 10, 11, 8, 1, tzinfo=UTC),
         )
+        history = [cron, first]
+        _serve_thread(adapter, history)
 
         await self._dispatch(
-            adapter, "@hermes_bot first", message_id="msg-1", thread_root="cron-1"
+            adapter, first.body, message_id="msg-1", thread_root="cron-1"
         )
         await self._dispatch(
             adapter, "@hermes_bot second", message_id="msg-2", thread_root="cron-1"
@@ -2684,18 +2706,13 @@ class TestThreadContext:
             actor_id="bot-user-id",
             created_at=datetime(2026, 10, 11, 8, 1, tzinfo=UTC),
         )
-        history = TimelinePage(events=[_thread_event(human), _thread_event(from_bot)])
-        adapter._chatto_client.get_thread_events = AsyncMock(
-            side_effect=[
-                TimelinePage(events=[_thread_event(human)]),
-                history,
-                history,
-            ]
-        )
+        history = [human]
+        _serve_thread(adapter, history)
 
         await self._dispatch(
             adapter, human.body, message_id="msg-1", thread_root="msg-1"
         )
+        history.append(from_bot)
         await self._dispatch(
             adapter, "@hermes_bot and now?", message_id="msg-2", thread_root="msg-1"
         )
@@ -2737,22 +2754,13 @@ class TestThreadContext:
             message_id="msg-2",
             created_at=datetime(2026, 10, 11, 8, 6, tzinfo=UTC),
         )
-        adapter._chatto_client.get_thread_events = AsyncMock(
-            side_effect=[
-                TimelinePage(events=[_thread_event(cron), _thread_event(first)]),
-                TimelinePage(
-                    events=[
-                        _thread_event(cron),
-                        _thread_event(first),
-                        _thread_event(second),
-                    ]
-                ),
-            ]
-        )
+        history = [cron, first]
+        _serve_thread(adapter, history)
 
         await self._dispatch(
             adapter, first.body, message_id="msg-1", thread_root="cron-1"
         )
+        history.append(second)
         await self._dispatch(
             adapter, second.body, message_id="msg-2", thread_root="cron-1"
         )
@@ -2907,36 +2915,36 @@ class TestThreadContext:
             "in the context, while repeating a turn the session already holds"
         )
 
-    async def test_an_empty_fetch_records_no_cursor_rather_than_an_unfindable_one(
+    async def test_an_empty_fetch_records_the_trigger_and_the_next_turn_recovers(
         self,
     ):
-        """An empty window names no position.
-
-        The trigger is not a usable stand-in for one: recording it leaves a cursor
-        the next fetch cannot find, which then reads as a thread that outgrew its
-        cursor and mutes it for good. Leaving the thread unrecorded means the next
-        turn tries again from the top.
+        """A fetch that names nothing still leaves an exact cursor behind.
+
+        The trigger is the turn's own message, so the session holds it whatever
+        the thread fetch returned — recording it is exact, not a guess. That is
+        only safe because the next fetch is *anchored at* the cursor: a cursor
+        names a boundary rather than a position inside the window, so one the
+        window did not contain is still usable, and the next turn resumes from
+        there instead of re-seeding the thread.
         """
-        store = _FakeSessionStore()
-        adapter = self._adapter()
-        self._seed_channel(adapter)
-        adapter.set_session_store(store)
-        # The thread fetch is empty and the root cannot be fetched either, so
-        # the window names no message at all. get_message answers per event: the
-        # trigger resolves (the turn dispatches normally), the root does not.
         trigger = _make_message(
             body="@hermes_bot anyone there?",
             message_id="msg-1",
             created_at=datetime(2026, 10, 11, 8, 0, tzinfo=UTC),
         )
-        adapter._chatto_client.get_thread_events = AsyncMock(
-            return_value=TimelinePage(events=[])
-        )
+        adapter = self._adapter()
+        self._seed_channel(adapter)
+        adapter.set_session_store(_FakeSessionStore())
+        # The thread fetch is empty and the root cannot be fetched either, so the
+        # window names no message at all. get_message answers per event: the
+        # trigger resolves (the turn dispatches normally), the root does not.
         adapter._chatto_client.get_message = AsyncMock(
             side_effect=lambda room_id, event_id: (
                 trigger if event_id == "msg-1" else None
             )
         )
+        history: list = []
+        _serve_thread(adapter, history)
         payload = _make_posted_payload(room_id="room-1", message_event_id="msg-1")
         payload.thread_root_event_id = "t1"
 
@@ -2945,22 +2953,109 @@ class TestThreadContext:
         assert adapter.handle_message.await_count == 1, (
             "the turn did not dispatch at all, so this proves nothing about the cursor"
         )
-        assert "t1" not in adapter._seeded_threads, (
-            "a cursor the fetched window cannot name was recorded; the next turn "
-            "would look for it, fail to find it, and take the thread silent"
-        )
-        # And the turn after it recovers instead of staying silent.
+        assert adapter._seeded_threads.get("t1") == "msg-1"
+
+        # And the next turn picks up from that cursor instead of re-seeding.
         human = _make_message(
             body="@hermes_bot still there?",
             message_id="msg-2",
             actor_id="user-1",
             created_at=datetime(2026, 10, 11, 8, 1, tzinfo=UTC),
         )
-        adapter._chatto_client.get_thread_events = AsyncMock(
-            return_value=TimelinePage(events=[_thread_event(human)])
-        )
+        history.append(human)
         await self._dispatch(adapter, human.body, message_id="msg-2", thread_root="t1")
         assert adapter._seeded_threads["t1"] == "msg-2"
+        assert "still there?" not in (
+            self._dispatched_event(adapter).channel_context or ""
+        ), (
+            "the turn's own trigger was delivered into its own context, so the "
+            "cursor did not pass it"
+        )
+
+    async def test_a_thread_longer_than_the_window_still_delivers_the_tool_output(self):
+        """The case the fetch anchor exists for.
+
+        The window is capped at THREAD_CONTEXT_MESSAGE_LIMIT, and an unanchored
+        page returns the *newest* messages in the thread — so once the cursor is
+        older than the window it is not in the page, there is no tail to find
+        inside it, and every later turn answers without ever seeing what the bot
+        said. That is the thread going permanently silent, not degrading.
+
+        Anchored, the server returns the oldest messages *past* the cursor, so the
+        window and the thing being delivered are the same window. A backlog larger
+        than the window then drains over turns instead of being lost: each turn
+        takes the next slice, and nothing is repeated.
+        """
+        limit = ChattoConstants.THREAD_CONTEXT_MESSAGE_LIMIT
+        adapter = self._adapter()
+        self._seed_channel(adapter)
+        store = _FakeSessionStore()
+        adapter.set_session_store(store)
+
+        def at(minute: int):
+            return datetime(2026, 10, 11, 8, minute, tzinfo=UTC)
+
+        root = _make_message(
+            body="@hermes_bot deploy thread", message_id="msg-root", created_at=at(0)
+        )
+        history = [root]
+        _serve_thread(adapter, history)
+        await self._dispatch(
+            adapter, root.body, message_id="msg-root", thread_root="msg-root"
+        )
+        store.create_session(
+            adapter._session_key_for(adapter.handle_message.await_args.args[0].source)
+        )
+
+        # Grow the thread well past the window with chatter the session already
+        # holds, then put the tool output beyond the window's reach.
+        for index in range(limit + 5):
+            history.append(
+                _make_message(
+                    body=f"chatter {index}",
+                    message_id=f"chatter-{index}",
+                    created_at=at(index + 1),
+                )
+            )
+        tool_post = _make_message(
+            body="Rolled the deploy back",
+            message_id="msg-tool",
+            actor_id="bot-user-id",
+            created_at=at(limit + 6),
+        )
+        ask = _make_message(
+            body="@hermes_bot status?", message_id="msg-ask", created_at=at(limit + 7)
+        )
+        history.extend([tool_post, ask])
+
+        await self._dispatch(
+            adapter, ask.body, message_id="msg-ask", thread_root="msg-root"
+        )
+
+        first = self._dispatched_event(adapter).channel_context or ""
+        assert "chatter 0" in first, (
+            "the anchored fetch did not come back with the oldest slice after the "
+            "cursor, which is the whole point of anchoring"
+        )
+        assert adapter._seeded_threads["msg-root"] == "chatter-19", (
+            "the cursor did not stop at the last message actually delivered, so "
+            "the next turn would skip the backlog"
+        )
+
+        nxt = _make_message(
+            body="@hermes_bot and now?", message_id="msg-next", created_at=at(limit + 8)
+        )
+        history.append(nxt)
+        await self._dispatch(
+            adapter, nxt.body, message_id="msg-next", thread_root="msg-root"
+        )
+
+        second = self._dispatched_event(adapter).channel_context or ""
+        assert "Rolled the deploy back" in second, (
+            "the backlog never drained: the tool output the session was missing "
+            "is still invisible after the thread grew past the window"
+        )
+        assert "chatter 0" not in second, "the delivered backlog was repeated"
 
     async def test_the_replying_message_is_not_seeded_into_its_own_context(self):
         """It is already the message text; repeating it in the block would tell