Rebuild the thread partition from the fetch instead of persisting a cursor

Code review reproduced two bugs in the cursor-based seeding, both mine:

- The cursor was left behind the turn's own trigger. The trigger is excluded
  from the context block but it IS in the session, so every turn handed the
  agent the previous question a second time, as an [Earlier in this thread]
  line beside the conversation that already contained it — in the headline
  scenario of a cron message, a human reply and the agent's answer.
- A thread whose first cursor never landed went silent forever after a
  restart. A first turn has no session yet, so the store refused that write;
  the cursor existed only in memory, and after a restart "session survives,
  cursor gone" was indistinguishable from "session predates this bookkeeping".

The second one is also why the persistence goes. Measured, not assumed:
set_session_metadata runs _update_entry -> _save() -> _persist_routing_data ->
replace_gateway_routing_entries, a full rewrite of the routing table under a
global save lock, per call. Correctness needs the cursor written on every turn,
so persisting it costs a full-table rewrite per turn — not a price worth
paying. What replaces it needs no writes at all, because the fetch is already
being made: _last_other_party_message returns the newest message that is not
ours, which is the last one the session certainly holds — every other person's
message arrives as its own turn, ours never does.

So the cursor stays in memory, a restart rebuilds the partition from the thread
itself, and neither bug has a hole to return through. Three tests pinned the
removed mechanism and are gone or rewritten; two new ones fail under mutation:
leaving the cursor behind the trigger, and treating a restart as leave-alone.

297 tests, all gates clean.

AGENTS.md

@@ -173,17 +173,16 @@ changes must keep them, or consciously renegotiate the docs.**
   is filtered as a self-event and never dispatched, so without the cursor it would
   be invisible to that thread for good, and the next human message would be
   answered by an agent that does not know what it said. `_seeded_threads` maps a
-  thread root to the last message ID delivered; the cap evicts the least recently
-  touched thread. The cursor also lives on the session
-  (`chatto.thread_seed_cursor` via `set_session_metadata`), so it outlives the
-  process — one opaque ID per thread, nothing of the thread's content mirrored
-  anywhere, and it dies with the session. It is written only when it moves: the
-  store saves the whole session index per call, and a first turn has no session
-  yet, so that first cursor never lands and the second turn writes it. A session
-  carrying no cursor predates this bookkeeping — history is in it, the cursor is
-  unknowable — so later turns deliver nothing rather than guess and duplicate a
-  block the session already holds. A cursor that
-  has fallen out of the fetched window means the thread outgrew
+  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. It is **not**
+  persisted: `set_session_metadata` rewrites the whole routing index per call,
+  which is not a price worth paying once per turn. A restart therefore drops it,
+  and `_last_other_party_message` rebuilds the partition from the fetch the turn
+  makes anyway: the newest message that is not ours is the last one the session
+  certainly holds, because every other person's message arrives as its own turn.
+  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 lookup leaves the
   thread unseeded so its next turn retries. Not discarded on reconnect, unlike

README.md

