|
|
@@ -3192,27 +3192,28 @@ class TestSessionKey:
|
|
|
|
|
|
|
|
|
def _thread_page(
|
|
|
- messages, thread_root_event_id, *, limit=None, before=None, after=None
|
|
|
+ messages, thread_root_event_id, *, limit=None, before=None
|
|
|
):
|
|
|
"""``get_thread_events`` modelled on the server's own handler, not a fixture.
|
|
|
|
|
|
- Three shapes, and the adapter's cursor logic depends on all three:
|
|
|
+ Two shapes, and the adapter's cursor logic depends on both:
|
|
|
|
|
|
- unanchored: the root plus the **newest** ``limit`` replies,
|
|
|
- - ``after=cursor``: the **oldest** ``limit`` replies past the cursor, exclusive,
|
|
|
- and without the root — the server sets ``IncludeRoot`` false for an anchor,
|
|
|
- ``before``: the newest ``limit`` replies ahead of it.
|
|
|
|
|
|
+ There is deliberately no ``after`` shape: on the wire that parameter takes
|
|
|
+ opaque cursor tokens the server seals and hands out with a page — an event
|
|
|
+ ID is rejected with ``invalid cursor format``. The adapter's own seed cursor
|
|
|
+ is an event ID, so `_serve_thread` also installs `get_thread_events_around`
|
|
|
+ for it; a test that wanted the old event-ID ``after`` back would now fail
|
|
|
+ loudly here instead of passing against a mock that never told the truth.
|
|
|
+
|
|
|
A fixed ``TimelinePage`` fixture cannot stand in for this: it would answer an
|
|
|
anchored fetch with the thread's start, which is exactly the bug the anchor
|
|
|
exists to remove.
|
|
|
"""
|
|
|
ordered = list(messages)
|
|
|
- if after is not None:
|
|
|
- index = next((i for i, m in enumerate(ordered) if m.id == after), None)
|
|
|
- ahead = ordered[index + 1 :] if index is not None else []
|
|
|
- selected = ahead[:limit] if limit else ahead
|
|
|
- elif before is not None:
|
|
|
+ if before is not None:
|
|
|
index = next((i for i, m in enumerate(ordered) if m.id == before), None)
|
|
|
head = ordered[:index] if index is not None else ordered
|
|
|
selected = head[-limit:] if limit else head
|
|
|
@@ -3225,10 +3226,46 @@ def _thread_page(
|
|
|
return TimelinePage(events=[_thread_event(m) for m in selected])
|
|
|
|
|
|
|
|
|
+def _thread_page_around(messages, thread_root_event_id, event_id, *, limit=None):
|
|
|
+ """``get_thread_events_around`` modelled on the server's own handler.
|
|
|
+
|
|
|
+ The server (``GetThreadReplyEventsAround``) lists the thread's replies
|
|
|
+ excluding the root, builds a chronological window centred on the anchor —
|
|
|
+ ``(limit-1)/2`` entries before it, the rest after — and reports the anchor's
|
|
|
+ index + 1, or 0 when the anchor is the root itself. The adapter slices the
|
|
|
+ page at that index to get exactly the messages past its cursor.
|
|
|
+ """
|
|
|
+ replies = [m for m in messages if m.id != thread_root_event_id]
|
|
|
+ limit = limit or len(replies) or 1
|
|
|
+ if event_id == thread_root_event_id:
|
|
|
+ anchor_index, target_index = -1, 0
|
|
|
+ else:
|
|
|
+ anchor_index = next(
|
|
|
+ (i for i, m in enumerate(replies) if m.id == event_id), None
|
|
|
+ )
|
|
|
+ if anchor_index is None:
|
|
|
+ raise ChattoError("message not found")
|
|
|
+ target_index = anchor_index + 1
|
|
|
+ start = max(anchor_index - (limit - 1) // 2, 0)
|
|
|
+ end = start + limit
|
|
|
+ if end > len(replies):
|
|
|
+ end = len(replies)
|
|
|
+ start = max(end - limit, 0)
|
|
|
+ window = replies[start:end]
|
|
|
+ relative_target = target_index - start if anchor_index >= 0 else 0
|
|
|
+ return (
|
|
|
+ TimelinePage(events=[_thread_event(m) for m in window]),
|
|
|
+ relative_target,
|
|
|
+ )
|
|
|
+
|
|
|
+
|
|
|
def _serve_thread(adapter, messages, limit=None):
|
|
|
"""Point the adapter's thread fetch at a message list the anchor can walk.
|
|
|
|
|
|
``messages`` is the thread's whole history, oldest first, root included.
|
|
|
+ Both shapes the server serves a thread read with are installed: the
|
|
|
+ unanchored ``get_thread_events`` and the event-ID-anchored
|
|
|
+ ``get_thread_events_around`` the seed cursor walks.
|
|
|
"""
|
|
|
adapter._chatto_client.get_thread_events = AsyncMock(
|
|
|
side_effect=lambda room_id, thread_root_event_id, **kwargs: _thread_page(
|
|
|
@@ -3236,7 +3273,16 @@ def _serve_thread(adapter, messages, limit=None):
|
|
|
thread_root_event_id,
|
|
|
limit=kwargs.get("limit") or limit,
|
|
|
before=kwargs.get("before"),
|
|
|
- after=kwargs.get("after"),
|
|
|
+ )
|
|
|
+ )
|
|
|
+ adapter._chatto_client.get_thread_events_around = AsyncMock(
|
|
|
+ side_effect=lambda room_id, thread_root_event_id, event_id, **kwargs: (
|
|
|
+ _thread_page_around(
|
|
|
+ messages,
|
|
|
+ thread_root_event_id,
|
|
|
+ event_id,
|
|
|
+ limit=kwargs.get("limit") or limit,
|
|
|
+ )
|
|
|
)
|
|
|
)
|
|
|
|
|
|
@@ -3912,6 +3958,111 @@ class TestThreadContext:
|
|
|
|
|
|
assert "Nightly build" in self._dispatched_event(adapter).channel_context
|
|
|
|
|
|
+ async def test_a_stale_cursor_falls_back_to_the_blind_pass(self):
|
|
|
+ """A killable cursor: its message is deleted server-side, so the fetch
|
|
|
+ anchored on it fails every turn.
|
|
|
+
|
|
|
+ The old anchored-read contract would answer this with a permanent
|
|
|
+ contextless dispatch. Instead the turn fails open — it refetches
|
|
|
+ unanchored, re-delivers the bot's own messages like the blind pass a
|
|
|
+ session without a cursor gets, and records a fresh frontier so the dead
|
|
|
+ cursor is never read again.
|
|
|
+ """
|
|
|
+ store = _FakeSessionStore()
|
|
|
+ adapter = self._adapter()
|
|
|
+ self._seed_channel(adapter)
|
|
|
+ adapter.set_session_store(store)
|
|
|
+ source = MagicMock()
|
|
|
+ source.chat_type = "channel"
|
|
|
+ source.chat_id = "room-1"
|
|
|
+ source.thread_id = "cron-1"
|
|
|
+ source.prospective_thread_id = None
|
|
|
+ source.platform = Platform(ChattoConstants.PLATFORM_NAME)
|
|
|
+ source.user_id = "user-1"
|
|
|
+ source.profile = None
|
|
|
+ store.create_session(adapter._session_key_for(source))
|
|
|
+
|
|
|
+ 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),
|
|
|
+ )
|
|
|
+ from_bot = _make_message(
|
|
|
+ body="Deployed, all green",
|
|
|
+ message_id="msg-tool",
|
|
|
+ actor_id="bot-user-id",
|
|
|
+ created_at=datetime(2026, 10, 11, 8, 2, tzinfo=UTC),
|
|
|
+ )
|
|
|
+ history = [cron, from_bot]
|
|
|
+ _serve_thread(adapter, history)
|
|
|
+ # A cursor whose message the server no longer knows: the around fetch
|
|
|
+ # rejects it, the unanchored fetch does not care.
|
|
|
+ adapter._seeded_threads["cron-1"] = "msg-deleted"
|
|
|
+
|
|
|
+ block = await adapter._thread_context_for(
|
|
|
+ adapter._chatto_client,
|
|
|
+ room_id="room-1",
|
|
|
+ thread_root_event_id="cron-1",
|
|
|
+ current_message_id="msg-2",
|
|
|
+ source=source,
|
|
|
+ )
|
|
|
+
|
|
|
+ assert "Deployed, all green" in block, (
|
|
|
+ "the failed anchor muted the thread instead of falling back to the "
|
|
|
+ "unanchored fetch"
|
|
|
+ )
|
|
|
+ assert "status?" not in block, (
|
|
|
+ "the human messages became turns of their own and are in the "
|
|
|
+ "session; the fallback delivers the bot's output only"
|
|
|
+ )
|
|
|
+
|
|
|
+ # The fresh frontier is written over the dead one, so the next process
|
|
|
+ # does not re-read what the server just refused.
|
|
|
+ assert adapter._seeded_threads["cron-1"] == "msg-tool"
|
|
|
+ assert store.writes == 1
|
|
|
+
|
|
|
+ async def test_a_seeded_turn_never_sends_an_event_id_as_an_after_cursor(self):
|
|
|
+ """The wire does not accept one: GetThreadEvents' after/before take
|
|
|
+ opaque cursor tokens the server seals, and an event ID is rejected with
|
|
|
+ 'invalid cursor format'. The old mock agreed with the adapter about a
|
|
|
+ contract the server does not have — so the fake now rejects it, and this
|
|
|
+ test pins that the seeded second turn walks the around API instead."""
|
|
|
+ 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, 1, tzinfo=UTC),
|
|
|
+ )
|
|
|
+ history = [cron, first]
|
|
|
+ _serve_thread(adapter, history)
|
|
|
+
|
|
|
+ await self._dispatch(
|
|
|
+ adapter, first.body, message_id="msg-1", thread_root="cron-1"
|
|
|
+ )
|
|
|
+ adapter._chatto_client.get_thread_events.reset_mock()
|
|
|
+ await self._dispatch(
|
|
|
+ adapter, "@hermes_bot second", message_id="msg-2", thread_root="cron-1"
|
|
|
+ )
|
|
|
+
|
|
|
+ adapter._chatto_client.get_thread_events.assert_not_called()
|
|
|
+ call = adapter._chatto_client.get_thread_events_around.await_args
|
|
|
+ assert call is not None and call.args[2] == "msg-1", (
|
|
|
+ "the seeded turn did not anchor its fetch at the persisted cursor"
|
|
|
+ )
|
|
|
+ window_limit = 2 * ChattoConstants.THREAD_CONTEXT_MESSAGE_LIMIT + 1
|
|
|
+ assert call.kwargs.get("limit") == window_limit, (
|
|
|
+ "the around window is centred on the anchor, so it has to ask for "
|
|
|
+ "almost twice the delivery limit to still deliver a full slice"
|
|
|
+ )
|
|
|
+
|
|
|
async def test_the_root_is_fetched_when_the_timeline_omits_it(self):
|
|
|
"""A cron message is a plain room message until someone replies, so
|
|
|
whether the server lists it back as the thread's first event is not a
|
|
|
@@ -3963,21 +4114,14 @@ class TestThreadContext:
|
|
|
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)]),
|
|
|
- ]
|
|
|
- )
|
|
|
+ history = [human]
|
|
|
+ _serve_thread(adapter, history)
|
|
|
|
|
|
# First turn seeds the thread — it must NOT contain the later message.
|
|
|
await self._dispatch(
|
|
|
@@ -3987,8 +4131,9 @@ class TestThreadContext:
|
|
|
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.
|
|
|
+ # The tool's message, posted into the thread after that first turn; the
|
|
|
+ # next lookup sees it because the fetch is anchored at the cursor.
|
|
|
+ history.append(from_bot)
|
|
|
await self._dispatch(
|
|
|
adapter, "@hermes_bot any news?", message_id="msg-2", thread_root="msg-1"
|
|
|
)
|
|
|
@@ -4229,14 +4374,13 @@ class TestThreadContext:
|
|
|
async def test_an_empty_fetch_records_the_trigger_and_the_next_turn_recovers(
|
|
|
self,
|
|
|
):
|
|
|
- """A fetch that names nothing still leaves an exact cursor behind.
|
|
|
+ """ A fetch that names nothing still leaves an exact cursor behind.
|
|
|
|
|
|
The trigger is the turn's own message, so the session holds it whatever
|
|
|
the thread fetch returned — recording it is exact, not a guess. That is
|
|
|
- only safe because the next fetch is *anchored at* the cursor: a cursor
|
|
|
- names a boundary rather than a position inside the window, so one the
|
|
|
- window did not contain is still usable, and the next turn resumes from
|
|
|
- there instead of re-seeding the thread.
|
|
|
+ only safe because the next fetch is *anchored at* the cursor: the server
|
|
|
+ builds the window around it, so the cursor is always inside the page the
|
|
|
+ next turn walks forward from instead of re-seeding the thread.
|
|
|
"""
|
|
|
trigger = _make_message(
|
|
|
body="@hermes_bot anyone there?",
|
|
|
@@ -4266,6 +4410,10 @@ class TestThreadContext:
|
|
|
)
|
|
|
assert adapter._seeded_threads.get("t1") == "msg-1"
|
|
|
|
|
|
+ # The trigger exists server-side even though the first window did not
|
|
|
+ # name it — the next fetch anchors on it and finds it.
|
|
|
+ history.append(trigger)
|
|
|
+
|
|
|
# And the next turn picks up from that cursor instead of re-seeding.
|
|
|
human = _make_message(
|
|
|
body="@hermes_bot still there?",
|