|
|
@@ -1215,6 +1215,7 @@ class ChattoAdapter(BasePlatformAdapter):
|
|
|
room_id: str,
|
|
|
thread_root_event_id: str,
|
|
|
current_message_id: str,
|
|
|
+ source: Any,
|
|
|
) -> str:
|
|
|
"""The thread's earlier messages, delivered once per thread.
|
|
|
|
|
|
@@ -1230,10 +1231,24 @@ 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.
|
|
|
+
|
|
|
Best-effort by design, like the roster: a failed lookup leaves the
|
|
|
thread unseeded so the next turn retries.
|
|
|
"""
|
|
|
- if not thread_root_event_id or thread_root_event_id in self._seeded_threads:
|
|
|
+ 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)
|
|
|
return ""
|
|
|
history = await self._thread_history(
|
|
|
client, room_id=room_id, thread_root_event_id=thread_root_event_id
|
|
|
@@ -1313,6 +1328,20 @@ class ChattoAdapter(BasePlatformAdapter):
|
|
|
"""Record a thread whose history has been handed to its session."""
|
|
|
self._seeded_threads.append(thread_root_event_id)
|
|
|
|
|
|
+ def _thread_session_exists(self, source: Any) -> bool:
|
|
|
+ """Whether the gateway already has a session for this source.
|
|
|
+
|
|
|
+ The gateway's own routing index is the authority on "does this thread's
|
|
|
+ conversation already exist" — it is persistent (state.db), so unlike
|
|
|
+ the in-memory deque it answers correctly after a restart. Returns False
|
|
|
+ when no store is wired (tests, or a startup path before
|
|
|
+ ``set_session_store`` runs) so we fail open and seed.
|
|
|
+ """
|
|
|
+ store = getattr(self, "_session_store", None)
|
|
|
+ if store is None:
|
|
|
+ return False
|
|
|
+ return store.peek_session_id(self._session_key_for(source)) is not None
|
|
|
+
|
|
|
def _thread_context_block(self, messages: list[Message]) -> str:
|
|
|
"""Render earlier thread messages as the agent's reference block.
|
|
|
|
|
|
@@ -1734,10 +1763,23 @@ class ChattoAdapter(BasePlatformAdapter):
|
|
|
# repeat is not deduplication but the point (see _roster_for_room).
|
|
|
# The gateway prepends channel_context above the message text, so the
|
|
|
# roster never mingles with what the user actually wrote.
|
|
|
+ source = self.build_source(
|
|
|
+ chat_id=room_id,
|
|
|
+ chat_name=self._room_names.get(message.room_id),
|
|
|
+ chat_type=chat_type_for_room_kind(room_kind),
|
|
|
+ user_id=message.actor_id,
|
|
|
+ user_name=user.login, # use login, because display_name is changeable by anyone.
|
|
|
+ thread_id=thread_id,
|
|
|
+ message_id=source_message_id,
|
|
|
+ role_authorized=True,
|
|
|
+ )
|
|
|
+
|
|
|
# Channel context rides ahead of the message text as read-only
|
|
|
# reference: the roster names the audience, the thread block carries
|
|
|
# what the thread already said. DMs get neither — there is no audience
|
|
|
- # to name and no thread to catch up on.
|
|
|
+ # to name and no thread to catch up on. Built after the source, because
|
|
|
+ # the thread block asks the session store whether this thread's session
|
|
|
+ # already has the history (see _thread_context_for).
|
|
|
context_blocks: list[str] = []
|
|
|
if room_kind != RoomKind.DM:
|
|
|
logger.debug(
|
|
|
@@ -1757,21 +1799,11 @@ class ChattoAdapter(BasePlatformAdapter):
|
|
|
room_id=room_id,
|
|
|
thread_root_event_id=thread_root_event_id or "",
|
|
|
current_message_id=message.id,
|
|
|
+ source=source,
|
|
|
)
|
|
|
if thread_context:
|
|
|
context_blocks.append(thread_context)
|
|
|
|
|
|
- source = self.build_source(
|
|
|
- chat_id=room_id,
|
|
|
- chat_name=self._room_names.get(message.room_id),
|
|
|
- chat_type=chat_type_for_room_kind(room_kind),
|
|
|
- user_id=message.actor_id,
|
|
|
- user_name=user.login, # use login, because display_name is changeable by anyone.
|
|
|
- thread_id=thread_id,
|
|
|
- message_id=source_message_id,
|
|
|
- role_authorized=True,
|
|
|
- )
|
|
|
-
|
|
|
# prepare a MessageEvent
|
|
|
message_event = MessageEvent(
|
|
|
text=message_body,
|