|
|
@@ -48,6 +48,7 @@ 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/
|
|
|
@@ -72,10 +73,12 @@ try:
|
|
|
stream_events,
|
|
|
)
|
|
|
from chattolib.realtime_types import (
|
|
|
+ MessageEditedPayload,
|
|
|
MessagePostedPayload,
|
|
|
ReactionPayload,
|
|
|
)
|
|
|
from chattolib.types import (
|
|
|
+ Message,
|
|
|
PresenceStatus,
|
|
|
Room,
|
|
|
RoomKind,
|
|
|
@@ -275,6 +278,15 @@ class ChattoAdapter(BasePlatformAdapter):
|
|
|
# Event IDs already processed — chattolib may redeliver events across
|
|
|
# reconnects, so every inbound event is checked against this list.
|
|
|
self._seen: list[str] = []
|
|
|
+ # Message IDs this adapter has handed to the gateway (posted or
|
|
|
+ # edit-re-dispatched). Edits of anything on this list never start a
|
|
|
+ # fresh turn — that is the lock against re-answering settled
|
|
|
+ # conversations by editing old messages.
|
|
|
+ self._dispatched_ids: list[str] = []
|
|
|
+ # 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.
|
|
|
+ self._processing: dict[str, str] = {}
|
|
|
self._joined_room_ids: list[str] = []
|
|
|
# Rooms the server force-joined everyone into (Room.universal) — used
|
|
|
# only for [universal] tags in the joined-rooms log line, never for
|
|
|
@@ -916,12 +928,195 @@ class ChattoAdapter(BasePlatformAdapter):
|
|
|
message_body = message.body or ""
|
|
|
logger.debug("message: %s", message)
|
|
|
|
|
|
- attachments = list(message.attachments or [])
|
|
|
# A message carrying only an image/PDF has an empty body — dropping it
|
|
|
# here is what made attachments sent to Hermes disappear silently.
|
|
|
+ attachments = list(message.attachments or [])
|
|
|
if not message_body and not attachments:
|
|
|
return
|
|
|
|
|
|
+ event = await self._admit_and_build(
|
|
|
+ client,
|
|
|
+ room_id=payload.room_id,
|
|
|
+ message=message,
|
|
|
+ source_message_id=payload.message_event_id,
|
|
|
+ thread_root_event_id=payload.thread_root_event_id or None,
|
|
|
+ )
|
|
|
+ if event is None:
|
|
|
+ return
|
|
|
+
|
|
|
+ self._remember_dispatched(event.message_id or "")
|
|
|
+ logger.info("Chatto: dispatching message to Hermes")
|
|
|
+ await self.handle_message(event)
|
|
|
+ return
|
|
|
+
|
|
|
+ def _remember_dispatched(self, message_id: str) -> None:
|
|
|
+ """Record a message ID as handed to the gateway, capped like _seen."""
|
|
|
+ if not message_id:
|
|
|
+ return
|
|
|
+ self._dispatched_ids.append(message_id)
|
|
|
+ while len(self._dispatched_ids) > ChattoConstants.SEEN_CAP:
|
|
|
+ self._dispatched_ids.remove(self._dispatched_ids[0])
|
|
|
+
|
|
|
+ def _edit_is_fresh(self, message: Message) -> bool:
|
|
|
+ """Whether this edit is young enough to still be processed.
|
|
|
+
|
|
|
+ Age is measured against the posting time, so an edit to an hours-old
|
|
|
+ message cannot resurrect a settled conversation even when it arrives
|
|
|
+ right now.
|
|
|
+ """
|
|
|
+ if message.created_at is None or message.updated_at is None:
|
|
|
+ return True
|
|
|
+ age_seconds = (message.updated_at - message.created_at).total_seconds()
|
|
|
+ return age_seconds <= self.chatto_config.edit_window.value
|
|
|
+
|
|
|
+ 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.
|
|
|
+ """
|
|
|
+ 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
|
|
|
+ ),
|
|
|
+ )
|
|
|
+
|
|
|
+ async def _dispatch_message_edited(self, payload: MessageEditedPayload) -> None:
|
|
|
+ """Route an inbound edit according to the edit-dispatch contract.
|
|
|
+
|
|
|
+ Three outcomes for the edited message:
|
|
|
+
|
|
|
+ - currently being processed → cancel that turn and re-dispatch with
|
|
|
+ the corrected text (the cancelled turn reports 🚫 via its
|
|
|
+ CANCELLED outcome hook),
|
|
|
+ - never dispatched (e.g. a forgotten @mention added later) → re-run
|
|
|
+ the admission gates against the new body and answer for real,
|
|
|
+ - already answered → stay answered.
|
|
|
+
|
|
|
+ Edits whose text parses as a DM membership command are dropped: the
|
|
|
+ command ran when the message was posted and must not run again.
|
|
|
+ """
|
|
|
+ if not self.chatto_config.edit_dispatch.value:
|
|
|
+ return
|
|
|
+ # Read-only memberships cost no API call, mirroring the posted path.
|
|
|
+ if not self._is_respond_room(payload.room_id):
|
|
|
+ logger.debug("Chatto: edit from read-only room %s ignored", payload.room_id)
|
|
|
+ return
|
|
|
+
|
|
|
+ try:
|
|
|
+ client = await self._require_client()
|
|
|
+ except RuntimeError:
|
|
|
+ logger.warning("Chatto: dropping edit - no client available")
|
|
|
+ return
|
|
|
+ logger.debug("Chatto WS: 'message_edited' payload:%s", payload)
|
|
|
+
|
|
|
+ message = await payload.fetch_message(client=client)
|
|
|
+ if message is None or message.deleted_at:
|
|
|
+ return
|
|
|
+ message_body = message.body or ""
|
|
|
+ if not message_body and not message.attachments:
|
|
|
+ return
|
|
|
+ if not self._edit_is_fresh(message):
|
|
|
+ logger.debug(
|
|
|
+ "Chatto: edit of msg %s outside the edit window",
|
|
|
+ payload.message_event_id,
|
|
|
+ )
|
|
|
+ return
|
|
|
+
|
|
|
+ # Cheap triage before the admission pipeline: an edit to an
|
|
|
+ # already-settled message (dispatched, neither running nor queued)
|
|
|
+ # must not cost get_room/media calls or re-fire acknowledgements.
|
|
|
+ was_dispatched = payload.message_event_id in self._dispatched_ids
|
|
|
+ if (
|
|
|
+ was_dispatched
|
|
|
+ and payload.message_event_id not in self._processing.values()
|
|
|
+ and not any(
|
|
|
+ getattr(pending, "message_id", None) == payload.message_event_id
|
|
|
+ for pending in self._pending_messages.values()
|
|
|
+ )
|
|
|
+ ):
|
|
|
+ logger.debug(
|
|
|
+ "Chatto: edit of already-answered msg %s ignored",
|
|
|
+ payload.message_event_id,
|
|
|
+ )
|
|
|
+ return
|
|
|
+
|
|
|
+ event = await self._admit_and_build(
|
|
|
+ client,
|
|
|
+ room_id=payload.room_id,
|
|
|
+ message=message,
|
|
|
+ source_message_id=payload.message_event_id,
|
|
|
+ thread_root_event_id=message.thread_root_event_id or None,
|
|
|
+ allow_dm_commands=False,
|
|
|
+ )
|
|
|
+ if event is None:
|
|
|
+ return
|
|
|
+
|
|
|
+ session_key = self._session_key_for(event.source)
|
|
|
+ if self._processing.get(session_key) == payload.message_event_id:
|
|
|
+ logger.info(
|
|
|
+ "Chatto: msg %s edited mid-run - restarting the turn",
|
|
|
+ payload.message_event_id,
|
|
|
+ )
|
|
|
+ # The cancelled task's completion hook clears _processing and
|
|
|
+ # reports 🚫 before this coroutine moves on, because cancel
|
|
|
+ # awaits the task. Queued follow-ups must survive.
|
|
|
+ await self.cancel_session_processing(
|
|
|
+ session_key, release_guard=True, discard_pending=False
|
|
|
+ )
|
|
|
+ elif payload.message_event_id not in self._dispatched_ids:
|
|
|
+ logger.info(
|
|
|
+ "Chatto: msg %s was never dispatched - edit starts a fresh turn",
|
|
|
+ payload.message_event_id,
|
|
|
+ )
|
|
|
+ else:
|
|
|
+ pending = self._pending_messages.get(session_key)
|
|
|
+ if pending is not None and pending.message_id == payload.message_event_id:
|
|
|
+ # Still queued behind the running turn: correct it in place
|
|
|
+ # instead of answering stale wording later.
|
|
|
+ pending.text = event.text
|
|
|
+ logger.info(
|
|
|
+ "Chatto: queued msg %s updated to its edited text",
|
|
|
+ payload.message_event_id,
|
|
|
+ )
|
|
|
+ return
|
|
|
+ logger.debug(
|
|
|
+ "Chatto: edit of already-answered msg %s ignored",
|
|
|
+ payload.message_event_id,
|
|
|
+ )
|
|
|
+ return
|
|
|
+
|
|
|
+ self._remember_dispatched(event.message_id or "")
|
|
|
+ logger.info("Chatto: dispatching edited message to Hermes")
|
|
|
+ await self.handle_message(event)
|
|
|
+
|
|
|
+ async def _admit_and_build(
|
|
|
+ self,
|
|
|
+ client: ChattoClient,
|
|
|
+ *,
|
|
|
+ room_id: str,
|
|
|
+ message: Message,
|
|
|
+ source_message_id: str,
|
|
|
+ thread_root_event_id: str | None,
|
|
|
+ allow_dm_commands: bool = True,
|
|
|
+ ) -> MessageEvent | None:
|
|
|
+ """Run one hydrated inbound message through the admission pipeline.
|
|
|
+
|
|
|
+ Shared by the posted and the edited path: user resolution, auth,
|
|
|
+ mention gates, thread anchoring and media caching all behave
|
|
|
+ identically for both. Returns ``None`` for anything that must not
|
|
|
+ reach the agent. DM membership commands are executed here (side
|
|
|
+ effect) unless ``allow_dm_commands`` is False — edits pass False so
|
|
|
+ a corrected command line neither runs twice nor leaks to the agent.
|
|
|
+
|
|
|
+ The caller owns dispatching: a non-None result still needs
|
|
|
+ ``handle_message()``.
|
|
|
+ """
|
|
|
if message.actor_id in self._user_cache:
|
|
|
# try the user cache.
|
|
|
user = self._user_cache.get(message.actor_id)
|
|
|
@@ -929,41 +1124,44 @@ class ChattoAdapter(BasePlatformAdapter):
|
|
|
# get the user and update cache.
|
|
|
directory_member = await client.get_user(user_id=message.actor_id)
|
|
|
if directory_member is None:
|
|
|
- return
|
|
|
+ return None
|
|
|
user = directory_member.user
|
|
|
if user is None:
|
|
|
- return
|
|
|
+ return None
|
|
|
self._user_cache[user.id] = user
|
|
|
|
|
|
if user is None:
|
|
|
- return
|
|
|
+ return None
|
|
|
|
|
|
if not self._check_auth(user):
|
|
|
- return
|
|
|
+ return None
|
|
|
|
|
|
# Todo: use a function that either reads from cache or gets room kind again.
|
|
|
if self._room_kinds.get(message.room_id) is None:
|
|
|
room_viewer_state = await client.get_room(message.room_id)
|
|
|
if room_viewer_state is None:
|
|
|
- return
|
|
|
+ return None
|
|
|
if room_viewer_state.room is None:
|
|
|
- return
|
|
|
+ return None
|
|
|
self._room_kinds[message.room_id] = (
|
|
|
room_viewer_state.room.kind or RoomKind.UNSPECIFIED
|
|
|
)
|
|
|
|
|
|
room_kind = self._room_kinds.get(message.room_id)
|
|
|
+ message_body = message.body or ""
|
|
|
|
|
|
logger.debug("message_body: %s room_kind: %s", message_body, room_kind)
|
|
|
|
|
|
# Membership commands ride in over DMs only: they change what the bot
|
|
|
# listens to and must never reach the agent pipeline or the mention
|
|
|
# gates.
|
|
|
- if room_kind == RoomKind.DM and await self._handle_dm_command(
|
|
|
- message.room_id,
|
|
|
- message_body,
|
|
|
- ):
|
|
|
- return
|
|
|
+ if room_kind == RoomKind.DM:
|
|
|
+ if allow_dm_commands:
|
|
|
+ if await self._handle_dm_command(room_id, message_body):
|
|
|
+ return None
|
|
|
+ elif message_body.startswith("/"):
|
|
|
+ logger.debug("Chatto: edited DM command %r not re-run", message_body)
|
|
|
+ return None
|
|
|
|
|
|
# require_mention deliberately gates channels only: in a channel the bot
|
|
|
# is one of many listeners and must be addressed, whereas a DM is already
|
|
|
@@ -983,7 +1181,7 @@ class ChattoAdapter(BasePlatformAdapter):
|
|
|
"Discarding message. Bot was not mentionend but require_mention is '%s'.",
|
|
|
self.chatto_config.require_mention.value,
|
|
|
)
|
|
|
- return
|
|
|
+ return None
|
|
|
|
|
|
logger.debug("mentioned: %s", mentioned)
|
|
|
|
|
|
@@ -1001,25 +1199,25 @@ class ChattoAdapter(BasePlatformAdapter):
|
|
|
logger.info("Chatto: message addresses someone else, acknowledging only")
|
|
|
if self.chatto_config.reactions.value:
|
|
|
await self.add_reaction(message.room_id, message.id, "🫥")
|
|
|
- return
|
|
|
+ return None
|
|
|
|
|
|
# Thread anchoring — if the incoming message is inside a Chatto thread, we
|
|
|
# keep that thread by default; otherwise leave thread_id unset so
|
|
|
# replies land at the root.
|
|
|
thread_id = (
|
|
|
- payload.thread_root_event_id or None
|
|
|
- ) # we could also take "payload.room_id" but then, we're in a thread already.
|
|
|
+ thread_root_event_id or None
|
|
|
+ ) # we could also take the room id but then, we're in a thread already.
|
|
|
if not thread_id and room_kind != RoomKind.DM:
|
|
|
thread_id = message.id
|
|
|
|
|
|
source = self.build_source(
|
|
|
- chat_id=payload.room_id,
|
|
|
+ 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=payload.message_event_id,
|
|
|
+ message_id=source_message_id,
|
|
|
role_authorized=True,
|
|
|
)
|
|
|
|
|
|
@@ -1037,12 +1235,13 @@ class ChattoAdapter(BasePlatformAdapter):
|
|
|
|
|
|
# Attachments — download and hand the local cache paths to the gateway,
|
|
|
# which runs vision enrichment / document extraction off media_urls.
|
|
|
+ attachments = list(message.attachments or [])
|
|
|
(
|
|
|
message_event.media_urls,
|
|
|
message_event.media_types,
|
|
|
media_kinds,
|
|
|
) = await self._cache_attachments(
|
|
|
- payload.room_id,
|
|
|
+ room_id,
|
|
|
attachments,
|
|
|
)
|
|
|
if media_kinds:
|
|
|
@@ -1054,9 +1253,7 @@ class ChattoAdapter(BasePlatformAdapter):
|
|
|
message_event.message_type = MessageType.TEXT
|
|
|
|
|
|
logger.debug("Chatto: MessageEvent: %s", message_event)
|
|
|
- logger.info("Chatto: dispatching message to Hermes")
|
|
|
- await self.handle_message(message_event)
|
|
|
- return
|
|
|
+ return message_event
|
|
|
|
|
|
async def _forward_reaction(
|
|
|
self,
|
|
|
@@ -1121,6 +1318,18 @@ class ChattoAdapter(BasePlatformAdapter):
|
|
|
|
|
|
await self._dispatch_message_posted(event_payload)
|
|
|
|
|
|
+ elif (edited_payload := event.get("message_edited")) is not None:
|
|
|
+ # Same self-event filter: our own streaming edits echo back here.
|
|
|
+ # Redeliveries of edits we made are also caught by _mark_seen in
|
|
|
+ # edit_message().
|
|
|
+ actor_id = event.actor_id
|
|
|
+ if actor_id and self.me and actor_id == self.me.id:
|
|
|
+ return
|
|
|
+
|
|
|
+ await self._dispatch_message_edited(
|
|
|
+ cast(MessageEditedPayload, edited_payload)
|
|
|
+ )
|
|
|
+
|
|
|
elif event.kind in ("reaction_added", "reaction_removed"):
|
|
|
reaction_payload = event.get(event.kind)
|
|
|
if reaction_payload is not None:
|
|
|
@@ -1140,7 +1349,6 @@ class ChattoAdapter(BasePlatformAdapter):
|
|
|
"user_typing",
|
|
|
"notification_created",
|
|
|
"new_direct_message_notification",
|
|
|
- "message_edited",
|
|
|
"message_retracted",
|
|
|
):
|
|
|
logger.debug(
|
|
|
@@ -1875,10 +2083,20 @@ class ChattoAdapter(BasePlatformAdapter):
|
|
|
return chat_id, message_id
|
|
|
|
|
|
async def on_processing_start(self, event: MessageEvent) -> None:
|
|
|
- """Add an 👀 (eyes) reaction to the incoming message.
|
|
|
+ """Record the turn as open, then add an 👀 (eyes) reaction.
|
|
|
+
|
|
|
+ The record is what lets an edit of this very message be recognized as
|
|
|
+ a mid-run correction. It must be maintained even when reactions are
|
|
|
+ disabled — tracking and decorating are independent concerns.
|
|
|
|
|
|
BasePlatformAdapter override
|
|
|
"""
|
|
|
+ session_key = (
|
|
|
+ self._session_key_for(event.source) if event.source is not None else ""
|
|
|
+ )
|
|
|
+ if session_key and event.message_id:
|
|
|
+ self._processing[session_key] = str(event.message_id)
|
|
|
+
|
|
|
if not self.chatto_config.reactions.value:
|
|
|
return
|
|
|
|
|
|
@@ -1900,10 +2118,20 @@ class ChattoAdapter(BasePlatformAdapter):
|
|
|
event: MessageEvent,
|
|
|
outcome: ProcessingOutcome,
|
|
|
) -> None:
|
|
|
- """Swap the 👀 reaction for ✅ (success) or ❌ (failure).
|
|
|
+ """Close the turn's record, then swap 👀 for ✅/❌/🚫.
|
|
|
+
|
|
|
+ Fires for every outcome, including CANCELLED (mid-run edit
|
|
|
+ correction) — this is what re-arms `_processing` before the corrected
|
|
|
+ turn is dispatched.
|
|
|
|
|
|
BasePlatformAdapter override
|
|
|
"""
|
|
|
+ session_key = (
|
|
|
+ self._session_key_for(event.source) if event.source is not None else ""
|
|
|
+ )
|
|
|
+ if session_key and event.message_id:
|
|
|
+ self._processing.pop(session_key, None)
|
|
|
+
|
|
|
if not self.chatto_config.reactions.value:
|
|
|
return
|
|
|
chat_id, message_id = self._event_room_and_message_id(event)
|
