adapter.py 60 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541154215431544154515461547
  1. """
  2. Chatto Platform Adapter for Hermes Agent.
  3. A plugin-based gateway adapter that connects to a Chatto server
  4. (self-hosted team chat) and relays messages to/from the Hermes agent.
  5. The adapter uses the chattolib library for all Chatto API interactions,
  6. including both outbound messaging and realtime WebSocket connections.
  7. """
  8. from __future__ import annotations
  9. import inspect
  10. import random
  11. from gateway.platforms.helpers import MessageDeduplicator
  12. # Put the vendored dependencies for THIS platform on sys.path before importing
  13. # anything from chattolib. Imported relatively as part of the plugin package and
  14. # absolutely when this module is loaded standalone (e.g. by the tests).
  15. try:
  16. from .vendor_path import setup_vendor_path
  17. except ImportError: # pragma: no cover - depends on how the module is loaded
  18. from vendor_path import setup_vendor_path
  19. setup_vendor_path()
  20. import asyncio
  21. import hashlib
  22. import logging
  23. import mimetypes
  24. import os
  25. from datetime import datetime, timezone
  26. from typing import Any, Dict, List, Literal, Optional, Tuple, cast
  27. from urllib.parse import urlsplit
  28. logger = logging.getLogger(__name__)
  29. from gateway.platforms.base import (
  30. BasePlatformAdapter,
  31. SendResult,
  32. MessageEvent,
  33. MessageType,
  34. ProcessingOutcome,
  35. cache_media_bytes,
  36. get_inbound_media_max_bytes,
  37. validate_inbound_media_size,
  38. )
  39. from gateway.config import Platform, PlatformConfig
  40. # Chattolib imports (vendored)
  41. # Using vendored chattolib from vendor/chattolib/
  42. # See vendor_chattolib.sh for how to update the vendored copy
  43. # Absolute imports — vendor/ is on sys.path (see above) and chattolib's own
  44. # modules import each other absolutely. Mixing in relative ".vendor.chattolib"
  45. # imports would load a second, distinct copy of every module, so isinstance()
  46. # checks across the two copies would silently fail.
  47. try:
  48. from chattolib.client import (
  49. ChattoClient,
  50. )
  51. from chattolib.exceptions import (
  52. ChattoAuthError,
  53. ChattoError,
  54. )
  55. from chattolib.realtime import (
  56. ChattoRealtimeError,
  57. ChattoRealtimeCloseError, RealtimeEvent,
  58. stream_events
  59. )
  60. from chattolib.realtime_types import (
  61. MessagePostedPayload
  62. )
  63. from chattolib.types import (
  64. PresenceStatus, RoomKind, User
  65. )
  66. except ImportError as e:
  67. # Fail loudly: continuing here only defers the failure to a confusing
  68. # NameError somewhere deep in the adapter.
  69. logger.error("Chatto: failed to import vendored chattolib: %s", e)
  70. raise
  71. try:
  72. from .platform_config import (
  73. ChattoConfiguration, ChattoConstants,
  74. )
  75. except ImportError: # pragma: no cover - loaded as a top-level module (tests)
  76. from platform_config import (
  77. ChattoConfiguration, ChattoConstants,
  78. )
  79. # --------------------------------------------------------------------------- #
  80. # Adapter
  81. # --------------------------------------------------------------------------- #
  82. def hermes_adapter_factory(config: PlatformConfig):
  83. """Factory wrapper that constructs ChattoAdapter from a PlatformConfig."""
  84. return ChattoAdapter(config)
  85. class ChattoAdapter(BasePlatformAdapter):
  86. """Chatto platform adapter — receives messages via WebSocket realtime,
  87. sends via ConnectRPC."""
  88. _SPLIT_THRESHOLD = 9900
  89. # Read by BasePlatformAdapter.max_message_length_for_chat(), which the
  90. # gateway and the stream consumer use to chunk outgoing messages. Without
  91. # it they fall back to 4096 and split Chatto messages far earlier than
  92. # necessary — send() itself already truncates at 10000.
  93. MAX_MESSAGE_LENGTH = ChattoConstants.MAX_MESSAGE_LENGTH
  94. splits_long_messages = True
  95. supports_code_blocks: bool = True
  96. supports_status_text: bool = True # client.update_custom_status
  97. def __init__(self, pconfig: PlatformConfig):
  98. """Signature needs to be compatible with BasePlatformAdapter.__init__ """
  99. super().__init__(config=pconfig, platform=Platform(ChattoConstants.PLATFORM_NAME))
  100. # "extra" has been pre-processed by Hermes-Framework to be a dict of extra config values insode PlatformConfig.
  101. # --- Configuration from our configuration data class with some logic ---
  102. self.chatto_config: ChattoConfiguration = ChattoConfiguration(pconfig)
  103. # ------ State -------
  104. # SDK runtime handle (injected by Hermes); annotate for Pylance
  105. self.sdk: Any = getattr(self, "sdk", None)
  106. # Our own user, filled in by connect(). Events arriving before connect()
  107. # completes must not blow up on an undefined attribute.
  108. self.me: Optional[User] = None
  109. # --- Runtime state ---
  110. self._user_id: str = ""
  111. self._user_display: str = ""
  112. self._room_names: Dict[str, str] = {}
  113. self._room_kinds: Dict[str, RoomKind] = {}
  114. self._our_thread_roots: set = set() # thread root event IDs we created
  115. self._our_message_ids: set = set() # message IDs we sent (for thread root detection)
  116. self._seen: list[str] = [] # Plain RealtimeEvent-id list
  117. self._resume_cursor: Optional[str] = None
  118. self._watch_room_ids: List[str] = []
  119. self._ws_task: Optional[asyncio.Task] = None
  120. self._ws_ready: Optional[asyncio.Event] = None
  121. self._ws_active = False
  122. self._ws_ref = None # reference to open websocket for dynamic resubscribe
  123. # Persistent typing indicator loops per room
  124. self._typing_tasks: Dict[str, asyncio.Task] = {}
  125. # Member directory cache: user_id -> user info dict
  126. self._user_cache: Dict[str, User] = {}
  127. # Chattolib client cache and lock for async access.
  128. self._chatto_client: Optional[ChattoClient] = None
  129. self._chatto_client_lock: asyncio.Lock = asyncio.Lock()
  130. # Dedup — chattolib may redeliver events across reconnects.
  131. self._dedup = MessageDeduplicator()
  132. def _list_functions(self):
  133. for name, member in inspect.getmembers(self, predicate=callable):
  134. # filtert dunder-Methoden falls gewünscht
  135. if not name.startswith('__'):
  136. logger.info("functions: %s", name)
  137. # ------------------------------------------------------------------ #
  138. # Auth
  139. # ------------------------------------------------------------------ #
  140. async def _get_chatto_client(self: ChattoAdapter) -> Optional[ChattoClient]:
  141. """Get or create a ChattoClient instance."""
  142. if self._chatto_client is not None:
  143. return self._chatto_client
  144. async with self._chatto_client_lock:
  145. if self._chatto_client is not None:
  146. return self._chatto_client
  147. try:
  148. assert(self.chatto_config.login.value) # now we can assume, _login is available.
  149. client = await self._open_client(
  150. base_url=self.chatto_config.base_url.value,
  151. login=self.chatto_config.login.value,
  152. password=self.chatto_config.password.value,
  153. token=self.chatto_config.token.value,
  154. )
  155. self._chatto_client = client
  156. self._token = client.token
  157. logger.info("Chatto: logged in as '%s' via chattolib", self.chatto_config.login.value)
  158. return client
  159. except ChattoAuthError as e:
  160. logger.error("Chatto: authentication failed: %s", e)
  161. return None
  162. except (ChattoError, ValueError) as e:
  163. logger.error("Chatto: failed to create client: %s", e)
  164. return None
  165. async def _require_client(self) -> ChattoClient:
  166. """Return a ChattoClient or raise RuntimeError if unavailable.
  167. Use this helper when the caller expects a client to exist and
  168. wants a single canonical failure path. Methods that prefer a
  169. soft-fail can catch RuntimeError and return gracefully.
  170. """
  171. client = await self._get_chatto_client()
  172. if client is None:
  173. raise RuntimeError("Chatto client unavailable")
  174. return client
  175. async def _ensure_token(self) -> bool:
  176. """Ensure we have a logged-in Chatto client and token."""
  177. if self.chatto_config.token.value and isinstance(self._chatto_client, ChattoClient):
  178. return True
  179. client = await self._get_chatto_client()
  180. return client is not None
  181. async def _relogin(self) -> bool:
  182. """Force re-login (token expired)."""
  183. self._token = None
  184. return await self._ensure_token()
  185. # ------------------------------------------------------------------ #
  186. # Connection
  187. # ------------------------------------------------------------------ #
  188. async def _open_client(
  189. self,
  190. *,
  191. base_url: str,
  192. login: str,
  193. password: str,
  194. token: Optional[str] = None,
  195. ) -> ChattoClient:
  196. """Return a connected ``ChattoClient`` using token or login/password."""
  197. if token:
  198. return ChattoClient(token=token, base_url=base_url)
  199. return await ChattoClient.login(login, password, base_url=base_url)
  200. async def connect(self, *, is_reconnect: bool = False) -> bool:
  201. """Connect to Chatto and start the realtime event stream.
  202. BasePlatformAdapter override
  203. """
  204. logger.info("Chatto: connecting...")
  205. if not await self._ensure_token():
  206. return False
  207. try:
  208. client = await self._require_client()
  209. except RuntimeError:
  210. self._set_fatal_error("connect_failed", "Chatto client not available", retryable=True)
  211. return False
  212. # Get our first own user info
  213. try:
  214. self.me = await client.me()
  215. except Exception as exc:
  216. logger.error("Chatto: failed to get user info: %s", exc)
  217. self._set_fatal_error(
  218. "chatto_auth_failed",
  219. f"Chatto auth failed: {exc}",
  220. retryable=False,
  221. )
  222. try:
  223. assert(self._chatto_client)
  224. await self._chatto_client.close()
  225. finally:
  226. self._chatto_client = None
  227. return False
  228. # Broadcast online presence so the bot appears online in the member list
  229. try:
  230. await client.update_presence(status=PresenceStatus.ONLINE)
  231. except Exception:
  232. logger.debug("Chatto: update_presence(online) failed on connect", exc_info=True)
  233. self._closing = False
  234. # Start background realtime WS event stream loop.
  235. self._ws_ready = asyncio.Event()
  236. self._ws_task = asyncio.create_task(
  237. self._chattolib_event_loop(), name="chatto-event-stream",
  238. )
  239. self._mark_connected()
  240. self._list_functions()
  241. logger.info(
  242. "Chatto: connected and authentiated to %s using login:'%s' (display_name:'%s' id: '%s')",
  243. self.chatto_config.base_url.value,
  244. self.me.login, self.me.display_name, self.me.id
  245. )
  246. return True
  247. async def disconnect(self) -> None:
  248. """Stop WebSocket, liveness probe, typing tasks, and clear state.
  249. BasePlatformAdapter override
  250. """
  251. # Broadcast offline presence before tearing down
  252. try:
  253. client = await self._require_client()
  254. await client.update_presence(status=PresenceStatus.OFFLINE)
  255. except Exception:
  256. logger.debug("Chatto: update_presence(PresenceStatus.OFFLINE) failed on disconnect", exc_info=True)
  257. self._ws_active = False
  258. self._closing = True
  259. # Cancel all typing tasks
  260. for chat_id in list(self._typing_tasks.keys()):
  261. await self.stop_typing(chat_id)
  262. if self._ws_task and not self._ws_task.done():
  263. self._ws_task.cancel()
  264. try:
  265. await self._ws_task
  266. except (asyncio.CancelledError, Exception):
  267. pass
  268. self._ws_task = None
  269. if self._chatto_client:
  270. try:
  271. await self._chatto_client.close()
  272. except Exception:
  273. logger.exception("Chatto: error closing client")
  274. finally:
  275. self._chatto_client = None
  276. self._token = None
  277. logger.info("Chatto: disconnected")
  278. self._mark_disconnected()
  279. async def _seed_room(self, room_id: str) -> None:
  280. """Seed high-water mark from the newest events so a restart doesn't replay history."""
  281. try:
  282. try:
  283. client = await self._require_client()
  284. timeline_page = await client.get_room_events(room_id)
  285. except RuntimeError:
  286. logger.debug("Chatto: _seed_room aborted - no client available for %s", room_id)
  287. return
  288. for ev in timeline_page.events:
  289. if ev.id:
  290. self._mark_seen(ev.id)
  291. logger.debug("Chatto: seeded room %s with %d events", room_id, len(timeline_page.events))
  292. except Exception as e:
  293. logger.debug("Chatto: get room events failed for %s: %s", room_id, e)
  294. # ------------------------------------------------------------------ #
  295. # Realtime Event List
  296. # ------------------------------------------------------------------ #
  297. def _mark_seen(self, event_id: str) -> None:
  298. self._seen.append(event_id)
  299. while len(self._seen) > ChattoConstants.SEEN_CAP:
  300. self._seen.remove(self._seen[0]) # fastest removal of first item in a list.
  301. def _is_seen(self, event_id: str) -> bool:
  302. return event_id in self._seen
  303. # ------------------------------------------------------------------ #
  304. # WebSocket Realtime Transport
  305. # ------------------------------------------------------------------ #
  306. def _check_auth(self, user: User) -> bool:
  307. """We roll our own auth-systen on the against the chatto_config'ured allowed_users etc.
  308. because.. Hermes authz_mixin.py IS NOT SANE.
  309. """
  310. if self.chatto_config.allow_all_users.value:
  311. return True
  312. if user.login in self.chatto_config.allowed_users.value:
  313. return True
  314. if user.id in self.chatto_config.allowed_users.value:
  315. return True
  316. logger.info("Chatto: rejecting message from unauthorized user '%s' (%s)", user.login, user.id)
  317. return False
  318. # ------------------------------------------------------------------ #
  319. # Inbound attachments
  320. # ------------------------------------------------------------------ #
  321. async def _download_attachment_bytes(self, url: str) -> bytes:
  322. """Download an attachment, refusing to buffer more than the gateway cap.
  323. The Content-Length header is checked first so an oversized asset is
  324. rejected before a single chunk is read; the running total is re-checked
  325. as chunks arrive, because a missing or lying header must not smuggle an
  326. unbounded body past the cap.
  327. """
  328. import httpx
  329. max_bytes = get_inbound_media_max_bytes()
  330. chunks: List[bytes] = []
  331. total = 0
  332. async with httpx.AsyncClient(
  333. timeout=ChattoConstants.HTTP_TIMEOUT, follow_redirects=True,
  334. ) as http:
  335. async with http.stream("GET", url) as resp:
  336. resp.raise_for_status()
  337. declared = resp.headers.get("content-length")
  338. if declared:
  339. try:
  340. declared_size = int(declared)
  341. except ValueError:
  342. logger.debug("Chatto: ignoring invalid Content-Length %r", declared)
  343. else:
  344. validate_inbound_media_size(
  345. declared_size, media_type="attachment", max_bytes=max_bytes,
  346. )
  347. async for chunk in resp.aiter_bytes():
  348. total += len(chunk)
  349. validate_inbound_media_size(
  350. total, media_type="attachment", max_bytes=max_bytes,
  351. )
  352. chunks.append(chunk)
  353. return b"".join(chunks)
  354. async def _cache_attachments(
  355. self, room_id: str, attachments: List[Any],
  356. ) -> Tuple[List[str], List[str], List[str]]:
  357. """Download message attachments into the gateway media cache.
  358. Returns ``(media_urls, media_types, media_kinds)`` — the paths are
  359. agent-visible cache paths, exactly what ``cache_media_bytes`` yields for
  360. every other platform. A failing attachment is logged and skipped: the
  361. message itself still reaches the agent.
  362. """
  363. media_urls: List[str] = []
  364. media_types: List[str] = []
  365. media_kinds: List[str] = []
  366. for att in attachments or []:
  367. asset_url = getattr(att, "asset_url", None)
  368. url = getattr(asset_url, "url", "") if asset_url else ""
  369. filename = getattr(att, "filename", "") or ""
  370. content_type = getattr(att, "content_type", "") or ""
  371. if not url:
  372. # Videos are announced before transcoding finishes, so the
  373. # signed URL can legitimately be missing on arrival.
  374. logger.info(
  375. "Chatto: attachment '%s' has no asset URL yet, skipping", filename,
  376. )
  377. continue
  378. try:
  379. data = await self._download_attachment_bytes(url)
  380. cached = cache_media_bytes(
  381. data, filename=filename, mime_type=content_type,
  382. )
  383. except Exception as e:
  384. logger.warning(
  385. "Chatto: failed to cache attachment '%s' (%s): %s",
  386. filename, content_type, e,
  387. )
  388. continue
  389. if cached is None:
  390. logger.warning(
  391. "Chatto: attachment '%s' (%s) could not be cached, skipping",
  392. filename, content_type,
  393. )
  394. continue
  395. media_urls.append(cached.path)
  396. media_types.append(cached.media_type)
  397. media_kinds.append(cached.kind)
  398. return media_urls, media_types, media_kinds
  399. @staticmethod
  400. def _message_type_for_media_kinds(media_kinds: List[str]) -> MessageType:
  401. """Pick the MessageType for a set of cached attachment kinds."""
  402. if "document" in media_kinds:
  403. return MessageType.DOCUMENT
  404. if "image" in media_kinds:
  405. return MessageType.PHOTO
  406. if "video" in media_kinds:
  407. return MessageType.VIDEO
  408. if "audio" in media_kinds:
  409. return MessageType.AUDIO
  410. return MessageType.TEXT
  411. async def _dispatch_message_posted(self, payload: MessagePostedPayload) -> None:
  412. try:
  413. client = await self._require_client()
  414. except RuntimeError:
  415. logger.warning("Chatto: chattolib event loop aborted - no client available")
  416. return
  417. logger.info("Chatto WS: 'message_posted' event_payload:%s", payload)
  418. message = await payload.fetch_message(client=client)
  419. if message is None or message.deleted_at:
  420. return
  421. message_body = message.body or ""
  422. attachments = list(message.attachments or [])
  423. # A message carrying only an image/PDF has an empty body — dropping it
  424. # here is what made attachments sent to Hermes disappear silently.
  425. if not message_body and not attachments:
  426. return
  427. if message.actor_id in self._user_cache:
  428. # try the user cache.
  429. user = self._user_cache.get(message.actor_id)
  430. else:
  431. # get the user and update cache.
  432. directory_member = await client.get_user(user_id=message.actor_id)
  433. if directory_member is None:
  434. return
  435. user = directory_member.user
  436. if user is None:
  437. return
  438. self._user_cache[user.id] = user
  439. if user is None:
  440. return
  441. if not self._check_auth(user):
  442. return
  443. # For DMs, always respond. For rooms with require_mention, only respond when mentioned.
  444. # Strip the mention from the text for the agent
  445. # Todo: use a function that either reads from cache or gets room kind again.
  446. if self._room_kinds.get(message.room_id) is None:
  447. room_viewer_state = await client.get_room(message.room_id)
  448. if room_viewer_state is None:
  449. return
  450. if room_viewer_state.room is None:
  451. return
  452. self._room_kinds[message.room_id] = room_viewer_state.room.kind or RoomKind.UNSPECIFIED
  453. room_kind = self._room_kinds.get(message.room_id)
  454. logger.info("message_body: %s room_kind: %s", message_body, room_kind)
  455. mentioned = False
  456. if (room_kind == RoomKind.CHANNEL and self.chatto_config.require_mention.value and self.me):
  457. if self.me.login and not mentioned:
  458. mentioned = bool(f"@{self.me.login}" in message_body)
  459. if self.me.display_name and not mentioned:
  460. mentioned = bool(f"@{self.me.display_name}" in message_body)
  461. if mentioned is False:
  462. logger.debug("Discarding message. Bot was not mentionend but require_mention is '%s'.", self.chatto_config.require_mention.value)
  463. return
  464. logger.info("mentioned: %s", mentioned)
  465. # Thread anchoring — if the incoming message is inside a thread, we
  466. # keep that thread by default; otherwise leave thread_id unset so
  467. # replies land at the root.
  468. thread_id = payload.thread_root_event_id or None
  469. if not thread_id and room_kind != RoomKind.DM and self.chatto_config.auto_thread.value:
  470. thread_id = message.id
  471. source = self.build_source(
  472. chat_id=payload.room_id,
  473. chat_name=self._room_names.get(message.room_id),
  474. chat_type="dm" if room_kind == RoomKind.DM else room_kind or RoomKind.UNSPECIFIED, # Only "dm" seems to be a reserved keyword from Base adapter class.,
  475. user_id=message.actor_id,
  476. user_name=user.login, # use login, because display_name is changeable by anyone.
  477. thread_id=thread_id,
  478. message_id=payload.message_event_id,
  479. role_authorized=True,
  480. )
  481. msg_type = MessageType.COMMAND if (message_body.lstrip().startswith("/")) else MessageType.TEXT
  482. my_body = message_body.lstrip() if msg_type == MessageType.COMMAND else message_body
  483. # Attachments — download and hand the local cache paths to the gateway,
  484. # which runs vision enrichment / document extraction off media_urls.
  485. media_urls, media_types, media_kinds = await self._cache_attachments(
  486. payload.room_id, attachments,
  487. )
  488. if media_kinds and msg_type != MessageType.COMMAND:
  489. # Same precedence as the Teams/Signal adapters: document-context
  490. # injection gates strictly on DOCUMENT, image handling keys off the
  491. # per-path image/* MIME regardless of message_type.
  492. msg_type = self._message_type_for_media_kinds(media_kinds)
  493. message_event = MessageEvent(
  494. text=my_body,
  495. source=source,
  496. message_id=message.id,
  497. message_type=msg_type,
  498. media_urls=media_urls,
  499. media_types=media_types,
  500. timestamp=message.created_at or datetime.now(timezone.utc),
  501. raw_message=message,
  502. reply_to_message_id=message.id,
  503. reply_to_text=message_body,
  504. reply_to_author_id=user.id,
  505. reply_to_author_name=user.login,
  506. )
  507. logger.info("Chatto: Dispatching MessageEvent to Hermes: %s", message_event)
  508. await self.handle_message(message_event)
  509. return
  510. async def _handle_realtime_event(self, event: RealtimeEvent) -> None:
  511. if self._is_seen(event.id):
  512. return
  513. if event.actor_id is None:
  514. return
  515. logger.info("EVENT happened: '%s' from %s", event.kind, event.actor_id)
  516. if (event_payload := event.get("message_posted")) is not None:
  517. # Self-event filter — the actor_id on the envelope is authoritative
  518. # (chattolib does NOT filter this itself; see chatto-bridge notes).
  519. actor_id = event.actor_id
  520. if actor_id and self.me and actor_id == self.me.id:
  521. return
  522. await self._dispatch_message_posted(event_payload)
  523. # confirmed:
  524. elif event.kind in ("mention_notification", "projection_event", "presence_changed", "notification_dismissed",
  525. "room_marked_as_read", "user_typing", "notification_created", "new_direct_message_notification",
  526. 'reaction_removed', 'reaction_added'):
  527. logger.debug("Chatto: '%s' event received. Not yet implemented or not needed.", event.kind)
  528. else:
  529. logger.error("Chatto: unknown event kind: '%s'", event.kind)
  530. async def _chattolib_event_loop(self) -> None:
  531. """Event loop using chattolib's stream_events.
  532. This replaces the manual WebSocket loop with chattolib's high-level
  533. stream_events() which provides pre-decoded RealtimeEvent objects.
  534. """
  535. delay = ChattoConstants.WS_RECONNECT_INITIAL_BACKOFF
  536. while not self._closing:
  537. try:
  538. client = await self._require_client()
  539. await self._refresh_rooms()
  540. logger.info("Chatto: starting chattolib event stream with %d rooms", len(self._watch_room_ids))
  541. async for event in stream_events(client):
  542. if self._closing:
  543. return
  544. await self._handle_realtime_event(event)
  545. # Iterator exited cleanly — treat as a normal close and reconnect
  546. # with the local backoff (no server hint available).
  547. logger.info("Chatto: realtime stream ended, reconnecting")
  548. except asyncio.CancelledError:
  549. return
  550. except ChattoRealtimeCloseError as exc:
  551. if not exc.reconnect:
  552. logger.error(
  553. "Chatto: realtime closed by server (%s: %s), not reconnecting",
  554. exc.code, exc.message,
  555. )
  556. return
  557. wait = max(exc.retry_after_ms / 1000.0, ChattoConstants.WS_RECONNECT_INITIAL_BACKOFF)
  558. logger.warning(
  559. "Chatto: realtime closed by server (%s), reconnecting in %.1fs",
  560. exc.code, wait,
  561. )
  562. delay = ChattoConstants.WS_RECONNECT_INITIAL_BACKOFF # server hint supersedes local backoff
  563. await self._sleep_interruptible(wait)
  564. continue
  565. except ChattoRealtimeError as exc:
  566. if getattr(exc, "fatal", False):
  567. logger.error("Chatto: fatal realtime error (%s): %s", exc.code, exc.message)
  568. return
  569. logger.warning(
  570. "Chatto: realtime error (%s: %s), reconnecting in %.1fs",
  571. exc.code, exc.message, delay,
  572. )
  573. except Exception as exc:
  574. logger.warning(
  575. "Chatto: unexpected realtime error: %s, reconnecting in %.1fs",
  576. exc, delay,
  577. )
  578. if self._closing:
  579. return
  580. jitter = delay * 0.2 * random.random()
  581. await self._sleep_interruptible(delay + jitter)
  582. delay = min(delay * 2, ChattoConstants.WS_RECONNECT_MAX_BACKOFF)
  583. async def _sleep_interruptible(self, seconds: float) -> None:
  584. """Sleep in short slices so disconnect() cancels promptly."""
  585. end = asyncio.get_running_loop().time() + seconds
  586. while not self._closing:
  587. remaining = end - asyncio.get_running_loop().time()
  588. if remaining <= 0:
  589. return
  590. await asyncio.sleep(min(remaining, 0.5))
  591. async def _refresh_rooms(self) -> None:
  592. """Refresh room list via ConnectRPC, join and seed any newly discovered rooms."""
  593. try:
  594. client = await self._require_client()
  595. except RuntimeError:
  596. logger.warning("Chatto WS: _refresh_rooms aborted - no client available")
  597. return
  598. try:
  599. rooms_list = await client.list_rooms()
  600. new_room_ids: List[str] = []
  601. for room_with_state in rooms_list:
  602. if not room_with_state:
  603. continue
  604. room_obj = room_with_state.room or None
  605. if not room_obj:
  606. continue
  607. self._room_names[room_obj.id] = room_obj.name
  608. self._room_kinds[room_obj.id] = room_obj.kind
  609. if room_with_state.viewer_state.is_member and room_obj.id not in self._watch_room_ids:
  610. new_room_ids.append(room_obj.id)
  611. if not new_room_ids:
  612. return
  613. logger.info("Chatto WS: discovered %d new room(s): %s", len(new_room_ids), new_room_ids)
  614. for rid in new_room_ids:
  615. if self._room_kinds.get(rid) != RoomKind.DM:
  616. await client.join_room(rid) # but list_rooms() would not return any if we were not already joined?!
  617. await self._seed_room(rid)
  618. self._watch_room_ids.append(rid)
  619. watch_room_names: list[str] = []
  620. for rid in self._watch_room_ids:
  621. watch_room_names.append(self._room_names[rid] + " (" + rid + ")")
  622. logger.info("Chatto WS: Watching %d room(s): %s", len(self._watch_room_ids), ", ".join(watch_room_names))
  623. except Exception:
  624. logger.warning("Chatto WS: _refresh_rooms failed", exc_info=True)
  625. # ------------------------------------------------------------------ #
  626. # Read state & notification dismissal (best-effort, Chatto-unique)
  627. # ------------------------------------------------------------------ #
  628. # Best-effort: mark all watched rooms as read (room_id may be undefined here)
  629. for _rid in list(self._watch_room_ids):
  630. try:
  631. await client.mark_room_as_read(room_id=_rid)
  632. await client.dismiss_all_notifications()
  633. except Exception:
  634. logger.debug("Chatto: mark_room_as_read failed for %s", _rid, exc_info=True)
  635. # ------------------------------------------------------------------ #
  636. # Sending (ConnectRPC — unchanged from polling version)
  637. # ------------------------------------------------------------------ #
  638. async def send(
  639. self,
  640. chat_id: str,
  641. content: str,
  642. reply_to: Optional[str] = None,
  643. metadata: Optional[Dict[str, Any]] = None,
  644. ) -> SendResult:
  645. """Send a message to a Chatto room.
  646. Long messages are split into chunks via ``truncate_message`` and
  647. each chunk is sent as a separate CreateMessage call. The first
  648. chunk's message ID is returned as ``message_id``.
  649. When ``auto_thread`` is enabled and the incoming message was a
  650. regular room message (not already in a thread), the first chunk is
  651. sent as a room message and its ID becomes the thread root. Subsequent
  652. chunks are sent in that thread. This mirrors Discord's auto_thread
  653. behavior.
  654. BasePlatformAdapter override
  655. """
  656. if not content:
  657. return SendResult(success=False, error="Empty message")
  658. formatted = self.format_message(content) if hasattr(self, "format_message") else content
  659. chunks = self.truncate_message(formatted, ChattoConstants.MAX_MESSAGE_LENGTH)
  660. # Thread support — resolve thread_id once
  661. # DM rooms don't support threads, so skip threading for DMs
  662. thread_id = (metadata or {}).get("thread_id")
  663. # Only use reply_to as thread_id if auto_thread is enabled.
  664. # When auto_thread=false, responses go directly in the room
  665. # without threading under the incoming message.
  666. if reply_to and self.chatto_config.auto_thread.value:
  667. # reply_to might be the incoming message ID. If we already have
  668. # thread_id from metadata, keep it (it's the thread root).
  669. # Only use reply_to as thread_id if we don't already have one.
  670. if not thread_id:
  671. thread_id = reply_to
  672. # Check if this is a DM room — DMs don't support threads
  673. room_kind = self._room_kinds.get(chat_id)
  674. is_dm = room_kind == RoomKind.DM
  675. if is_dm:
  676. thread_id = None
  677. # Auto-thread: by default, Chatto creates a thread for replies to room
  678. # messages (not DMs, not already in a thread). This keeps conversations
  679. # organized in the room. Can be disabled via extra.auto_thread=false.
  680. use_auto_thread = self.chatto_config.auto_thread.value and not thread_id and not is_dm
  681. message_ids: List[str] = []
  682. last_resp: Optional[Any] = None
  683. last_error: Optional[str] = None
  684. retryable = False
  685. try:
  686. client = await self._require_client()
  687. except RuntimeError:
  688. return SendResult(success=False, error="Chatto client not available", retryable=True)
  689. for i, chunk in enumerate(chunks):
  690. try:
  691. msg_obj = await client.post_message(
  692. room_id=chat_id,
  693. body=chunk,
  694. thread_root_event_id=str(thread_id) if thread_id else "",
  695. )
  696. except ChattoError as e:
  697. last_error = str(e)
  698. retryable = True
  699. break
  700. except Exception as e:
  701. last_error = str(e)
  702. retryable = True
  703. break
  704. last_resp = msg_obj
  705. self._mark_seen(msg_obj.id)
  706. message_ids.append(msg_obj.id)
  707. self._our_message_ids.add(msg_obj.id)
  708. # If we sent a message WITHOUT a thread_id, this message could
  709. # become a thread root if someone replies to it
  710. if not thread_id:
  711. self._our_thread_roots.add(msg_obj.id)
  712. # Auto-thread: first chunk becomes the thread root,
  713. # subsequent chunks go in the thread
  714. if use_auto_thread and i == 0 and not thread_id:
  715. thread_id = msg_obj.id
  716. # Nothing got through at all — report the failure instead of a phantom success.
  717. if not message_ids:
  718. return SendResult(
  719. success=False,
  720. error=last_error or "Chatto: message could not be sent",
  721. retryable=retryable,
  722. )
  723. first_id = message_ids[0]
  724. # ------------------------------------------------------------------ #
  725. # Thread following (best-effort, Chatto-unique)
  726. # ------------------------------------------------------------------ #
  727. if thread_id:
  728. try:
  729. await client.follow_thread(chat_id, thread_id)
  730. except Exception:
  731. logger.debug("Chatto: follow_thread failed for %s/%s", chat_id, thread_id, exc_info=True)
  732. # A later chunk failed after earlier ones went out: partial delivery.
  733. if last_error:
  734. logger.warning(
  735. "Chatto: sent %d/%d chunk(s) to %s before failing: %s",
  736. len(message_ids), len(chunks), chat_id, last_error,
  737. )
  738. return SendResult(success=True, message_id=first_id, raw_response=last_resp)
  739. # Overridden from BaseAdapter:
  740. async def send_typing(self, chat_id: str, metadata=None) -> None:
  741. """Start a persistent typing indicator for a room.
  742. Sends a typing ping every 10 seconds (Chatto's indicator likely
  743. lasts ~8-10s). The background loop runs until ``stop_typing()``
  744. is called or the task is cancelled.
  745. BasePlatformAdapter override
  746. """
  747. if chat_id in self._typing_tasks:
  748. return # already running
  749. async def _typing_loop() -> None:
  750. try:
  751. while True:
  752. try:
  753. try:
  754. client = await self._require_client()
  755. except RuntimeError:
  756. return
  757. await client.update_typing_indicator(room_id=str(chat_id))
  758. except asyncio.CancelledError:
  759. return
  760. except Exception:
  761. pass
  762. await asyncio.sleep(10)
  763. except asyncio.CancelledError:
  764. pass
  765. finally:
  766. self._typing_tasks.pop(chat_id, None)
  767. self._typing_tasks[chat_id] = asyncio.create_task(_typing_loop())
  768. async def stop_typing(self, chat_id: str) -> None:
  769. """Stop the persistent typing indicator for a room.
  770. BasePlatformAdapter override
  771. """
  772. task = self._typing_tasks.pop(chat_id, None)
  773. if task:
  774. task.cancel()
  775. try:
  776. await task
  777. except (asyncio.CancelledError, Exception):
  778. pass
  779. async def get_chat_info(self, chat_id: str) -> Dict[str, Any]:
  780. """Get information about a chat/room.
  781. BasePlatformAdapter override
  782. """
  783. name = self._room_names.get(chat_id, chat_id)
  784. kind = self._room_kinds.get(chat_id)
  785. chat_type = "dm" if kind == RoomKind.DM else "group"
  786. return {
  787. "name": name,
  788. "type": chat_type,
  789. }
  790. # ------------------------------------------------------------------ #
  791. # Reactions
  792. # ------------------------------------------------------------------ #
  793. @staticmethod
  794. def _emoji_to_shortcode(emoji: str) -> str:
  795. """Convert a unicode emoji to a Chatto shortcode name.
  796. If the emoji is already a shortcode (no unicode mapping found),
  797. return it as-is.
  798. """
  799. shortcode = ChattoConstants.EMOJI_TO_SHORTCODE.get(emoji)
  800. if shortcode:
  801. return shortcode
  802. # Already a shortcode like "thumbsup" — return as-is
  803. return emoji
  804. async def add_reaction(self, room_id: str, message_id: str, emoji: str) -> bool:
  805. """Add a reaction to a message via MessageService/AddReaction."""
  806. shortcode = self._emoji_to_shortcode(emoji)
  807. try:
  808. try:
  809. client = await self._require_client()
  810. except RuntimeError:
  811. logger.warning("Chatto: AddReaction — client unavailable")
  812. return False
  813. result = await client.add_reaction(
  814. room_id=room_id,
  815. message_event_id=message_id,
  816. emoji=shortcode,
  817. )
  818. return result
  819. except ChattoError as e:
  820. logger.warning("Chatto: AddReaction failed: %s", e)
  821. return False
  822. except Exception as e:
  823. logger.warning("Chatto: AddReaction error: %s", e)
  824. return False
  825. async def remove_reaction(self, chat_id: str, message_id: str, emoji: str) -> bool:
  826. """Remove a reaction from a message via MessageService/RemoveReaction."""
  827. shortcode = self._emoji_to_shortcode(emoji)
  828. try:
  829. try:
  830. client = await self._require_client()
  831. except RuntimeError:
  832. logger.warning("Chatto: RemoveReaction — client unavailable")
  833. return False
  834. result = await client.remove_reaction(
  835. room_id=str(chat_id),
  836. message_event_id=str(message_id),
  837. emoji=shortcode,
  838. )
  839. return result
  840. except ChattoError as e:
  841. logger.warning("Chatto: RemoveReaction failed: %s", e)
  842. return False
  843. except Exception as e:
  844. logger.warning("Chatto: RemoveReaction error: %s", e)
  845. return False
  846. # ------------------------------------------------------------------ #
  847. # DM initiation (Chatto-unique)
  848. # ------------------------------------------------------------------ #
  849. async def start_dm(self, user_id: str) -> Optional[str]:
  850. """Start a direct message with a user via RoomService/StartDM.
  851. Returns the room ID on success, or None on failure.
  852. """
  853. if not user_id:
  854. return None
  855. try:
  856. try:
  857. client = await self._require_client()
  858. except RuntimeError:
  859. return None
  860. room = await client.start_dm(participant_ids=[str(user_id)])
  861. self._room_names[room.id] = room.name
  862. self._room_kinds[room.id] = room.kind
  863. return room.id
  864. except ChattoError as e:
  865. logger.debug("Chatto: StartDM failed: %s", e)
  866. return None
  867. except Exception as e:
  868. logger.debug("Chatto: StartDM error: %s", e)
  869. return None
  870. # ------------------------------------------------------------------ #
  871. # Room creation (Chatto-unique)
  872. # ------------------------------------------------------------------ #
  873. async def create_room(
  874. self,
  875. name: str,
  876. description: str = "",
  877. group_id: str = "",
  878. universal: bool = True,
  879. ) -> Optional[str]:
  880. """Create an ad-hoc room via RoomService/CreateRoom.
  881. Returns the room ID on success, or None on failure.
  882. """
  883. try:
  884. try:
  885. client = await self._require_client()
  886. except RuntimeError:
  887. return None
  888. room = await client.create_room(
  889. name=name,
  890. group_id=group_id or "",
  891. description=description,
  892. universal=universal,
  893. )
  894. rid = str(room.id) if room else ""
  895. if rid:
  896. self._room_names[rid] = room.name
  897. self._room_kinds[rid] = room.kind
  898. return rid
  899. logger.debug("Chatto: CreateRoom returned no room id")
  900. return None
  901. except ChattoError as e:
  902. logger.debug("Chatto: CreateRoom failed: %s", e)
  903. return None
  904. except Exception as e:
  905. logger.debug("Chatto: CreateRoom error: %s", e)
  906. return None
  907. # ------------------------------------------------------------------ #
  908. # Processing lifecycle hooks (reactions-based, like Discord)
  909. # ------------------------------------------------------------------ #
  910. def _event_room_and_message_id(self, event: MessageEvent) -> Tuple[str, str]:
  911. """Extract room_id and message_id from a MessageEvent."""
  912. message_id = event.message_id or ""
  913. chat_id = event.source.chat_id
  914. return chat_id, message_id
  915. async def on_processing_start(self, event: MessageEvent) -> None:
  916. """Add an 👀 (eyes) reaction to the incoming message.
  917. BasePlatformAdapter override
  918. """
  919. logger.info("self.chatto_config.reactions.value: %s", self.chatto_config.reactions.value)
  920. if not self.chatto_config.reactions.value:
  921. return
  922. chat_id, message_id = self._event_room_and_message_id(event)
  923. if not chat_id or not message_id:
  924. logger.warning("Chatto: on_processing_start — empty chat_id or message_id, skipping reaction")
  925. return
  926. await self.add_reaction(chat_id, message_id, "👀")
  927. async def on_processing_complete(
  928. self, event: MessageEvent, outcome: ProcessingOutcome
  929. ) -> None:
  930. """Swap the 👀 reaction for ✅ (success) or ❌ (failure).
  931. BasePlatformAdapter override
  932. """
  933. if not self.chatto_config.reactions.value:
  934. return
  935. chat_id, message_id = self._event_room_and_message_id(event)
  936. if not chat_id or not message_id:
  937. return
  938. # Remove the processing eyes reaction
  939. await self.remove_reaction(chat_id, message_id, "👀")
  940. # Add the outcome reaction
  941. if outcome == ProcessingOutcome.SUCCESS:
  942. await self.add_reaction(chat_id, message_id, "✅")
  943. elif outcome == ProcessingOutcome.FAILURE:
  944. await self.add_reaction(chat_id, message_id, "❌")
  945. # ------------------------------------------------------------------ #
  946. # Asset upload (chunked)
  947. # ------------------------------------------------------------------ #
  948. async def _upload_asset(self, room_id: str, file_path: str) -> Optional[str]:
  949. """Upload a file via the chunked AssetUploadService.
  950. Returns the asset ID on success, or None on failure.
  951. """
  952. try:
  953. with open(file_path, "rb") as f:
  954. file_data = f.read()
  955. except Exception as e:
  956. logger.error("Chatto: failed to read file %s — %s", file_path, e)
  957. return None
  958. if not file_data:
  959. logger.error("Chatto: file %s is empty", file_path)
  960. return None
  961. file_size = len(file_data)
  962. file_name = os.path.basename(file_path)
  963. mime_type = mimetypes.guess_type(file_path)[0] or "application/octet-stream"
  964. sha256_hash = hashlib.sha256(file_data).hexdigest()
  965. try:
  966. try:
  967. client = await self._require_client()
  968. except RuntimeError:
  969. logger.error("Chatto: upload aborted - no client available")
  970. return None
  971. # Step 1: Create upload session
  972. upload = await client.create_upload(
  973. room_id=room_id,
  974. filename=file_name,
  975. size=file_size,
  976. sha256=sha256_hash,
  977. content_type=mime_type,
  978. )
  979. upload_id = str(getattr(cast(Any, upload), "id", ""))
  980. if not upload_id:
  981. logger.error("Chatto: CreateUpload returned no upload ID")
  982. return None
  983. # Step 2: Upload chunks
  984. offset = 0
  985. while offset < file_size:
  986. chunk = file_data[offset:offset + ChattoConstants.UPLOAD_CHUNK_SIZE]
  987. chunk_sha256 = hashlib.sha256(chunk).hexdigest()
  988. await client.upload_chunk(
  989. upload_id=upload_id,
  990. offset=offset,
  991. content=chunk,
  992. chunk_sha256=chunk_sha256,
  993. )
  994. offset += len(chunk)
  995. # Step 3: Complete upload
  996. upload, asset = await client.complete_upload(upload_id=upload_id)
  997. if not asset:
  998. logger.error("Chatto: CompleteUpload returned no asset")
  999. return None
  1000. asset_id = str(getattr(cast(Any, asset), "id", ""))
  1001. logger.info("Chatto: uploaded %s as asset %s (%d bytes)", file_name, asset_id, file_size)
  1002. return asset_id
  1003. except ChattoError as e:
  1004. logger.error("Chatto: upload failed: %s", e)
  1005. return None
  1006. except Exception as e:
  1007. logger.error("Chatto: upload error: %s", e)
  1008. return None
  1009. async def send_image_file(
  1010. self,
  1011. chat_id: str,
  1012. file_path: str,
  1013. caption: Optional[str] = None,
  1014. reply_to: Optional[str] = None,
  1015. metadata: Optional[Dict[str, Any]] = None,
  1016. ) -> SendResult:
  1017. """Send a local image file via the chunked upload API. Do not change signature.
  1018. BasePlatformAdapter override
  1019. """
  1020. # Validate the path is safe
  1021. safe_path = self.validate_media_delivery_path(file_path)
  1022. if not safe_path:
  1023. logger.warning("Chatto: send_image_file — unsafe path %s", file_path)
  1024. text = "⚠️ Couldn't deliver the image attachment."
  1025. if caption:
  1026. text = f"{caption}\n{text}"
  1027. return await self.send(chat_id, text, reply_to=reply_to, metadata=metadata)
  1028. asset_id = await self._upload_asset(str(chat_id), safe_path)
  1029. if not asset_id:
  1030. # Fallback to a notice
  1031. text = "⚠️ Couldn't deliver the image attachment."
  1032. if caption:
  1033. text = f"{caption}\n{text}"
  1034. return await self.send(chat_id, text, reply_to=reply_to, metadata=metadata)
  1035. thread_id = (metadata or {}).get("thread_id")
  1036. if reply_to:
  1037. thread_id = reply_to
  1038. try:
  1039. try:
  1040. client = await self._require_client()
  1041. except RuntimeError:
  1042. return SendResult(success=False, error="Chatto client not available", retryable=True)
  1043. msg = await client.post_message(
  1044. room_id=str(chat_id),
  1045. body=caption or "",
  1046. attachment_asset_ids=[asset_id],
  1047. thread_root_event_id=str(thread_id) if thread_id else "",
  1048. )
  1049. self._mark_seen(msg.id)
  1050. return SendResult(success=True, message_id=msg.id, raw_response=msg)
  1051. except ChattoError as e:
  1052. return SendResult(success=False, error=str(e), retryable=True)
  1053. except Exception as e:
  1054. return SendResult(success=False, error=str(e), retryable=False)
  1055. async def send_image(
  1056. self,
  1057. chat_id: str,
  1058. image_url: str,
  1059. caption: Optional[str] = None,
  1060. reply_to: Optional[str] = None,
  1061. metadata: Optional[Dict[str, Any]] = None,
  1062. ) -> SendResult:
  1063. """Send an image to a Chatto room.
  1064. Tries to download the image from the URL and upload it as a native
  1065. attachment. Falls back to sending the URL as a link (Chatto renders
  1066. link previews) if the download fails.
  1067. BasePlatformAdapter override
  1068. """
  1069. # Try downloading and uploading as attachment
  1070. try:
  1071. import tempfile
  1072. import urllib.request as _urllib_request
  1073. # Download to a temp file
  1074. parsed = urlsplit(image_url)
  1075. url_path = parsed.path
  1076. ext = os.path.splitext(url_path)[1] or ".png"
  1077. tmp_fd, tmp_path = tempfile.mkstemp(suffix=ext, prefix="chatto_img_")
  1078. try:
  1079. os.close(tmp_fd)
  1080. req = _urllib_request.Request(image_url, headers={"User-Agent": "Hermes/1.0"})
  1081. try:
  1082. import ssl
  1083. ctx = ssl.create_default_context()
  1084. except Exception:
  1085. ctx = None
  1086. with _urllib_request.urlopen(req, timeout=ChattoConstants.HTTP_TIMEOUT, context=ctx) as resp:
  1087. with open(tmp_path, "wb") as f:
  1088. f.write(resp.read())
  1089. # Upload as attachment
  1090. result = await self.send_image_file(
  1091. chat_id, tmp_path, caption=caption,
  1092. reply_to=reply_to, metadata=metadata,
  1093. )
  1094. if result.success:
  1095. return result
  1096. finally:
  1097. try:
  1098. os.unlink(tmp_path)
  1099. except OSError:
  1100. pass
  1101. except Exception as e:
  1102. logger.debug("Chatto: send_image download/upload failed, falling back to link: %s", e)
  1103. # Fallback: send as link (Chatto renders link previews)
  1104. text = image_url
  1105. if caption:
  1106. text = f"{caption}\n{image_url}"
  1107. return await self.send(chat_id, text, reply_to=reply_to, metadata=metadata)
  1108. # ---------------------------------------------------------------------------
  1109. # Cron / out-of-process delivery
  1110. # ---------------------------------------------------------------------------
  1111. async def hermes_standalone_sender_fn(
  1112. pconfig,
  1113. chat_id,
  1114. message,
  1115. *,
  1116. thread_id=None,
  1117. media_files=None,
  1118. force_document=False,
  1119. ) -> SendResult:
  1120. """Deliver a message to Chatto without a running gateway adapter. Do not modify signature.
  1121. Used by cron / scheduled routines that run out-of-process. Creates a
  1122. short-lived chattolib client, posts, and closes.
  1123. """
  1124. chatto_config: ChattoConfiguration = ChattoConfiguration(pconfig=pconfig)
  1125. # Create a temporary client for standalone sending — we need a base URL plus
  1126. # either a token or a full login/password pair.
  1127. has_credentials = bool(
  1128. chatto_config.token.value
  1129. or (chatto_config.login.value and chatto_config.password.value)
  1130. )
  1131. if not chatto_config.base_url.value or not has_credentials:
  1132. return SendResult(success=False, error="Chatto: base URL or credentials missing")
  1133. client: ChattoClient
  1134. try:
  1135. if chatto_config.token.value:
  1136. client = ChattoClient(base_url=chatto_config.base_url.value, token=chatto_config.token.value)
  1137. else:
  1138. client = await ChattoClient.login(
  1139. base_url=chatto_config.base_url.value,
  1140. login=chatto_config.login.value,
  1141. password=chatto_config.password.value,
  1142. )
  1143. except Exception as exc:
  1144. return SendResult(success=False, error=f"Chatto login failed: {exc}")
  1145. try:
  1146. kwargs: Dict[str, Any] = {}
  1147. if chatto_config.auto_thread.value and thread_id:
  1148. kwargs["in_reply_to"] = thread_id
  1149. if media_files and media_files.get("attachment_asset_ids"):
  1150. kwargs["attachment_asset_ids"] = list(media_files["attachment_asset_ids"])
  1151. try:
  1152. posted = await client.post_message(chat_id, message, **kwargs)
  1153. except Exception as exc:
  1154. return SendResult(success=False, error=str(exc))
  1155. return SendResult(success=True, message_id=getattr(posted, "id", "") or None)
  1156. finally:
  1157. try:
  1158. await client.close()
  1159. except Exception as exc:
  1160. logger.error(
  1161. "Chatto standalone: error closing short-lived client (perhaps already closed): %s", exc,
  1162. )
  1163. def hermes_validate_config(config: PlatformConfig) -> bool:
  1164. """"
  1165. Function name should be the same as register argument name with "hermes_" prefix, so we
  1166. know that it is needed for plugin register(). Do not change signature.
  1167. - config
  1168. Check whether Chatto Plugin is configured. Compare to hermes_is_connected()."""
  1169. chatto_config = ChattoConfiguration(pconfig=config)
  1170. if len(chatto_config.allowed_users.value) > 0 and chatto_config.allow_all_users.value:
  1171. logger.info("Chatto: Conflicting configuration. Either use 'allowed_users' or 'allow_all_users' but not both.")
  1172. return False
  1173. if chatto_config.base_url.value:
  1174. if (chatto_config.token.value is not None) or (chatto_config.login.value and chatto_config.password.value):
  1175. return True
  1176. else:
  1177. logger.error("Chatto: Minimally, either token or login/password must be set.")
  1178. else:
  1179. logger.error("Chatto: base_url must be set.")
  1180. return False
  1181. def hermes_check_fn() -> bool:
  1182. """Check if Chatto is configured and dependencies are available.
  1183. Add real logic?! Or just .. there are no dependencies.. always return true. Really. Docs suck."""
  1184. try:
  1185. import chattolib.client # noqa: F401 — vendored dependency probe
  1186. return True
  1187. except ImportError:
  1188. return False
  1189. # ---------------------------------------------------------------------------
  1190. # is_connected probe
  1191. # ---------------------------------------------------------------------------
  1192. def hermes_is_connected(config: PlatformConfig) -> bool:
  1193. """Check whether Chatto Plugin is connected. But where to? To the Hermes Agent? To Chatto Server?
  1194. The Hermes Agent plugin docs suck and it seems there are many functions to do the same."""
  1195. return bool(hermes_validate_config(config) and config.enabled)
  1196. def hermes_setup_fn() -> None:
  1197. """Interactive setup wizard for Chatto. Is called by and only works in Hermes CLI context.
  1198. Function name should be the same as register argument name with "hermes_" prefix, so we
  1199. know that it is needed for plugin register().
  1200. """
  1201. from hermes_cli.setup import (
  1202. prompt,
  1203. prompt_yes_no,
  1204. save_env_value,
  1205. get_env_value,
  1206. print_header,
  1207. print_info,
  1208. print_warning,
  1209. print_success,
  1210. )
  1211. url = prompt(
  1212. "Chatto server URL (e.g. https://chat.example.com) or leave blank for default ChattoHQ on chat.chatto.run:")
  1213. if url:
  1214. save_env_value(ChattoConfiguration.base_url.env_name, url)
  1215. login = prompt("Chatto login (username):")
  1216. if login:
  1217. save_env_value(ChattoConfiguration.login.env_name, login)
  1218. password = prompt("Chatto password:", password=True)
  1219. if password:
  1220. save_env_value(ChattoConfiguration.password.env_name, password)
  1221. channels = prompt("Room IDs to watch (comma-separated, or empty for all):")
  1222. if channels:
  1223. save_env_value(ChattoConfiguration.channels_list.env_name, channels)
  1224. home = prompt("Home room ID for notifications (or empty):")
  1225. if home:
  1226. save_env_value(ChattoConfiguration.home_channel.env_name, home)
  1227. allow_all = prompt_yes_no("Allow all users to talk? (true/false):")
  1228. if allow_all:
  1229. save_env_value(ChattoConfiguration.allow_all_users.env_name, str(allow_all))
  1230. print_success("\n✓ Chatto configured. Restart the gateway to activate.")
  1231. def hermes_env_enablement_fn() -> Optional[dict]:
  1232. """Seed PlatformConfig.extra from env vars.
  1233. Returns a dict compatible with the PlatformConfig merge hook (or None
  1234. when no env-provided values are present).
  1235. Called by the platform registry during load_gateway_config().
  1236. Return None when the platform isn't minimally configured — the
  1237. caller then skips auto-enabling. Return a dict to seed extras.
  1238. The special 'home_channel' key is extracted and becomes a proper
  1239. HomeChannel dataclass on the PlatformConfig; every other key is
  1240. merged into PlatformConfig.extra.
  1241. Function name should be the same as register argument name with "hermes_" prefix, so we
  1242. know that it is needed for plugin register().
  1243. """
  1244. # Seed keys must be the config.yaml "extra" keys (config_key), NOT the env
  1245. # var names — ChattoConfiguration reads extra[config_key].
  1246. seed: Dict[str, Any] = {
  1247. ChattoConfiguration.base_url.config_key: (
  1248. os.getenv(ChattoConfiguration.base_url.env_name) or ChattoClient.DEFAULT_BASE_URL
  1249. ).strip(),
  1250. }
  1251. for field in ChattoConfiguration.fields():
  1252. if field.field_name == ChattoConfiguration.base_url.field_name:
  1253. continue
  1254. env_value = os.getenv(field.env_name)
  1255. if env_value:
  1256. seed[field.config_key] = env_value.strip()
  1257. logger.debug("seed: %s", {k: v for k, v in seed.items() if k not in ("token", "password")})
  1258. return seed
  1259. # ---------------------------------------------------------------------------
  1260. # Plugin registration entry point
  1261. # ---------------------------------------------------------------------------
  1262. def register(ctx) -> None:
  1263. """Plugin entry point — called by the Hermes plugin system."""
  1264. logger.info("Registering Chatto platform plugin on Hermes Agent")
  1265. logger.info("ChattoConfiguration.allowed_users.env_name: %s", ChattoConfiguration.allowed_users.env_name)
  1266. ctx.register_platform(
  1267. name=ChattoConstants.PLATFORM_NAME, # this will be the config.yaml key.
  1268. label=ChattoConstants.PLATFORM_LABEL,
  1269. adapter_factory=hermes_adapter_factory,
  1270. check_fn=hermes_check_fn,
  1271. validate_config=hermes_validate_config,
  1272. is_connected=hermes_is_connected,
  1273. install_hint=ChattoConstants.INSTALL_HINT,
  1274. env_enablement_fn=hermes_env_enablement_fn,
  1275. setup_fn=hermes_setup_fn,
  1276. cron_deliver_env_var=ChattoConfiguration.home_channel.env_name,
  1277. standalone_sender_fn=hermes_standalone_sender_fn,
  1278. allowed_users_env=ChattoConfiguration.allowed_users.env_name,
  1279. allow_all_env=ChattoConfiguration.allow_all_users.env_name,
  1280. max_message_length=ChattoConstants.MAX_MESSAGE_LENGTH,
  1281. emoji="💬",
  1282. allow_update_command=True,
  1283. pii_safe=False,
  1284. platform_hint=(
  1285. "You are chatting in Chatto (a self-hosted or cloud-hosted team or community chat server). "
  1286. "Markdown is supported. Users _may_ address you by @-mentioning your name. If configured, " \
  1287. "you also react without a @-mention. Direct messages reach you without a mention."
  1288. "Keep responses conversational."
  1289. ),
  1290. )