Keep the thread frontier on the session; the message list cannot supply it

The previous commit rebuilt a lost cursor from the fetch as "the newest message
that is not ours is the last one the session holds". That premise is wrong, and
the fetch it reasoned over is the shape a thread reply always produces: the
window ends with the turn's own trigger, a person's message that became this
turn seconds ago. So the rebuilt partition landed on the trigger, the tail came
out empty, and every bot message before it was skipped — after a restart, the
agent's own output in a thread went missing precisely when it was needed.
Excluding the trigger does not fix it either: the *previous* turn's trigger is
indistinguishable from a message the session holds, so the next restart skips
again. A session's frontier is simply not a function of the message list.

So the frontier is stored where it can be read exactly, on the session under
chatto.thread_seed_cursor. set_session_metadata rewrites the whole routing index
per call, so the write is spent only where it buys something: a turn that
delivered the bot's own output — the one thing no later turn can recover from
the list — and the turn after a first turn whose cursor found no session to land
on. A turn that only carried the trigger writes nothing.

One case remains where no cursor exists anywhere: the first turn seeded nothing
(so there was nothing to write), no session existed yet, and the process
restarted before the second turn. A thread is blind exactly once. That pass
re-delivers the bot's own messages, leaves people's messages to their own turns,
and records the frontier so no later process is guessing. It can duplicate one
block, in the rarer variant where that first seed did carry own output; losing
the output instead is the common case and permanent, so the duplicate is the
cheaper error.

Tests use _FakeSessionStore instead of MagicMock: the refusal that a first turn
runs into (no session under that key) is load-bearing, and a mock returns a
Mock for every call, so it could never have shown it. Five mutations, all caught:
delivering nothing in the blind pass, ignoring the persisted cursor, writing on
every turn, not recognising own output as unrecoverable, and letting the blind
pass deliver other people's messages too.

AGENTS.md

@@ -177,18 +177,28 @@ 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. 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
-  the roster: thread history is immutable, so a refetch could only duplicate what
-  the session already has. Note there
+  would hand the next turn the previous question a second time. It is **also**
+  written to the session (`chatto.thread_seed_cursor`), because a restart drops
+  the in-memory map while the session survives — and the frontier is **not
+  derivable from the message list**, which is the trap: the newest message that
+  is not ours is the turn's own trigger (a person's message that just became this
+  turn), so rebuilding a partition from the fetch skips every bot message before
+  it, and the *previous* turn's trigger looks just as much like a message the
+  session holds as the bot's own output looks like one it does not. Persisting
+  costs a full routing-index rewrite per call, so it is paid only where it buys
+  something: on a turn that delivered the bot's own output (the one thing a later
+  turn cannot recover from the list), and on the turn after a first turn whose
+  cursor found no session to land on (`_unpersisted_threads`). A thread can
+  therefore be *blind* exactly once — first 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 so no process after it is guessing. 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
+  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
   is no session-reset path on this platform — `/join` and `/leave` are the
   only commands and they ride DMs — so an emptied session cannot strand a
   thread the way it would elsewhere; do not assume one exists.

README.md

@@ -220,7 +220,7 @@ This matters most for a reply to something the bot posted itself. A cron job's m
 
 Messages in the thread from people outside `CHATTO_ALLOWED_USERS` are prefixed `(unverified)`. Only the message that triggers a turn passes the auth gate, so the rest of a shared thread is reference text, not trusted input; an author the adapter has never seen is marked unverified too. With `CHATTO_ALLOW_ALL_USERS=true` nothing is marked.
 
-Seeding happens once per session, not per message and not per restart: later turns in the thread need no block because the session already holds that history, and neither does the first turn after the Hermes gateway restarts. If the lookup fails, the turn proceeds without it and the next turn in that thread retries. Direct messages never thread and never seed.
+Seeding happens once per session, not per message and not per restart: later turns in the thread need no block because the session already holds that history, and neither does the first turn after the Hermes gateway restarts — the plugin remembers how far each thread's session reached. If the lookup fails, the turn proceeds without it and the next turn in that thread retries. Direct messages never thread and never seed.
 
 ### Room Member Roster
 
@@ -386,7 +386,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. The gateway restarting in between does not change that: the adapter rebuilds where the thread left off from the thread itself.
+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: where the thread left off is remembered on the session, not just in the running process.
 
 ## Troubleshooting
 

adapter.py

@@ -38,7 +38,7 @@ from dataclasses import dataclass, field
 from datetime import UTC, datetime
 from difflib import SequenceMatcher
 from enum import StrEnum
