Seed the thread tail, so tool messages reach the session they were sent into

Reported: a message posted with chatto_send_room into an existing thread never
appears in that thread's Hermes session. It cannot — the bot's own realtime echo
is filtered as a self-event (adapter.py, _handle_realtime_event), so it is never
dispatched, and seeding ran once per thread. Every message the agent posted into
a live thread by tool was therefore invisible to it for good: the next human
message in that thread was answered by an agent that did not know what it had
just said.

Seeding now delivers what the session has not seen yet. _seeded_threads maps a
thread root to the last message ID handed over; the first turn seeds the whole
thread, a later turn seeds only the tail after that cursor. Delivering the
thread again instead would duplicate context on every turn, which is what the
once-per-thread rule existed to prevent — so the invariant the tests now hold is
"no repeated context", not "one lookup".

Two cases deliver nothing rather than guess, both logged: a cursor from an
earlier process (it died with that process, so _thread_session_exists records a
None instead of re-seeding a live session), and a cursor the thread has outgrown
past THREAD_CONTEXT_MESSAGE_LIMIT.

Mutation-checked: with the tail path removed the new test fails with an empty
context; with the cursor not advancing after a delivery the new anti-duplication
test fails. The latter gap was found by mutating first — nothing pinned that
property, so it now has its own test.

Cost: one get_thread_events per turn into an already-seeded thread, the same
order as the roster fetch every channel turn already does.

AGENTS.md

@@ -166,15 +166,24 @@ changes must keep them, or consciously renegotiate the docs.**
   is labelled `… (this bot)` so the agent recognises it; authors off the
   allowlist are labelled `(unverified)` — auth gates the trigger message
   only, and the rest of a shared thread is reference text (fail closed for
-  authors never seen). Seeding is once per SESSION, not per thread and not per process: the
-  in-memory deque alone would re-seed every thread in the room after each
-  restart (the marker dies with the process, the session does not), so
-  `_thread_session_exists` asks the gateway's persistent routing index via
-  `peek_session_id`, which answers correctly across restarts. The deque stays
-  as a timing pre-check for two turns racing before the first session write.
-  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 holds. Note there
+  authors never seen). Seeding delivers **what the session has not seen yet**, keyed on a
+  cursor rather than done once: the first turn seeds the whole thread, and a later
+  turn seeds only what landed after the cursor. That is what makes the agent's own
+  output reachable — a message it posts into a live thread with `chatto_send_room`
+  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. Across a restart that cursor is gone while the session is not,
+  so `_thread_session_exists` asks the gateway's persistent routing index via
+  `peek_session_id`, records a `None` cursor — seeded by an earlier process, tail
+  unknowable — and later turns deliver nothing rather than guess. 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

@@ -385,6 +385,8 @@ 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.
+
 ## Troubleshooting
 
 **First step for every issue below:** look at what the gateway logged around the time things went wrong:

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, NamedTuple
+from typing import Any, Final, NamedTuple
 from urllib.parse import unquote, urlsplit
 
 import httpx
@@ -292,6 +292,28 @@ class _RoomRoster:
     unfetched: int = 0
 
 
+_UNSEEDED: Final = object()
+"""Sentinel: this process has delivered nothing for that thread yet.
+
+Distinct from a recorded ``None``, which means an *earlier* process seeded the
+thread and its cursor is unknowable here.
+"""
+
+
+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)
@@ -348,15 +370,22 @@ class ChattoAdapter(BasePlatformAdapter):
         self._roster_announced: deque[str] = deque(maxlen=ChattoConstants.SEEN_CAP)
         # Announced rooms' membership projection (see _RoomRoster).
         self._rosters: dict[str, _RoomRoster] = {}
-        # Thread roots whose earlier messages have already been handed to their
-        # session. Thread sessions start empty, so the first turn into a thread
-        # seeds what came before it — a cron message whose echo we filtered as
-        # a self-event, or chatter that predates the bot joining. Seeding again
-        # would duplicate context the session already holds, so it happens once
-        # per thread. Unlike the roster this is deliberately NOT discarded on
-        # reconnect: thread history is immutable, so a refetch could only
-        # repeat what the session already has.
-        self._seeded_threads: deque[str] = deque(maxlen=ChattoConstants.SEEN_CAP)
+        # Thread roots whose history has been handed to their session, mapped to
+        # the last message ID that was delivered. Thread sessions start empty, so
+        # the first turn into a thread seeds what came before it — a cron message
+        # whose echo we filtered as a self-event, or chatter that predates the
+        # bot joining. A None value means an earlier process seeded it and this
+        # one cannot tell how far it got.
+        #
+        # The cursor is what makes later turns safe to seed again: the agent's
+        # own messages never reach a session (their echo is filtered as a
+        # self-event), so a reply it posts into a live thread via
+        # chatto_send_room is invisible to that thread forever unless the tail
+        # after the cursor is delivered. Re-delivering everything instead would
+        # duplicate context on every turn. Insertion-ordered and capped like the
+        # seen-ring; unlike the roster this is deliberately NOT discarded on
+        # reconnect, since thread history is immutable.
+        self._seeded_threads: dict[str, str | None] = {}
         # session_key -> message ID currently being processed there. Written
         # by on_processing_start, cleared by on_processing_complete; an edit
         # landing on the recorded ID is a mid-run correction.
