|
|
@@ -56,7 +56,6 @@ from gateway.platforms.base import (
|
|
|
get_inbound_media_max_bytes,
|
|
|
validate_inbound_media_size,
|
|
|
)
|
|
|
-from gateway.session import build_session_key
|
|
|
|
|
|
# Chattolib imports (vendored)
|
|
|
# Using vendored chattolib from vendor/chattolib/
|
|
|
@@ -387,8 +386,7 @@ class ChattoAdapter(BasePlatformAdapter):
|
|
|
# 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.
|
|
|
+ # bot joining.
|
|
|
#
|
|
|
# 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
|
|
|
@@ -398,17 +396,17 @@ class ChattoAdapter(BasePlatformAdapter):
|
|
|
# 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] = {}
|
|
|
- # 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
|
|
|
# the cursor past the other's messages and that session is seeded once
|
|
|
# and never again. The flag is off by default and this is the price of
|
|
|
- # switching it on; keying the map by session key would fix it and is
|
|
|
- # noted next to _thread_context_for.
|
|
|
+ # switching it on; keying the map by session key would fix it.
|
|
|
+ self._seeded_threads: dict[str, str] = {}
|
|
|
+ # 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()
|
|
|
# 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.
|
|
|
@@ -1314,13 +1312,10 @@ class ChattoAdapter(BasePlatformAdapter):
|
|
|
session_key = self._session_key_for(source)
|
|
|
|
|
|
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,
|
|
|
+ self._seeded_threads[thread_root_event_id],
|
|
|
current_message_id,
|
|
|
session_key=session_key,
|
|
|
)
|
|
|
@@ -1359,26 +1354,24 @@ class ChattoAdapter(BasePlatformAdapter):
|
|
|
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)
|
|
|
+ if cursor is not None:
|
|
|
+ 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,
|
|
|
+ self._advance_thread_cursor(
|
|
|
+ session_key, thread_root_event_id, history, earlier, current_message_id
|
|
|
)
|
|
|
return self._thread_context_block(earlier)
|
|
|
|
|
|
@staticmethod
|
|
|
def _seeded_through(
|
|
|
history: list[Message], delivered: list[Message], current_message_id: str
|
|
|
- ) -> str:
|
|
|
- """How far this turn's session now reaches, for the cursor.
|
|
|
+ ) -> str | None:
|
|
|
+ """How far this turn's session now reaches, or None if the window cannot say.
|
|
|
|
|
|
The trigger is excluded from the *block* — it is the message text — but
|
|
|
the session holds it, so the cursor passes it too. Leaving it behind
|
|
|
@@ -1387,10 +1380,41 @@ class ChattoAdapter(BasePlatformAdapter):
|
|
|
contains it. Only when the fetched window does not list the trigger: a
|
|
|
cursor beyond the window would be unreachable, so fall back to what was
|
|
|
actually delivered.
|
|
|
+
|
|
|
+ None when neither is in the window — an empty fetch. The trigger is *not*
|
|
|
+ a usable fallback there: recording it leaves a cursor the next fetch
|
|
|
+ cannot find, which then reads as a thread that grew past its cursor and
|
|
|
+ mutes it for good.
|
|
|
"""
|
|
|
if any(m.id == current_message_id for m in history):
|
|
|
return current_message_id
|
|
|
- return delivered[-1].id if delivered else current_message_id
|
|
|
+ return delivered[-1].id if delivered else None
|
|
|
+
|
|
|
+ def _advance_thread_cursor(
|
|
|
+ self,
|
|
|
+ session_key: str,
|
|
|
+ thread_root_event_id: str,
|
|
|
+ history: list[Message],
|
|
|
+ delivered: list[Message],
|
|
|
+ current_message_id: str,
|
|
|
+ ) -> None:
|
|
|
+ """Record the frontier this turn established, if the window holds one.
|
|
|
+
|
|
|
+ A turn that cannot name a position leaves the thread unrecorded, which is
|
|
|
+ the state its next turn retries from — strictly better than recording a
|
|
|
+ cursor that can never be found again.
|
|
|
+ """
|
|
|
+ cursor = self._seeded_through(history, delivered, current_message_id)
|
|
|
+ if cursor is None:
|
|
|
+ logger.debug(
|
|
|
+ "Chatto: thread %s delivered nothing the fetched window can "
|
|
|
+ "name a cursor from; left unrecorded for the next turn",
|
|
|
+ thread_root_event_id,
|
|
|
+ )
|
|
|
+ return
|
|
|
+ self._remember_thread_cursor(
|
|
|
+ session_key, thread_root_event_id, cursor, delivered=delivered
|
|
|
+ )
|
|
|
|
|
|
def _seed_thread_tail(
|
|
|
self,
|
|
|
@@ -1424,11 +1448,8 @@ 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._remember_thread_cursor(
|
|
|
- session_key,
|
|
|
- thread_root_event_id,
|
|
|
- self._seeded_through(history, unseen, current_message_id),
|
|
|
- delivered=unseen,
|
|
|
+ self._advance_thread_cursor(
|
|
|
+ session_key, thread_root_event_id, history, unseen, current_message_id
|
|
|
)
|
|
|
return self._thread_context_block(unseen)
|
|
|
|
|
|
@@ -1497,7 +1518,7 @@ class ChattoAdapter(BasePlatformAdapter):
|
|
|
self,
|
|
|
session_key: str,
|
|
|
thread_root_event_id: str,
|
|
|
- cursor: str | None,
|
|
|
+ cursor: str,
|
|
|
*,
|
|
|
delivered: list[Message],
|
|
|
) -> None:
|
|
|
@@ -1527,7 +1548,7 @@ class ChattoAdapter(BasePlatformAdapter):
|
|
|
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
|
|
|
+ self, session_key: str, thread_root_event_id: str, cursor: str
|
|
|
) -> None:
|
|
|
"""Mirror the cursor onto the session, remembering a refusal.
|
|
|
|
|
|
@@ -1587,13 +1608,10 @@ class ChattoAdapter(BasePlatformAdapter):
|
|
|
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 | None
|
|
|
+ self, thread_root_event_id: str, last_delivered_id: str
|
|
|
) -> 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.
|
|
|
"""
|
|
|
@@ -1786,19 +1804,16 @@ class ChattoAdapter(BasePlatformAdapter):
|
|
|
def _session_key_for(self, source) -> str:
|
|
|
"""The gateway's own session key for this source.
|
|
|
|
|
|
- Built with exactly the inputs ``handle_message`` uses, so lookups in
|
|
|
- ``_processing`` and calls to ``cancel_session_processing`` hit the
|
|
|
- same session the gateway is running.
|
|
|
+ Delegates to the base class rather than calling ``build_session_key``
|
|
|
+ itself. A hand-built key has to be right about two things this one was
|
|
|
+ not: the routing identity has to be pinned first (``_canonicalize``), and
|
|
|
+ the key has to carry the serving profile's namespace. Without the second,
|
|
|
+ a multiplexed gateway keys every bot into ``agent:main:`` and the derived
|
|
|
+ key names a session nobody is running — which made
|
|
|
+ ``_thread_session_exists`` answer False for a live thread session, and
|
|
|
+ would make ``cancel_session_processing`` a no-op on an edited message.
|
|
|
"""
|
|
|
- return build_session_key(
|
|
|
- source,
|
|
|
- group_sessions_per_user=self.config.extra.get(
|
|
|
- "group_sessions_per_user", True
|
|
|
- ),
|
|
|
- thread_sessions_per_user=self.config.extra.get(
|
|
|
- "thread_sessions_per_user", False
|
|
|
- ),
|
|
|
- )
|
|
|
+ return self._source_session_key(source)
|
|
|
|
|
|
async def _dispatch_message_edited(self, payload: Any) -> None:
|
|
|
"""Route an inbound edit according to the edit-dispatch contract.
|
|
|
@@ -4052,9 +4067,7 @@ def _pconfig_or_empty(pconfig: PlatformConfig | None) -> PlatformConfig:
|
|
|
return GatewayPlatformConfig(enabled=True, extra={})
|
|
|
|
|
|
|
|
|
-def _room_is_writable(
|
|
|
- cfg: ChattoConfiguration | None, room: Room, room_id: str
|
|
|
-) -> bool:
|
|
|
+def _room_is_writable(cfg: ChattoConfiguration | None, room: Room) -> bool:
|
|
|
"""Whether the bot may post into this room on its own initiative.
|
|
|
|
|
|
Delegates to ``_room_answers_in`` so the outbound gate is the *same*
|
|
|
@@ -4062,21 +4075,27 @@ def _room_is_writable(
|
|
|
be postable, and two copies of that rule would drift apart until the send
|
|
|
tool became a way around the read-only promise.
|
|
|
"""
|
|
|
- return _room_answers_in(cfg, room_id, room.kind)
|
|
|
+ return _room_answers_in(cfg, room.id, room.kind)
|
|
|
|
|
|
|
|
|
def _not_joined_refusal(room: Room) -> _ToolRefusal:
|
|
|
"""The refusal for a room the bot can see but has not joined.
|
|
|
|
|
|
- The two room kinds need different advice, and giving a channel the DM
|
|
|
- remedy sends the user down a path that cannot work: a direct message to the
|
|
|
- bot is not how you get it into a channel. For a channel the paths are an
|
|
|
- invite in Chatto, or the ``/join`` the bot answers over DM.
|
|
|
+ The room kinds need different advice, and giving a channel the DM remedy
|
|
|
+ sends the user down a path that cannot work: a direct message to the bot is
|
|
|
+ not how you get it into a channel. For a channel the paths are an invite in
|
|
|
+ Chatto, or the ``/join`` the bot answers over DM.
|
|
|
+
|
|
|
+ The test is on ``DM`` rather than on ``CHANNEL`` on purpose: a server that
|
|
|
+ never sets ``kind`` reports ``UNSPECIFIED``, and this plugin counts an unknown
|
|
|
+ kind as a channel everywhere else (``_room_answers_in``, ``_refresh_rooms``).
|
|
|
+ Keying on ``CHANNEL`` would hand exactly the advice this function exists to
|
|
|
+ avoid to every self-hosted server on an older or looser build.
|
|
|
"""
|
|
|
remedy = (
|
|
|
- f"Invite the bot account in Chatto, or DM it `/join {room.id}`"
|
|
|
- if room.kind == RoomKind.CHANNEL
|
|
|
- else "Have someone start a DM with the bot account"
|
|
|
+ "Have someone start a DM with the bot account"
|
|
|
+ if room.kind == RoomKind.DM
|
|
|
+ else f"Invite the bot account in Chatto, or DM it `/join {room.id}`"
|
|
|
)
|
|
|
return _refuse(
|
|
|
f"Room '{room.name or room.id}' ({room.id}) is not one the bot has "
|
|
|
@@ -4398,7 +4417,7 @@ async def _send_room(args: dict) -> str:
|
|
|
|
|
|
# Posting into a room the plugin treats as read-only would break a
|
|
|
# promise the README makes about unlisted rooms staying silent.
|
|
|
- if not _room_is_writable(cfg, room, room.id):
|
|
|
+ if not _room_is_writable(cfg, room):
|
|
|
raise _read_only_room_refusal(room.name or room.id)
|
|
|
|
|
|
if create_thread and room.kind == RoomKind.DM:
|