@@ -385,7 +385,7 @@ Credentials never come from the model: no tool takes a server URL or a token as
 
 Long messages are split at the same limit as every other send (see [Message Splitting](#message-splitting)). If a chunk fails after earlier ones were delivered, the tool reports what it already sent plus a `partial: true` marker, so a retry can repeat only the missing part.
 
-A message the tool posts is the bot's own, so Chatto echoes it back and the adapter drops it as a self-event — it never becomes a turn of its own. In a thread, the next human message still reaches the agent with what it said: each turn seeds the thread with whatever landed since the last turn, so a `chatto_send_room` reply is in the session the next time someone speaks there. That progress marker is kept on the session, so it survives a gateway restart.
+A message the tool posts is the bot's own, so Chatto echoes it back and the adapter drops it as a self-event — it never becomes a turn of its own. In a thread, the next human message still reaches the agent with what it said: each turn seeds the thread with whatever landed since the last turn, so a `chatto_send_room` reply is in the session the next time someone speaks there. The gateway restarting in between does not change that: the adapter rebuilds where the thread left off from the thread itself.
 
 ## Troubleshooting
 

adapter.py

@@ -295,15 +295,9 @@ class _RoomRoster:
 _UNSEEDED: Final = object()
 """Sentinel: this process has delivered nothing for that thread yet.
 
-Distinct from a recorded ``None``, which means the cursor is unknowable — an
-earlier process seeded the thread before this bookkeeping existed.
-"""
-
-_CURSOR_META_KEY = "chatto.thread_seed_cursor"
-"""Session-metadata key holding the last message ID seeded into that session.
-
-One opaque ID per thread, on the session the platform already owns: nothing of
-the thread's content is mirrored anywhere, and the value dies with the session.
+Only in memory. It is never persisted: the session store rewrites the whole
+routing index per call, which is not a price worth paying once per turn, and a
+cursor the process can rebuild from a fetch it already makes buys nothing.
 """
 
 
@@ -1284,22 +1278,7 @@ class ChattoAdapter(BasePlatformAdapter):
         if not thread_root_event_id:
             return ""
 
-        session_key = self._session_key_for(source)
         delivered_to = self._seeded_threads.get(thread_root_event_id, _UNSEEDED)
-        if delivered_to is _UNSEEDED:
-            # A cursor written by an earlier process outlives it: it lives on the
-            # session, so a restart keeps it and the thread can go on delivering
-            # its tail instead of going silent for good.
-            delivered_to = self._persisted_cursor(session_key, _UNSEEDED)
-            if delivered_to is not _UNSEEDED:
-                self._seeded_threads[thread_root_event_id] = delivered_to
-            elif self._thread_session_exists(source):
-                # This session predates the cursor bookkeeping entirely, so its
-                # history is in the session but the cursor is unknowable. Later
-                # turns deliver nothing rather than guess at one and duplicate a
-                # block the session already has.
-                self._record_thread_seeded(thread_root_event_id, None, session_key)
-                return ""
 
         history = await self._thread_history(
             client, room_id=room_id, thread_root_event_id=thread_root_event_id
@@ -1311,30 +1290,71 @@ class ChattoAdapter(BasePlatformAdapter):
             )
             return ""
 
-        if isinstance(delivered_to, str) or delivered_to is None:
-            return self._seed_thread_tail(
-                thread_root_event_id,
-                history,
-                delivered_to,
-                current_message_id,
-                session_key,
-            )
+        if delivered_to is _UNSEEDED:
+            if not self._thread_session_exists(source):
+                # 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._record_thread_seeded(
+                    thread_root_event_id,
+                    self._seeded_through(history, earlier, current_message_id),
+                )
+                return self._thread_context_block(earlier)
+            # An earlier process answered in this thread and its session
+            # survived, so the cursor died with that process. Rather than
+            # persist one (the store rewrites the whole routing index per call,
+            # which is not a price worth paying per turn), reconstruct the
+            # partition from the fetch already being done: the newest message
+            # that is not ours is the last one the session certainly holds,
+            # because every other person's message arrives as its own turn.
+            delivered_to = self._last_other_party_message(history)
+            if delivered_to is None:
+                return ""
+        elif not isinstance(delivered_to, str):
+            return ""
 
-        earlier = [m for m in history if m.id != current_message_id]
-        self._record_thread_seeded(
-            thread_root_event_id,
-            earlier[-1].id if earlier else current_message_id,
-            session_key,
+        return self._seed_thread_tail(
+            thread_root_event_id, history, delivered_to, current_message_id
         )
-        return self._thread_context_block(earlier)
+
+    def _last_other_party_message(self, history: list[Message]) -> str | None:
+        """The newest message in ``history`` that is not the bot's own, or None.
+
+        The partition point a lost cursor is rebuilt from. Every other person's
+        message became a turn of its own and is in the session; the bot's own
+        messages are the ones that never are, so everything after this point is
+        what the session is missing.
+        """
+        me = self.me
+        for message in reversed(history):
+            if me is None or message.actor_id != me.id:
+                return message.id
+        return None
+
+    @staticmethod
+    def _seeded_through(
+        history: list[Message], delivered: list[Message], current_message_id: str
+    ) -> str:
+        """How far this turn's session now reaches, for the cursor.
+
+        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.
+        """
+        if any(m.id == current_message_id for m in history):
+            return current_message_id
+        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 | None | object,
+        delivered_to: str,
         current_message_id: str,
-        session_key: str,
     ) -> str:
         """Deliver only what landed in an already-seeded thread since the cursor.
 
@@ -1343,8 +1363,6 @@ class ChattoAdapter(BasePlatformAdapter):
         self-event. Without it the next human message in that thread is answered
         by an agent that does not know what it said.
         """
-        if not isinstance(delivered_to, str):
-            return ""
         tail = _messages_after(history, delivered_to)
         if tail is None:
             # The cursor is no longer in the fetched window, so the thread grew
@@ -1358,9 +1376,13 @@ class ChattoAdapter(BasePlatformAdapter):
             )
             return ""
         unseen = [m for m in tail if m.id != current_message_id]