@@ -1247,13 +1276,17 @@ class ChattoAdapter(BasePlatformAdapter):
         """
         if not thread_root_event_id:
             return ""
-        if thread_root_event_id in self._seeded_threads:
-            return ""
-        if self._thread_session_exists(source):
-            # This thread already has a session, so its history is already in
-            # that session — recording it keeps a later turn from re-asking.
-            self._mark_thread_seeded(thread_root_event_id)
+
+        delivered_to = self._seeded_threads.get(thread_root_event_id, _UNSEEDED)
+        if delivered_to is _UNSEEDED and self._thread_session_exists(source):
+            # An earlier process seeded this thread and its session is still
+            # there, so that session already holds the history. Recording it
+            # keeps a later turn from re-asking — but the cursor died with that
+            # process, so later turns deliver no tail rather than guess at one
+            # and duplicate a block the session already has.
+            self._record_thread_seeded(thread_root_event_id, None)
             return ""
+
         history = await self._thread_history(
             client, room_id=room_id, thread_root_event_id=thread_root_event_id
         )
@@ -1263,10 +1296,52 @@ class ChattoAdapter(BasePlatformAdapter):
                 thread_root_event_id,
             )
             return ""
-        self._mark_thread_seeded(thread_root_event_id)
+
+        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
+            )
+
         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
+        )
         return self._thread_context_block(earlier)
 
+    def _seed_thread_tail(
+        self,
+        thread_root_event_id: str,
+        history: list[Message],
+        delivered_to: str | None | object,
+        current_message_id: str,
+    ) -> str:
+        """Deliver only 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.
+        """
+        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
+            # 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]
+        if not unseen:
+            return ""
+        self._record_thread_seeded(thread_root_event_id, unseen[-1].id)
+        return self._thread_context_block(unseen)
+
     async def _thread_history(
         self, client: ChattoClient, *, room_id: str, thread_root_event_id: str
     ) -> list[Message] | None:
@@ -1328,9 +1403,20 @@ class ChattoAdapter(BasePlatformAdapter):
             return []
         return [root] if root is not None and not root.deleted_at else []
 
-    def _mark_thread_seeded(self, thread_root_event_id: str) -> None:
-        """Record a thread whose history has been handed to its session."""
-        self._seeded_threads.append(thread_root_event_id)
+    def _record_thread_seeded(
+        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`` when an earlier process seeded the thread: the cursor died with
+        it, so later turns must deliver nothing rather than re-seed blindly.
+        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)))
 
     def _thread_session_exists(self, source: Any) -> bool:
         """Whether the gateway already has a session for this source.

test_adapter.py

@@ -2158,9 +2158,11 @@ class TestThreadContext:
         context = self._dispatched_event(adapter).channel_context
         assert "Nightly build finished" in context
 
-    async def test_second_turn_in_the_thread_seeds_only_once(self):
+    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."""
+        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."""
         adapter = self._adapter()
         self._seed_channel(adapter)
         cron = _make_message(
@@ -2180,7 +2182,6 @@ class TestThreadContext:
             adapter, "@hermes_bot second", message_id="msg-2", thread_root="cron-1"
         )
 
-        assert adapter._chatto_client.get_thread_events.await_count == 1
         second = adapter.handle_message.await_args.args[0]
         assert "Nightly build finished" not in (second.channel_context or "")
 
@@ -2213,7 +2214,7 @@ class TestThreadContext:
         # session store still resolves the key — it is persistent.
         after = self._adapter()
         self._seed_channel(after)
-        assert list(after._seeded_threads) == []
+        assert dict(after._seeded_threads) == {}
         after.set_session_store(
             MagicMock(peek_session_id=MagicMock(return_value="sess-7"))
         )
