|
|
@@ -31,6 +31,7 @@ import os
|
|
|
import re
|
|
|
import tempfile
|
|
|
from collections import deque
|
|
|
+from dataclasses import dataclass, field
|
|
|
from datetime import UTC, datetime
|
|
|
from difflib import SequenceMatcher
|
|
|
from enum import StrEnum
|
|
|
@@ -81,9 +82,9 @@ try:
|
|
|
MessagePostedPayload,
|
|
|
PresenceChangedPayload,
|
|
|
ReactionPayload,
|
|
|
+ RoomEventPayload,
|
|
|
)
|
|
|
from chattolib.types import (
|
|
|
- DirectoryMember,
|
|
|
Message,
|
|
|
MessageAttachment,
|
|
|
PresenceStatus,
|
|
|
@@ -274,6 +275,21 @@ _PRESENCE_LABELS = {
|
|
|
}
|
|
|
|
|
|
|
|
|
+@dataclass
|
|
|
+class _RoomRoster:
|
|
|
+ """One announced room's roster projection.
|
|
|
+
|
|
|
+ A miniature of the server's room membership: which user IDs belong to
|
|
|
+ the room (the users themselves live in the shared ``_user_cache``) and
|
|
|
+ how many directory entries were beyond the fetch limit, rendered as
|
|
|
+ ``… and N more``. Presence changes patch the cached users in place;
|
|
|
+ only a membership change or reconnect discards this and refetches.
|
|
|
+ """
|
|
|
+
|
|
|
+ member_ids: set[str] = field(default_factory=set)
|
|
|
+ unfetched: int = 0
|
|
|
+
|
|
|
+
|
|
|
def hermes_adapter_factory(config: PlatformConfig):
|
|
|
"""Construct a ChattoAdapter from a PlatformConfig."""
|
|
|
return ChattoAdapter(config)
|
|
|
@@ -323,12 +339,17 @@ class ChattoAdapter(BasePlatformAdapter):
|
|
|
self._dispatched_ids: deque[str] = deque(maxlen=ChattoConstants.SEEN_CAP)
|
|
|
# Rooms whose member roster has already been announced to the agent via
|
|
|
# channel_context — the roster is room-scoped, so the first turn in a
|
|
|
- # room carries it and only a presence change re-delivers it.
|
|
|
+ # room carries it. Together with _rosters/_roster_text this is our
|
|
|
+ # miniature projection of server state: presence changes patch the
|
|
|
+ # cached users in place and the next turn re-renders from cache;
|
|
|
+ # membership events and reconnects discard a room so its next turn
|
|
|
+ # refetches once.
|
|
|
self._roster_announced: deque[str] = deque(maxlen=ChattoConstants.SEEN_CAP)
|
|
|
- # Rooms whose announced roster is stale after a presence_changed event
|
|
|
- # and must be re-fetched on the next turn there. Lazy: the refresh
|
|
|
- # happens when an inbound message actually arrives, not on the event.
|
|
|
- self._roster_dirty: set[str] = set()
|
|
|
+ # Announced rooms' membership projection (see _RoomRoster).
|
|
|
+ self._rosters: dict[str, _RoomRoster] = {}
|
|
|
+ # Announced rooms' last delivered roster text — the dedup key that
|
|
|
+ # keeps presence churn from repeating an unchanged channel_context.
|
|
|
+ self._roster_text: dict[str, 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.
|
|
|
@@ -884,6 +905,8 @@ class ChattoAdapter(BasePlatformAdapter):
|
|
|
return f"Chatto refused to leave {label}."
|
|
|
if room_obj.id in self._joined_room_ids:
|
|
|
self._joined_room_ids.remove(room_obj.id)
|
|
|
+ # We are no longer part of this audience; keep no roster for it.
|
|
|
+ self._evict_roster(room_obj.id)
|
|
|
return f"Left {label}."
|
|
|
|
|
|
def _room_join_hint(self, room_id: str) -> str:
|
|
|
@@ -1071,13 +1094,17 @@ class ChattoAdapter(BasePlatformAdapter):
|
|
|
return True
|
|
|
return self._room_policy(room_id) != RoomPolicy.SILENT
|
|
|
|
|
|
- async def _room_roster_context(self, client: ChattoClient, room_id: str) -> str:
|
|
|
- """The channel's member roster as one context line, "" on any failure.
|
|
|
+ async def _room_roster_context(
|
|
|
+ self, client: ChattoClient, room_id: str
|
|
|
+ ) -> tuple[_RoomRoster, str] | None:
|
|
|
+ """Fetch the room's member roster as a fresh projection + rendered line.
|
|
|
|
|
|
The agent only ever sees its prompt: without this block it cannot know
|
|
|
who else is in a channel, because unaddressed messages are dropped by
|
|
|
- mention gating long before they could teach it a name. Best-effort by
|
|
|
- design — a directory hiccup must never cost the turn itself.
|
|
|
+ mention gating long before they could teach it a name. Fetched users go
|
|
|
+ into _user_cache so presence patching and mention resolution share one
|
|
|
+ store. Returns ``None`` on failure or when nothing renderable remains —
|
|
|
+ best-effort by design, a directory hiccup must never cost the turn.
|
|
|
"""
|
|
|
try:
|
|
|
members, page = await client.list_room_members(
|
|
|
@@ -1087,17 +1114,25 @@ class ChattoAdapter(BasePlatformAdapter):
|
|
|
logger.debug(
|
|
|
"Chatto: could not list members of room %s", room_id, exc_info=True
|
|
|
)
|
|
|
- return ""
|
|
|
+ return None
|
|
|
logger.debug(
|
|
|
"Chatto: roster for room %s: %d members fetched, total_count=%d",
|
|
|
room_id,
|
|
|
len(members),
|
|
|
page.total_count,
|
|
|
)
|
|
|
+
|
|
|
+ member_ids: set[str] = set()
|
|
|
+ users: list[User] = []
|
|
|
+ for member in members:
|
|
|
+ user = member.user
|
|
|
+ if user is None or user.deleted:
|
|
|
+ continue
|
|
|
+ member_ids.add(user.id)
|
|
|
+ self._user_cache[user.id] = user
|
|
|
+ users.append(user)
|
|
|
entries = [
|
|
|
- entry
|
|
|
- for entry in (self._roster_entry(member) for member in members)
|
|
|
- if entry
|
|
|
+ entry for entry in (self._roster_entry(user) for user in users) if entry
|
|
|
]
|
|
|
if not entries:
|
|
|
logger.debug(
|
|
|
@@ -1106,20 +1141,22 @@ class ChattoAdapter(BasePlatformAdapter):
|
|
|
room_id,
|
|
|
len(members),
|
|
|
)
|
|
|
- return ""
|
|
|
+ return None
|
|
|
|
|
|
unfetched = max(0, page.total_count - len(members))
|
|
|
more = f" … and {unfetched} more" if unfetched else ""
|
|
|
- return ", ".join(entries) + more
|
|
|
+ return (
|
|
|
+ _RoomRoster(member_ids=member_ids, unfetched=unfetched),
|
|
|
+ ", ".join(entries) + more,
|
|
|
+ )
|
|
|
|
|
|
- def _roster_entry(self, member: DirectoryMember) -> str:
|
|
|
+ def _roster_entry(self, user: User) -> str:
|
|
|
"""One roster entry, e.g. ``@bob (Bob Example, online)``.
|
|
|
|
|
|
Skips deleted users and our own account (the agent knows itself);
|
|
|
display name and presence are optional parts of the parenthetical.
|
|
|
"""
|
|
|
- user = member.user
|
|
|
- if user is None or user.deleted:
|
|
|
+ if user.deleted:
|
|
|
return ""
|
|
|
if self.me is not None and user.id == self.me.id:
|
|
|
return ""
|
|
|
@@ -1136,31 +1173,81 @@ class ChattoAdapter(BasePlatformAdapter):
|
|
|
entry += f" ({', '.join(details)})"
|
|
|
return entry
|
|
|
|
|
|
- async def _roster_for_room(self, client: ChattoClient, *, room_id: str) -> str:
|
|
|
- """Roster context for this room — once, then only on presence change.
|
|
|
+ def _roster_line(self, room_id: str) -> str:
|
|
|
+ """Render the room's roster line from its cached projection."""
|
|
|
+ roster = self._rosters.get(room_id)
|
|
|
+ if roster is None:
|
|
|
+ return ""
|
|
|
+ users = [self._user_cache.get(member_id) for member_id in roster.member_ids]
|
|
|
+ entries = [
|
|
|
+ entry
|
|
|
+ for entry in (
|
|
|
+ self._roster_entry(user) for user in users if user is not None
|
|
|
+ )
|
|
|
+ if entry
|
|
|
+ ]
|
|
|
+ more = f" … and {roster.unfetched} more" if roster.unfetched else ""
|
|
|
+ return ", ".join(entries) + more
|
|
|
+
|
|
|
+ def _announce_roster(self, room_id: str, roster: _RoomRoster, text: str) -> None:
|
|
|
+ """Record a freshly fetched roster as the room's announced projection.
|
|
|
|
|
|
- The roster is room-scoped, so the first turn in a room carries it via
|
|
|
- channel_context. After that the room stays announced and further turns
|
|
|
- bring nothing until a presence_changed event marks it stale
|
|
|
- (``_roster_dirty``), which makes the next turn re-fetch and re-deliver
|
|
|
- it. Bounded like _dispatched_ids; a room counts as announced only once
|
|
|
- a roster was actually fetched: a failed lookup leaves it unmarked, so
|
|
|
- the next turn through this gate can still bring the roster.
|
|
|
+ Bounded like _dispatched_ids: appending past the cap drops the oldest
|
|
|
+ room's projection alongside its deque entry.
|
|
|
"""
|
|
|
- if room_id in self._roster_announced and room_id not in self._roster_dirty:
|
|
|
- logger.debug(
|
|
|
- "Chatto: roster for room %s already announced and fresh",
|
|
|
- room_id,
|
|
|
- )
|
|
|
- return ""
|
|
|
- roster = await self._room_roster_context(client, room_id)
|
|
|
- if not roster:
|
|
|
- self._roster_dirty.discard(room_id)
|
|
|
- return ""
|
|
|
if room_id not in self._roster_announced:
|
|
|
+ oldest = (
|
|
|
+ self._roster_announced.popleft()
|
|
|
+ if len(self._roster_announced) == self._roster_announced.maxlen
|
|
|
+ else None
|
|
|
+ )
|
|
|
self._roster_announced.append(room_id)
|
|
|
- self._roster_dirty.discard(room_id)
|
|
|
- return roster
|
|
|
+ if oldest is not None:
|
|
|
+ self._rosters.pop(oldest, None)
|
|
|
+ self._roster_text.pop(oldest, None)
|
|
|
+ self._rosters[room_id] = roster
|
|
|
+ self._roster_text[room_id] = text
|
|
|
+
|
|
|
+ def _evict_roster(self, room_id: str) -> None:
|
|
|
+ """Drop all roster state for a room.
|
|
|
+
|
|
|
+ Called on membership changes (user_joined/left_room), when we leave a
|
|
|
+ room ourselves, and on reconnect — protocol v1 has no presence snapshot
|
|
|
+ on subscribe, so a discarded cache is what forces one honest refetch.
|
|
|
+ """
|
|
|
+ try:
|
|
|
+ self._roster_announced.remove(room_id)
|
|
|
+ except ValueError:
|
|
|
+ pass
|
|
|
+ self._rosters.pop(room_id, None)
|
|
|
+ self._roster_text.pop(room_id, None)
|
|
|
+
|
|
|
+ async def _roster_for_room(self, client: ChattoClient, *, room_id: str) -> str:
|
|
|
+ """Roster context for this room, kept fresh through the projection.
|
|
|
+
|
|
|
+ The first turn in a room fetches and announces the roster; afterwards
|
|
|
+ the cached members are re-rendered and re-delivered only when the line
|
|
|
+ actually changed — presence patches land in _user_cache without any
|
|
|
+ API call, so churn like rapid away/offline flips costs dictionary
|
|
|
+ writes only. Membership events or a reconnect evict the room, making
|
|
|
+ its next turn refetch once. An unannounced room retries on its next
|
|
|
+ turn until a lookup succeeds.
|
|
|
+ """
|
|
|
+ if room_id in self._roster_announced:
|
|
|
+ line = self._roster_line(room_id)
|
|
|
+ if line == self._roster_text.get(room_id):
|
|
|
+ logger.debug("Chatto: roster for room %s unchanged", room_id)
|
|
|
+ return ""
|
|
|
+ logger.debug("Chatto: roster for room %s changed - redelivering", room_id)
|
|
|
+ self._roster_text[room_id] = line
|
|
|
+ return line
|
|
|
+
|
|
|
+ fetched = await self._room_roster_context(client, room_id)
|
|
|
+ if fetched is None:
|
|
|
+ return ""
|
|
|
+ roster, text = fetched
|
|
|
+ self._announce_roster(room_id, roster, text)
|
|
|
+ return text
|
|
|
|
|
|
async def _dispatch_message_posted(self, payload: MessagePostedPayload) -> None:
|
|
|
# Respond-room gate first: read-only memberships must not cost a
|
|
|
@@ -1455,10 +1542,10 @@ class ChattoAdapter(BasePlatformAdapter):
|
|
|
)
|
|
|
|
|
|
# The first turn in a room is where the agent learns who is in the
|
|
|
- # channel — later turns bring the roster back only after a presence
|
|
|
- # change (see _roster_for_room). The gateway prepends channel_context
|
|
|
- # above the message text, so the roster never mingles with what the user
|
|
|
- # actually wrote.
|
|
|
+ # channel — later turns re-deliver it only when the cached projection
|
|
|
+ # renders a different line (see _roster_for_room). The gateway prepends
|
|
|
+ # channel_context above the message text, so the roster never mingles
|
|
|
+ # with what the user actually wrote.
|
|
|
roster = ""
|
|
|
if room_kind != RoomKind.DM:
|
|
|
logger.debug(
|
|
|
@@ -1469,7 +1556,7 @@ class ChattoAdapter(BasePlatformAdapter):
|
|
|
if not roster:
|
|
|
logger.debug(
|
|
|
"Chatto: dispatching channel message without roster "
|
|
|
- "(lookup failed, empty, or already announced and fresh)"
|
|
|
+ "(lookup failed, empty, or unchanged)"
|
|
|
)
|
|
|
|
|
|
source = self.build_source(
|
|
|
@@ -1599,34 +1686,48 @@ class ChattoAdapter(BasePlatformAdapter):
|
|
|
)
|
|
|
|
|
|
elif (presence := event.get("presence_changed")) is not None:
|
|
|
- # A participant's presence changed — any room whose roster the agent
|
|
|
- # has already seen is now stale and must be re-delivered on the next
|
|
|
- # turn there. Our own 60s presence refresh must not churn every room,
|
|
|
- # so ignore self events; rooms we never announced need no refresh.
|
|
|
- user_id = cast(PresenceChangedPayload, presence).user_id
|
|
|
- if self.me is not None and user_id == self.me.id:
|
|
|
- logger.debug("Chatto: ignoring own presence change for roster refresh")
|
|
|
- elif self._roster_announced:
|
|
|
- self._roster_dirty.update(self._roster_announced)
|
|
|
+ # A participant's presence changed — patch the cached user in place
|
|
|
+ # and let the next turn in any room they are announced for re-render
|
|
|
+ # its roster line. No flags, no API calls: churn like rapid
|
|
|
+ # away/offline flips costs dictionary writes only. Our own 60s
|
|
|
+ # presence refresh must not feed the agent a self-roster, so self
|
|
|
+ # events are ignored; users we never cached enter with the next
|
|
|
+ # roster fetch carrying current status anyway.
|
|
|
+ presence_payload = cast(PresenceChangedPayload, presence)
|
|
|
+ if self.me is not None and presence_payload.user_id == self.me.id:
|
|
|
+ logger.debug("Chatto: ignoring own presence change")
|
|
|
+ elif (user := self._user_cache.get(presence_payload.user_id)) is not None:
|
|
|
+ user.presence_status = presence_payload.status
|
|
|
logger.debug(
|
|
|
- "Chatto: marked %d announced room(s) stale after presence "
|
|
|
- "change of user %s",
|
|
|
- len(self._roster_dirty),
|
|
|
- user_id,
|
|
|
+ "Chatto: patched presence of %s to %s",
|
|
|
+ presence_payload.user_id,
|
|
|
+ presence_payload.status.name,
|
|
|
)
|
|
|
|
|
|
+ elif event.kind in ("user_joined_room", "user_left_room"):
|
|
|
+ # Membership moved — the room's roster projection is now wrong, so
|
|
|
+ # discard it; the next turn there refetches once and re-delivers.
|
|
|
+ room_event = cast(RoomEventPayload, event.get(event.kind))
|
|
|
+ if room_event is not None and room_event.room_id in self._roster_announced:
|
|
|
+ logger.debug(
|
|
|
+ "Chatto: %s invalidated the roster of room %s",
|
|
|
+ event.kind,
|
|
|
+ room_event.room_id,
|
|
|
+ )
|
|
|
+ self._evict_roster(room_event.room_id)
|
|
|
+
|
|
|
# confirmed:
|
|
|
+ # NOTE: "projection_event" (and "caught_up") belong to realtime protocol
|
|
|
+ # v2 on unreleased Chatto main — chattolib speaks v1 and can never
|
|
|
+ # deliver them here. Revisit when vendored chattolib gains v2 typing.
|
|
|
elif event.kind in (
|
|
|
"mention_notification",
|
|
|
- "projection_event",
|
|
|
"notification_dismissed",
|
|
|
"room_marked_as_read",
|
|
|
"user_typing",
|
|
|
"notification_created",
|
|
|
"new_direct_message_notification",
|
|
|
"message_retracted",
|
|
|
- "user_joined_room",
|
|
|
- "user_left_room",
|
|
|
"thread_created",
|
|
|
"thread_follow_changed",
|
|
|
"room_updated",
|
|
|
@@ -1803,12 +1904,22 @@ class ChattoAdapter(BasePlatformAdapter):
|
|
|
)
|
|
|
|
|
|
async def _refresh_rooms(self) -> None:
|
|
|
- """Refresh room list via ConnectRPC, join and seed any newly discovered rooms."""
|
|
|
+ """Refresh room list via ConnectRPC, join and seed any newly discovered rooms.
|
|
|
+
|
|
|
+ Runs once per (re)connect, so this is also the bootstrap point for the
|
|
|
+ roster projection: protocol v1 sends no presence snapshot on subscribe,
|
|
|
+ meaning cached presence could only heal on the next random change.
|
|
|
+ Discarding it makes each announced room's first turn refetch fresh
|
|
|
+ data — lazily, so quiet rooms stay free of API calls.
|
|
|
+ """
|
|
|
client = await self._get_chatto_client()
|
|
|
if client is None:
|
|
|
logger.warning("Chatto WS: _refresh_rooms aborted - no client available")
|
|
|
return
|
|
|
|
|
|
+ for announced in list(self._roster_announced):
|
|
|
+ self._evict_roster(announced)
|
|
|
+
|
|
|
try:
|
|
|
rooms_list = await client.list_rooms()
|
|
|
member_ids: set[str] = set()
|
|
|
@@ -1843,6 +1954,7 @@ class ChattoAdapter(BasePlatformAdapter):
|
|
|
]
|
|
|
for rid in stale_room_ids:
|
|
|
self._joined_room_ids.remove(rid)
|
|
|
+ self._evict_roster(rid)
|
|
|
if stale_room_ids:
|
|
|
logger.info(
|
|
|
"Chatto WS: no longer a member of %d room(s): %s",
|
|
|
@@ -3135,10 +3247,10 @@ def hermes_env_enablement_fn() -> dict:
|
|
|
know that it is needed for plugin register().
|
|
|
"""
|
|
|
seed: dict[str, Any] = {}
|
|
|
- for field in ChattoConfiguration.fields():
|
|
|
- env_value = os.getenv(field.env_name)
|
|
|
+ for config_field in ChattoConfiguration.fields():
|
|
|
+ env_value = os.getenv(config_field.env_name)
|
|
|
if env_value:
|
|
|
- seed[field.config_key] = env_value.strip()
|
|
|
+ seed[config_field.config_key] = env_value.strip()
|
|
|
|
|
|
logger.debug(
|
|
|
"seed: %s", {k: v for k, v in seed.items() if k not in ("token", "password")}
|