Просмотр исходного кода

Keep the thread cursor on the session, so a restart does not silence it

The cursor was process-local, so a gateway restart dropped it while the session
survived: the thread fell into the "seeded by an earlier process" branch and
delivered nothing ever again. A chatto_send_room message posted into that thread
while the gateway was down stayed invisible to it for the rest of its life.

It now lives on the session (chatto.thread_seed_cursor via set_session_metadata),
which is persistent, scoped per profile, and dies with the session. One opaque
message ID per thread — nothing of the thread's content is mirrored anywhere.
Worth noting what this is not: it is not the Discord cursor. Discord partitions
at its own last message and excludes it, which works there because Discord's only
self-messages are agent replies that the turn already wrote to the session.
Chatto has a second kind — tool and cron posts, never in any session — and that
partition would swallow exactly the messages this exists to surface.

Written only when it moves: set_session_metadata saves the whole session index per
call, so an unchanged cursor has nothing to say. A thread's first turn has no
session yet and its cursor therefore never lands; the second turn writes it.

Two mutations, both caught. Ignoring the persisted cursor fails the restart test.
Writing on every turn instead of only on change failed nothing at first — the
test exercised a turn where the cursor could not move anyway, so it proved
nothing. Rewritten to drive a turn that advances the cursor and one that does
not; it now fails when the write happens on the unchanged one.

Also: a session carrying no cursor is one this bookkeeping predates, so it still
delivers nothing rather than guessing — re-seeding would repeat context the
session already holds.
Paul Klumpp 2 недель назад
Родитель
Сommit
060fb3386a
4 измененных файлов с 224 добавлено и 28 удалено
  1. 9 4
      AGENTS.md
  2. 1 1
      README.md
  3. 88 18
      adapter.py
  4. 126 5
      test_adapter.py

+ 9 - 4
AGENTS.md

@@ -174,10 +174,15 @@ changes must keep them, or consciously renegotiate the docs.**
   be invisible to that thread for good, and the next human message would be
   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
   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
   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
