1"""Native Tinyrelay adapter for Hermes Agent.23Tiny is a Nostr relay connection. The adapter intentionally uses structured events for4interactive prompts; no Buzz compatibility or prose parsing is involved.5"""67from __future__ import annotations89import asyncio10from contextvars import ContextVar11import json12import mimetypes13import hashlib14import os15import re16import time17from datetime import datetime18from pathlib import Path19from typing import Any2021from gateway.config import Platform, PlatformConfig22from gateway.platforms.base import BasePlatformAdapter, SendResult23from gateway.platforms.event import MessageEvent, MessageType2425try:26 from .client import TinyRPC27 from .interactions import InteractionMixin28 from .media import incoming_media29 from .ingress import EventTracker, event_thread, reply_parent30except ImportError: # Hermes plugin loader may import adapter.py as a top-level module.31 from client import TinyRPC32 from interactions import InteractionMixin33 from media import incoming_media34 from ingress import EventTracker, event_thread, reply_parent3536_HEX = re.compile(r"\b[0-9a-f]{64}\b", re.I)37_MENTION = re.compile(r"(?<![0-9A-Za-z])@([0-9a-f]{64})(?![0-9A-Za-z])", re.I)383940def _tags(event: dict, name: str) -> list[list[str]]:41 return [tag for tag in event.get("tags", []) if len(tag) >= 2 and tag[0] == name]424344def _tag(event: dict, name: str) -> str | None:45 found = _tags(event, name)46 return found[0][1] if found else None474849def _explicit_mentions(text: str) -> list[str]:50 """Return only public keys explicitly written as @mentions in text."""51 return list(dict.fromkeys(match.group(1).lower() for match in _MENTION.finditer(text or "")))525354def _mention_tags(text: str) -> list[list[str]]:55 return [["p", pubkey] for pubkey in _explicit_mentions(text)]565758class TinyAdapter(InteractionMixin, BasePlatformAdapter):59 """Native room messages and interactions for Hermes."""6061 supports_code_blocks = True62 supports_async_delivery = True63 MAX_MESSAGE_LENGTH = 1200064 MAX_INTERACTION_OPTIONS = 126566 def __init__(self, config: PlatformConfig):67 super().__init__(config, Platform("tiny"))68 extra = config.extra or {}69 self.relay_url = str(extra.get("relay_url") or os.environ.get("TINY_RELAY_URL", "")).strip()70 self.cli_path = str(extra.get("cli_path") or os.environ.get("TINY_CLI_PATH", "tinyagent"))71 self.rooms = [x.strip() for x in str(extra.get("rooms") or extra.get("channels") or os.environ.get("TINY_CHANNELS", "")).split(",") if x.strip()]72 home = extra.get("home_channel") or os.environ.get("TINY_HOME_CHANNEL", "")73 self.home_room = str(home).strip() or (self.rooms[0] if self.rooms else "")74 self.allowed = {x.strip().lower() for x in str(extra.get("allowed_users") or os.environ.get("TINY_ALLOWED_USERS", "")).split(",") if x.strip()}75 require_mention = extra.get("require_mention", os.environ.get("TINY_REQUIRE_MENTION", "true"))76 self.require_mention = str(require_mention).strip().lower() not in {"false", "0", "no", "off"}77 self.rpc: TinyRPC | None = None78 self.pubkey = ""79 self._subscription = "tinyagent"80 self._loop_task: asyncio.Task | None = None81 # request event id -> (room, Hermes key, interaction kind, expiry, option ids)82 self._pending = {}83 self._tracker = None84 self._sent = {}85 self._session_assignees: dict[str, str] = {}86 self._processing: set[str] = set()87 self._reply_routes: dict[str, dict] = {}88 self._turn_routes: dict[str, dict] = {}89 self._turn_route = ContextVar("tinyagent_turn_route", default=None)909192 def name(self) -> str:93 return "Tiny"9495 def set_message_handler(self, handler):96 # Bind routing when Hermes actually starts the turn, not when a queued97 # message arrives. Its public prompt hooks omit reply anchors for Tiny.98 async def routed(event):99 session = self._event_session_key(event)100 route = {**self._reply_routes.get(event.message_id, {}), "session": session}101 token = self._turn_route.set(route)102 self._turn_routes[session] = route103 try:104 return await handler(event)105 finally:106 if self._turn_routes.get(session) is route:107 self._turn_routes.pop(session, None)108 self._turn_route.reset(token)109 super().set_message_handler(routed)110111 async def _process_message_background(self, event, session_key):112 # Include Hermes' final chunks and attachment sends, which run after the113 # message handler returns. Each queued turn gets its own task context.114 token = self._turn_route.set({**self._reply_routes.get(event.message_id, {}), "session": session_key})115 try:116 return await super()._process_message_background(event, session_key)117 finally:118 self._turn_route.reset(token)119120 async def _reply_tags(self, room, reply_to=None, metadata=None):121 metadata = metadata or {}122 current = self._turn_route.get() or {}123 if current.get("room") != room:124 current = {}125 reply = reply_to or metadata.get("reply_to_message_id") or metadata.get("message_id") or current.get("reply_to_message_id")126 if reply and reply != current.get("reply_to_message_id"):127 current = {}128 route = self._reply_routes.get(reply, {})129 if route.get("room") != room:130 route = {}131 thread = metadata.get("thread_id") or route.get("thread_id") or current.get("thread_id")132 root = thread or route.get("root")133 if reply and not root:134 # Synthetic sends may refer to messages received before this process.135 rows = await self.rpc.query({"ids": [reply], "#h": [room], "limit": 1})136 parent = rows[0] if rows else None137 root = (event_thread(parent) if parent else None) or reply138 if parent and parent.get("kind") in (11, 12):139 thread = root140 tags = [["e", str(root), "", "root"]] if root else []141 if reply and reply != root:142 tags.append(["e", str(reply), "", "reply"])143 return tags, 12 if thread else 9144145 async def connect(self, *, is_reconnect: bool = False) -> bool:146 self.rpc = TinyRPC(self.relay_url, cli_path=self.cli_path)147 identity = await self.rpc.call("identity")148 self.pubkey = str(identity.get("pubkey", ""))149 if not self.pubkey:150 raise RuntimeError("Tiny identity did not return a public key")151 self._tracker = EventTracker(self.relay_url, self.pubkey, self.rooms)152 filter: dict[str, Any] = {"kinds": [9, 11, 12, 1111], **self._tracker.filter_since()}153 if self.rooms:154 filter["#h"] = self.rooms155 await self.rpc.subscribe(self._subscription, filter, self._on_event)156 self._mark_connected()157 return True158159 async def disconnect(self) -> None:160 if self.rpc:161 with_context = self.rpc.unsubscribe(self._subscription)162 await asyncio.gather(with_context, return_exceptions=True)163 await self.rpc.close()164 self.rpc = None165 self._mark_disconnected()166167 async def _on_event(self, event: dict) -> None:168 event_id = event.get("id") if isinstance(event, dict) else None169 if event_id in self._processing:170 return171 self._processing.add(event_id)172 try:173 await self._receive_event(event)174 finally:175 self._processing.discard(event_id)176177 async def _receive_event(self, event: dict) -> None:178 if not isinstance(event, dict) or event.get("pubkey", "").lower() == self.pubkey.lower():179 return180 if event.get("kind") in (1111, 7):181 await self._on_answer(event)182 return183 room = _tag(event, "h")184 author = str(event.get("pubkey", ""))185 if not room or (self.rooms and room not in self.rooms) or not self._authorized(author):186 return187 if self._tracker and not self._tracker.should_accept(event, now=int(time.time())):188 return189 root = event_thread(event)190 parent_id = reply_parent(event)191 parent = None192 mentioned = self.pubkey in [row[1] for row in _tags(event, "p")]193 if parent_id:194 rows = await self.rpc.query({"ids": [parent_id], "#h": [room], "limit": 1})195 parent = rows[0] if rows else None196 own_reply = bool(parent and parent.get("pubkey") == self.pubkey)197 if self.require_mention and not mentioned and not own_reply:198 return199 event_id = str(event.get("id", ""))200 # Kind 11/12 carry Tiny thread sessions. Kind 9 remains the room session even when201 # it carries an e tag, matching Hermes' room routing contract.202 thread_id = event_id if event.get("kind") == 11 else root if event.get("kind") == 12 else None203 source = self.build_source(chat_id=room, chat_name=room, chat_type="group", user_id=author,204 thread_id=thread_id, message_id=event_id)205 media_urls, media_types = await incoming_media(self.rpc, event, self.relay_url)206 message_type = MessageType.TEXT207 if media_types:208 major = media_types[0].split("/", 1)[0]209 message_type = {"image": MessageType.PHOTO, "audio": MessageType.AUDIO,210 "video": MessageType.VIDEO}.get(major, MessageType.DOCUMENT)211 message = MessageEvent(text=str(event.get("content", "")), message_type=message_type, source=source,212 raw_message=event, message_id=event_id,213 media_urls=media_urls, media_types=media_types,214 media_text_inlined=[False] * len(media_urls),215 timestamp=datetime.fromtimestamp(int(event.get("created_at", time.time()))),216 reply_to_message_id=parent_id, reply_to_is_own_message=own_reply,217 reply_to_text=parent.get("content") if parent else None)218 self._session_assignees[self._event_session_key(message)] = author219 self._reply_routes[event_id] = {"room": room, "root": root or event_id,220 "reply_to_message_id": event_id, "thread_id": thread_id, "user_id": author}221 if len(self._reply_routes) > 10000:222 self._reply_routes.pop(next(iter(self._reply_routes)))223 if self._tracker:224 self._tracker.mark(event)225 await self.handle_message(message)226227 def _authorized(self, author: str) -> bool:228 return author.lower() in self.allowed229230231 def _mentions(content: str) -> list[str]:232 return _explicit_mentions(content)233234 async def _publish(self, room: str, content: str, tags: list[list[str]], *, kind: int = 9) -> SendResult:235 if not self.rpc:236 return SendResult(success=False, error="Tiny is disconnected", retryable=True)237 try:238 result = await self.rpc.publish({"kind": kind, "content": content, "tags": [["h", room], *tags]})239 event = result if isinstance(result, dict) else {}240 return SendResult(success=True, message_id=event.get("id"), raw_response=event)241 except Exception as exc:242 return SendResult(success=False, error=str(exc), retryable=True)243244 async def send(self, chat_id: str, content: str, reply_to: str | None = None, metadata: dict | None = None) -> SendResult:245 tags, kind = await self._reply_tags(chat_id, reply_to, metadata)246 mentions = _explicit_mentions(content)247 for pubkey in (metadata or {}).get("mentions", []):248 if _HEX.fullmatch(str(pubkey)):249 mentions.append(str(pubkey).lower())250 for pubkey in dict.fromkeys(mentions):251 tags.append(["p", pubkey])252 tags.append(["mention", pubkey])253 result = await self._publish(chat_id, content, tags, kind=kind)254 if result.success:255 self._sent[result.message_id] = tags256 if len(self._sent) > 1000:257 self._sent.pop(next(iter(self._sent)))258 return result259260 async def edit_message(self, chat_id: str, message_id: str, content: str, *, finalize: bool = False) -> SendResult:261 tags = [["e", message_id]]262 for key in _explicit_mentions(content):263 tags.extend([["p", key], ["mention", key]])264 result = await self._publish(chat_id, content, tags, kind=40003)265 if result.success:266 result.message_id = message_id267 return result268269 async def delete_message(self, chat_id: str, message_id: str) -> bool:270 # Tiny's room deletion is a moderator action; this member connector cannot claim deletion.271 return False272273 async def _send_file(self, chat_id: str, file_path: str, caption: str | None = None,274 media_type: str | None = None, metadata: dict | None = None,275 reply_to: str | None = None, file_name: str | None = None) -> SendResult:276 if not self.rpc:277 return SendResult(success=False, error="Tiny is disconnected", retryable=True)278 try:279 path = Path(file_path)280 if not 0 < path.stat().st_size <= 32 << 20:281 return SendResult(success=False, error="Tiny attachments must be between 1 byte and 32 MiB")282 data = await asyncio.to_thread(path.read_bytes)283 mime = media_type or mimetypes.guess_type(path.name)[0] or "application/octet-stream"284 filename = Path(file_name).name if file_name else path.name285 descriptor = await self.rpc.upload(chat_id, filename, mime, data)286 target = descriptor.get("url") or descriptor.get("path") or descriptor.get("sha256")287 if not target:288 return SendResult(success=False, error="Tiny upload returned no descriptor")289 digest = descriptor.get("sha256") or descriptor.get("hash")290 if digest != hashlib.sha256(data).hexdigest():291 return SendResult(success=False, error="Tiny returned a different attachment hash")292 fields = ["url " + str(target), "m " + mime, "filename " + filename]293 if digest:294 fields.append("x " + str(digest))295 fields.append("size " + str(len(data)))296 tags, kind = await self._reply_tags(chat_id, reply_to, metadata)297 tags.append(["imeta", *fields])298 for key in _explicit_mentions(caption or ""):299 tags.extend([["p", key], ["mention", key]])300 return await self._publish(chat_id, caption or "", tags, kind=kind)301 except Exception as exc:302 return SendResult(success=False, error=str(exc), retryable=True)303304 async def send_image_file(self, chat_id: str, image_path: str, caption: str | None = None,305 reply_to=None, metadata=None, **kwargs) -> SendResult:306 return await self._send_file(chat_id, image_path, caption,307 mimetypes.guess_type(image_path)[0] or "image/png", metadata, reply_to)308309 async def send_document(self, chat_id: str, file_path: str, caption: str | None = None,310 file_name=None, reply_to=None, metadata=None, **kwargs) -> SendResult:311 return await self._send_file(chat_id, file_path, caption, metadata=metadata, reply_to=reply_to, file_name=file_name)312313 async def send_video(self, chat_id: str, video_path: str, caption: str | None = None,314 reply_to=None, metadata=None, **kwargs) -> SendResult:315 return await self._send_file(chat_id, video_path, caption, metadata=metadata, reply_to=reply_to)316317 async def send_voice(self, chat_id: str, audio_path: str, caption: str | None = None,318 reply_to=None, metadata=None, **kwargs) -> SendResult:319 return await self._send_file(chat_id, audio_path, caption,320 mimetypes.guess_type(audio_path)[0] or "audio/ogg", metadata, reply_to)321322 async def send_typing(self, chat_id: str, metadata=None) -> None:323 await self._publish(chat_id, "", [["expiration", str(int(time.time()) + 30)], ["status", "typing"]], kind=20002)324325 async def get_chat_info(self, chat_id: str) -> dict[str, Any]:326 return {"name": chat_id, "type": "group"}327328329def check_requirements() -> bool:330 return bool(os.environ.get("TINY_RELAY_URL") and os.environ.get("TINY_PRIVATE_KEY"))331332333def validate_config(config: PlatformConfig) -> bool:334 extra = config.extra or {}335 return bool(extra.get("relay_url") or os.environ.get("TINY_RELAY_URL"))336337338def _env_enablement() -> dict | None:339 relay = os.environ.get("TINY_RELAY_URL", "").strip()340 key = os.environ.get("TINY_PRIVATE_KEY", "").strip()341 if not relay or not key:342 return None343 result = {"relay_url": relay}344 for env, key_name in (("TINY_HOME_CHANNEL", "home_channel"), ("TINY_ALLOWED_USERS", "allowed_users"),345 ("TINY_CLI_PATH", "cli_path"), ("TINY_REQUIRE_MENTION", "require_mention")):346 if os.environ.get(env):347 result[key_name] = os.environ[env]348 return result349350351async def _standalone_send(pconfig, chat_id: str, message: str, *, thread_id=None, media_files=None,352 force_document=False):353 extra = getattr(pconfig, "extra", {}) or {}354 rpc = TinyRPC(str(extra.get("relay_url") or os.environ.get("TINY_RELAY_URL", "")),355 cli_path=str(extra.get("cli_path") or os.environ.get("TINY_CLI_PATH", "tinyagent")))356 try:357 await rpc.connect()358 adapter = TinyAdapter(pconfig)359 adapter.rpc = rpc360 media = list(media_files or [])361 last_id = None362 if message and len(media) != 1:363 result = await adapter.send(chat_id, message, metadata={"thread_id": thread_id})364 if not result.success:365 return {"error": result.error or "Tiny send failed"}366 last_id = result.message_id367 for item in media:368 path = str(item[0] if isinstance(item, (list, tuple)) else item)369 is_voice = bool(item[1]) if isinstance(item, (list, tuple)) and len(item) > 1 else False370 mime = mimetypes.guess_type(path)[0] or ("audio/ogg" if is_voice else "application/octet-stream")371 if force_document:372 mime = "application/octet-stream"373 result = await adapter._send_file(chat_id, path, message if len(media) == 1 else None,374 mime, {"thread_id": thread_id})375 if not result.success:376 return {"error": result.error or "Tiny media send failed"}377 last_id = result.message_id or last_id378 return {"success": True, "message_id": last_id, "media_delivered": bool(media)}379 except Exception as exc:380 return {"error": str(exc)}381 finally:382 await rpc.close()383384385def register(ctx):386 ctx.register_platform(387 name="tiny", label="Tiny", adapter_factory=TinyAdapter,388 check_fn=check_requirements, validate_config=validate_config,389 required_env=["TINY_RELAY_URL", "TINY_PRIVATE_KEY"],390 install_hint="Requires the tinyagent helper binary on PATH or at TINY_CLI_PATH",391 env_enablement_fn=_env_enablement, cron_deliver_env_var="TINY_HOME_CHANNEL",392 standalone_sender_fn=_standalone_send, allowed_users_env="TINY_ALLOWED_USERS",393 max_message_length=12000, platform_hint="You are collaborating through Tinyrelay. Only explicit @ mentions notify a person. Use @ followed by their 64-character public key when you need their attention; ordinary updates stay quiet.",394 emoji="◈", pii_safe=False, allow_update_command=True,395 )