@@ -2259,9 +2260,9 @@ class TestThreadContext:
             "Nightly build finished" in self._dispatched_event(adapter).channel_context
         )
 
-    async def test_without_a_session_store_the_deque_still_guards_repeat_seeding(self):
+    async def test_without_a_session_store_the_cursor_still_guards_duplication(self):
         """No store wired (tests, or a gateway path that has not registered it
-        yet) must not disable seeding's once-per-thread behaviour."""
+        yet) must not disable the record of what has already been delivered."""
         adapter = self._adapter()
         self._seed_channel(adapter)
         cron = _make_message(
@@ -2281,7 +2282,8 @@ class TestThreadContext:
             adapter, "@hermes_bot second", message_id="msg-2", thread_root="cron-1"
         )
 
-        assert adapter._chatto_client.get_thread_events.await_count == 1
+        second = adapter.handle_message.await_args.args[0]
+        assert "Nightly build finished" not in (second.channel_context or "")
 
     async def test_the_bot_knows_the_seeded_cron_message_is_its_own(self):
         """The cron message is the bot's output that it never saw arrive —
@@ -2366,7 +2368,7 @@ class TestThreadContext:
             adapter, "@hermes_bot first", message_id="msg-1", thread_root="cron-1"
         )
 
-        assert list(adapter._seeded_threads) == []
+        assert dict(adapter._seeded_threads) == {}
 
         adapter._chatto_client.get_thread_events = AsyncMock(
             return_value=TimelinePage(events=[_thread_event(cron)])
@@ -2412,6 +2414,105 @@ class TestThreadContext:
         context = self._dispatched_event(adapter).channel_context
         assert "Nightly build finished" in context
 
+    async def test_a_tool_message_posted_into_a_live_thread_reaches_the_session(self):
+        """A thread the agent is already answering in, into which a
+        chatto_send_room reply lands. That message is the bot's own, so its
+        realtime echo is filtered as a self-event and never dispatched — the
+        thread session therefore does not contain it, and the next human
+        message in that thread is answered by an agent that never saw what it
+        said. Seeding covers this only on the FIRST turn; a thread already
+        answered in gets nothing."""
+        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),
+        )
+        # The tool's message, posted into the thread after that first turn.
+        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._chatto_client.get_thread_events = AsyncMock(
+            side_effect=[
+                # The first turn's history predates the tool post; a later
+                # lookup, if one happened, would see it.
+                TimelinePage(events=[_thread_event(human)]),
+                TimelinePage(events=[_thread_event(human), _thread_event(from_bot)]),
+            ]
+        )
+
+        # First turn seeds the thread — it must NOT contain the later message.
+        await self._dispatch(
+            adapter, human.body, message_id="msg-1", thread_root="msg-1"
+        )
+        assert "All green" not in (
+            self._dispatched_event(adapter).channel_context or ""
+        )
+
+        # Second turn: the thread is already answered in, so seeding is skipped
+        # — and the tool's message is nowhere in the session.
+        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, (
+            "a message the bot posted into this thread via chatto_send_room is "
+            "missing from the thread session: its echo is filtered as a self-"
+            "event, and seeding already happened on the first turn"
+        )
+
+    async def test_a_delivered_tail_is_not_delivered_again(self):
+        """The cursor must advance past what it delivered. Without that, every
+        later turn in the thread repeats the same block — the duplication the
+        once-per-thread rule existed to prevent, now reintroduced per turn."""
+        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),
+        )
+        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,
+            ]
+        )
+
+        await self._dispatch(
+            adapter, human.body, message_id="msg-1", thread_root="msg-1"
+        )
+        await self._dispatch(
+            adapter, "@hermes_bot and now?", message_id="msg-2", thread_root="msg-1"
+        )
+        assert "All green, deploying now" in (
+            self._dispatched_event(adapter).channel_context or ""
+        )
+
+        adapter.handle_message.reset_mock()
+        await self._dispatch(
+            adapter, "@hermes_bot still?", message_id="msg-3", thread_root="msg-1"
+        )
+        third = adapter.handle_message.await_args.args[0].channel_context or ""
+        assert "All green, deploying now" not in third, (
+            "the tail was delivered again on a later turn — the cursor did not "
+            "advance, so every turn repeats it"
+        )
+
     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."""
@@ -2541,7 +2642,7 @@ class TestThreadContext:
         await self._dispatch(adapter, "@hermes_bot hi")
 
         adapter._chatto_client.get_thread_events.assert_not_called()
-        assert list(adapter._seeded_threads) == []
+        assert dict(adapter._seeded_threads) == {}
 
 
 class TestEditDispatch: