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 logging13import mimetypes14import hashlib15import 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 TinyRPC, scoped_secret27 from .interactions import InteractionMixin28 from .media import incoming_media29 from .ingress import EventTracker, IdentityLock, event_thread, reply_parent30 from . import tasks as taskmod31 from .tasks import TaskState, answer_tags, request_root, request_text32except ImportError: # Hermes plugin loader may import adapter.py as a top-level module.33 from client import TinyRPC, scoped_secret34 from interactions import InteractionMixin35 from media import incoming_media36 from ingress import EventTracker, IdentityLock, event_thread, reply_parent37 import tasks as taskmod38 from tasks import TaskState, answer_tags, request_root, request_text3940logger = logging.getLogger(__name__)4142_HEX = re.compile(r"\b[0-9a-f]{64}\b", re.I)43_MENTION = re.compile(r"(?<![0-9A-Za-z])@([0-9a-f]{64})(?![0-9A-Za-z])", re.I)44# Buzz's convention: this exact message from an allowed user stops the current turn.45CANCEL_COMMAND = "!cancel"464748def _tags(event: dict, name: str) -> list[list[str]]:49 return [tag for tag in event.get("tags", []) if len(tag) >= 2 and tag[0] == name]505152def _tag(event: dict, name: str) -> str | None:53 found = _tags(event, name)54 return found[0][1] if found else None555657def _explicit_mentions(text: str) -> list[str]:58 """Return only public keys explicitly written as @mentions in text."""59 return list(dict.fromkeys(match.group(1).lower() for match in _MENTION.finditer(text or "")))606162def _mention_tags(text: str) -> list[list[str]]:63 return [["p", pubkey] for pubkey in _explicit_mentions(text)]646566class TinyAdapter(InteractionMixin, BasePlatformAdapter):67 """Native room messages and interactions for Hermes."""6869 supports_code_blocks = True70 supports_async_delivery = True71 MAX_MESSAGE_LENGTH = 1200072 MAX_INTERACTION_OPTIONS = 1273 # Helper-death recovery backoff in seconds; Hermes's own watcher starts at 30s.74 RECONNECT_MIN = 1.075 RECONNECT_MAX = 30.076 SWEEP_INTERVAL = 60.077 # Long-task cards: at most one 43003 progress event per task in this many seconds.78 PROGRESS_INTERVAL = 15.079 MAX_TASKS = 20080 # Waits before the second and third attempt to publish a task's terminal event.81 TERMINAL_RETRY_DELAYS = (1.0, 3.0)8283 def __init__(self, config: PlatformConfig):84 super().__init__(config, Platform("tiny"))85 extra = config.extra or {}86 self.relay_url = str(extra.get("relay_url") or scoped_secret("TINY_RELAY_URL")).strip()87 self.cli_path = str(extra.get("cli_path") or scoped_secret("TINY_CLI_PATH", "tinyagent"))88 self.rooms = [x.strip() for x in str(extra.get("rooms") or extra.get("channels") or scoped_secret("TINY_CHANNELS")).split(",") if x.strip()]89 home = extra.get("home_channel") or getattr(getattr(config, "home_channel", None), "chat_id", None) \90 or scoped_secret("TINY_HOME_CHANNEL")91 if isinstance(home, dict):92 home = home.get("chat_id", "")93 self.home_room = str(home).strip() or (self.rooms[0] if self.rooms else "")94 # Allowed users come from the profile scope, so a multiplexed profile never95 # inherits another profile's operators from the process environment.96 self.allowed = {x.strip().lower() for x in str(extra.get("allowed_users") or scoped_secret("TINY_ALLOWED_USERS")).split(",") if x.strip()}97 require_mention = extra.get("require_mention", scoped_secret("TINY_REQUIRE_MENTION", "true"))98 self.require_mention = str(require_mention).strip().lower() not in {"false", "0", "no", "off"}99 self.rpc: TinyRPC | None = None100 self.pubkey = ""101 self._subscription = "tinyagent"102 self._loop_task: asyncio.Task | None = None103 # request event id -> (room, Hermes key, interaction kind, expiry, option ids)104 self._pending = {}105 self._tracker = None106 self._sent = {}107 self._session_assignees: dict[str, str] = {}108 self._processing: set[str] = set()109 self._reply_routes: dict[str, dict] = {}110 self._turn_routes: dict[str, dict] = {}111 self._turn_route = ContextVar("tinyagent_turn_route", default=None)112 self._lock: IdentityLock | None = None113 self._closed = False114 # The helper whose subscription is being set up; its death is reported by _establish.115 self._establishing: TinyRPC | None = None116 self._reconnect_task: asyncio.Task | None = None117 self._last_sweep = float("-inf")118 # request id -> long-task card state, in intake order.119 self._tasks: dict[str, TaskState] = {}120121122 def name(self) -> str:123 return "Tiny"124125 def set_message_handler(self, handler):126 # Bind routing when Hermes actually starts the turn, not when a queued127 # message arrives. Its public prompt hooks omit reply anchors for Tiny.128 async def routed(event):129 session = self._event_session_key(event)130 route = {**self._reply_routes.get(event.message_id, {}), "session": session}131 token = self._turn_route.set(route)132 self._turn_routes[session] = route133 task = self._tasks.get(event.message_id)134 try:135 response = await handler(event)136 except BaseException as exc:137 # The failure hook fires before Hermes posts its notice; keep the text now.138 if task is not None and not isinstance(exc, asyncio.CancelledError):139 task.note_error(str(exc) or type(exc).__name__)140 raise141 else:142 if task is not None and isinstance(response, str):143 task.final_text = response144 return response145 finally:146 if self._turn_routes.get(session) is route:147 self._turn_routes.pop(session, None)148 self._turn_route.reset(token)149 super().set_message_handler(routed)150151 async def _process_message_background(self, event, session_key):152 # Include Hermes' final chunks and attachment sends, which run after the153 # message handler returns. Each queued turn gets its own task context.154 token = self._turn_route.set({**self._reply_routes.get(event.message_id, {}), "session": session_key})155 try:156 return await super()._process_message_background(event, session_key)157 finally:158 self._turn_route.reset(token)159160 async def _reply_tags(self, room, reply_to=None, metadata=None):161 metadata = metadata or {}162 current = self._turn_route.get() or {}163 if current.get("room") != room:164 current = {}165 reply = reply_to or metadata.get("reply_to_message_id") or metadata.get("message_id") or current.get("reply_to_message_id")166 if reply and reply != current.get("reply_to_message_id"):167 current = {}168 route = self._reply_routes.get(reply, {})169 if route.get("room") != room:170 route = {}171 thread = metadata.get("thread_id") or route.get("thread_id") or current.get("thread_id")172 # A session thread named after a received event (a task request or a kind 11173 # opener) replies in the thread that event belongs to.174 anchor = self._reply_routes.get(str(thread or ""), {})175 if anchor.get("room") == room and anchor.get("thread_id"):176 thread = anchor["thread_id"]177 root = thread or route.get("root")178 if reply and not root:179 # Synthetic sends may refer to messages received before this process.180 rows = await self.rpc.query({"ids": [reply], "#h": [room], "limit": 1})181 parent = rows[0] if rows else None182 root = (event_thread(parent) if parent else None) or reply183 if parent and parent.get("kind") in (11, 12):184 thread = root185 tags = [["e", str(root), "", "root"]] if root else []186 if reply and reply != root:187 tags.append(["e", str(reply), "", "reply"])188 return tags, 12 if thread else 9189190 async def connect(self, *, is_reconnect: bool = False) -> bool:191 """Start the helper and subscribe. Hermes's watcher calls this on a fresh adapter192 with is_reconnect=True; helper death inside a live adapter reuses _establish."""193 self._closed = False194 try:195 await self._establish()196 except BaseException:197 if self._lock is not None and self.rpc is None:198 # No session survived the failure; free the identity for the next attempt.199 self._lock.release()200 self._lock = None201 raise202 return True203204 async def _establish(self) -> None:205 """Spawn a helper, prove the identity, take its lock and subscribe from the cursor.206207 The session owns the new helper before the subscription opens: the helper may208 deliver events before it acknowledges the subscribe call, and those events209 already need ``self.rpc``. A failure before the subscription is acknowledged210 restores the previous helper reference and closes the new one."""211 rpc = TinyRPC(self.relay_url, cli_path=self.cli_path, on_exit=self._on_helper_exit)212 previous = self.rpc213 self._establishing = rpc214 try:215 identity = await rpc.call("identity")216 pubkey = str((identity or {}).get("pubkey", ""))217 if not pubkey:218 raise RuntimeError("Tiny identity did not return a public key")219 if self._lock is None or self.pubkey.lower() != pubkey.lower():220 if self._lock is not None:221 self._lock.release()222 lock = IdentityLock(self.relay_url, pubkey)223 lock.acquire()224 self._lock = lock225 self.pubkey = pubkey226 if self._tracker is None or self._tracker.identity != pubkey.lower():227 self._tracker = EventTracker(self.relay_url, pubkey, self.rooms)228 filter: dict[str, Any] = {"kinds": [9, 11, 12, 1111, taskmod.KIND_JOB_REQUEST, taskmod.KIND_JOB_CANCEL],229 **self._tracker.filter_since()}230 if self.rooms:231 filter["#h"] = self.rooms232 self.rpc = rpc233 await rpc.subscribe(self._subscription, filter, self._on_event)234 if not rpc.alive:235 raise RuntimeError("Tiny helper exited while subscribing")236 except BaseException:237 if self.rpc is rpc:238 self.rpc = previous239 self._establishing = None240 await rpc.close()241 raise242 self._establishing = None243 if previous is not None and previous is not rpc:244 await previous.close()245 self._mark_connected()246247 def _on_helper_exit(self, rpc) -> None:248 """TinyRPC reports the helper died on its own; recover in the background."""249 if self._closed or rpc is not self.rpc or rpc is self._establishing:250 return251 logger.warning("Tiny helper exited; reconnecting")252 self._mark_disconnected()253 if self._reconnect_task is None or self._reconnect_task.done():254 self._reconnect_task = asyncio.create_task(self._reconnect_loop())255256 async def _reconnect_loop(self) -> None:257 """Bounded exponential backoff until the helper is back or the adapter closes.258259 The adapter stays installed in Hermes throughout, so process-local waiters and260 the identity lock survive; the gateway's reconnect watcher only builds a new261 adapter for fatal errors, which this path never raises.262 """263 delay = self.RECONNECT_MIN264 while not self._closed:265 await asyncio.sleep(delay)266 if self._closed:267 return268 try:269 await self._establish()270 except asyncio.CancelledError:271 raise272 except Exception as exc:273 delay = min(delay * 2, self.RECONNECT_MAX)274 logger.warning("Tiny helper reconnect failed: %s; retrying in %.0fs", exc, delay)275 continue276 logger.info("Tiny helper reconnected from cursor %s", self._tracker.cursor if self._tracker else "?")277 return278279 async def disconnect(self) -> None:280 self._closed = True281 task, self._reconnect_task = self._reconnect_task, None282 if task is not None and task is not asyncio.current_task():283 task.cancel()284 await asyncio.gather(task, return_exceptions=True)285 rpc, self.rpc = self.rpc, None286 if rpc is not None:287 if rpc.alive:288 await asyncio.gather(rpc.unsubscribe(self._subscription), return_exceptions=True)289 await rpc.close()290 if self._lock is not None:291 self._lock.release()292 self._lock = None293 for task in self._tasks.values():294 self._stop_progress_timer(task)295 self._mark_disconnected()296297 async def _on_event(self, event: dict) -> None:298 event_id = event.get("id") if isinstance(event, dict) else None299 if event_id in self._processing:300 return301 self._processing.add(event_id)302 now = time.monotonic()303 if now - self._last_sweep >= self.SWEEP_INTERVAL:304 self._last_sweep = now305 self._sweep_pending()306 try:307 await self._receive_event(event)308 finally:309 self._processing.discard(event_id)310311 async def _receive_event(self, event: dict) -> None:312 if not isinstance(event, dict) or event.get("pubkey", "").lower() == self.pubkey.lower():313 return314 if event.get("kind") in (1111, 7):315 await self._on_answer(event)316 return317 room = _tag(event, "h")318 author = str(event.get("pubkey", ""))319 if not room or (self.rooms and room not in self.rooms) or not self._authorized(author):320 return321 if self._tracker and not self._tracker.should_accept(event, now=int(time.time())):322 return323 kind = event.get("kind")324 if kind == taskmod.KIND_JOB_REQUEST:325 await self._on_task_request(event, room, author)326 return327 if kind == taskmod.KIND_JOB_CANCEL:328 await self._on_task_cancel(event, author)329 if self._tracker:330 self._tracker.mark(event)331 return332 if kind in (9, 12) and self._is_cancel_command(event):333 await self._on_cancel_command(event, room, author)334 if self._tracker:335 self._tracker.mark(event)336 return337 root = event_thread(event)338 parent_id = reply_parent(event)339 parent = None340 mentioned = self.pubkey in [row[1] for row in _tags(event, "p")]341 if parent_id:342 rows = await self.rpc.query({"ids": [parent_id], "#h": [room], "limit": 1})343 parent = rows[0] if rows else None344 own_reply = bool(parent and parent.get("pubkey") == self.pubkey)345 if self.require_mention and not mentioned and not own_reply:346 return347 event_id = str(event.get("id", ""))348 # Kind 11/12 carry Tiny thread sessions. Kind 9 remains the room session even when349 # it carries an e tag, matching Hermes' room routing contract.350 thread_id = event_id if event.get("kind") == 11 else root if event.get("kind") == 12 else None351 source = self.build_source(chat_id=room, chat_name=room, chat_type="group", user_id=author,352 thread_id=thread_id, message_id=event_id)353 media_urls, media_types = await incoming_media(self.rpc, event, self.relay_url)354 message_type = MessageType.TEXT355 if media_types:356 major = media_types[0].split("/", 1)[0]357 message_type = {"image": MessageType.PHOTO, "audio": MessageType.AUDIO,358 "video": MessageType.VIDEO}.get(major, MessageType.DOCUMENT)359 message = MessageEvent(text=str(event.get("content", "")), message_type=message_type, source=source,360 raw_message=event, message_id=event_id,361 media_urls=media_urls, media_types=media_types,362 media_text_inlined=[False] * len(media_urls),363 timestamp=datetime.fromtimestamp(int(event.get("created_at", time.time()))),364 reply_to_message_id=parent_id, reply_to_is_own_message=own_reply,365 reply_to_text=parent.get("content") if parent else None)366 self._session_assignees[self._event_session_key(message)] = author367 self._reply_routes[event_id] = {"room": room, "root": root or event_id,368 "reply_to_message_id": event_id, "thread_id": thread_id, "user_id": author}369 if len(self._reply_routes) > 10000:370 self._reply_routes.pop(next(iter(self._reply_routes)))371 await self._deliver(message, event)372373 async def _deliver(self, message: MessageEvent, event: dict) -> bool:374 """Hand a turn to Hermes and return whether it was admitted.375376 The event counts as delivered only once Hermes started or queued a turn for it377 (its admission receipt) or consumed it inline; a failed or refused hand-off378 leaves it eligible for redelivery until the attempt budget is spent. A Hermes379 without the receipt admits by returning."""380 event_id = str(event.get("id", ""))381 try:382 await self.handle_message(message)383 except BaseException as exc:384 if self._tracker and not isinstance(exc, asyncio.CancelledError):385 if self._tracker.fail(event):386 logger.error("Tiny event %s retired after repeated failures", event_id, exc_info=exc)387 raise388 admitted = getattr(message, "_gateway_accepted", None)389 if admitted is False and not self._consumed_inline(message):390 if self._tracker and self._tracker.fail(event):391 logger.error("Tiny event %s retired: Hermes refused it repeatedly", event_id)392 else:393 logger.warning("Tiny event %s was not admitted by Hermes; it stays eligible for redelivery", event_id)394 return False395 if self._tracker:396 self._tracker.mark(event)397 return True398399 def _consumed_inline(self, message: MessageEvent) -> bool:400 """True for events Hermes dispatches inline without an admission receipt: slash401 commands during a busy session and text answers to a pending question."""402 if message.get_command() is not None:403 return True404 try:405 from tools import clarify_gateway406 return clarify_gateway.get_pending_for_session(self._event_session_key(message),407 include_choice_prompts=True) is not None408 except Exception:409 return False410411 # -- long-task cards -------------------------------------------------412413 async def _on_task_request(self, event: dict, room: str, author: str) -> None:414 """Turn a 43001 that asks this key into a thread turn anchored at the request."""415 request_id = str(event.get("id", ""))416 if self.pubkey not in [row[1] for row in _tags(event, "p")]:417 if self._tracker:418 self._tracker.mark(event)419 return420 # Restart safety: a request this key already finished (or the requester421 # cancelled) is not run again.422 done = await self.rpc.query({"kinds": list(taskmod.TERMINAL_KINDS), "#e": [request_id],423 "#h": [room], "limit": 1})424 if done or request_id in self._tasks:425 if self._tracker:426 self._tracker.mark(event)427 return428 root = request_root(event)429 # Every request runs in its own session, named by the request id, so Hermes430 # never merges two requests into one turn. Replies still land in the thread431 # the request belongs to (see _reply_tags).432 source = self.build_source(chat_id=room, chat_name=room, chat_type="group", user_id=author,433 thread_id=request_id, message_id=request_id)434 message = MessageEvent(text=request_text(event), message_type=MessageType.TEXT, source=source,435 raw_message=event, message_id=request_id,436 timestamp=datetime.fromtimestamp(int(event.get("created_at", time.time()))))437 session = self._event_session_key(message)438 task = TaskState(request_id, room, author, root, session)439 self._tasks[request_id] = task440 while len(self._tasks) > self.MAX_TASKS:441 self._forget_task(next(iter(self._tasks)))442 self._session_assignees[session] = author443 self._reply_routes[request_id] = {"room": room, "root": root, "reply_to_message_id": request_id,444 "thread_id": root, "user_id": author}445 try:446 admitted = await self._deliver(message, event)447 except BaseException:448 self._forget_task(request_id)449 raise450 if not admitted:451 self._forget_task(request_id)452453 def _forget_task(self, request_id: str) -> None:454 task = self._tasks.pop(request_id, None)455 if task is not None:456 self._stop_progress_timer(task)457458 def _task_for_send(self, room: str, tags: list[list[str]]) -> TaskState | None:459 """Return the task a send belongs to: the request it answers, else the task of460 the current turn, else the running task of the thread it lands in."""461 marked = {row[3]: row[1] for row in tags if len(row) >= 4 and row[0] == "e"}462 for key in ("reply", "root"):463 task = self._tasks.get(str(marked.get(key) or ""))464 if task is not None and not task.terminal and task.room == room:465 return task466 current = self._turn_route.get() or {}467 if current.get("room") == room and current.get("session"):468 task = self._task_for_session(current["session"])469 if task is not None:470 return task471 root = marked.get("root")472 if not root:473 return None474 matches = [task for task in self._tasks.values()475 if not task.terminal and task.room == room and task.thread_root == root]476 return next((task for task in matches if task.accepted), matches[0] if matches else None)477478 def _task_for_message(self, message_id: str) -> TaskState | None:479 return next((task for task in self._tasks.values()480 if not task.terminal and message_id in task.classes), None)481482 def _task_for_session(self, session: str) -> TaskState | None:483 matches = [task for task in self._tasks.values() if not task.terminal and task.session_key == session]484 return next((task for task in matches if task.accepted), matches[0] if matches else None)485486 def _observe(self, task: TaskState | None, message_id: str | None, content: str, cls: str, *,487 final: bool = False, file_name: str | None = None) -> None:488 if task is None or not message_id:489 return490 task.observe(message_id, content, cls, final=final, file_name=file_name)491 self._schedule_progress(task)492493 def _clock(self) -> float:494 return time.monotonic()495496 def _schedule_progress(self, task: TaskState) -> None:497 """Publish pending progress now, or once the throttle window has passed."""498 if task.terminal or not task.accepted or not task.has_progress or task.timer is not None:499 return500 delay = task.progress_due(self._clock(), self.PROGRESS_INTERVAL)501 if delay <= 0:502 task.timer = asyncio.ensure_future(self._flush_progress(task.request_id))503 return504 loop = asyncio.get_running_loop()505 task.timer = loop.call_later(delay, self._progress_timer_fired, task.request_id)506507 def _progress_timer_fired(self, request_id: str) -> None:508 task = self._tasks.get(request_id)509 if task is None:510 return511 task.timer = asyncio.ensure_future(self._flush_progress(request_id))512513 def _stop_progress_timer(self, task: TaskState) -> None:514 timer, task.timer = task.timer, None515 try:516 current = asyncio.current_task()517 except RuntimeError:518 current = None519 if timer is not None and timer is not current:520 timer.cancel()521522 async def _stop_task_turn(self, task: TaskState) -> None:523 """Stop the Hermes turn of a task: cancel it when it runs, drop it when still queued."""524 if task.accepted:525 await self.cancel_session_processing(task.session_key)526 return527 pending = getattr(self, "_pending_messages", {}).get(task.session_key)528 if pending is not None and str(getattr(pending, "message_id", "")) == task.request_id:529 self._pending_messages.pop(task.session_key, None)530531 async def _flush_progress(self, request_id: str) -> None:532 task = self._tasks.get(request_id)533 if task is None:534 return535 if task.timer is not None and task.timer is not asyncio.current_task():536 task.timer.cancel()537 try:538 if task.terminal or not task.has_progress:539 return540 content = task.take_progress(self._clock())541 result = await self._publish(task.room, content, answer_tags(task), kind=taskmod.KIND_JOB_PROGRESS)542 if not result.success:543 logger.warning("Tiny task %s progress failed: %s", request_id[:8], result.error)544 finally:545 task.timer = None546 if task.has_progress:547 self._schedule_progress(task)548549 async def _publish_terminal(self, task: TaskState, kind: int, content: str, extra: list[list[str]] = ()) -> None:550 """Publish exactly one terminal event for a task, retrying a failed send.551552 The task stays known until the event is on the relay or the attempts are553 spent, so nothing else can end it twice."""554 if not task.end():555 return556 self._stop_progress_timer(task)557 tags = [*answer_tags(task), *extra]558 try:559 for attempt, delay in enumerate((0.0, *self.TERMINAL_RETRY_DELAYS), 1):560 if delay:561 await asyncio.sleep(delay)562 result = await self._publish(task.room, content, tags, kind=kind)563 if result.success:564 return565 logger.warning("Tiny task %s terminal event %d failed (attempt %d): %s",566 task.request_id[:8], kind, attempt, result.error)567 logger.error("Tiny task %s terminal event %d gave up after %d attempts", task.request_id[:8], kind,568 1 + len(self.TERMINAL_RETRY_DELAYS))569 finally:570 self._forget_task(task.request_id)571572 async def on_processing_start(self, event: MessageEvent) -> None:573 await super().on_processing_start(event)574 task = self._tasks.get(str(event.message_id or ""))575 if task is None or task.terminal or task.accepted:576 return577 task.accepted = True578 result = await self._publish(task.room, "", answer_tags(task), kind=taskmod.KIND_JOB_ACCEPTED)579 if not result.success:580 logger.warning("Tiny task %s accepted event failed: %s", task.request_id[:8], result.error)581582 async def on_processing_complete(self, event: MessageEvent, outcome) -> None:583 await super().on_processing_complete(event, outcome)584 task = self._tasks.get(str(event.message_id or ""))585 if task is None:586 return587 name = str(getattr(outcome, "value", outcome)).lower()588 if task.terminal:589 return590 merged = self._merged_tasks(task)591 # Hermes 0.21.1 grades a turn without counting attachment delivery, so a592 # file-only reply reports failure. The adapter saw the posts itself: with no593 # handler error and no failed send, what reached the room is the result.594 if name == "success" or (name == "failure" and task.delivered and not task.error):595 content, extra = task.result()596 await self._publish_terminal(task, taskmod.KIND_JOB_RESULT, content, extra)597 elif name == "cancelled":598 await self._publish_terminal(task, taskmod.KIND_JOB_ERROR, "Cancelled")599 else:600 await self._publish_terminal(task, taskmod.KIND_JOB_ERROR, task.failure())601 for other, carrier in merged:602 await self._publish_terminal(other, taskmod.KIND_JOB_ERROR, f"Merged into task {carrier}")603604 def _merged_tasks(self, task: TaskState) -> list[tuple[TaskState, str]]:605 """Return tasks Hermes folded into another turn of this task's session.606607 Sessions are unique per request, so this is a safety net: a request that608 shares the finished task's session, never started and is not the session's609 queued follow-up has lost its own turn. Its text travelled with the queued610 follow-up when there is one, else with the finished turn."""611 queued = getattr(self, "_pending_messages", {}).get(task.session_key)612 store = getattr(self, "_text_debounce_store", None)613 state = store().get(task.session_key) if callable(store) else None614 if queued is None and state is not None:615 queued = getattr(state, "event", None)616 queued_id = str(getattr(queued, "message_id", "") or "")617 carrier = queued_id or task.request_id618 return [(other, carrier) for other in list(self._tasks.values())619 if other is not task and not other.terminal and not other.accepted620 and other.session_key == task.session_key and other.request_id != queued_id]621622 async def _on_task_cancel(self, event: dict, author: str) -> None:623 """A 43005 from the requester ends the task; the cancel itself is the terminal event."""624 task = self._tasks.get(str(_tag(event, "e") or ""))625 if task is None or task.terminal or author != task.requester:626 return627 task.end()628 self._stop_progress_timer(task)629 await self._stop_task_turn(task)630 self._forget_task(task.request_id)631632 def _is_cancel_command(self, event: dict) -> bool:633 return (str(event.get("content", "")).strip() == CANCEL_COMMAND634 and self.pubkey in [row[1] for row in _tags(event, "p")])635636 async def _on_cancel_command(self, event: dict, room: str, author: str) -> None:637 """Buzz's !cancel: stop the current turn of the room or thread session instead of638 forwarding it. A task turn ends with a 43006 naming who cancelled it."""639 event_id = str(event.get("id", ""))640 thread_id = event_thread(event) if event.get("kind") == 12 else None641 source = self.build_source(chat_id=room, chat_name=room, chat_type="group", user_id=author,642 thread_id=thread_id, message_id=event_id)643 session = self._event_session_key(MessageEvent(text=CANCEL_COMMAND, source=source, message_id=event_id))644 task = self._task_for_session(session)645 if task is None:646 root = event_thread(event)647 if root:648 task = next((entry for entry in self._tasks.values()649 if not entry.terminal and entry.room == room and entry.thread_root == root), None)650 elif session not in getattr(self, "_active_sessions", {}):651 # A bare room-level !cancel with no turn of its own stops the room's newest task.652 running = [entry for entry in self._tasks.values()653 if not entry.terminal and entry.room == room and entry.accepted]654 task = running[-1] if running else None655 if task is not None:656 await self._publish_terminal(task, taskmod.KIND_JOB_ERROR, f"Cancelled by {author}")657 await self._stop_task_turn(task)658 return659 await self.cancel_session_processing(session)660661 async def create_handoff_thread(self, parent_chat_id: str, name: str) -> str | None:662 """Open a real thread in the room so a handed-off session continues there."""663 result = await self._publish(parent_chat_id, str(name or "").strip() or "Handoff", [], kind=11)664 if not result.success or not result.message_id:665 logger.warning("Tiny handoff thread failed: %s", result.error)666 return None667 return str(result.message_id)668669 def _authorized(self, author: str) -> bool:670 return author.lower() in self.allowed671672673 def _mentions(content: str) -> list[str]:674 return _explicit_mentions(content)675676 async def _publish(self, room: str, content: str, tags: list[list[str]], *, kind: int = 9) -> SendResult:677 if not self.rpc:678 return SendResult(success=False, error="Tiny is disconnected", retryable=True)679 try:680 result = await self.rpc.publish({"kind": kind, "content": content, "tags": [["h", room], *tags]})681 event = result if isinstance(result, dict) else {}682 return SendResult(success=True, message_id=event.get("id"), raw_response=event)683 except Exception as exc:684 return SendResult(success=False, error=str(exc), retryable=True)685686 async def send(self, chat_id: str, content: str, reply_to: str | None = None, metadata: dict | None = None) -> SendResult:687 tags, kind = await self._reply_tags(chat_id, reply_to, metadata)688 mentions = _explicit_mentions(content)689 for pubkey in (metadata or {}).get("mentions", []):690 if _HEX.fullmatch(str(pubkey)):691 mentions.append(str(pubkey).lower())692 for pubkey in dict.fromkeys(mentions):693 tags.append(["p", pubkey])694 tags.append(["mention", pubkey])695 result = await self._publish(chat_id, content, tags, kind=kind)696 task = self._task_for_send(chat_id, tags)697 if result.success:698 self._sent[result.message_id] = tags699 if len(self._sent) > 1000:700 self._sent.pop(next(iter(self._sent)))701 # Hermes marks a streamed draft with expect_edits and a final reply with notify;702 # anything else in the thread is tool chrome or a status line.703 meta = metadata or {}704 final = bool(meta.get("notify"))705 cls = taskmod.TEXT if final or meta.get("expect_edits") else taskmod.CHROME706 self._observe(task, result.message_id, content, cls, final=final)707 elif task is not None:708 task.note_error(result.error or "Tiny send failed")709 return result710711 async def edit_message(self, chat_id: str, message_id: str, content: str, *, finalize: bool = False) -> SendResult:712 tags = [["e", message_id]]713 for key in _explicit_mentions(content):714 tags.extend([["p", key], ["mention", key]])715 result = await self._publish(chat_id, content, tags, kind=40003)716 task = self._task_for_message(str(message_id))717 if result.success:718 result.message_id = message_id719 self._observe(task, str(message_id), content, taskmod.CHROME, final=finalize)720 elif task is not None:721 task.note_error(result.error or "Tiny edit failed")722 return result723724 async def delete_message(self, chat_id: str, message_id: str) -> bool:725 # Tiny's room deletion is a moderator action; this member connector cannot claim deletion.726 return False727728 async def _send_file(self, chat_id: str, file_path: str, caption: str | None = None,729 media_type: str | None = None, metadata: dict | None = None,730 reply_to: str | None = None, file_name: str | None = None) -> SendResult:731 if not self.rpc:732 return SendResult(success=False, error="Tiny is disconnected", retryable=True)733 try:734 path = Path(file_path)735 if not 0 < path.stat().st_size <= 32 << 20:736 return SendResult(success=False, error="Tiny attachments must be between 1 byte and 32 MiB")737 data = await asyncio.to_thread(path.read_bytes)738 mime = media_type or mimetypes.guess_type(path.name)[0] or "application/octet-stream"739 filename = Path(file_name).name if file_name else path.name740 descriptor = await self.rpc.upload(chat_id, filename, mime, data)741 target = descriptor.get("url") or descriptor.get("path") or descriptor.get("sha256")742 if not target:743 return SendResult(success=False, error="Tiny upload returned no descriptor")744 digest = descriptor.get("sha256") or descriptor.get("hash")745 if digest != hashlib.sha256(data).hexdigest():746 return SendResult(success=False, error="Tiny returned a different attachment hash")747 fields = ["url " + str(target), "m " + mime, "filename " + filename]748 if digest:749 fields.append("x " + str(digest))750 fields.append("size " + str(len(data)))751 tags, kind = await self._reply_tags(chat_id, reply_to, metadata)752 tags.append(["imeta", *fields])753 for key in _explicit_mentions(caption or ""):754 tags.extend([["p", key], ["mention", key]])755 result = await self._publish(chat_id, caption or "", tags, kind=kind)756 task = self._task_for_send(chat_id, tags)757 if result.success:758 self._observe(task, result.message_id, caption or "", taskmod.FILE, file_name=filename)759 elif task is not None:760 task.note_error(result.error or "Tiny upload failed")761 return result762 except Exception as exc:763 return SendResult(success=False, error=str(exc), retryable=True)764765 async def send_image_file(self, chat_id: str, image_path: str, caption: str | None = None,766 reply_to=None, metadata=None, **kwargs) -> SendResult:767 return await self._send_file(chat_id, image_path, caption,768 mimetypes.guess_type(image_path)[0] or "image/png", metadata, reply_to)769770 async def send_document(self, chat_id: str, file_path: str, caption: str | None = None,771 file_name=None, reply_to=None, metadata=None, **kwargs) -> SendResult:772 return await self._send_file(chat_id, file_path, caption, metadata=metadata, reply_to=reply_to, file_name=file_name)773774 async def send_video(self, chat_id: str, video_path: str, caption: str | None = None,775 reply_to=None, metadata=None, **kwargs) -> SendResult:776 return await self._send_file(chat_id, video_path, caption, metadata=metadata, reply_to=reply_to)777778 async def send_voice(self, chat_id: str, audio_path: str, caption: str | None = None,779 reply_to=None, metadata=None, **kwargs) -> SendResult:780 return await self._send_file(chat_id, audio_path, caption,781 mimetypes.guess_type(audio_path)[0] or "audio/ogg", metadata, reply_to)782783 async def send_typing(self, chat_id: str, metadata=None) -> None:784 await self._publish(chat_id, "", [["expiration", str(int(time.time()) + 30)], ["status", "typing"]], kind=20002)785786 async def get_chat_info(self, chat_id: str) -> dict[str, Any]:787 return {"name": chat_id, "type": "group"}788789790def check_requirements() -> bool:791 return bool(scoped_secret("TINY_RELAY_URL") and scoped_secret("TINY_PRIVATE_KEY"))792793794def validate_config(config: PlatformConfig) -> bool:795 extra = config.extra or {}796 return bool(extra.get("relay_url") or scoped_secret("TINY_RELAY_URL"))797798799def _env_enablement() -> dict | None:800 relay = scoped_secret("TINY_RELAY_URL").strip()801 key = scoped_secret("TINY_PRIVATE_KEY").strip()802 if not relay or not key:803 return None804 result = {"relay_url": relay}805 for env, key_name in (("TINY_ALLOWED_USERS", "allowed_users"), ("TINY_CLI_PATH", "cli_path"),806 ("TINY_REQUIRE_MENTION", "require_mention")):807 if scoped_secret(env):808 result[key_name] = scoped_secret(env)809 home = scoped_secret("TINY_HOME_CHANNEL").strip()810 if home:811 # Hermes lifts this dict into a HomeChannel for cron and standalone delivery.812 result["home_channel"] = {"chat_id": home, "name": "Home"}813 return result814815816async def _standalone_send(pconfig, chat_id: str, message: str, *, thread_id=None, media_files=None,817 force_document=False):818 extra = getattr(pconfig, "extra", {}) or {}819 rpc = TinyRPC(str(extra.get("relay_url") or scoped_secret("TINY_RELAY_URL")),820 cli_path=str(extra.get("cli_path") or scoped_secret("TINY_CLI_PATH", "tinyagent")))821 try:822 await rpc.connect()823 adapter = TinyAdapter(pconfig)824 adapter.rpc = rpc825 media = list(media_files or [])826 last_id = None827 if message and len(media) != 1:828 result = await adapter.send(chat_id, message, metadata={"thread_id": thread_id})829 if not result.success:830 return {"error": result.error or "Tiny send failed"}831 last_id = result.message_id832 for item in media:833 path = str(item[0] if isinstance(item, (list, tuple)) else item)834 is_voice = bool(item[1]) if isinstance(item, (list, tuple)) and len(item) > 1 else False835 mime = mimetypes.guess_type(path)[0] or ("audio/ogg" if is_voice else "application/octet-stream")836 if force_document:837 mime = "application/octet-stream"838 result = await adapter._send_file(chat_id, path, message if len(media) == 1 else None,839 mime, {"thread_id": thread_id})840 if not result.success:841 return {"error": result.error or "Tiny media send failed"}842 last_id = result.message_id or last_id843 return {"success": True, "message_id": last_id, "media_delivered": bool(media)}844 except Exception as exc:845 return {"error": str(exc)}846 finally:847 await rpc.close()848849850def register(ctx):851 ctx.register_platform(852 name="tiny", label="Tiny", adapter_factory=TinyAdapter,853 check_fn=check_requirements, validate_config=validate_config,854 required_env=["TINY_RELAY_URL", "TINY_PRIVATE_KEY"],855 install_hint="Requires the tinyagent helper binary on PATH or at TINY_CLI_PATH",856 env_enablement_fn=_env_enablement, cron_deliver_env_var="TINY_HOME_CHANNEL",857 standalone_sender_fn=_standalone_send, allowed_users_env="TINY_ALLOWED_USERS",858 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.",859 emoji="◈", pii_safe=False, allow_update_command=True,860 )