Each task request runs in its own session so Hermes cannot merge two requests into one turn, and a request Hermes still merges ends with an error naming the carrier. The session is assigned before subscribing so events delivered ahead of the acknowledgment cannot reach a missing helper. A file-only task whose deliveries the adapter observed publishes a result even when Hermes core grades the turn as failed. Events are marked seen only after Hermes issues an admission receipt, terminal cards are retried, the private key and settings resolve through the profile's secret scope, identity locks live in a root shared by all profiles, and the replay cache keeps every ID inside its window.
- commit
- 46d32538d616d6bc85850f22eb02a8e1ad9032ee
- parent
- 64d4ccea8f3d2ee0c28d75a72c2d59714dd6c325
- author
- Dami <dami.etoile@gmail.com>
| M | docs/ | 8 | +- |
| M | tinyagent/ | 6 | +- |
| M | tinyagent/ | 181 | +++++++++++++++---- |
| M | tinyagent/ | 35 | +++- |
| M | tinyagent/ | 63 | ++++++- |
| M | tinyagent/ | 5 | + |
| M | tinyagent/ | 282 | ++++++++++++++++++++++++++++++- |
| M | tinyagent/ | 72 | +++++++- |
| M | tinyagent/ | 165 | +++++++++++++++++- |
@@ -21,17 +21,17 @@ Answers are kind 1111 NIP-22 replies naming the request. For a single selection,
2121 2222 ## Long task cards2323 24-A long task is a kind 43001 request that names the agent key in a `p` tag and carries the room in `h`. The Hermes adapter accepts a request from an allowed user in a room it watches, runs it as one turn in a thread anchored at the request (or at the root the request names in a marked `e` tag) and answers with the task kinds. Every answer carries `e` with the request id, `p` with the requester's key and `h` with the room, so clients derive the card state from the newest events.24+A long task is a kind 43001 request that names the agent key in a `p` tag and carries the room in `h`. The Hermes adapter accepts a request from an allowed user in a room it watches, runs it as one turn in a thread anchored at the request (or at the root the request names in a marked `e` tag) and answers with the task kinds. Each request runs in its own Hermes session, so requests that share a thread run as separate turns and each keeps its own card; replies to a request in an existing thread carry that thread's root and name the request in a marked reply `e` tag. Every answer carries `e` with the request id, `p` with the requester's key and `h` with the room, so clients derive the card state from the newest events.2525 2626 | Kind | When the adapter publishes it | Content |2727 | --- | --- | --- |2828 | 43002 accepted | Once, when Hermes starts processing the request, not on receipt. A queued request is accepted when its turn begins. | Empty. |2929 | 43003 progress | While the turn runs, at most once every 15 seconds per task. Tool calls are never published one by one. | The newest line of intermediate assistant text, then the newest tool line, each clipped to 200 characters. Nothing is published when nothing changed. |30-| 43004 result | Once, when the turn ends successfully. | The first line of the final reply, clipped to 200 characters, with an extra `e` tag naming the thread message that holds the full text. Posted files add a `Posted name, name` line and one `e` tag per file message. A turn that posted nothing carries the text Hermes returned, else `Done`. |31-| 43006 error | Once, when the turn fails or is stopped without a requester cancel. | The error text Hermes exposes, `The task failed` when there is none, `Cancelled` when Hermes stopped the turn, or `Cancelled by <pubkey>` when an allowed user sent `!cancel`. |30+| 43004 result | Once, when the turn ends successfully, or when Hermes reports a failure but the adapter itself posted a reply or a file for the task without a handler error or a failed send. | The first line of the final reply, clipped to 200 characters, with an extra `e` tag naming the thread message that holds the full text. Posted files add a `Posted name, name` line and one `e` tag per file message. A turn that posted nothing carries the text Hermes returned, else `Done`. |31+| 43006 error | Once, when the turn fails or is stopped without a requester cancel. | The error text Hermes exposes, `The task failed` when there is none, `Cancelled` when Hermes stopped the turn, `Cancelled by <pubkey>` when an allowed user sent `!cancel`, or `Merged into task <id>` when Hermes folded the request into another request's turn. |3232 3333 Ordinary replies during the task stay kind 12 messages under the request root, so the room timeline keeps the conversation and the card summarises it. The final text is never repeated in the result; the card links it.3434 35-A task gets exactly one terminal event from the adapter. A kind 43005 cancel from the requester ends the task on its own: the adapter stops the turn and publishes nothing further. A `!cancel` message mentioning the agent in the thread, or in the room while the task runs, stops the turn and ends the task with a 43006 naming the sender. On start, the adapter checks the relay for an existing result, error or cancel before running a request, so a restart does not run a finished task again.35+A task gets exactly one terminal event from the adapter. A result or error the relay refuses is published again after 1 second and once more after 3 seconds; the task stays known until the event is on the relay or the third attempt fails. A kind 43005 cancel from the requester ends the task on its own: the adapter stops the turn and publishes nothing further. A `!cancel` message mentioning the agent in the thread, or in the room while the task runs, stops the turn and ends the task with a 43006 naming the sender. On start, the adapter checks the relay for an existing result, error or cancel before running a request, so a restart does not run a finished task again.3636 3737 Hermes handoffs open a kind 11 thread in the home room whose content is the handoff name; the handed-off session continues there as kind 12 replies.@@ -35,7 +35,9 @@ By default, Hermes responds when an allowed user mentions the agent or replies t
3535 3636 `TINY_PRIVATE_KEY` is read from the environment by the helper and is never a command-line argument. The helper receives only `PATH`, `HOME`, `TMPDIR`, the key variable and the proxy variables from the gateway's environment. Give that key membership and an agent grant for the selected rooms on Tiny, including attachment access. Use a relay with the [native interaction extension](../docs/extensions/tinyagent.md) for interactive cards. Hermes loads the adapter through its normal plugin system. Attachments use authenticated Tinyrelay endpoints, with size and hash checks.3737 38-One gateway runs one Tiny identity at a time. On connect, the adapter takes an exclusive lock for the relay and key under `$HERMES_HOME/state/tinyagent/` and releases it on disconnect. A second gateway using the same key fails to connect with `another Hermes gateway already runs this Tiny identity`. The same directory keeps the delivery cursor, so a restart resumes where the previous run stopped instead of replaying old messages.38+One gateway runs one Tiny identity at a time. On connect, the adapter takes an exclusive lock for the relay and key and releases it on disconnect. The lock lives under `state/tinyagent/locks/` in the Hermes root shared by every profile of an installation (the parent of `profiles/`, or the Hermes home itself), with `tinyagent-locks/` in the system temporary directory as the fallback when that root is not writable, so two profiles cannot run the same key either. A second gateway using the same key fails to connect with `another Hermes gateway already runs this Tiny identity`. Each profile keeps its own delivery cursor under `$HERMES_HOME/state/tinyagent/`, so a restart resumes where the previous run stopped instead of replaying old messages. A message counts as delivered only once Hermes has started or queued a turn for it; a message Hermes refused stays eligible for redelivery for three attempts.39+40+In a multiplexed gateway, the adapter reads `TINY_PRIVATE_KEY`, `TINY_ALLOWED_USERS` and the other `TINY_*` settings through Hermes's profile secret scope, so each profile hands the helper its own key and operator list; the key still reaches the helper only through its environment variable.3941 4042 If the helper process exits, Hermes reports the Tiny platform as disconnected and the adapter restarts the helper on its own, waiting 1 second before the first attempt and doubling the wait up to 30 seconds between failures. Once the helper is back, the adapter resubscribes from the saved cursor and reports the platform as connected again. Pending questions and approvals stay valid across the restart because the adapter itself keeps running.4143 @@ -52,7 +54,7 @@ tinyagent diagnose --relay https://relay.example.test --rooms room-id --kinds 9,
5254 5355 ## Long tasks5456 55-A member can hand Hermes a long task with a kind 43001 request that names the agent key in a `p` tag. The adapter treats the request like a mention: the subject and content become one turn, and the conversation runs as a thread anchored at the request (or at the thread the request names). Hermes's replies stay in that thread; the task card summarises them. The adapter publishes accepted, progress, result and error events as the turn runs, and a request that already has a result, error or cancel on the relay is not run again after a restart. The requester ends a task with a kind 43005 cancel; any allowed user can stop the current turn of a room or thread by sending `!cancel` with the agent mentioned. See [Tiny agent interactions](../docs/extensions/tinyagent.md#long-task-cards) for the events the adapter publishes and when.57+A member can hand Hermes a long task with a kind 43001 request that names the agent key in a `p` tag. The adapter treats the request like a mention: the subject and content become one turn, and the conversation runs as a thread anchored at the request (or at the thread the request names). Every request gets its own Hermes session, so several requests in one thread run as separate turns and are never merged into one. Hermes's replies stay in that thread; the task card summarises them. The adapter publishes accepted, progress, result and error events as the turn runs, retries a failed result or error publication twice before giving up, and does not run again a request that already has a result, error or cancel on the relay after a restart. The requester ends a task with a kind 43005 cancel; any allowed user can stop the current turn of a room or thread by sending `!cancel` with the agent mentioned. See [Tiny agent interactions](../docs/extensions/tinyagent.md#long-task-cards) for the events the adapter publishes and when.5658 5759 ## Relay tools in Hermes5860 @@ -12,7 +12,6 @@ import json
1212 import logging1313 import mimetypes1414 import hashlib15-import os1615 import re1716 import time1817 from datetime import datetime@@ -24,14 +23,14 @@ from gateway.platforms.base import BasePlatformAdapter, SendResult
2423 from gateway.platforms.event import MessageEvent, MessageType2524 2625 try:27- from .client import TinyRPC26+ from .client import TinyRPC, scoped_secret2827 from .interactions import InteractionMixin2928 from .media import incoming_media3029 from .ingress import EventTracker, IdentityLock, event_thread, reply_parent3130 from . import tasks as taskmod3231 from .tasks import TaskState, answer_tags, request_root, request_text3332 except ImportError: # Hermes plugin loader may import adapter.py as a top-level module.34- from client import TinyRPC33+ from client import TinyRPC, scoped_secret3534 from interactions import InteractionMixin3635 from media import incoming_media3736 from ingress import EventTracker, IdentityLock, event_thread, reply_parent@@ -78,20 +77,24 @@ class TinyAdapter(InteractionMixin, BasePlatformAdapter):
7877 # Long-task cards: at most one 43003 progress event per task in this many seconds.7978 PROGRESS_INTERVAL = 15.08079 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)8182 8283 def __init__(self, config: PlatformConfig):8384 super().__init__(config, Platform("tiny"))8485 extra = config.extra or {}85- self.relay_url = str(extra.get("relay_url") or os.environ.get("TINY_RELAY_URL", "")).strip()86- self.cli_path = str(extra.get("cli_path") or os.environ.get("TINY_CLI_PATH", "tinyagent"))87- 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()]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()]8889 home = extra.get("home_channel") or getattr(getattr(config, "home_channel", None), "chat_id", None) \89- or os.environ.get("TINY_HOME_CHANNEL", "")90+ or scoped_secret("TINY_HOME_CHANNEL")9091 if isinstance(home, dict):9192 home = home.get("chat_id", "")9293 self.home_room = str(home).strip() or (self.rooms[0] if self.rooms else "")93- self.allowed = {x.strip().lower() for x in str(extra.get("allowed_users") or os.environ.get("TINY_ALLOWED_USERS", "")).split(",") if x.strip()}94- require_mention = extra.get("require_mention", os.environ.get("TINY_REQUIRE_MENTION", "true"))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"))9598 self.require_mention = str(require_mention).strip().lower() not in {"false", "0", "no", "off"}9699 self.rpc: TinyRPC | None = None97100 self.pubkey = ""@@ -108,8 +111,10 @@ class TinyAdapter(InteractionMixin, BasePlatformAdapter):
108111 self._turn_route = ContextVar("tinyagent_turn_route", default=None)109112 self._lock: IdentityLock | None = None110113 self._closed = False114+ # The helper whose subscription is being set up; its death is reported by _establish.115+ self._establishing: TinyRPC | None = None111116 self._reconnect_task: asyncio.Task | None = None112- self._last_sweep = 0.0117+ self._last_sweep = float("-inf")113118 # request id -> long-task card state, in intake order.114119 self._tasks: dict[str, TaskState] = {}115120 @@ -164,6 +169,11 @@ class TinyAdapter(InteractionMixin, BasePlatformAdapter):
164169 if route.get("room") != room:165170 route = {}166171 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"]167177 root = thread or route.get("root")168178 if reply and not root:169179 # Synthetic sends may refer to messages received before this process.@@ -181,12 +191,26 @@ class TinyAdapter(InteractionMixin, BasePlatformAdapter):
181191 """Start the helper and subscribe. Hermes's watcher calls this on a fresh adapter182192 with is_reconnect=True; helper death inside a live adapter reuses _establish."""183193 self._closed = False184- await self._establish()194+ 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+ raise185202 return True186203 187204 async def _establish(self) -> None:188- """Spawn a helper, prove the identity, take its lock and subscribe from the cursor."""205+ """Spawn a helper, prove the identity, take its lock and subscribe from the cursor.206+207+ 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."""189211 rpc = TinyRPC(self.relay_url, cli_path=self.cli_path, on_exit=self._on_helper_exit)212+ previous = self.rpc213+ self._establishing = rpc190214 try:191215 identity = await rpc.call("identity")192216 pubkey = str((identity or {}).get("pubkey", ""))@@ -205,18 +229,24 @@ class TinyAdapter(InteractionMixin, BasePlatformAdapter):
205229 **self._tracker.filter_since()}206230 if self.rooms:207231 filter["#h"] = self.rooms232+ self.rpc = rpc208233 await rpc.subscribe(self._subscription, filter, self._on_event)234+ if not rpc.alive:235+ raise RuntimeError("Tiny helper exited while subscribing")209236 except BaseException:237+ if self.rpc is rpc:238+ self.rpc = previous239+ self._establishing = None210240 await rpc.close()211241 raise212- previous, self.rpc = self.rpc, rpc242+ self._establishing = None213243 if previous is not None and previous is not rpc:214244 await previous.close()215245 self._mark_connected()216246 217247 def _on_helper_exit(self, rpc) -> None:218248 """TinyRPC reports the helper died on its own; recover in the background."""219- if self._closed or rpc is not self.rpc:249+ if self._closed or rpc is not self.rpc or rpc is self._establishing:220250 return221251 logger.warning("Tiny helper exited; reconnecting")222252 self._mark_disconnected()@@ -340,10 +370,13 @@ class TinyAdapter(InteractionMixin, BasePlatformAdapter):
340370 self._reply_routes.pop(next(iter(self._reply_routes)))341371 await self._deliver(message, event)342372 343- async def _deliver(self, message: MessageEvent, event: dict) -> None:344- """Hand a turn to Hermes. The event counts as delivered only once Hermes has345- accepted it; a failed hand-off leaves it eligible for redelivery until the346- attempt budget is spent."""373+ async def _deliver(self, message: MessageEvent, event: dict) -> bool:374+ """Hand a turn to Hermes and return whether it was admitted.375+376+ 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."""347380 event_id = str(event.get("id", ""))348381 try:349382 await self.handle_message(message)@@ -352,8 +385,28 @@ class TinyAdapter(InteractionMixin, BasePlatformAdapter):
352385 if self._tracker.fail(event):353386 logger.error("Tiny event %s retired after repeated failures", event_id, exc_info=exc)354387 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 False355395 if self._tracker:356396 self._tracker.mark(event)397+ return True398+399+ 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 False357410 358411 # -- long-task cards -------------------------------------------------359412 @@ -373,8 +426,11 @@ class TinyAdapter(InteractionMixin, BasePlatformAdapter):
373426 self._tracker.mark(event)374427 return375428 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).376432 source = self.build_source(chat_id=room, chat_name=room, chat_type="group", user_id=author,377- thread_id=root, message_id=request_id)433+ thread_id=request_id, message_id=request_id)378434 message = MessageEvent(text=request_text(event), message_type=MessageType.TEXT, source=source,379435 raw_message=event, message_id=request_id,380436 timestamp=datetime.fromtimestamp(int(event.get("created_at", time.time()))))@@ -387,10 +443,12 @@ class TinyAdapter(InteractionMixin, BasePlatformAdapter):
387443 self._reply_routes[request_id] = {"room": room, "root": root, "reply_to_message_id": request_id,388444 "thread_id": root, "user_id": author}389445 try:390- await self._deliver(message, event)446+ admitted = await self._deliver(message, event)391447 except BaseException:392448 self._forget_task(request_id)393449 raise450+ if not admitted:451+ self._forget_task(request_id)394452 395453 def _forget_task(self, request_id: str) -> None:396454 task = self._tasks.pop(request_id, None)@@ -398,8 +456,19 @@ class TinyAdapter(InteractionMixin, BasePlatformAdapter):
398456 self._stop_progress_timer(task)399457 400458 def _task_for_send(self, room: str, tags: list[list[str]]) -> TaskState | None:401- """Return the running task whose thread a send lands in, if any."""402- root = next((row[1] for row in tags if len(row) >= 4 and row[0] == "e" and row[3] == "root"), 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")403472 if not root:404473 return None405474 matches = [task for task in self._tasks.values()@@ -478,14 +547,27 @@ class TinyAdapter(InteractionMixin, BasePlatformAdapter):
478547 self._schedule_progress(task)479548 480549 async def _publish_terminal(self, task: TaskState, kind: int, content: str, extra: list[list[str]] = ()) -> None:481- """Publish exactly one terminal event for a task, whatever happens to the send."""550+ """Publish exactly one terminal event for a task, retrying a failed send.551+552+ The task stays known until the event is on the relay or the attempts are553+ spent, so nothing else can end it twice."""482554 if not task.end():483555 return484556 self._stop_progress_timer(task)485- result = await self._publish(task.room, content, [*answer_tags(task), *extra], kind=kind)486- if not result.success:487- logger.error("Tiny task %s terminal event %d failed: %s", task.request_id[:8], kind, result.error)488- self._forget_task(task.request_id)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)489571 490572 async def on_processing_start(self, event: MessageEvent) -> None:491573 await super().on_processing_start(event)@@ -504,15 +586,38 @@ class TinyAdapter(InteractionMixin, BasePlatformAdapter):
504586 return505587 name = str(getattr(outcome, "value", outcome)).lower()506588 if task.terminal:507- self._forget_task(task.request_id)508589 return509- if name == "success":590+ 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):510595 content, extra = task.result()511596 await self._publish_terminal(task, taskmod.KIND_JOB_RESULT, content, extra)512597 elif name == "cancelled":513598 await self._publish_terminal(task, taskmod.KIND_JOB_ERROR, "Cancelled")514599 else:515600 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}")603+604+ def _merged_tasks(self, task: TaskState) -> list[tuple[TaskState, str]]:605+ """Return tasks Hermes folded into another turn of this task's session.606+607+ 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]516621 517622 async def _on_task_cancel(self, event: dict, author: str) -> None:518623 """A 43005 from the requester ends the task; the cancel itself is the terminal event."""@@ -683,25 +788,25 @@ class TinyAdapter(InteractionMixin, BasePlatformAdapter):
683788 684789 685790 def check_requirements() -> bool:686- return bool(os.environ.get("TINY_RELAY_URL") and os.environ.get("TINY_PRIVATE_KEY"))791+ return bool(scoped_secret("TINY_RELAY_URL") and scoped_secret("TINY_PRIVATE_KEY"))687792 688793 689794 def validate_config(config: PlatformConfig) -> bool:690795 extra = config.extra or {}691- return bool(extra.get("relay_url") or os.environ.get("TINY_RELAY_URL"))796+ return bool(extra.get("relay_url") or scoped_secret("TINY_RELAY_URL"))692797 693798 694799 def _env_enablement() -> dict | None:695- relay = os.environ.get("TINY_RELAY_URL", "").strip()696- key = os.environ.get("TINY_PRIVATE_KEY", "").strip()800+ relay = scoped_secret("TINY_RELAY_URL").strip()801+ key = scoped_secret("TINY_PRIVATE_KEY").strip()697802 if not relay or not key:698803 return None699804 result = {"relay_url": relay}700805 for env, key_name in (("TINY_ALLOWED_USERS", "allowed_users"), ("TINY_CLI_PATH", "cli_path"),701806 ("TINY_REQUIRE_MENTION", "require_mention")):702- if os.environ.get(env):703- result[key_name] = os.environ[env]704- home = os.environ.get("TINY_HOME_CHANNEL", "").strip()807+ if scoped_secret(env):808+ result[key_name] = scoped_secret(env)809+ home = scoped_secret("TINY_HOME_CHANNEL").strip()705810 if home:706811 # Hermes lifts this dict into a HomeChannel for cron and standalone delivery.707812 result["home_channel"] = {"chat_id": home, "name": "Home"}@@ -711,8 +816,8 @@ def _env_enablement() -> dict | None:
711816 async def _standalone_send(pconfig, chat_id: str, message: str, *, thread_id=None, media_files=None,712817 force_document=False):713818 extra = getattr(pconfig, "extra", {}) or {}714- rpc = TinyRPC(str(extra.get("relay_url") or os.environ.get("TINY_RELAY_URL", "")),715- cli_path=str(extra.get("cli_path") or os.environ.get("TINY_CLI_PATH", "tinyagent")))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")))716821 try:717822 await rpc.connect()718823 adapter = TinyAdapter(pconfig)@@ -15,6 +15,29 @@ _HELPER_ENV = ("PATH", "HOME", "TMPDIR", "HTTP_PROXY", "HTTPS_PROXY", "NO_PROXY"
1515 "http_proxy", "https_proxy", "no_proxy")1616 1717 18+def scoped_secret(name: str, default: str = "", *, strict: bool = False) -> str:19+ """Resolve a TINY_* setting through Hermes's profile secret scope.20+21+ A multiplexed gateway serves several profiles from one process and keeps each22+ profile's ``.env`` in a context-local scope rather than in ``os.environ``; reading23+ through ``get_secret`` hands a profile its own key. Outside Hermes, or on a Hermes24+ without the scope API, the process environment is the source. ``strict`` lets the25+ fail-closed error of an unscoped read in multiplex mode propagate; passive probes26+ use the default instead.27+ """28+ try:29+ from agent.secret_scope import get_secret30+ except ImportError:31+ return os.environ.get(name, default)32+ try:33+ value = get_secret(name, default)34+ except RuntimeError:35+ if strict:36+ raise37+ return default38+ return default if value is None else str(value)39+40+1841 class TinyRPC:1942 def __init__(self, relay_url: str, key_env: str = "TINY_PRIVATE_KEY", cli_path: str = "tinyagent",2043 on_exit: Callable[["TinyRPC"], Any] | None = None):@@ -34,9 +57,15 @@ class TinyRPC:
3457 self._exited = False3558 3659 def helper_env(self) -> dict[str, str]:37- """Return the scoped environment handed to the helper process."""38- keep = set(_HELPER_ENV) | {self.key_env}39- return {name: value for name, value in os.environ.items() if name in keep}60+ """Return the scoped environment handed to the helper process.61+62+ The signing key is resolved through the active profile scope and reaches the63+ helper only through its environment variable."""64+ env = {name: value for name, value in os.environ.items() if name in _HELPER_ENV}65+ key = scoped_secret(self.key_env, "", strict=True)66+ if key:67+ env[self.key_env] = key68+ return env4069 4170 4271 def alive(self) -> bool:@@ -20,6 +20,8 @@ from typing import Iterable
2020 _EVENT_ID = re.compile(r"^[0-9a-f]{64}$", re.I)2121 _REPLAY_WINDOW = 3002222 _FUTURE_SKEW = 30023+# Hard ceiling on remembered IDs, far above anything the replay window holds in practice.24+MAX_SEEN = 1000002325 # A poison event is retired after this many failed hand-offs to Hermes.2426 MAX_ATTEMPTS = 32527 _MAX_ATTEMPT_ENTRIES = 256@@ -43,10 +45,44 @@ def state_root(state_dir: str | Path | None = None) -> Path:
4345 4446 4547 def adapter_state_dir(root: str | Path | None = None) -> Path:46- """Return the directory holding tracker state and identity locks."""48+ """Return the directory holding tracker state for the active Hermes home."""4749 return state_root(root) / "state" / "tinyagent"4850 4951 52+def shared_root() -> Path:53+ """Return the Hermes root every profile shares: the parent of ``profiles`` when the54+ process home is a named profile, else the process home itself."""55+ try:56+ from hermes_constants import get_default_hermes_root57+ return Path(get_default_hermes_root()).expanduser()58+ except Exception:59+ home = Path(os.environ.get("HERMES_HOME", "").strip() or "~/.hermes").expanduser()60+ return home.parent.parent if home.parent.name == "profiles" else home61+62+63+def lock_dir(root: str | Path | None = None) -> Path:64+ """Return the directory holding identity locks.65+66+ Locks live under the shared Hermes root so two profiles cannot run the same67+ identity. When that root cannot be written, the system temporary directory68+ holds them instead.69+ """70+ if root:71+ return adapter_state_dir(root) / "locks"72+ candidate = shared_root() / "state" / "tinyagent" / "locks"73+ try:74+ candidate.mkdir(parents=True, exist_ok=True)75+ os.chmod(candidate, 0o700)76+ if os.access(candidate, os.W_OK):77+ return candidate78+ except OSError:79+ pass80+ fallback = Path(tempfile.gettempdir()) / "tinyagent-locks"81+ fallback.mkdir(parents=True, exist_ok=True)82+ logger.warning("Tiny identity locks use %s because %s is not writable", fallback, candidate)83+ return fallback84+85+5086 def _tag_values(event: dict, name: str) -> list[list[str]]:5187 return [tag for tag in event.get("tags", []) if isinstance(tag, list) and len(tag) >= 2 and tag[0] == name]5288 @@ -62,16 +98,21 @@ def _expiration(event: dict) -> int | None:
6298 6399 64100 class EventTracker:65- """Bounded persistent event cursor and ID set for one Tiny subscription.101+ """Persistent event cursor and ID set for one Tiny subscription.66102 67103 State file (JSON): ``floor`` and ``cursor`` are Unix seconds; ``seen`` maps a68104 delivered event ID to its ``created_at``; ``attempts`` maps an event ID to the69105 number of failed hand-offs. Older files stored ``seen`` as a list of IDs and70106 load with every listed ID pinned to the cursor.107+108+ ``seen`` keeps every ID inside the replay window (``cursor - 300`` seconds, never109+ below ``floor``); the window is what bounds correctness, so a burst of events in110+ one second is never forgotten. ``max_seen`` is only a hard ceiling: when it is hit111+ the oldest seconds are retired, the floor moves past them and a warning is logged.71112 """72113 73114 def __init__(self, relay: str, identity: str, rooms: Iterable[str], *, state_dir: str | Path | None = None,74- now: int | None = None, max_seen: int = 10000):115+ now: int | None = None, max_seen: int = MAX_SEEN):75116 self.relay, self.identity = str(relay), str(identity).lower()76117 self.rooms = tuple(sorted(str(room) for room in rooms))77118 key = hashlib.sha256((self.relay + "\0" + self.identity + "\0" + "\0".join(self.rooms)).encode()).hexdigest()[:24]@@ -127,7 +168,7 @@ class EventTracker:
127168 return max(self.floor, self.cursor - _REPLAY_WINDOW)128169 129170 def _prune(self) -> None:130- """Drop IDs the window already rejects; raise the floor only when still over budget."""171+ """Drop IDs the window already rejects; the hard ceiling alone may raise the floor."""131172 start = self._window_start()132173 self.seen = {key: stamp for key, stamp in self.seen.items() if stamp >= start}133174 if len(self.seen) > self.max_seen:@@ -138,7 +179,7 @@ class EventTracker:
138179 threshold += 1139180 self.seen = {key: stamp for key, stamp in self.seen.items() if stamp >= threshold}140181 self.floor = max(self.floor, threshold)141- logger.warning("Tiny replay cache exceeded %d IDs; events before %d are now ignored",182+ logger.warning("Tiny replay cache hit its hard ceiling of %d IDs; events before %d are now ignored",142183 self.max_seen, self.floor)143184 self.attempts = {key: count for key, count in self.attempts.items() if key not in self.seen}144185 while len(self.attempts) > _MAX_ATTEMPT_ENTRIES:@@ -204,11 +245,15 @@ class EventTracker:
204245 205246 206247 class IdentityLock:207- """Exclusive per-identity lock so one gateway consumes a Tiny key at a time."""248+ """Exclusive per-identity lock so one gateway consumes a Tiny key at a time.249+250+ The lock file is keyed by relay and public key and lives in ``lock_dir()``, which251+ every profile of one Hermes installation shares.252+ """208253 209254 def __init__(self, relay: str, pubkey: str, *, state_dir: str | Path | None = None):210255 key = hashlib.sha256((str(relay) + "\0" + str(pubkey).lower()).encode()).hexdigest()[:24]211- self.path = adapter_state_dir(state_dir) / f"{key}.lock"256+ self.path = lock_dir(state_dir) / f"{key}.lock"212257 self._fd: int | None = None213258 214259 @@ -266,5 +311,5 @@ def is_reply_to_own(event: dict, own_ids: set[str]) -> bool:
266311 return any(tag[1] in own_ids for tag in _tag_values(event, "e"))267312 268313 269-__all__ = ["EventTracker", "IdentityLock", "MAX_ATTEMPTS", "event_thread", "reply_parent", "is_reply_to_own",270- "state_root", "adapter_state_dir"]314+__all__ = ["EventTracker", "IdentityLock", "MAX_ATTEMPTS", "MAX_SEEN", "event_thread", "reply_parent",315+ "is_reply_to_own", "state_root", "adapter_state_dir", "shared_root", "lock_dir"]@@ -125,6 +125,11 @@ class TaskState:
125125 if text:126126 self.error = text127127 128+129+ def delivered(self) -> bool:130+ """True once a reply or a file of this task reached the room."""131+ return any(cls in (TEXT, FILE) for cls in self.classes.values())132+128133 # -- progress --------------------------------------------------------129134 130135 @@ -20,6 +20,11 @@ BOT = "a" * 64
2020 OTHER = "c" * 642121 2222 23+def admit(message):24+ """What Hermes's handle_message records once it starts or queues a turn."""25+ message._gateway_accepted = True26+27+2328 class FakeRPC:2429 def __init__(self):2530 self.events = []@@ -475,7 +480,9 @@ def test_helper_death_reconnects_from_persisted_cursor(adapter, helper, tmp_path
475480 # A tracker with history: the floor is old and the cursor moves with delivered events.476481 instance._tracker = EventTracker("http://relay", BOT, ["room"], now=now - 1000)477482 received = []478- async def handle(message): received.append(message)483+ async def handle(message):484+ received.append(message)485+ admit(message)479486 instance.handle_message = handle480487 481488 async def run():@@ -587,7 +594,9 @@ def test_event_is_marked_only_after_hermes_accepts_it(adapter, tmp_path):
587594 asyncio.run(adapter._on_event(poison))588595 assert len(attempts) == 3589596 handled = []590- async def accept(message): handled.append(message.message_id)597+ async def accept(message):598+ handled.append(message.message_id)599+ admit(message)591600 adapter.handle_message = accept592601 good = {**poison, "id": "e" * 64, "content": "hello"}593602 asyncio.run(adapter._on_event(good))@@ -597,6 +606,275 @@ def test_event_is_marked_only_after_hermes_accepts_it(adapter, tmp_path):
597606 assert json.loads(adapter._tracker.path.read_text())["seen"] == {poison["id"]: now, good["id"]: now}598607 599608 609+def test_event_refused_without_an_admission_receipt_stays_deliverable(adapter, tmp_path):610+ # Hermes returns normally but records no receipt: no turn was started or queued611+ # (no handler installed, for example). The event is a failed attempt, not seen.612+ from ingress import EventTracker613+ now = int(time.time())614+ adapter._tracker = EventTracker("http://relay", BOT, ["room"], state_dir=tmp_path, now=now - 10)615+ event = {"kind": 9, "id": "d" * 64, "pubkey": OWNER, "created_at": now, "content": "hello",616+ "tags": [["h", "room"], ["p", BOT]]}617+ calls = []618+ async def refused(message):619+ calls.append(message.message_id)620+ adapter.handle_message = refused621+ asyncio.run(adapter._on_event(event))622+ assert calls == [event["id"]]623+ assert adapter._tracker.should_accept(event) and adapter._tracker.attempts == {event["id"]: 1}624+ asyncio.run(adapter._on_event(event))625+ asyncio.run(adapter._on_event(event))626+ assert len(calls) == 3 and not adapter._tracker.should_accept(event)627+ # The real handle_message with no gateway handler installed is exactly that case.628+ adapter.handle_message = TinyAdapter.handle_message.__get__(adapter)629+ adapter._message_handler = None630+ other = {**event, "id": "e" * 64}631+ asyncio.run(adapter._on_event(other))632+ assert adapter._tracker.attempts == {other["id"]: 1} and adapter._tracker.should_accept(other)633+ # A slash command during a busy session is dispatched inline without a receipt.634+ async def inline(message):635+ calls.append(message.message_id)636+ adapter.handle_message = inline637+ command = {**event, "id": "f" * 64, "content": "/new"}638+ asyncio.run(adapter._on_event(command))639+ assert not adapter._tracker.should_accept(command) and command["id"] not in adapter._tracker.attempts640+ # An older Hermes without the receipt admits by returning.641+ async def legacy(message):642+ message._gateway_accepted = None643+ adapter.handle_message = legacy644+ old = {**event, "id": "1" * 64}645+ asyncio.run(adapter._on_event(old))646+ assert not adapter._tracker.should_accept(old)647+648+649+def test_task_request_refused_by_hermes_is_not_kept(adapter, tmp_path):650+ from ingress import EventTracker651+ now = int(time.time())652+ adapter._tracker = EventTracker("http://relay", BOT, ["room"], state_dir=tmp_path, now=now - 10)653+ async def refused(message):654+ pass655+ adapter.handle_message = refused656+ event = {"kind": 43001, "id": "d" * 64, "pubkey": OWNER, "created_at": now, "content": "Run it",657+ "tags": [["h", "room"], ["p", BOT]]}658+ asyncio.run(adapter._on_event(event))659+ assert adapter._tasks == {} and adapter._tracker.should_accept(event)660+ assert adapter._tracker.attempts == {event["id"]: 1}661+662+663+class RacingProcess:664+ """A helper that delivers subscription frames before acknowledging the subscribe call,665+ which the Go helper's scheduling permits."""666+667+ def __init__(self, *, events=(), subscribe_error=None, die_on_subscribe=False):668+ from test_client import FakeProcess669+ self.inner = FakeProcess()670+ self.stdout, self.stdin = self.inner.stdout, self.inner.stdin671+ self.stdin.process = self672+ self.events, self.subscribe_error, self.die_on_subscribe = list(events), subscribe_error, die_on_subscribe673+ self.terminated = False674+675+676+ def returncode(self):677+ return self.inner.returncode678+679+ def on_request(self, request):680+ method = request["method"]681+ if method == "identity":682+ reply = {"id": request["id"], "result": {"pubkey": BOT}}683+ elif method == "query":684+ reply = {"id": request["id"], "result": []}685+ elif method == "subscribe" and self.subscribe_error:686+ reply = {"id": request["id"], "error": {"message": self.subscribe_error}}687+ else:688+ reply = {"id": request["id"], "result": {}}689+ if method != "subscribe":690+ self.stdout.feed_data((json.dumps(reply) + "\n").encode())691+ return692+ name = request["params"]["subscription"]693+ frames = [{"event": "connected", "subscription": name}]694+ frames += [{"event": "event", "subscription": name, "data": event} for event in self.events]695+ self.stdout.feed_data("".join(json.dumps(frame) + "\n" for frame in frames).encode())696+ if self.die_on_subscribe:697+ self.inner.die()698+ return699+ asyncio.get_running_loop().call_later(0.01, self.stdout.feed_data, (json.dumps(reply) + "\n").encode())700+701+ def terminate(self):702+ self.terminated = True703+ self.inner.terminate()704+705+ async def wait(self):706+ return self.inner.returncode707+708+709+class Racing:710+ """Plans the next helper process; the process itself is built inside the running loop."""711+712+ def __init__(self):713+ self.kwargs = {}714+ self.process = None715+716+ def __call__(self, **kwargs):717+ self.kwargs = kwargs718+ return self719+720+721+ def terminated(self):722+ return self.process is not None and self.process.terminated723+724+725+ def returncode(self):726+ return None if self.process is None else self.process.returncode727+728+729+730+def racing(monkeypatch):731+ import client as module732+ plan = Racing()733+ async def create(*args, **kwargs):734+ plan.process = RacingProcess(**plan.kwargs)735+ return plan.process736+ monkeypatch.setattr(module.asyncio, "create_subprocess_exec", create)737+ return plan738+739+740+def test_events_delivered_before_the_subscribe_acknowledgment_reach_the_session(adapter, racing):741+ now = int(time.time())742+ request = {"kind": 43001, "id": "d" * 64, "pubkey": OWNER, "created_at": now, "content": "Run it",743+ "tags": [["h", "room"], ["p", BOT]]}744+ hello = {"kind": 9, "id": "e" * 64, "pubkey": OWNER, "created_at": now, "content": "hello",745+ "tags": [["h", "room"], ["p", BOT]]}746+ process = racing(events=[request, hello])747+ instance = live_adapter()748+ received, errors = [], []749+ async def handle(message):750+ received.append(message.message_id)751+ admit(message)752+ instance.handle_message = handle753+ original = instance._on_event754+ async def guarded(event):755+ try:756+ await original(event)757+ except Exception as exc:758+ errors.append(f"{type(exc).__name__}: {exc}")759+ raise760+ instance._on_event = guarded761+762+ async def run():763+ await instance.connect()764+ for _ in range(20):765+ await asyncio.sleep(0)766+ assert errors == [] and sorted(received) == sorted([request["id"], hello["id"]])767+ assert instance.is_connected and instance.rpc is not None and instance.rpc.alive768+ assert instance._tracker.attempts == {} and set(instance._tracker.seen) == {request["id"], hello["id"]}769+ assert "d" * 64 in instance._tasks770+ await instance.disconnect()771+ assert process.terminated772+ asyncio.run(run())773+774+775+776+def test_failed_subscribe_rolls_the_session_back(adapter, racing, failure):777+ now = int(time.time())778+ hello = {"kind": 9, "id": "e" * 64, "pubkey": OWNER, "created_at": now, "content": "hello",779+ "tags": [["h", "room"], ["p", BOT]]}780+ process = racing(events=[hello], subscribe_error="relay refused" if failure == "refused" else None,781+ die_on_subscribe=failure == "died")782+ instance = live_adapter()783+ received = []784+ async def handle(message):785+ received.append(message.message_id)786+ admit(message)787+ instance.handle_message = handle788+789+ async def run():790+ with pytest.raises(RuntimeError):791+ await instance.connect()792+ for _ in range(10):793+ await asyncio.sleep(0)794+ # The first connect owns nothing afterwards: no helper, no reconnect loop, no lock.795+ assert instance.rpc is None and not instance.is_connected and instance._lock is None796+ assert instance._reconnect_task is None and instance._establishing is None797+ assert process.terminated or process.returncode is not None798+ # The event the helper pushed before failing was handled with the helper it came from.799+ assert received == [hello["id"]] or received == []800+ await instance.disconnect()801+ asyncio.run(run())802+803+804+def test_reconnect_keeps_the_previous_helper_reference_until_the_new_one_subscribes(adapter, helper):805+ instance = live_adapter()806+807+ async def run():808+ await instance.connect()809+ first = helper.instances[-1]810+ first.die()811+ helper.fail_identity = True812+ for _ in range(8):813+ await asyncio.sleep(0)814+ # Every failed attempt leaves the dead helper as the session's reference, so a815+ # late callback from it is refused as stale rather than crashing on None.816+ assert instance.rpc is first and not instance.is_connected817+ helper.fail_identity = False818+ await asyncio.wait_for(instance._reconnect_task, 2)819+ assert instance.rpc is helper.instances[-1] and instance.rpc is not first and instance.is_connected820+ await instance.disconnect()821+ asyncio.run(run())822+823+824+def test_profile_scope_hands_the_helper_its_own_key(adapter, monkeypatch):825+ from agent.secret_scope import reset_secret_scope, set_multiplex_active, set_secret_scope826+ import adapter as module827+ from client import TinyRPC828+ monkeypatch.setenv("TINY_PRIVATE_KEY", "default-profile-key")829+ monkeypatch.setenv("TINY_ALLOWED_USERS", OTHER)830+ monkeypatch.setenv("TINY_RELAY_URL", "http://default")831+ set_multiplex_active(True)832+ token = set_secret_scope({"TINY_PRIVATE_KEY": "secondary-profile-key", "TINY_ALLOWED_USERS": OWNER,833+ "TINY_RELAY_URL": "http://secondary"})834+ try:835+ env = TinyRPC("http://secondary").helper_env()836+ assert env["TINY_PRIVATE_KEY"] == "secondary-profile-key"837+ assert set(env) <= {"PATH", "HOME", "TMPDIR", "TINY_PRIVATE_KEY", "HTTP_PROXY", "HTTPS_PROXY", "NO_PROXY",838+ "http_proxy", "https_proxy", "no_proxy"}839+ scoped = TinyAdapter(PlatformConfig(enabled=True, extra={}))840+ assert scoped.allowed == {OWNER} and scoped.relay_url == "http://secondary"841+ assert module._env_enablement()["allowed_users"] == OWNER842+ finally:843+ reset_secret_scope(token)844+ try:845+ # Multiplexing with no scope fails closed for the key; passive probes stay quiet.846+ with pytest.raises(RuntimeError):847+ TinyRPC("http://secondary").helper_env()848+ assert TinyAdapter(PlatformConfig(enabled=True, extra={})).allowed == set()849+ assert module.check_requirements() is False850+ finally:851+ set_multiplex_active(False)852+ plain = TinyAdapter(PlatformConfig(enabled=True, extra={}))853+ assert plain.allowed == {OTHER} and TinyRPC("http://default").helper_env()["TINY_PRIVATE_KEY"] == "default-profile-key"854+855+856+def test_identity_lock_is_shared_by_profile_homes_of_one_installation(adapter, helper, tmp_path, monkeypatch):857+ from hermes_constants import reset_hermes_home_override, set_hermes_home_override858+ monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes"))859+ first, second = live_adapter(), live_adapter()860+861+ async def run():862+ token = set_hermes_home_override(tmp_path / "hermes" / "profiles" / "a")863+ try:864+ await first.connect()865+ finally:866+ reset_hermes_home_override(token)867+ token = set_hermes_home_override(tmp_path / "hermes" / "profiles" / "b")868+ try:869+ with pytest.raises(RuntimeError, match="another Hermes gateway already runs this Tiny identity"):870+ await second.connect()871+ finally:872+ reset_hermes_home_override(token)873+ assert first._lock.path.parent == tmp_path / "hermes" / "state" / "tinyagent" / "locks"874+ await first.disconnect()875+ asyncio.run(run())876+877+600878 def test_env_enablement_seeds_a_home_channel(adapter, monkeypatch):601879 import adapter as module602880 from gateway.config import GatewayConfig, HomeChannel, Platform@@ -1,5 +1,6 @@
11 import json22 import hashlib3+import os34 import subprocess45 import sys56 import textwrap@@ -32,18 +33,35 @@ def test_persisted_cursor_overlaps_timestamp_and_deduplicates(tmp_path):
3233 assert not second.should_accept(event("one", 101))3334 3435 35-def test_expiration_and_bounded_seen(tmp_path):36+def test_expiration_and_the_hard_ceiling(tmp_path, caplog):3637 tracker = EventTracker("r", "a", ["lab"], state_dir=tmp_path, now=100, max_seen=2)3738 assert not tracker.should_accept(event("expired", 100, [["expiration", "100"]]))38- for index in range(3):39- tracker.mark(event(str(index), 101 + index))40- # The persisted set stays bounded. Over budget, the oldest second (101) is retired41- # and the floor moves to the oldest retained second (102), so the two newest IDs42- # remain deduplicated instead of being forgotten.39+ with caplog.at_level("WARNING", logger="ingress"):40+ for index in range(3):41+ tracker.mark(event(str(index), 101 + index))42+ # Only the hard ceiling retires IDs. Over it, the oldest second (101) is retired,43+ # the floor moves to the oldest retained second (102) and the hit is logged, so the44+ # two newest IDs remain deduplicated instead of being forgotten.4345 state = json.loads(tracker.path.read_text())4446 assert len(state["seen"]) == 24547 assert set(state["seen"].values()) == {102, 103}4648 assert state["floor"] == 10249+ assert any("hard ceiling of 2" in record.getMessage() for record in caplog.records)50+51+52+def test_same_second_burst_stays_inside_the_window(tmp_path):53+ # The replay window bounds correctness: every ID in it is kept whatever the54+ # burst size, so an unseen event from a busy second is still accepted and the55+ # seen ones are still rejected. The default ceiling sits far above the window.56+ assert ingress.MAX_SEEN == 10000057+ tracker = EventTracker("r", "a", ["lab"], state_dir=tmp_path, now=100)58+ for index in range(4):59+ tracker.mark(event(f"burst-{index}", 101))60+ assert tracker.floor == 100 and len(tracker.seen) == 461+ assert tracker.should_accept(event("unseen", 101), now=101)62+ assert all(not tracker.should_accept(event(f"burst-{index}", 101), now=101) for index in range(4))63+ restored = EventTracker("r", "a", ["lab"], state_dir=tmp_path, now=101)64+ assert len(restored.seen) == 4 and not restored.should_accept(event("burst-0", 101), now=101)4765 4866 4967 def test_nip10_root_reply_and_own_parent():@@ -162,6 +180,48 @@ def test_state_root_uses_hermes_active_home(tmp_path, monkeypatch):
162180 assert ingress.state_root() == tmp_path / "profile"163181 164182 183+def test_identity_lock_is_shared_across_profile_homes(tmp_path, monkeypatch):184+ # Two profiles of one Hermes installation resolve the same lock file, so they185+ # cannot run the same identity at once; tracker state stays per profile home.186+ root = tmp_path / "hermes"187+ monkeypatch.setenv("HERMES_HOME", str(root / "profiles" / "a"))188+ first = IdentityLock("http://relay", "a" * 64)189+ tracker_a = EventTracker("r", "a" * 64, ["lab"], now=100)190+ monkeypatch.setenv("HERMES_HOME", str(root / "profiles" / "b"))191+ second = IdentityLock("http://relay", "a" * 64)192+ tracker_b = EventTracker("r", "a" * 64, ["lab"], now=100)193+ assert first.path == second.path == root / "state" / "tinyagent" / "locks" / first.path.name194+ assert tracker_a.path.parent == root / "profiles" / "a" / "state" / "tinyagent"195+ assert tracker_b.path.parent == root / "profiles" / "b" / "state" / "tinyagent"196+ first.acquire()197+ with pytest.raises(RuntimeError, match="another Hermes gateway already runs this Tiny identity"):198+ second.acquire()199+ first.release()200+ second.acquire()201+ assert second.held202+ second.release()203+ monkeypatch.setenv("HERMES_HOME", str(root))204+ assert IdentityLock("http://relay", "a" * 64).path == first.path205+206+207+def test_identity_lock_falls_back_to_the_temp_dir_when_the_root_is_read_only(tmp_path, monkeypatch):208+ if hasattr(os, "geteuid") and os.geteuid() == 0:209+ pytest.skip("root can write anywhere")210+ frozen = tmp_path / "frozen"211+ frozen.mkdir()212+ frozen.chmod(0o500)213+ monkeypatch.setattr(ingress, "shared_root", lambda: frozen)214+ monkeypatch.setattr(ingress.tempfile, "gettempdir", lambda: str(tmp_path / "tmp"))215+ try:216+ lock = IdentityLock("http://relay", "a" * 64)217+ assert lock.path.parent == tmp_path / "tmp" / "tinyagent-locks"218+ lock.acquire()219+ assert lock.held220+ lock.release()221+ finally:222+ frozen.chmod(0o700)223+224+165225 def test_identity_lock_is_exclusive_in_process(tmp_path):166226 first = IdentityLock("http://relay", "A" * 64, state_dir=tmp_path)167227 second = IdentityLock("http://relay", "a" * 64, state_dir=tmp_path)@@ -32,6 +32,9 @@ class FakeRPC:
3232 self.rows = []3333 self.terminal = []3434 self.fail_kinds = set()35+ # kind -> number of publishes of that kind still to refuse36+ self.fail_times = {}37+ self.attempts = []3538 3639 async def query(self, filter):3740 if "ids" in filter:@@ -42,7 +45,9 @@ class FakeRPC:
4245 return []4346 4447 async def publish(self, event):45- if event["kind"] in self.fail_kinds:48+ self.attempts.append(event["kind"])49+ if event["kind"] in self.fail_kinds or self.fail_times.get(event["kind"], 0) > 0:50+ self.fail_times[event["kind"]] = self.fail_times.get(event["kind"], 0) - 14651 raise RuntimeError("relay refused the event")4752 result = {"id": f"{len(self.events) + 1:064x}", "created_at": int(time.time()), "pubkey": BOT, **event}4853 self.events.append(result)@@ -96,10 +101,16 @@ async def settle(rounds=5):
96101 await asyncio.sleep(0)97102 98103 104+def admit(message):105+ """What Hermes's handle_message records once it starts or queues a turn."""106+ message._gateway_accepted = True107+108+99109 def capture(adapter):100110 received = []101111 async def handle(message):102112 received.append(message)113+ admit(message)103114 adapter.handle_message = handle104115 return received105116 @@ -115,9 +126,17 @@ def test_request_becomes_a_thread_turn_anchored_at_the_request(adapter):
115126 assert task.requester == OWNER and task.thread_root == REQUEST and not task.accepted116127 assert adapter._session_assignees[task.session_key] == OWNER117128 assert adapter.rpc.events == []129+ # A request inside an existing thread keeps its own session, named by the request,130+ # while its replies sit in that thread under the request.118131 asyncio.run(adapter._on_event(request("e" * 64, root="7" * 64, content="Continue here")))119- assert received[1].text == "Continue here" and received[1].source.thread_id == "7" * 64132+ assert received[1].text == "Continue here" and received[1].source.thread_id == "e" * 64120133 assert adapter._tasks["e" * 64].thread_root == "7" * 64134+ assert adapter._tasks["e" * 64].session_key != task.session_key135+ reply = asyncio.run(adapter.send("room", "On it", reply_to="e" * 64, metadata={"thread_id": "e" * 64}))136+ assert reply.success and adapter.rpc.events[-1]["kind"] == 12137+ assert ["e", "7" * 64, "", "root"] in adapter.rpc.events[-1]["tags"]138+ assert ["e", "e" * 64, "", "reply"] in adapter.rpc.events[-1]["tags"]139+ assert adapter._tasks["e" * 64].classes == {reply.message_id: "chrome"}121140 122141 123142 @@ -196,11 +215,31 @@ def pipeline(adapter, handler):
196215 return await handler(message)197216 adapter.set_message_handler(wrapped)198217 async def process(message):218+ admit(message)199219 await adapter._process_message_background(message, adapter._event_session_key(message))200220 adapter.handle_message = process201221 return seen202222 203223 224+def grade_like_0_21_1(adapter):225+ """Hermes 0.21.1 does not count attachment delivery: a file-only turn grades as failure."""226+ original = adapter.on_processing_complete227+ async def complete(event, outcome):228+ posted_text = any(event["kind"] == 12 and not any(row[0] == "imeta" for row in event["tags"])229+ for event in adapter.rpc.events)230+ if outcome == ProcessingOutcome.SUCCESS and not posted_text:231+ outcome = ProcessingOutcome.FAILURE232+ await original(event, outcome)233+ adapter.on_processing_complete = complete234+235+236+async def finish_turns(adapter, timeout=5):237+ tasks = list(adapter._session_tasks.values())238+ if tasks:239+ await asyncio.wait_for(asyncio.gather(*tasks, return_exceptions=True), timeout)240+ await settle()241+242+204243 def test_success_publishes_one_result_linking_the_thread_reply(adapter):205244 async def handler(message):206245 return "Build passed on main.\nAll 42 tests green."@@ -220,18 +259,43 @@ def test_success_publishes_one_result_linking_the_thread_reply(adapter):
220259 assert len(adapter.rpc.kinds(43004)) == 1221260 222261 223-def test_file_only_result_names_and_links_the_files(adapter, tmp_path):262+263+def test_file_only_result_names_and_links_the_files(adapter, tmp_path, hermes):224264 note = tmp_path / "report.txt"225265 note.write_text("build log")226266 async def handler(message):227267 return f"MEDIA:{note}"228268 pipeline(adapter, handler)269+ if hermes == "0.21.1":270+ grade_like_0_21_1(adapter)229271 asyncio.run(adapter._on_event(request()))230272 files = [event for event in adapter.rpc.events if any(row[0] == "imeta" for row in event["tags"])]231273 assert len(files) == 1 and files[0]["kind"] == 12232274 results = adapter.rpc.kinds(43004)233275 assert len(results) == 1 and results[0]["content"] == "Posted report.txt"234276 assert ["e", files[0]["id"]] in results[0]["tags"]277+ assert adapter.rpc.kinds(43006) == [] and REQUEST not in adapter._tasks278+279+280+281+def test_true_failures_still_publish_an_error(adapter, shape):282+ async def handler(message):283+ return "Build passed."284+ seen = pipeline(adapter, handler)285+ if shape == "nothing-delivered":286+ # Hermes graded the turn as failed and the adapter saw no reply reach the room.287+ original = adapter.on_processing_complete288+ async def complete(event, outcome):289+ adapter._tasks[REQUEST].classes.clear()290+ await original(event, ProcessingOutcome.FAILURE)291+ adapter.on_processing_complete = complete292+ else:293+ adapter.rpc.fail_kinds = {12}294+ asyncio.run(adapter._on_event(request()))295+ errors = adapter.rpc.kinds(43006)296+ assert len(errors) == 1 and adapter.rpc.kinds(43004) == [] and REQUEST not in adapter._tasks297+ expected = "The task failed" if shape == "nothing-delivered" else "relay refused the event"298+ assert errors[0]["content"] == expected235299 236300 237301 def test_failure_publishes_one_error_with_the_exposed_text(adapter):@@ -247,18 +311,105 @@ def test_failure_publishes_one_error_with_the_exposed_text(adapter):
247311 assert adapter.rpc.kinds(43004) == [] and REQUEST not in adapter._tasks248312 249313 250-def test_terminal_publication_happens_once_even_when_the_send_fails(adapter):314+def test_terminal_publication_is_retried_and_happens_once(adapter):251315 async def handler(message):252316 return "Done"253317 seen = pipeline(adapter, handler)254- adapter.rpc.fail_kinds = {43004}318+ adapter.TERMINAL_RETRY_DELAYS = (0, 0)319+ adapter.rpc.fail_times = {43004: 2}255320 asyncio.run(adapter._on_event(request()))256- assert adapter.rpc.kinds(43004) == [] and adapter.rpc.kinds(43006) == []321+ assert adapter.rpc.attempts.count(43004) == 3322+ assert len(adapter.rpc.kinds(43004)) == 1 and adapter.rpc.kinds(43006) == []257323 assert REQUEST not in adapter._tasks258- adapter.rpc.fail_kinds = set()259324 asyncio.run(adapter.on_processing_complete(seen[0], ProcessingOutcome.SUCCESS))260325 asyncio.run(adapter.on_processing_complete(seen[0], ProcessingOutcome.FAILURE))326+ assert len(adapter.rpc.kinds(43004)) == 1 and adapter.rpc.kinds(43006) == []327+328+329+def test_terminal_publication_gives_up_after_three_attempts(adapter):330+ async def handler(message):331+ return "Done"332+ seen = pipeline(adapter, handler)333+ adapter.TERMINAL_RETRY_DELAYS = (0, 0)334+ adapter.rpc.fail_kinds = {43004}335+ forgotten = []336+ original = adapter._forget_task337+ def forget(request_id):338+ forgotten.append(request_id in adapter._tasks)339+ original(request_id)340+ adapter._forget_task = forget341+ asyncio.run(adapter._on_event(request()))342+ assert adapter.rpc.attempts.count(43004) == 3343+ # The task stayed known until the last attempt, then was dropped exactly once.344+ assert forgotten[0] and REQUEST not in adapter._tasks261345 assert adapter.rpc.kinds(43004) == [] and adapter.rpc.kinds(43006) == []346+ adapter.rpc.fail_kinds = set()347+ asyncio.run(adapter.on_processing_complete(seen[0], ProcessingOutcome.SUCCESS))348+ assert adapter.rpc.kinds(43004) == []349+350+351+def test_requests_sharing_a_thread_run_as_separate_turns(adapter):352+ # Three requests in one thread arrive while the first still runs. Each gets its353+ # own session, so Hermes never folds two requests into one turn: three handler354+ # calls, three accepted events, three results, no orphaned card.355+ adapter._busy_text_mode = "queue"356+ adapter._busy_text_debounce_seconds = 0357+ release = asyncio.Event()358+ seen = []359+ async def handler(message):360+ seen.append(message)361+ if len(seen) == 1:362+ await release.wait()363+ return f"Finished {message.text}"364+ adapter.set_message_handler(handler)365+ common = "7" * 64366+ ids = [f"{index + 10:064x}" for index in range(3)]367+ async def run():368+ for index, request_id in enumerate(ids):369+ await adapter._on_event(request(request_id, root=common, content=f"job {index}"))370+ await settle()371+ assert [message.text for message in seen] == ["job 0", "job 1", "job 2"]372+ assert adapter._pending_messages == {}373+ release.set()374+ await finish_turns(adapter)375+ asyncio.run(run())376+ assert sorted(tag(event, "e") for event in adapter.rpc.kinds(43002)) == ids377+ results = adapter.rpc.kinds(43004)378+ assert sorted(tag(event, "e") for event in results) == ids379+ assert adapter.rpc.kinds(43006) == [] and adapter._tasks == {}380+ replies = [event for event in adapter.rpc.events if event["kind"] == 12]381+ assert len(replies) == 3382+ assert all(["e", common, "", "root"] in event["tags"] for event in replies)383+ assert sorted(row[1] for event in replies for row in event["tags"] if len(row) > 3 and row[3] == "reply") == ids384+ for result in results:385+ reply = next(event for event in replies if ["e", tag(result, "e"), "", "reply"] in event["tags"])386+ assert ["e", reply["id"]] in result["tags"] and result["content"] == reply["content"]387+388+389+def test_requests_merged_by_hermes_end_with_an_error_naming_the_carrier(adapter):390+ # Safety net for a Hermes that still coalesces two requests into one turn: the391+ # request that lost its turn gets an error card naming the turn that carried it.392+ received = capture(adapter)393+ ids = [f"{index + 10:064x}" for index in range(3)]394+ async def run():395+ for index, request_id in enumerate(ids):396+ await adapter._on_event(request(request_id, content=f"job {index}"))397+ first, second, third = (adapter._tasks[request_id] for request_id in ids)398+ second.session_key = third.session_key = first.session_key399+ await adapter.on_processing_start(received[0])400+ # The third request is still queued under the session: it keeps its card.401+ adapter._pending_messages[first.session_key] = received[2]402+ await adapter.on_processing_complete(received[0], ProcessingOutcome.SUCCESS)403+ errors = adapter.rpc.kinds(43006)404+ assert [tag(event, "e") for event in errors] == [ids[1]]405+ assert errors[0]["content"] == f"Merged into task {ids[2]}"406+ assert [tag(event, "e") for event in adapter.rpc.kinds(43004)] == [ids[0]]407+ assert ids[0] not in adapter._tasks and ids[1] not in adapter._tasks and ids[2] in adapter._tasks408+ adapter._pending_messages.clear()409+ await adapter.on_processing_start(received[2])410+ await adapter.on_processing_complete(received[2], ProcessingOutcome.SUCCESS)411+ assert adapter._tasks == {} and len(adapter.rpc.kinds(43004)) == 2412+ asyncio.run(run())262413 263414 264415 def slow_turn(adapter):