-from typing import Any, Final, NamedTuple
+from typing import Any, Final, NamedTuple, TypeVar
 from urllib.parse import unquote, urlsplit
 
 import httpx
@@ -293,14 +293,26 @@ class _RoomRoster:
 
 
 _UNSEEDED: Final = object()
-"""Sentinel: this process has delivered nothing for that thread yet.
+"""Sentinel: nothing is known yet about how far this thread's session reaches.
 
-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.
+The in-memory map starts every thread here. The session's own copy starts here
+too — a session carrying no cursor is one whose frontier cannot be established
+(see _thread_context_for), which is not the same as a thread this process has
+not seeded.
 """
 
 
+_CURSOR_META_KEY: Final = "chatto.thread_seed_cursor"
+"""Session-metadata key holding the last message ID a thread's session reached.
+
+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.
+"""
+
+
+_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.
 
@@ -387,6 +399,9 @@ class ChattoAdapter(BasePlatformAdapter):
         # seen-ring; unlike the roster this is deliberately NOT discarded on
         # reconnect, since thread history is immutable.
         self._seeded_threads: dict[str, str | None] = {}
+        # Threads whose cursor is on its way to the session but has not landed:
+        # a thread's first turn has no session yet, and the second turn retries.
+        self._unpersisted_threads: set[str] = set()
         # Keyed by thread root, which assumes one session per thread. With
         # thread_sessions_per_user turned on, each participant gets their own
         # session while this map stays shared, so one participant's turn moves
@@ -1272,12 +1287,13 @@ class ChattoAdapter(BasePlatformAdapter):
         that thread does not exist yet, so there is nothing to fetch. DMs never
         thread and never seed.
 
-        Seeding is asked of the session store, not just the in-memory deque:
-        the marker is process-local and dies on every restart, while the
-        thread's session lives in state.db — so a restart would otherwise
-        re-seed every thread in the room and duplicate a context the session
-        already holds. The deque remains as a cheap timing pre-check for two
-        turns racing before the first session is written.
+        The cursor is kept twice: in memory for this process, and on the session
+        for the next one. Only the second one survives a restart, and it is not
+        a redundant copy — a session's frontier is *not* derivable from the
+        thread's messages, because the bot's own messages belong to no session
+        while the previous turn's trigger looks exactly like one the session
+        holds. Rebuilding a partition from the fetch therefore lands on the
+        wrong message and skips the very output this exists to deliver.
 
         Best-effort by design, like the roster: a failed lookup leaves the
         thread unseeded so the next turn retries.
@@ -1285,8 +1301,6 @@ class ChattoAdapter(BasePlatformAdapter):
         if not thread_root_event_id:
             return ""
 
-        delivered_to = self._seeded_threads.get(thread_root_event_id, _UNSEEDED)
-
         history = await self._thread_history(
             client, room_id=room_id, thread_root_event_id=thread_root_event_id
         )
@@ -1297,46 +1311,68 @@ class ChattoAdapter(BasePlatformAdapter):
             )
             return ""
 
-        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 ""
+        session_key = self._session_key_for(source)
 
-        return self._seed_thread_tail(
-            thread_root_event_id, history, delivered_to, current_message_id
-        )
+        if thread_root_event_id in self._seeded_threads:
+            delivered_to = self._seeded_threads[thread_root_event_id]
+            if not isinstance(delivered_to, str):
+                return ""
+            return self._seed_thread_tail(
+                thread_root_event_id,
+                history,
+                delivered_to,
+                current_message_id,
+                session_key=session_key,
+            )
 
-    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.
+        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,
+            )
 
-        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
+        if self._thread_session_exists(source):
+            # A session that survived a restart but carries no cursor: a thread
+            # whose first turn had nothing to seed wrote no cursor (there was no
+            # session to write it to) and never got a second turn to retry on.
+            #
+            # Its frontier is not derivable from the message list, so this
+            # delivers the one thing that is provably at risk — the bot's own
+            # messages, which belong to no session — and leaves the people's
+            # messages alone, since those became turns. Duplicating the former
+            # is possible (a first seed that did carry them, same lost second
+            # turn) and is bounded to that one block; losing it is the common
+            # case and permanent. The cursor is written either way, so the thread
+            # is blind once, not for the rest of its life.
+            own = [m for m in history if m.id != current_message_id and self._is_own(m)]
+            logger.debug(
+                "Chatto: thread %s has a session without a seed cursor; "
+                "re-delivering %d own message(s)",
+                thread_root_event_id,
+                len(own),
+            )
+            cursor = self._seeded_through(history, own, current_message_id)
+            self._record_thread_seeded(thread_root_event_id, cursor)
+            self._persist_thread_cursor(session_key, thread_root_event_id, cursor)
+            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._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(
@@ -1362,6 +1398,8 @@ class ChattoAdapter(BasePlatformAdapter):
         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.
 
@@ -1386,9 +1424,11 @@ class ChattoAdapter(BasePlatformAdapter):
         # 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(
+        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)
 
@@ -1453,11 +1493,107 @@ class ChattoAdapter(BasePlatformAdapter):
             return []
         return [root] if root is not None and not root.deleted_at else []
 
+    def _remember_thread_cursor(
+        self,
+        session_key: str,
+        thread_root_event_id: str,
+        cursor: str | None,
+        *,
+        delivered: list[Message],
+    ) -> None:
+        """Record how far a thread's session reaches, and mirror it if it must.
+
+        The in-memory map is authoritative while this process lives. The session
+        copy exists for one thing — a restart, where the map is empty and the
+        session survived — and it is written when either of two things holds:
+
+        - the turn put the bot's own output into the session. That is the one
+          thing a later turn cannot recover from the message list, so it is the
+          one cursor that has to outlive us. Every other turn's frontier is the
+          newest message that is not ours, and a restart that loses such a cursor
+          delivers nothing, which is right because that turn delivered nothing
+          worth losing.
+        - an earlier turn's cursor is still waiting for a session to land on. A
+          thread's first turn has none — and its seed is exactly the case above,
+          a cron message delivered to a session that did not exist yet. By the
+          second turn the session exists.
+
+        Writing unconditionally would cost a full routing-table rewrite per turn
+        (see _persist_thread_cursor) to save a case that costs nothing.
+        """
+        self._record_thread_seeded(thread_root_event_id, cursor)
+        unrecoverable = any(self._is_own(message) for message in delivered)
+        if unrecoverable or thread_root_event_id in self._unpersisted_threads:
+            self._persist_thread_cursor(session_key, thread_root_event_id, cursor)
+
+    def _persist_thread_cursor(
+        self, session_key: str, thread_root_event_id: str, cursor: str | None
+    ) -> None:
+        """Mirror the cursor onto the session, remembering a refusal.
+
+        ``set_session_metadata`` rewrites the whole routing index on every call,
+        which is why the caller restricts this to the turns that need it. A
+        refusal costs nothing here — the in-memory cursor carries the thread
+        until the next process — so it is remembered and retried on the next turn
+        rather than logged away, because a cursor that never lands is exactly the
+        cursor a restart needs.
+        """
+        store = getattr(self, "_session_store", None)
+        writer = getattr(store, "set_session_metadata", None)
+        if not callable(writer):
+            return
+        try:
+            landed = bool(writer(session_key, _CURSOR_META_KEY, cursor))
+        except Exception:
+            logger.debug(
+                "Chatto: persisting the seed cursor for %s failed",
+                session_key,
+                exc_info=True,
+            )
+            return
+        if landed:
+            self._unpersisted_threads.discard(thread_root_event_id)
+        else:
+            # No session under that key yet, which is the expected shape of a
+            # thread's first turn.
+            self._unpersisted_threads.add(thread_root_event_id)
+
+    def _persisted_cursor(self, session_key: str, default: _T) -> str | _T:
+        """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 "this thread's frontier is unknown", 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:
+            # Duck-typed: the store is optional and may not carry metadata at
+            # all, so its value comes back as whatever it is.
+            value: Any = 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 value if value is not None else default
+
+    def _is_own(self, message: Message) -> bool:
+        """Whether this message is one the bot itself posted."""
+        return self.me is not None and message.actor_id == self.me.id
+
     def _record_thread_seeded(
-        self, thread_root_event_id: str, last_delivered_id: str
+        self, thread_root_event_id: str, last_delivered_id: str | None
     ) -> None:
         """Note that this thread's session holds its history up to that message.
 
+        ``None`` records a frontier that cannot be established, which is not the
+        same as "not seeded": the thread stays silent rather than guessing.
+
         Re-inserting keeps the map insertion-ordered, which is what the cap below
         evicts from: the least recently touched thread.
         """