-        if not unseen:
-            return ""
-        self._record_thread_seeded(thread_root_event_id, unseen[-1].id, session_key)
+        # 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._record_thread_seeded(
+            thread_root_event_id,
+            self._seeded_through(history, unseen, current_message_id),
+        )
         return self._thread_context_block(unseen)
 
     async def _thread_history(
@@ -1425,68 +1447,17 @@ class ChattoAdapter(BasePlatformAdapter):
         return [root] if root is not None and not root.deleted_at else []
 
     def _record_thread_seeded(
-        self, thread_root_event_id: str, last_delivered_id: str | None, session_key: str
+        self, thread_root_event_id: str, last_delivered_id: str
     ) -> None:
         """Note that this thread's session holds its history up to that message.
 
-        ``None`` when the cursor is unknowable — an earlier process seeded the
-        thread before this bookkeeping existed. Re-inserting keeps the map
-        insertion-ordered, which is what the cap below evicts from: the least
-        recently touched thread.
+        Re-inserting keeps the map insertion-ordered, which is what the cap below
+        evicts from: the least recently touched thread.
         """
         self._seeded_threads.pop(thread_root_event_id, None)
         self._seeded_threads[thread_root_event_id] = last_delivered_id
         while len(self._seeded_threads) > ChattoConstants.SEEN_CAP:
             self._seeded_threads.pop(next(iter(self._seeded_threads)))
-        self._persist_cursor(session_key, last_delivered_id)
-
-    def _persisted_cursor(self, session_key: str, default: Any) -> Any:
-        """The cursor a previous process left on this session, or ``default``.
-
-        Best-effort bookkeeping, so every failure answers ``default``: a missing
-        store, a store without metadata support, a rejected read. That degrades
-        to seeding as if the thread were new, never to a failed turn.
-        """
-        store = getattr(self, "_session_store", None)
-        reader = getattr(store, "get_session_metadata", None)
-        if not callable(reader):
-            return default
-        try:
-            value = reader(session_key, _CURSOR_META_KEY, None)
-        except Exception:
-            logger.debug(
-                "Chatto: reading the seed cursor for %s failed",
-                session_key,
-                exc_info=True,
-            )
-            return default
-        return default if value is None else value
-
-    def _persist_cursor(self, session_key: str, cursor: str | None) -> None:
-        """Mirror the cursor onto the session so it outlives this process.
-
-        Only when it moves: the store saves the whole session index on every
-        call, and an unchanged cursor has nothing to say. A refusal costs
-        nothing — the in-memory cursor carries the thread until it moves, and
-        that is the state this call is only a backup of. There is no session
-        yet on a thread's first turn, which is why the first cursor never lands.
-        """
-        store = getattr(self, "_session_store", None)
-        writer = getattr(store, "set_session_metadata", None)
-        if not callable(writer):
-            return
-        try:
-            if not writer(session_key, _CURSOR_META_KEY, cursor):
-                logger.debug(
-                    "Chatto: session %s declined the seed cursor (no session yet?)",
-                    session_key,
-                )
-        except Exception:
-            logger.debug(
-                "Chatto: persisting the seed cursor for %s failed",
-                session_key,
-                exc_info=True,
-            )
 
     def _thread_session_exists(self, source: Any) -> bool:
         """Whether the gateway already has a session for this source.

test_adapter.py

@@ -2185,12 +2185,13 @@ class TestThreadContext:
         second = adapter.handle_message.await_args.args[0]
         assert "Nightly build finished" not in (second.channel_context or "")
 
-    async def test_a_restart_does_not_reseed_a_thread_whose_session_survived(self):
-        """A session that carries no cursor is one this bookkeeping predates.
+    async def test_a_restart_rebuilds_the_partition_instead_of_reseeding(self):
+        """A restart drops the in-memory cursor while the session survives.
 
-        Its history is in the session and the cursor is unknowable, so the block
-        is not re-delivered: re-seeding would repeat context the session already
-        holds, which is the one thing the cursor exists to prevent.
+        It is rebuilt from the fetch already being done: the newest message that
+        is not ours is the last one the session certainly holds, so everything
+        after it — our own tool output — is delivered, and the human messages
+        the session already has are not repeated.
         """
         before = self._adapter()
         self._seed_channel(before)
@@ -2200,6 +2201,17 @@ class TestThreadContext:
             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),
+        )
+        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)])
         )
@@ -2210,150 +2222,37 @@ class TestThreadContext:
             before.handle_message.await_args.args[0].source
         )
 
-        # A fresh adapter: the deque is gone, as it is after any restart. The
+        # A fresh adapter: the cursor is gone, as it is after any restart. The
         # session store still resolves the key — it is persistent.
         after = self._adapter()
         self._seed_channel(after)
         assert dict(after._seeded_threads) == {}
         after.set_session_store(
-            MagicMock(
-                peek_session_id=MagicMock(return_value="sess-7"),
-                # A session that predates cursor bookkeeping carries none — the
-                # case this asserts: history is in the session, cursor unknown.
-                get_session_metadata=MagicMock(return_value=None),
-                set_session_metadata=MagicMock(return_value=True),
-            )
+            MagicMock(peek_session_id=MagicMock(return_value="sess-7"))
         )
         after._chatto_client.get_thread_events = AsyncMock(
-            return_value=TimelinePage(events=[_thread_event(cron)])
+            return_value=TimelinePage(
+                events=[
+                    _thread_event(cron),
+                    _thread_event(human),
+                    _thread_event(from_bot),
+                ]
+            )
         )
 
         await self._dispatch(
             after, "@hermes_bot after restart", message_id="msg-2", thread_root="cron-1"
         )
 
-        after._chatto_client.get_thread_events.assert_not_called()
-        assert after._chatto_client.get_thread_events.await_count == 0
         resumed = after.handle_message.await_args.args[0]
-        assert "Nightly build finished" not in (resumed.channel_context or "")
+        context = resumed.channel_context or ""
+        assert "Deployed, all green" in context
+        assert "and the deploy?" not in context, (
+            "the human message the session already holds was delivered again"
+        )
         # Sanity: the key the adapter asks about is the one the gateway used.
         assert after._session_store.peek_session_id.call_args.args == (session_key,)
 