+  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
   has fallen out of the fetched window means the thread outgrew
   `THREAD_CONTEXT_MESSAGE_LIMIT`: deliver nothing rather than re-seed from the
   `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
   top, which would duplicate what the session holds. A failed lookup leaves the

+ 1 - 1
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.
 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.
+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.
 
 
 ## Troubleshooting
 ## Troubleshooting
 
 

+ 88 - 18
adapter.py

@@ -295,8 +295,15 @@ class _RoomRoster:
 _UNSEEDED: Final = object()
 _UNSEEDED: Final = object()
 """Sentinel: this process has delivered nothing for that thread yet.
 """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.
+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.
 """
 """
 
 
 
 
@@ -1277,15 +1284,22 @@ class ChattoAdapter(BasePlatformAdapter):
         if not thread_root_event_id:
         if not thread_root_event_id:
             return ""
             return ""
 
 
+        session_key = self._session_key_for(source)
         delivered_to = self._seeded_threads.get(thread_root_event_id, _UNSEEDED)
         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 ""
+        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(
         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
@@ -1299,12 +1313,18 @@ class ChattoAdapter(BasePlatformAdapter):
 
 
         if isinstance(delivered_to, str) or delivered_to is None:
         if isinstance(delivered_to, str) or delivered_to is None:
             return self._seed_thread_tail(
             return self._seed_thread_tail(
-                thread_root_event_id, history, delivered_to, current_message_id
+                thread_root_event_id,
+                history,
+                delivered_to,
+                current_message_id,
+                session_key,
             )
             )
 
 
         earlier = [m for m in history if m.id != current_message_id]
         earlier = [m for m in history if m.id != current_message_id]
         self._record_thread_seeded(
         self._record_thread_seeded(
-            thread_root_event_id, earlier[-1].id if earlier else current_message_id
+            thread_root_event_id,
+            earlier[-1].id if earlier else current_message_id,
+            session_key,
         )
         )
         return self._thread_context_block(earlier)
         return self._thread_context_block(earlier)
 
 
@@ -1314,6 +1334,7 @@ class ChattoAdapter(BasePlatformAdapter):
         history: list[Message],
         history: list[Message],
         delivered_to: str | None | object,
         delivered_to: str | None | object,
         current_message_id: str,
         current_message_id: str,
+        session_key: str,
     ) -> str:
     ) -> str:
         """Deliver only what landed in an already-seeded thread since the cursor.
         """Deliver only what landed in an already-seeded thread since the cursor.
 
 
@@ -1339,7 +1360,7 @@ class ChattoAdapter(BasePlatformAdapter):
         unseen = [m for m in tail if m.id != current_message_id]
         unseen = [m for m in tail if m.id != current_message_id]
         if not unseen:
         if not unseen:
             return ""
             return ""
-        self._record_thread_seeded(thread_root_event_id, unseen[-1].id)
+        self._record_thread_seeded(thread_root_event_id, unseen[-1].id, session_key)
         return self._thread_context_block(unseen)
         return self._thread_context_block(unseen)
 
 
     async def _thread_history(
     async def _thread_history(
@@ -1404,19 +1425,68 @@ class ChattoAdapter(BasePlatformAdapter):
         return [root] if root is not None and not root.deleted_at else []
         return [root] if root is not None and not root.deleted_at else []
 
 
     def _record_thread_seeded(
     def _record_thread_seeded(
-        self, thread_root_event_id: str, last_delivered_id: str | None
+        self, thread_root_event_id: str, last_delivered_id: str | None, session_key: str
     ) -> None:
     ) -> None:
         """Note that this thread's session holds its history up to that message.
         """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.
+        ``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.
         """
         """
         self._seeded_threads.pop(thread_root_event_id, None)
         self._seeded_threads.pop(thread_root_event_id, None)
         self._seeded_threads[thread_root_event_id] = last_delivered_id
         self._seeded_threads[thread_root_event_id] = last_delivered_id
         while len(self._seeded_threads) > ChattoConstants.SEEN_CAP:
         while len(self._seeded_threads) > ChattoConstants.SEEN_CAP:
             self._seeded_threads.pop(next(iter(self._seeded_threads)))
             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:
     def _thread_session_exists(self, source: Any) -> bool:
         """Whether the gateway already has a session for this source.
         """Whether the gateway already has a session for this source.

+ 126 - 5
test_adapter.py

@@ -2186,11 +2186,11 @@ class TestThreadContext:
         assert "Nightly build finished" not in (second.channel_context or "")
         assert "Nightly build finished" not in (second.channel_context or "")
 
 
     async def test_a_restart_does_not_reseed_a_thread_whose_session_survived(self):
     async def test_a_restart_does_not_reseed_a_thread_whose_session_survived(self):
-        """The seeding marker is in-memory, the session is in state.db.
+        """A session that carries no cursor is one this bookkeeping predates.
 
 
-        After a gateway restart the deque is empty while the thread's session
-        still holds everything the first turn seeded. Asking the session store
-        — not the deque — is what keeps the block one-shot across restarts.
+        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.
         """
         """
         before = self._adapter()
         before = self._adapter()
         self._seed_channel(before)
         self._seed_channel(before)
@@ -2216,7 +2216,13 @@ class TestThreadContext:
         self._seed_channel(after)
         self._seed_channel(after)
         assert dict(after._seeded_threads) == {}
         assert dict(after._seeded_threads) == {}
         after.set_session_store(
         after.set_session_store(
-            MagicMock(peek_session_id=MagicMock(return_value="sess-7"))
+            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),
+            )
         )
         )
         after._chatto_client.get_thread_events = AsyncMock(
         after._chatto_client.get_thread_events = AsyncMock(
             return_value=TimelinePage(events=[_thread_event(cron)])
             return_value=TimelinePage(events=[_thread_event(cron)])
@@ -2233,6 +2239,121 @@ class TestThreadContext:
         # Sanity: the key the adapter asks about is the one the gateway used.
         # 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.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):
     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
         """No session bound to the key means the session is empty — the exact
         situation seeding exists for. Fails open: seed rather than answer a
         situation seeding exists for. Fails open: seed rather than answer a