test_adapter.py

@@ -2098,6 +2098,42 @@ def _thread_event(message):
     )
 
 
+class _FakeSessionStore:
+    """An in-memory session store with the real store's shape.
+
+    Modelled on ``gateway.session`` rather than mocked wholesale, because two of
+    its behaviours are what the seed-cursor bookkeeping is built on:
+    ``set_session_metadata`` answers False when no session exists under that key
+    yet (the refusal a thread's first turn runs into), and metadata lives on the
+    session, so a second adapter handed this same instance sees what the first
+    one wrote — which is what surviving a restart means here.
+    """
+
+    def __init__(self):
+        self.sessions: dict[str, str] = {}
+        self.metadata: dict[tuple[str, str], object] = {}
+        self.writes = 0
+
+    def create_session(self, session_key: str, session_id: str = "sess-1") -> None:
+        """What the gateway does once it has actually run a turn."""
+        self.sessions[session_key] = session_id
+
+    def peek_session_id(self, session_key: str):
+        return self.sessions.get(session_key)
+
+    def get_session_metadata(self, session_key: str, key: str, default=None):
+        if session_key not in self.sessions:
+            return default
+        return self.metadata.get((session_key, key), default)
+
+    def set_session_metadata(self, session_key: str, key: str, value) -> bool:
+        if session_key not in self.sessions:
+            return False
+        self.writes += 1
+        self.metadata[(session_key, key)] = value
+        return True
+
+
 class TestThreadContext:
     """A thread session starts empty, so the first turn into a Chatto thread
     seeds the messages that came before it.
@@ -2185,16 +2221,17 @@ 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_rebuilds_the_partition_instead_of_reseeding(self):
+    async def test_a_restart_resumes_from_the_cursor_the_session_kept(self):
         """A restart drops the in-memory cursor while the session survives.
 
-        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.
+        The session's own copy is what outlives the process, so the thread goes
+        on delivering its tail from exactly where it stopped — our own tool
+        output arrives, and the human messages the session already holds do not.
         """
+        store = _FakeSessionStore()
         before = self._adapter()
         self._seed_channel(before)
+        before.set_session_store(store)
         cron = _make_message(
             body="Nightly build finished",
             message_id="cron-1",
@@ -2221,14 +2258,86 @@ class TestThreadContext:
         session_key = before._session_key_for(
             before.handle_message.await_args.args[0].source
         )
+        # 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),
+                ]
+            )
+        )
+        await self._dispatch(
+            before, "@hermes_bot thanks", message_id="msg-2", thread_root="cron-1"
+        )
+        assert store.writes == 1, (
+            "delivering our own output is the one turn whose cursor has to "
+            "outlive the process, so it is the one turn that must write — and "
+            "the pending cursor from turn one rides along on it rather than "
+            "costing a second routing-table rewrite"
+        )
+        later_tool_post = _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),
+                ]
+            )
+        )
+        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),
+                ]
+            )
+        )
+        await self._dispatch(
+            before, quiet.body, message_id="msg-4", thread_root="cron-1"
+        )
+        assert store.writes == 2, (
+            "a turn that delivered only the trigger wrote again — the cursor is "
+            "written when it cannot be recovered, not once per turn"
+        )
 
-        # A fresh adapter: the cursor is gone, as it is after any restart. The
-        # session store still resolves the key — it is persistent.
+        # A fresh adapter: the in-memory cursor is gone, as it is after any
+        # restart. The store is the same one, which is the point.
         after = self._adapter()
         self._seed_channel(after)
+        after.set_session_store(store)
         assert dict(after._seeded_threads) == {}
-        after.set_session_store(
-            MagicMock(peek_session_id=MagicMock(return_value="sess-7"))
+        after_me = _make_message(
+            body="Standing by",
+            message_id="msg-tool2",
+            actor_id="bot-user-id",
+            created_at=datetime(2026, 10, 11, 8, 9, tzinfo=UTC),
+        )
+        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(
@@ -2236,22 +2345,29 @@ class TestThreadContext:
                     _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),
                 ]
             )
         )
 
         await self._dispatch(
-            after, "@hermes_bot after restart", message_id="msg-2", thread_root="cron-1"
+            after, newest.body, message_id="msg-5", thread_root="cron-1"
         )
 
-        resumed = after.handle_message.await_args.args[0]
-        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"
+        context = self._dispatched_event(after).channel_context or ""
+        assert "Standing by" in context, (
+            "what landed after the persisted cursor was not delivered"
         )
+        assert "Rolled back" not in context, (
+            "the resumed cursor is behind where it was written: the thread was "
+            "seeded from the top and repeated what the session already holds"
+        )
+        assert "and the deploy?" not in context
         # 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,)
+        assert after._session_store.peek_session_id(session_key) == "sess-1"
 
     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
@@ -2259,9 +2375,7 @@ class TestThreadContext:
         reply the agent cannot read."""
         adapter = self._adapter()
         self._seed_channel(adapter)
-        adapter.set_session_store(
-            MagicMock(peek_session_id=MagicMock(return_value=None))
-        )
+        adapter.set_session_store(_FakeSessionStore())
         cron = _make_message(
             body="Nightly build finished",
             message_id="cron-1",
@@ -2583,29 +2697,99 @@ class TestThreadContext:
             "left behind the message the session already holds"
         )
 
-    async def test_a_lost_cursor_is_rebuilt_after_a_restart(
+    async def test_a_session_without_a_cursor_redelivers_only_the_bot_output(
         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."""
+        """The blind case: a session that survived with no cursor on it.
+
+        A thread's first turn has no session to write a cursor to, so a thread
+        seeded with nothing (a reply opening a fresh thread) leaves one behind
+        only if the process restarts before the second turn. There is no way to
+        reconstruct where the session stops, so this re-delivers the bot's own
+        messages — the ones no session holds and no later turn can recover —
+        leaves the people's messages to their own turns, and writes the cursor so
+        the next process is not guessing either.
+        """
+        store = _FakeSessionStore()
         before = self._adapter()
         self._seed_channel(before)
+        before.set_session_store(store)
         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),
         )
+        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"
+        )
+        session_key = before._session_key_for(
+            before.handle_message.await_args.args[0].source
+        )
+        store.create_session(session_key)
+        assert store.writes == 0, (
+            "a thread's first turn cannot write a cursor — there is no session "
+            "yet — so it must not pretend to"
+        )
+
         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))
+        after = self._adapter()
+        self._seed_channel(after)
+        after.set_session_store(store)
+        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: with no cursor to go on, the bot's own "
+            "output is delivered once by hand rather than never"
+        )
+        assert "status?" not in context, (
+            "the human message became a turn of its own and is in the session; "
+            "the blind pass delivers the bot's output only"
+        )
+        assert store.writes == 1 and after._seeded_threads["msg-1"] == "msg-tool", (
+            "the blind pass is a once-per-thread event: it records the frontier "
+            "it just established, so the next restart resumes normally. Not the "
+            "trigger — that message is not in the fetched window, and a cursor "
+            "beyond the window is unreachable on the next turn"
+        )
+
+    async def test_a_restart_keeps_the_bot_output_when_the_trigger_is_in_the_page(
+        self,
+    ):
+        """The shape a thread reply actually produces: the fetched window ends
+        with the turn's own trigger.
+
+        Rebuilding the partition as "the newest message that is not ours" lands
+        on that trigger — a person's message that became a turn just now — so the
+        tail comes out empty and every bot message before it is skipped. That is
+        the whole feature failing at exactly the moment it was added for, and it
+        is why the frontier is read off the session instead of guessed.
+        """
+        store = _FakeSessionStore()
+        before = self._adapter()
+        self._seed_channel(before)
+        before.set_session_store(store)
+        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),
         )
         before._chatto_client.get_thread_events = AsyncMock(
             return_value=TimelinePage(events=[_thread_event(human)])
@@ -2613,26 +2797,48 @@ class TestThreadContext:
         await self._dispatch(
             before, human.body, message_id="msg-1", thread_root="msg-1"
         )
+        store.create_session(
+            before._session_key_for(before.handle_message.await_args.args[0].source)
+        )
 
-        # Restart: a fresh adapter, the session survived, the cursor with it?
+        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),
+        )
+        trigger = _make_message(
+            body="@hermes_bot any news?",
+            message_id="msg-2",
+            created_at=datetime(2026, 10, 11, 8, 2, tzinfo=UTC),
+        )
         after = self._adapter()
         self._seed_channel(after)
-        after.set_session_store(
-            MagicMock(peek_session_id=MagicMock(return_value="sess-1"))
-        )
+        after.set_session_store(store)
         after._chatto_client.get_thread_events = AsyncMock(
             return_value=TimelinePage(
-                events=[_thread_event(human), _thread_event(from_bot)]
+                events=[
+                    _thread_event(human),
+                    _thread_event(from_bot),
+                    _thread_event(trigger),
+                ]
             )
         )
+
         await self._dispatch(
-            after, "@hermes_bot any news?", message_id="msg-2", thread_root="msg-1"
+            after, trigger.body, 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"
+            "the bot's own output in the thread was lost after a restart: the "
+            "thread went silent because the trigger is the newest message that "
+            "is not ours"
+        )
+        assert "status?" not in context, (
+            "delivering our own output is how this thread's blind pass resolves "
+            "the cursor — re-seeding the whole thread would also put the output "
+            "in the context, while repeating a turn the session already holds"
         )
 
     async def test_the_replying_message_is_not_seeded_into_its_own_context(self):