-    async def test_a_persisted_cursor_survives_a_restart(self):
-        """The cursor lives on the session, so a restart keeps it. Without that,
-        a thread answered in before the restart falls back to delivering
-        nothing at all — and a chatto_send_room message posted into it while the
-        gateway was down stays invisible for the rest of the thread's life."""
-        adapter = self._adapter()
-        self._seed_channel(adapter)
-        human = _make_message(
-            body="@hermes_bot status?",
-            message_id="msg-1",
-            actor_id="user-1",
-            created_at=datetime(2026, 10, 11, 8, 0, tzinfo=UTC),
-        )
-        from_bot = _make_message(
-            body="All green, deploying now",
-            message_id="msg-tool",
-            actor_id="bot-user-id",
-            created_at=datetime(2026, 10, 11, 8, 1, tzinfo=UTC),
-        )
-        adapter.set_session_store(
-            MagicMock(
-                peek_session_id=MagicMock(return_value="sess-7"),
-                get_session_metadata=MagicMock(return_value="msg-1"),
-                set_session_metadata=MagicMock(return_value=True),
-            )
-        )
-        adapter._chatto_client.get_thread_events = AsyncMock(
-            return_value=TimelinePage(
-                events=[_thread_event(human), _thread_event(from_bot)]
-            )
-        )
-
-        await self._dispatch(
-            adapter, "@hermes_bot any news?", message_id="msg-2", thread_root="msg-1"
-        )
-
-        context = self._dispatched_event(adapter).channel_context or ""
-        assert "All green, deploying now" in context, (
-            "the persisted cursor was not honoured — a thread answered in "
-            "before the restart delivers nothing again"
-        )
-
-    async def test_the_cursor_is_written_to_the_session_only_when_it_moves(self):
-        """A write per turn would mean a full session-index save per turn in a
-        live thread. The cursor has to be persisted exactly when it advances."""
-        adapter = self._adapter()
-        self._seed_channel(adapter)
-        store = MagicMock(
-            peek_session_id=MagicMock(return_value=None),
-            get_session_metadata=MagicMock(return_value=None),
-            set_session_metadata=MagicMock(return_value=True),
-        )
-        adapter.set_session_store(store)
-        human = _make_message(
-            body="@hermes_bot hi",
-            message_id="m-0",
-            actor_id="user-1",
-            created_at=datetime(2026, 10, 11, 8, 0, tzinfo=UTC),
-        )
-        from_bot = _make_message(
-            body="All green",
-            message_id="m-tool",
-            actor_id="bot-user-id",
-            created_at=datetime(2026, 10, 11, 8, 1, tzinfo=UTC),
-        )
-        adapter._chatto_client.get_thread_events = AsyncMock(
-            side_effect=[
-                TimelinePage(events=[_thread_event(human)]),
-                TimelinePage(events=[_thread_event(human), _thread_event(from_bot)]),
-                TimelinePage(events=[_thread_event(human), _thread_event(from_bot)]),
-            ]
-        )
-
-        await self._dispatch(
-            adapter, "@hermes_bot first", message_id="msg-1", thread_root="m-0"
-        )
-        assert store.set_session_metadata.call_count == 1
-
-        # This turn delivers the tool message, so the cursor moves — a write.
-        store.set_session_metadata.reset_mock()
-        await self._dispatch(
-            adapter, "@hermes_bot news?", message_id="msg-2", thread_root="m-0"
-        )
-        assert "All green" in (self._dispatched_event(adapter).channel_context or "")
-        store.set_session_metadata.assert_called_once()
-
-        # Nothing new this time: the cursor does not move, so nothing is written.
-        store.set_session_metadata.reset_mock()
-        await self._dispatch(
-            adapter, "@hermes_bot more?", message_id="msg-3", thread_root="m-0"
-        )
-        store.set_session_metadata.assert_not_called()
-
-    async def test_a_session_store_that_refuses_the_cursor_costs_nothing(self):
-        """The store is best-effort bookkeeping. A False from it must not fail
-        the turn — the in-memory cursor still carries the thread meanwhile."""
-        adapter = self._adapter()
-        self._seed_channel(adapter)
-        adapter.set_session_store(
-            MagicMock(
-                peek_session_id=MagicMock(return_value="sess-7"),
-                get_session_metadata=MagicMock(return_value=None),
-                set_session_metadata=MagicMock(return_value=False),
-            )
-        )
-        adapter._chatto_client.get_thread_events = AsyncMock(
-            return_value=TimelinePage(events=[])
-        )
-
-        await self._dispatch(
-            adapter, "@hermes_bot hi", message_id="msg-1", thread_root="m-0"
-        )
-
-        assert self._dispatched_event(adapter) is not None
-
     async def test_a_thread_whose_session_is_unknown_still_seeds(self):
         """No session bound to the key means the session is empty — the exact
         situation seeding exists for. Fails open: seed rather than answer a
@@ -2634,6 +2533,108 @@ class TestThreadContext:
             "advance, so every turn repeats it"
         )
 
+    async def test_the_previous_turns_trigger_is_not_re_delivered(self):
+        """The trigger is not seeded into its own context — but it IS in the
+        session, as the message that turn was built from. So the cursor has to
+        advance past it too, or the next turn hands the agent the previous
+        question a second time, as an [Earlier in this thread] line next to the
+        conversation that already contains it."""
+        adapter = self._adapter()
+        self._seed_channel(adapter)
+        cron = _make_message(
+            body="Nightly build finished",
+            message_id="cron-1",
+            actor_id="bot-user-id",
+            created_at=datetime(2026, 10, 11, 8, 0, tzinfo=UTC),
+        )
+        first = _make_message(
+            body="@hermes_bot and the deploy?",
+            message_id="msg-1",
+            created_at=datetime(2026, 10, 11, 8, 5, tzinfo=UTC),
+        )
+        second = _make_message(
+            body="@hermes_bot any news?",
+            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),
+                    ]
+                ),
+            ]
+        )
+
+        await self._dispatch(
+            adapter, first.body, message_id="msg-1", thread_root="cron-1"
+        )
+        await self._dispatch(
+            adapter, second.body, message_id="msg-2", thread_root="cron-1"
+        )
+
+        context = self._dispatched_event(adapter).channel_context or ""
+        assert "and the deploy?" not in context, (
+            "turn 1's own trigger came back in turn 2's context: the cursor was "
+            "left behind the message the session already holds"
+        )
+
+    async def test_a_lost_cursor_is_rebuilt_after_a_restart(
+        self,
+    ):
+        """A thread whose in-memory cursor is gone and whose session survived.
+
+        The cursor is rebuilt from the fetch rather than trusted, so the thread
+        keeps delivering its own output instead of going silent for good."""
+        before = self._adapter()
+        self._seed_channel(before)
+        human = _make_message(
+            body="@hermes_bot status?",
+            message_id="msg-1",
+            actor_id="user-1",
+            created_at=datetime(2026, 10, 11, 8, 0, tzinfo=UTC),
+        )
+        from_bot = _make_message(
+            body="All green, deploying now",
+            message_id="msg-tool",
+            actor_id="bot-user-id",
+            created_at=datetime(2026, 10, 11, 8, 1, tzinfo=UTC),
+        )
+        before.set_session_store(
+            MagicMock(peek_session_id=MagicMock(return_value=None))
+        )
+        before._chatto_client.get_thread_events = AsyncMock(
+            return_value=TimelinePage(events=[_thread_event(human)])
+        )
+        await self._dispatch(
+            before, human.body, message_id="msg-1", thread_root="msg-1"
+        )
+
+        # Restart: a fresh adapter, the session survived, the cursor with it?
+        after = self._adapter()
+        self._seed_channel(after)
+        after.set_session_store(
+            MagicMock(peek_session_id=MagicMock(return_value="sess-1"))
+        )
+        after._chatto_client.get_thread_events = AsyncMock(
+            return_value=TimelinePage(
+                events=[_thread_event(human), _thread_event(from_bot)]
+            )
+        )
+        await self._dispatch(
+            after, "@hermes_bot any news?", message_id="msg-2", thread_root="msg-1"
+        )
+
+        context = self._dispatched_event(after).channel_context or ""
+        assert "All green, deploying now" in context, (
+            "the thread went silent after a restart: a lost cursor must be "
+            "rebuilt from the fetch, never treated as a session to leave alone"
+        )
+
     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
         the agent the user said it twice."""