The replay cache no longer advances its floor past live seconds when it fills: IDs outside the replay window are dropped first, and the floor moves only when the cache is still over budget. Events are marked as seen only after Hermes accepts them, with a bounded attempt count so a poison event retires after three failures. When the Go helper dies the adapter marks the platform disconnected, respawns the helper with backoff and resubscribes from the persisted cursor. The helper now receives a minimal environment, state lives under the active Hermes home, and an identity lock stops two gateways from running one key. Home-channel auto-configuration returns the structured object Hermes expects, typing resumes after an approval resolves, and pending request cards are bounded and swept.
- commit
- 242a31ff4c0fe2c17967727d91dd85f709863600
- parent
- c1fefe784b4facc2e6b2f730fe9bdfb7e86ab721
- author
- Dami <dami.etoile@gmail.com>
| M | tinyagent/ | 6 | +- |
| M | tinyagent/ | 136 | +++++++++++++-- |
| M | tinyagent/ | 127 | ++++++++++----- |
| M | tinyagent/ | 158 | +++++++++++++++--- |
| M | tinyagent/ | 16 | ++ |
| M | tinyagent/ | 269 | ++++++++++++++++++++++++++++++ |
| A | tinyagent/ | 172 | +++++++++++++++++++ |
| M | tinyagent/ | 148 | ++++++++++++++++- |
@@ -33,7 +33,11 @@ hermes gateway run
3333 3434 By default, Hermes responds when an allowed user mentions the agent or replies to one of its messages. Set `TINY_REQUIRE_MENTION=false` to respond to all messages from allowed users in the selected rooms. This controls when Hermes responds; notifications still require an explicit mention by the agent.3535 36-`TINY_PRIVATE_KEY` is read from the environment by the helper and is never a command-line argument. 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.36+`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.37+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.39+40+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.3741 3842 The helper exposes a JSON-lines RPC mode for the adapter. Its public commands are:3943 @@ -9,6 +9,7 @@ from __future__ import annotations
99 import asyncio1010 from contextvars import ContextVar1111 import json12+import logging1213 import mimetypes1314 import hashlib1415 import os@@ -26,12 +27,14 @@ try:
2627 from .client import TinyRPC2728 from .interactions import InteractionMixin2829 from .media import incoming_media29- from .ingress import EventTracker, event_thread, reply_parent30+ from .ingress import EventTracker, IdentityLock, event_thread, reply_parent3031 except ImportError: # Hermes plugin loader may import adapter.py as a top-level module.3132 from client import TinyRPC3233 from interactions import InteractionMixin3334 from media import incoming_media34- from ingress import EventTracker, event_thread, reply_parent35+ from ingress import EventTracker, IdentityLock, event_thread, reply_parent36+37+logger = logging.getLogger(__name__)3538 3639 _HEX = re.compile(r"\b[0-9a-f]{64}\b", re.I)3740 _MENTION = re.compile(r"(?<![0-9A-Za-z])@([0-9a-f]{64})(?![0-9A-Za-z])", re.I)@@ -62,6 +65,10 @@ class TinyAdapter(InteractionMixin, BasePlatformAdapter):
6265 supports_async_delivery = True6366 MAX_MESSAGE_LENGTH = 120006467 MAX_INTERACTION_OPTIONS = 1268+ # Helper-death recovery backoff in seconds; Hermes's own watcher starts at 30s.69+ RECONNECT_MIN = 1.070+ RECONNECT_MAX = 30.071+ SWEEP_INTERVAL = 60.06572 6673 def __init__(self, config: PlatformConfig):6774 super().__init__(config, Platform("tiny"))@@ -69,7 +76,10 @@ class TinyAdapter(InteractionMixin, BasePlatformAdapter):
6976 self.relay_url = str(extra.get("relay_url") or os.environ.get("TINY_RELAY_URL", "")).strip()7077 self.cli_path = str(extra.get("cli_path") or os.environ.get("TINY_CLI_PATH", "tinyagent"))7178 self.rooms = [x.strip() for x in str(extra.get("rooms") or extra.get("channels") or os.environ.get("TINY_CHANNELS", "")).split(",") if x.strip()]72- home = extra.get("home_channel") or os.environ.get("TINY_HOME_CHANNEL", "")79+ home = extra.get("home_channel") or getattr(getattr(config, "home_channel", None), "chat_id", None) \80+ or os.environ.get("TINY_HOME_CHANNEL", "")81+ if isinstance(home, dict):82+ home = home.get("chat_id", "")7383 self.home_room = str(home).strip() or (self.rooms[0] if self.rooms else "")7484 self.allowed = {x.strip().lower() for x in str(extra.get("allowed_users") or os.environ.get("TINY_ALLOWED_USERS", "")).split(",") if x.strip()}7585 require_mention = extra.get("require_mention", os.environ.get("TINY_REQUIRE_MENTION", "true"))@@ -87,6 +97,10 @@ class TinyAdapter(InteractionMixin, BasePlatformAdapter):
8797 self._reply_routes: dict[str, dict] = {}8898 self._turn_routes: dict[str, dict] = {}8999 self._turn_route = ContextVar("tinyagent_turn_route", default=None)100+ self._lock: IdentityLock | None = None101+ self._closed = False102+ self._reconnect_task: asyncio.Task | None = None103+ self._last_sweep = 0.090104 91105 92106 def name(self) -> str:@@ -143,25 +157,87 @@ class TinyAdapter(InteractionMixin, BasePlatformAdapter):
143157 return tags, 12 if thread else 9144158 145159 async def connect(self, *, is_reconnect: bool = False) -> bool:146- self.rpc = TinyRPC(self.relay_url, cli_path=self.cli_path)147- identity = await self.rpc.call("identity")148- self.pubkey = str(identity.get("pubkey", ""))149- if not self.pubkey:150- raise RuntimeError("Tiny identity did not return a public key")151- self._tracker = EventTracker(self.relay_url, self.pubkey, self.rooms)152- filter: dict[str, Any] = {"kinds": [9, 11, 12, 1111], **self._tracker.filter_since()}153- if self.rooms:154- filter["#h"] = self.rooms155- await self.rpc.subscribe(self._subscription, filter, self._on_event)156- self._mark_connected()160+ """Start the helper and subscribe. Hermes's watcher calls this on a fresh adapter161+ with is_reconnect=True; helper death inside a live adapter reuses _establish."""162+ self._closed = False163+ await self._establish()157164 return True158165 166+ async def _establish(self) -> None:167+ """Spawn a helper, prove the identity, take its lock and subscribe from the cursor."""168+ rpc = TinyRPC(self.relay_url, cli_path=self.cli_path, on_exit=self._on_helper_exit)169+ try:170+ identity = await rpc.call("identity")171+ pubkey = str((identity or {}).get("pubkey", ""))172+ if not pubkey:173+ raise RuntimeError("Tiny identity did not return a public key")174+ if self._lock is None or self.pubkey.lower() != pubkey.lower():175+ if self._lock is not None:176+ self._lock.release()177+ lock = IdentityLock(self.relay_url, pubkey)178+ lock.acquire()179+ self._lock = lock180+ self.pubkey = pubkey181+ if self._tracker is None or self._tracker.identity != pubkey.lower():182+ self._tracker = EventTracker(self.relay_url, pubkey, self.rooms)183+ filter: dict[str, Any] = {"kinds": [9, 11, 12, 1111], **self._tracker.filter_since()}184+ if self.rooms:185+ filter["#h"] = self.rooms186+ await rpc.subscribe(self._subscription, filter, self._on_event)187+ except BaseException:188+ await rpc.close()189+ raise190+ previous, self.rpc = self.rpc, rpc191+ if previous is not None and previous is not rpc:192+ await previous.close()193+ self._mark_connected()194+195+ def _on_helper_exit(self, rpc) -> None:196+ """TinyRPC reports the helper died on its own; recover in the background."""197+ if self._closed or rpc is not self.rpc:198+ return199+ logger.warning("Tiny helper exited; reconnecting")200+ self._mark_disconnected()201+ if self._reconnect_task is None or self._reconnect_task.done():202+ self._reconnect_task = asyncio.create_task(self._reconnect_loop())203+204+ async def _reconnect_loop(self) -> None:205+ """Bounded exponential backoff until the helper is back or the adapter closes.206+207+ The adapter stays installed in Hermes throughout, so process-local waiters and208+ the identity lock survive; the gateway's reconnect watcher only builds a new209+ adapter for fatal errors, which this path never raises.210+ """211+ delay = self.RECONNECT_MIN212+ while not self._closed:213+ await asyncio.sleep(delay)214+ if self._closed:215+ return216+ try:217+ await self._establish()218+ except asyncio.CancelledError:219+ raise220+ except Exception as exc:221+ delay = min(delay * 2, self.RECONNECT_MAX)222+ logger.warning("Tiny helper reconnect failed: %s; retrying in %.0fs", exc, delay)223+ continue224+ logger.info("Tiny helper reconnected from cursor %s", self._tracker.cursor if self._tracker else "?")225+ return226+159227 async def disconnect(self) -> None:160- if self.rpc:161- with_context = self.rpc.unsubscribe(self._subscription)162- await asyncio.gather(with_context, return_exceptions=True)163- await self.rpc.close()164- self.rpc = None228+ self._closed = True229+ task, self._reconnect_task = self._reconnect_task, None230+ if task is not None and task is not asyncio.current_task():231+ task.cancel()232+ await asyncio.gather(task, return_exceptions=True)233+ rpc, self.rpc = self.rpc, None234+ if rpc is not None:235+ if rpc.alive:236+ await asyncio.gather(rpc.unsubscribe(self._subscription), return_exceptions=True)237+ await rpc.close()238+ if self._lock is not None:239+ self._lock.release()240+ self._lock = None165241 self._mark_disconnected()166242 167243 async def _on_event(self, event: dict) -> None:@@ -169,6 +245,10 @@ class TinyAdapter(InteractionMixin, BasePlatformAdapter):
169245 if event_id in self._processing:170246 return171247 self._processing.add(event_id)248+ now = time.monotonic()249+ if now - self._last_sweep >= self.SWEEP_INTERVAL:250+ self._last_sweep = now251+ self._sweep_pending()172252 try:173253 await self._receive_event(event)174254 finally:@@ -220,9 +300,17 @@ class TinyAdapter(InteractionMixin, BasePlatformAdapter):
220300 "reply_to_message_id": event_id, "thread_id": thread_id, "user_id": author}221301 if len(self._reply_routes) > 10000:222302 self._reply_routes.pop(next(iter(self._reply_routes)))303+ # The event counts as delivered only once Hermes has accepted it. A failed304+ # hand-off leaves it eligible for redelivery until the attempt budget is spent.305+ try:306+ await self.handle_message(message)307+ except BaseException as exc:308+ if self._tracker and not isinstance(exc, asyncio.CancelledError):309+ if self._tracker.fail(event):310+ logger.error("Tiny event %s retired after repeated failures", event_id, exc_info=exc)311+ raise223312 if self._tracker:224313 self._tracker.mark(event)225- await self.handle_message(message)226314 227315 def _authorized(self, author: str) -> bool:228316 return author.lower() in self.allowed@@ -341,10 +429,14 @@ def _env_enablement() -> dict | None:
341429 if not relay or not key:342430 return None343431 result = {"relay_url": relay}344- for env, key_name in (("TINY_HOME_CHANNEL", "home_channel"), ("TINY_ALLOWED_USERS", "allowed_users"),345- ("TINY_CLI_PATH", "cli_path"), ("TINY_REQUIRE_MENTION", "require_mention")):432+ for env, key_name in (("TINY_ALLOWED_USERS", "allowed_users"), ("TINY_CLI_PATH", "cli_path"),433+ ("TINY_REQUIRE_MENTION", "require_mention")):346434 if os.environ.get(env):347435 result[key_name] = os.environ[env]436+ home = os.environ.get("TINY_HOME_CHANNEL", "").strip()437+ if home:438+ # Hermes lifts this dict into a HomeChannel for cron and standalone delivery.439+ result["home_channel"] = {"chat_id": home, "name": "Home"}348440 return result349441 350442 @@ -9,12 +9,19 @@ import logging
99 import os1010 from typing import Any, Awaitable, Callable1111 12+# The helper receives only what it needs: a lookup path, a home for TLS and13+# temp files, its signing key and the proxy settings of the gateway.14+_HELPER_ENV = ("PATH", "HOME", "TMPDIR", "HTTP_PROXY", "HTTPS_PROXY", "NO_PROXY",15+ "http_proxy", "https_proxy", "no_proxy")16+1217 1318 class TinyRPC:14- def __init__(self, relay_url: str, key_env: str = "TINY_PRIVATE_KEY", cli_path: str = "tinyagent"):19+ def __init__(self, relay_url: str, key_env: str = "TINY_PRIVATE_KEY", cli_path: str = "tinyagent",20+ on_exit: Callable[["TinyRPC"], Any] | None = None):1521 self.relay_url = relay_url1622 self.key_env = key_env1723 self.cli_path = cli_path24+ self.on_exit = on_exit1825 self._proc: asyncio.subprocess.Process | None = None1926 self._reader_task: asyncio.Task | None = None2027 self._next_id = 0@@ -23,72 +30,110 @@ class TinyRPC:
2330 self._callback_tasks: set[asyncio.Task] = set()2431 self._ready: dict[str, asyncio.Event] = {}2532 self._write_lock = asyncio.Lock()33+ self._closing = False34+ self._exited = False35+36+ 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}40+41+42+ def alive(self) -> bool:43+ return self._proc is not None and self._proc.returncode is None and not self._exited2644 2745 async def connect(self) -> None:28- if self._proc and self._proc.returncode is None:29- return30- env = dict(os.environ)46+ if self._proc is not None:47+ if self.alive:48+ return49+ # A dead helper is not respawned in place: its subscriptions are gone,50+ # so the owner must rebuild the session through on_exit.51+ raise RuntimeError("tiny rpc exited")3152 # The key is inherited by the helper, never put in arguments or logs.3253 self._proc = await asyncio.create_subprocess_exec(3354 self.cli_path, "rpc", "--relay", self.relay_url, "--key-env", self.key_env,3455 stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE,35- stderr=asyncio.subprocess.DEVNULL, env=env, limit=64 * 1024 * 1024,56+ stderr=asyncio.subprocess.DEVNULL, env=self.helper_env(), limit=64 * 1024 * 1024,3657 )3758 self._reader_task = asyncio.create_task(self._read_loop())3859 3960 async def close(self) -> None:61+ self._closing = True4062 if self._reader_task:4163 self._reader_task.cancel()4264 await asyncio.gather(self._reader_task, return_exceptions=True)43- for task in self._callback_tasks:44- task.cancel()45- await asyncio.gather(*self._callback_tasks, return_exceptions=True)65+ if not self._exited:66+ # A deliberate close ends in-flight event handling; turns started by a helper67+ # that died on its own run to completion and are redelivered if they fail.68+ for task in self._callback_tasks:69+ task.cancel()70+ await asyncio.gather(*self._callback_tasks, return_exceptions=True)4671 self._callback_tasks.clear()4772 if self._proc and self._proc.returncode is None:4873 self._proc.terminate()4974 await self._proc.wait()50- error = RuntimeError("tiny rpc closed")75+ self._fail_pending(RuntimeError("tiny rpc closed"))76+ self._proc = None77+78+ def _fail_pending(self, error: Exception) -> None:5179 for future in self._pending.values():5280 if not future.done():5381 future.set_exception(error)5482 self._pending.clear()55- self._proc = None5683 5784 async def _read_loop(self) -> None:5885 assert self._proc and self._proc.stdout86+ died = False5987 try:60- async for line in self._proc.stdout:61- try:62- msg = json.loads(line)63- except (ValueError, UnicodeDecodeError):64- continue65- if msg.get("event") == "event":66- callback = self._subscriptions.get(str(msg.get("subscription", "")))67- if callback:68- task = asyncio.create_task(callback(msg.get("data") or {}))69- self._callback_tasks.add(task)70- task.add_done_callback(self._callback_done)71- continue72- if msg.get("event") == "connected":73- ready = self._ready.get(str(msg.get("subscription", "")))74- if ready:75- ready.set()76- continue77- request_id = msg.get("id")78- if isinstance(request_id, int) and request_id in self._pending:79- future = self._pending.pop(request_id)80- if future.done():81- continue82- if "error" in msg:83- future.set_exception(RuntimeError(str((msg["error"] or {}).get("message", "rpc error"))))84- else:85- future.set_result(msg.get("result"))88+ try:89+ await self._pump()90+ except asyncio.CancelledError:91+ raise92+ except Exception:93+ logging.getLogger(__name__).exception("Tiny helper stream failed")94+ # EOF or a broken pipe: the helper exited (or was closed) rather than being cancelled.95+ died = not self._closing8696 finally:87- error = RuntimeError("tiny rpc exited")88- for future in self._pending.values():89- if not future.done():90- future.set_exception(error)91- self._pending.clear()97+ self._fail_pending(RuntimeError("tiny rpc exited"))98+ if died and not self._exited:99+ # The helper went away on its own; tell the owner exactly once.100+ self._exited = True101+ if self.on_exit is not None:102+ try:103+ result = self.on_exit(self)104+ if asyncio.iscoroutine(result):105+ await result106+ except Exception:107+ logging.getLogger(__name__).exception("Tiny helper exit handler failed")108+109+ async def _pump(self) -> None:110+ assert self._proc and self._proc.stdout111+ async for line in self._proc.stdout:112+ try:113+ msg = json.loads(line)114+ except (ValueError, UnicodeDecodeError):115+ continue116+ if msg.get("event") == "event":117+ callback = self._subscriptions.get(str(msg.get("subscription", "")))118+ if callback:119+ task = asyncio.create_task(callback(msg.get("data") or {}))120+ self._callback_tasks.add(task)121+ task.add_done_callback(self._callback_done)122+ continue123+ if msg.get("event") == "connected":124+ ready = self._ready.get(str(msg.get("subscription", "")))125+ if ready:126+ ready.set()127+ continue128+ request_id = msg.get("id")129+ if isinstance(request_id, int) and request_id in self._pending:130+ future = self._pending.pop(request_id)131+ if future.done():132+ continue133+ if "error" in msg:134+ future.set_exception(RuntimeError(str((msg["error"] or {}).get("message", "rpc error"))))135+ else:136+ future.set_result(msg.get("result"))92137 93138 def _callback_done(self, task):94139 self._callback_tasks.discard(task)@@ -6,8 +6,10 @@ agent identity and subscribed rooms, so reconnects cannot replay old commands.
66 77 from __future__ import annotations88 9+import fcntl910 import hashlib1011 import json12+import logging1113 import os1214 import re1315 import tempfile@@ -18,6 +20,31 @@ from typing import Iterable
1820 _EVENT_ID = re.compile(r"^[0-9a-f]{64}$", re.I)1921 _REPLAY_WINDOW = 3002022 _FUTURE_SKEW = 30023+# A poison event is retired after this many failed hand-offs to Hermes.24+MAX_ATTEMPTS = 325+_MAX_ATTEMPT_ENTRIES = 25626+27+logger = logging.getLogger(__name__)28+29+30+def state_root(state_dir: str | Path | None = None) -> Path:31+ """Return the Hermes home used for adapter state.32+33+ The active Hermes home wins when the CLI package is importable (profiles may34+ override the environment); otherwise HERMES_HOME, then ~/.hermes.35+ """36+ if state_dir:37+ return Path(state_dir).expanduser()38+ try:39+ from hermes_cli.config import get_hermes_home40+ return Path(get_hermes_home()).expanduser()41+ except Exception:42+ return Path(os.environ.get("HERMES_HOME", "").strip() or "~/.hermes").expanduser()43+44+45+def adapter_state_dir(root: str | Path | None = None) -> Path:46+ """Return the directory holding tracker state and identity locks."""47+ return state_root(root) / "state" / "tinyagent"2148 2249 2350 def _tag_values(event: dict, name: str) -> list[list[str]]:@@ -35,21 +62,26 @@ def _expiration(event: dict) -> int | None:
3562 3663 3764 class EventTracker:38- """Bounded persistent event cursor and ID set for one Tiny subscription."""65+ """Bounded persistent event cursor and ID set for one Tiny subscription.66+67+ State file (JSON): ``floor`` and ``cursor`` are Unix seconds; ``seen`` maps a68+ delivered event ID to its ``created_at``; ``attempts`` maps an event ID to the69+ number of failed hand-offs. Older files stored ``seen`` as a list of IDs and70+ load with every listed ID pinned to the cursor.71+ """3972 4073 def __init__(self, relay: str, identity: str, rooms: Iterable[str], *, state_dir: str | Path | None = None,4174 now: int | None = None, max_seen: int = 10000):4275 self.relay, self.identity = str(relay), str(identity).lower()4376 self.rooms = tuple(sorted(str(room) for room in rooms))44- root = Path(state_dir or os.environ.get("HERMES_HOME", "~/.hermes")).expanduser()4577 key = hashlib.sha256((self.relay + "\0" + self.identity + "\0" + "\0".join(self.rooms)).encode()).hexdigest()[:24]46- self.path = root / "state" / "tinyagent" / f"{key}.json"78+ self.path = adapter_state_dir(state_dir) / f"{key}.json"4779 self.max_seen = max(1, int(max_seen))4880 self.now = int(time.time() if now is None else now)4981 self.floor = self.now5082 self.cursor = self.now51- self.seen: list[str] = []52- self._seen: set[str] = set()83+ self.seen: dict[str, int] = {}84+ self.attempts: dict[str, int] = {}5385 self._load()5486 # Establish the first-start floor immediately, including a clean restart before5587 # any event arrives. This prevents a later process from treating its own startup@@ -61,15 +93,23 @@ class EventTracker:
6193 data = json.loads(self.path.read_text())6294 self.floor = int(data.get("floor", data.get("cursor", self.floor)))6395 self.cursor = int(data.get("cursor", self.cursor))64- self.seen = [str(value) for value in data.get("seen", [])][-self.max_seen:]65- self._seen = set(self.seen)66- except (OSError, ValueError, TypeError):96+ seen = data.get("seen", {})97+ if isinstance(seen, list):98+ # Legacy format: IDs without timestamps stay in the replay window.99+ self.seen = {str(value): self.cursor for value in seen}100+ else:101+ self.seen = {str(key): int(value) for key, value in dict(seen).items()}102+ attempts = data.get("attempts", {})103+ self.attempts = {str(key): int(value) for key, value in dict(attempts).items()}104+ except (OSError, ValueError, TypeError, AttributeError):67105 return106+ self._prune()68107 69108 def _save(self) -> None:70109 self.path.parent.mkdir(parents=True, exist_ok=True)71110 os.chmod(self.path.parent, 0o700)72- payload = json.dumps({"floor": self.floor, "cursor": self.cursor, "seen": self.seen[-self.max_seen:]}, separators=(",", ":"))111+ payload = json.dumps({"floor": self.floor, "cursor": self.cursor, "seen": self.seen,112+ "attempts": self.attempts}, separators=(",", ":"))73113 fd, temporary = tempfile.mkstemp(prefix="tinyagent-", dir=self.path.parent)74114 try:75115 with os.fdopen(fd, "w") as stream:@@ -83,16 +123,37 @@ class EventTracker:
83123 except FileNotFoundError:84124 pass85125 126+ def _window_start(self) -> int:127+ return max(self.floor, self.cursor - _REPLAY_WINDOW)128+129+ def _prune(self) -> None:130+ """Drop IDs the window already rejects; raise the floor only when still over budget."""131+ start = self._window_start()132+ self.seen = {key: stamp for key, stamp in self.seen.items() if stamp >= start}133+ if len(self.seen) > self.max_seen:134+ stamps = sorted(self.seen.values(), reverse=True)135+ threshold = stamps[self.max_seen - 1]136+ if sum(1 for stamp in stamps if stamp >= threshold) > self.max_seen:137+ # More IDs share the oldest retained second than fit; retire that second.138+ threshold += 1139+ self.seen = {key: stamp for key, stamp in self.seen.items() if stamp >= threshold}140+ self.floor = max(self.floor, threshold)141+ logger.warning("Tiny replay cache exceeded %d IDs; events before %d are now ignored",142+ self.max_seen, self.floor)143+ self.attempts = {key: count for key, count in self.attempts.items() if key not in self.seen}144+ while len(self.attempts) > _MAX_ATTEMPT_ENTRIES:145+ self.attempts.pop(next(iter(self.attempts)))146+86147 def should_accept(self, event: dict, *, now: int | None = None) -> bool:87148 """Return true only for unseen, non-expired events at/after the cursor."""88149 event_id = str(event.get("id", ""))89- if not _EVENT_ID.fullmatch(event_id) or event_id in self._seen:150+ if not _EVENT_ID.fullmatch(event_id) or event_id in self.seen:90151 return False91152 try:92153 created = int(event.get("created_at"))93154 except (TypeError, ValueError):94155 return False95- if created < max(self.floor, self.cursor - _REPLAY_WINDOW):156+ if created < self._window_start():96157 return False97158 if created > int(time.time() if now is None else now) + _FUTURE_SKEW:98159 return False@@ -109,29 +170,79 @@ class EventTracker:
109170 if not _EVENT_ID.fullmatch(event_id):110171 return111172 try:112- self.cursor = max(self.cursor, int(event.get("created_at", self.cursor)))173+ created = int(event.get("created_at", self.cursor))113174 except (TypeError, ValueError):114- pass115- if event_id not in self._seen:116- self._seen.add(event_id)117- self.seen.append(event_id)118- if len(self.seen) > self.max_seen:119- # Retire the overlap instead of forgetting IDs still eligible for replay.120- self.floor = self.cursor + 1121- self._seen = set(self.seen[-self.max_seen:])122- self.seen = self.seen[-self.max_seen:]175+ created = self.cursor176+ self.cursor = max(self.cursor, created)177+ self.seen.setdefault(event_id, created)178+ self.attempts.pop(event_id, None)179+ self._prune()180+ self._save()181+182+ def fail(self, event: dict) -> bool:183+ """Count a failed hand-off; retire the event after MAX_ATTEMPTS and return True."""184+ event_id = str(event.get("id", ""))185+ if not _EVENT_ID.fullmatch(event_id):186+ return False187+ count = self.attempts.pop(event_id, 0) + 1188+ if count >= MAX_ATTEMPTS:189+ self.mark(event)190+ return True191+ self.attempts[event_id] = count192+ self._prune()123193 self._save()194+ return False124195 125196 def filter_since(self) -> dict:126197 """Return an overlapping subscription filter; same-timestamp IDs are retained."""127198 # Match should_accept's bounded overlap. A one-second overlap is insufficient128199 # when the helper returns events out of order or the gateway was offline.129- result = {"since": max(self.floor, self.cursor - _REPLAY_WINDOW)}200+ result = {"since": self._window_start()}130201 if self.rooms:131202 result["#h"] = list(self.rooms)132203 return result133204 134205 206+class IdentityLock:207+ """Exclusive per-identity lock so one gateway consumes a Tiny key at a time."""208+209+ def __init__(self, relay: str, pubkey: str, *, state_dir: str | Path | None = None):210+ key = hashlib.sha256((str(relay) + "\0" + str(pubkey).lower()).encode()).hexdigest()[:24]211+ self.path = adapter_state_dir(state_dir) / f"{key}.lock"212+ self._fd: int | None = None213+214+215+ def held(self) -> bool:216+ return self._fd is not None217+218+ def acquire(self) -> None:219+ if self._fd is not None:220+ return221+ self.path.parent.mkdir(parents=True, exist_ok=True)222+ os.chmod(self.path.parent, 0o700)223+ fd = os.open(self.path, os.O_RDWR | os.O_CREAT, 0o600)224+ try:225+ fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)226+ except OSError:227+ os.close(fd)228+ raise RuntimeError("another Hermes gateway already runs this Tiny identity") from None229+ try:230+ os.ftruncate(fd, 0)231+ os.write(fd, str(os.getpid()).encode())232+ except OSError:233+ pass234+ self._fd = fd235+236+ def release(self) -> None:237+ if self._fd is None:238+ return239+ fd, self._fd = self._fd, None240+ try:241+ fcntl.flock(fd, fcntl.LOCK_UN)242+ finally:243+ os.close(fd)244+245+135246 def event_thread(event: dict) -> str | None:136247 """Return canonical NIP-10 root thread ID, including marked e tags."""137248 marked_root = [tag[1] for tag in _tag_values(event, "e") if len(tag) >= 4 and tag[3] == "root"]@@ -155,4 +266,5 @@ def is_reply_to_own(event: dict, own_ids: set[str]) -> bool:
155266 return any(tag[1] in own_ids for tag in _tag_values(event, "e"))156267 157268 158-__all__ = ["EventTracker", "event_thread", "reply_parent", "is_reply_to_own"]269+__all__ = ["EventTracker", "IdentityLock", "MAX_ATTEMPTS", "event_thread", "reply_parent", "is_reply_to_own",270+ "state_root", "adapter_state_dir"]@@ -9,6 +9,9 @@ from types import SimpleNamespace
99 1010 from gateway.platforms.base import SendResult1111 12+# Open request cards kept per adapter; the oldest are dropped past this.13+MAX_PENDING = 50014+1215 1316 def tag(event, name):1417 return next((row[1] for row in event.get("tags", []) if len(row) > 1 and row[0] == name), None)@@ -88,9 +91,18 @@ class InteractionMixin:
8891 pending.request_kind = request_kind8992 result = await self._publish(pending.room, content, [*reply_tags, *tags], kind=request_kind)9093 if result.success and result.message_id:94+ self._sweep_pending()9195 self._pending[result.message_id] = pending96+ while len(self._pending) > MAX_PENDING:97+ self._pending.pop(next(iter(self._pending)))9298 return result9399 100+ def _sweep_pending(self, now=None):101+ """Drop expired request cards; the adapter also calls this from its receive path."""102+ now = int(time.time()) if now is None else now103+ for event_id in [key for key, entry in self._pending.items() if now >= entry.expires]:104+ self._pending.pop(event_id, None)105+94106 def _pending_request(self, room, session, callback, interaction, options, selection, metadata, ttl=300):95107 now = int(time.time())96108 route = self._request_route(session)@@ -134,6 +146,7 @@ class InteractionMixin:
134146 return await self._request(pending, f"{title}\n\n{message}", title)135147 136148 async def _on_answer(self, event):149+ self._sweep_pending()137150 event_id = tag(event, "e")138151 pending = self._pending.get(event_id)139152 if pending is None or not self._answer_matches(event, event_id, pending):@@ -157,6 +170,9 @@ class InteractionMixin:
157170 await self.send(pending.room, result, metadata=pending.metadata)158171 # Waiters are process-local in Hermes. Never apply a late answer to another request.159172 self._pending.pop(event_id, None)173+ if pending.interaction in ("approval", "confirmation"):174+ # Hermes pauses the typing indicator while it waits for an approval.175+ self.resume_typing_for_chat(pending.room)160176 161177 def _answer_matches(self, event, event_id, pending):162178 now = int(time.time())@@ -402,3 +402,272 @@ def test_public_approval_hook_preserves_scopes_on_older_hermes(adapter, monkeypa
402402 await adapter._on_answer(response(request, 'once'))403403 assert entry.event.is_set() and entry.result == 'once'404404 asyncio.run(run())405+406+407+class FakeHelper:408+ """Stands in for TinyRPC: answers identity, records filters and dies on demand."""409+410+ instances = []411+ fail_identity = False412+413+ def __init__(self, relay_url, key_env="TINY_PRIVATE_KEY", cli_path="tinyagent", on_exit=None):414+ self.relay_url, self.on_exit = relay_url, on_exit415+ self.filters, self.subscriptions, self.events = [], {}, []416+ self.closed = self.dead = False417+ FakeHelper.instances.append(self)418+419+420+ def alive(self):421+ return not (self.closed or self.dead)422+423+ async def call(self, method, params=None, timeout=30):424+ if FakeHelper.fail_identity:425+ raise RuntimeError("helper unavailable")426+ return {"pubkey": BOT} if method == "identity" else {}427+428+ async def subscribe(self, name, filter, callback):429+ self.filters.append(filter)430+ self.subscriptions[name] = callback431+432+ async def unsubscribe(self, name):433+ self.subscriptions.pop(name, None)434+435+ async def query(self, _filter):436+ return []437+438+ async def close(self):439+ self.closed = True440+441+ def die(self):442+ self.dead = True443+ self.on_exit(self)444+445+446+447+def helper(monkeypatch):448+ import adapter as module449+ FakeHelper.instances.clear()450+ FakeHelper.fail_identity = False451+ monkeypatch.setattr(module, "TinyRPC", FakeHelper)452+ yield FakeHelper453+ FakeHelper.fail_identity = False454+455+456+def live_adapter(marks=None):457+ instance = TinyAdapter(PlatformConfig(enabled=True, extra={"relay_url": "http://relay", "rooms": "room",458+ "allowed_users": OWNER}))459+ instance.RECONNECT_MIN = 0.01460+ instance.RECONNECT_MAX = 0.03461+ if marks is not None:462+ original = instance._mark_connected463+ def marked(**kwargs):464+ marks.append("connected")465+ original(**kwargs)466+ instance._mark_connected = marked467+ return instance468+469+470+def test_helper_death_reconnects_from_persisted_cursor(adapter, helper, tmp_path):471+ from ingress import EventTracker472+ marks = []473+ instance = live_adapter(marks)474+ now = int(time.time())475+ # A tracker with history: the floor is old and the cursor moves with delivered events.476+ instance._tracker = EventTracker("http://relay", BOT, ["room"], now=now - 1000)477+ received = []478+ async def handle(message): received.append(message)479+ instance.handle_message = handle480+481+ async def run():482+ assert await instance.connect()483+ first = helper.instances[-1]484+ assert instance.is_connected and instance._lock.held485+ assert first.filters == [{"kinds": [9, 11, 12, 1111], "since": now - 1000, "#h": ["room"]}]486+ await first.subscriptions["tinyagent"]({"kind": 9, "id": "d" * 64, "pubkey": OWNER, "created_at": now,487+ "content": "hello", "tags": [["h", "room"], ["p", BOT]]})488+ assert len(received) == 1 and instance._tracker.cursor == now489+ first.die()490+ assert not instance.is_connected491+ await asyncio.wait_for(instance._reconnect_task, 2)492+ second = helper.instances[-1]493+ assert second is not first and first.closed and instance.rpc is second494+ assert second.filters == [{"kinds": [9, 11, 12, 1111], "since": now - 300, "#h": ["room"]}]495+ assert instance.is_connected and instance._lock.held and marks == ["connected", "connected"]496+ # The same tracker persists across the reconnect: the delivered event stays deduplicated.497+ assert not instance._tracker.should_accept({"id": "d" * 64, "created_at": now, "tags": []})498+ await instance.disconnect()499+ assert not instance.is_connected and instance._lock is None and second.closed500+ asyncio.run(run())501+502+503+def test_reconnect_backs_off_and_disconnect_cancels_it(adapter, helper, monkeypatch):504+ import adapter as module505+ instance = live_adapter()506+ delays = []507+ real_sleep = asyncio.sleep508+ async def sleep(delay, *args):509+ delays.append(delay)510+ await real_sleep(0)511+ monkeypatch.setattr(module.asyncio, "sleep", sleep)512+513+ async def run():514+ await instance.connect()515+ first = helper.instances[-1]516+ helper.fail_identity = True517+ first.die()518+ task = instance._reconnect_task519+ for _ in range(12):520+ await real_sleep(0)521+ assert delays[:4] == [0.01, 0.02, 0.03, 0.03]522+ assert not instance.is_connected523+ await instance.disconnect()524+ assert task.cancelled() and instance._reconnect_task is None525+ spawned = len(helper.instances)526+ helper.fail_identity = False527+ for _ in range(5):528+ await real_sleep(0)529+ assert len(helper.instances) == spawned530+ assert not instance.is_connected531+ asyncio.run(run())532+533+534+def test_watcher_reconnect_uses_the_same_path(adapter, helper):535+ instance = live_adapter()536+537+ async def run():538+ await instance.connect()539+ first = helper.instances[-1]540+ tracker = instance._tracker541+ await instance.disconnect()542+ assert first.closed and not instance.is_connected543+ # Hermes's watcher calls connect(is_reconnect=True); the persisted cursor is reused.544+ assert await instance.connect(is_reconnect=True)545+ second = helper.instances[-1]546+ assert second is not first and instance.is_connected and instance._lock.held547+ assert second.filters[0]["since"] == tracker.filter_since()["since"]548+ await instance.disconnect()549+ asyncio.run(run())550+551+552+def test_identity_lock_refuses_a_second_gateway(adapter, helper):553+ first, second = live_adapter(), live_adapter()554+555+ async def run():556+ await first.connect()557+ with pytest.raises(RuntimeError, match="another Hermes gateway already runs this Tiny identity"):558+ await second.connect()559+ assert helper.instances[-1].closed and not second.is_connected and second._lock is None560+ await first.disconnect()561+ assert await second.connect()562+ assert second.is_connected563+ await second.disconnect()564+ asyncio.run(run())565+566+567+def test_event_is_marked_only_after_hermes_accepts_it(adapter, tmp_path):568+ from ingress import EventTracker569+ now = int(time.time())570+ adapter._tracker = EventTracker("http://relay", BOT, ["room"], state_dir=tmp_path, now=now - 10)571+ poison = {"kind": 9, "id": "d" * 64, "pubkey": OWNER, "created_at": now, "content": "boom",572+ "tags": [["h", "room"], ["p", BOT]]}573+ attempts = []574+ async def failing(message):575+ attempts.append(message.message_id)576+ raise RuntimeError("Hermes rejected the turn")577+ adapter.handle_message = failing578+ for attempt in (1, 2):579+ with pytest.raises(RuntimeError):580+ asyncio.run(adapter._on_event(poison))581+ assert adapter._tracker.should_accept(poison)582+ assert adapter._tracker.attempts == {poison["id"]: attempt}583+ with pytest.raises(RuntimeError):584+ asyncio.run(adapter._on_event(poison))585+ assert len(attempts) == 3586+ assert not adapter._tracker.should_accept(poison) and adapter._tracker.attempts == {}587+ asyncio.run(adapter._on_event(poison))588+ assert len(attempts) == 3589+ handled = []590+ async def accept(message): handled.append(message.message_id)591+ adapter.handle_message = accept592+ good = {**poison, "id": "e" * 64, "content": "hello"}593+ asyncio.run(adapter._on_event(good))594+ asyncio.run(adapter._on_event(good))595+ assert handled == [good["id"]]596+ assert not adapter._tracker.should_accept(good)597+ assert json.loads(adapter._tracker.path.read_text())["seen"] == {poison["id"]: now, good["id"]: now}598+599+600+def test_env_enablement_seeds_a_home_channel(adapter, monkeypatch):601+ import adapter as module602+ from gateway.config import GatewayConfig, HomeChannel, Platform603+ from gateway.config_env import _enable_plugin_platform604+ monkeypatch.setenv("TINY_RELAY_URL", "http://relay")605+ monkeypatch.setenv("TINY_PRIVATE_KEY", "k" * 64)606+ monkeypatch.setenv("TINY_HOME_CHANNEL", "home-room")607+ assert module._env_enablement()["home_channel"] == {"chat_id": "home-room", "name": "Home"}608+ class Ctx:609+ def register_platform(self, **kwargs):610+ platform_registry.register(PlatformEntry(**kwargs))611+ module.register(Ctx())612+ config = GatewayConfig()613+ _enable_plugin_platform(config, platform_registry.get("tiny"))614+ home = config.platforms[Platform("tiny")].home_channel615+ assert isinstance(home, HomeChannel) and home.chat_id == "home-room" and home.name == "Home"616+ assert "home_channel" not in config.platforms[Platform("tiny")].extra617+ assert TinyAdapter(config.platforms[Platform("tiny")]).home_room == "home-room"618+619+620+def test_answers_resume_typing_after_waits(adapter):621+ async def run():622+ entry = _ApprovalEntry({"command": "ls", "request_id": "approval-id"})623+ approval._gateway_queues["session"] = [entry]624+ adapter.pause_typing_for_chat("room")625+ await adapter.send_exec_approval("room", "ls", "session")626+ request = adapter.rpc.events[-1]627+ assert "room" in adapter._typing_paused628+ await adapter._on_answer(response(request, "once"))629+ assert entry.event.is_set() and "room" not in adapter._typing_paused630+ async def handler(choice): return None631+ slash_confirm.register("session", "confirm-id", "reload", handler)632+ adapter.pause_typing_for_chat("room")633+ await adapter.send_slash_confirm("room", "Reload?", "Confirm", "session", "confirm-id")634+ await adapter._on_answer(response(adapter.rpc.events[-1], "once"))635+ assert "room" not in adapter._typing_paused636+ asyncio.run(run())637+638+639+def test_pending_requests_are_bounded_and_swept(adapter, monkeypatch):640+ import interactions641+ monkeypatch.setattr(interactions, "MAX_PENDING", 3)642+ async def run():643+ for index in range(5):644+ clarify_gateway.register(f"q{index}", "session", "Pick", ["yes"])645+ await adapter.send_clarify("room", "Pick", ["yes"], f"q{index}", "session")646+ assert [entry.callback_id for entry in adapter._pending.values()] == ["q2", "q3", "q4"]647+ stale = next(iter(adapter._pending.values()))648+ stale.expires = int(time.time()) - 1649+ adapter._sweep_pending()650+ assert [entry.callback_id for entry in adapter._pending.values()] == ["q3", "q4"]651+ # An answer sweeps too, so an expired card never lingers.652+ list(adapter._pending.values())[0].expires = int(time.time()) - 1653+ await adapter._on_answer({"kind": 1111, "tags": [["e", "0" * 64]], "pubkey": OWNER,654+ "created_at": int(time.time()), "content": "c0"})655+ assert [entry.callback_id for entry in adapter._pending.values()] == ["q4"]656+ asyncio.run(run())657+658+659+def test_receive_path_sweeps_pending_once_a_minute(adapter, monkeypatch):660+ sweeps = []661+ monkeypatch.setattr(adapter, "_sweep_pending", lambda now=None: sweeps.append(now))662+ async def handle(message): pass663+ adapter.handle_message = handle664+ base = {"kind": 9, "pubkey": OWNER, "created_at": int(time.time()), "content": "hello",665+ "tags": [["h", "room"], ["p", BOT]]}666+ async def run():667+ await adapter._on_event({**base, "id": "d" * 64})668+ await adapter._on_event({**base, "id": "e" * 64})669+ asyncio.run(run())670+ assert len(sweeps) == 1671+ adapter._last_sweep = time.monotonic() - adapter.SWEEP_INTERVAL672+ asyncio.run(adapter._on_event({**base, "id": "f" * 64}))673+ assert len(sweeps) == 2@@ -0,0 +1,172 @@
1+"""TinyRPC process handling against a fake helper; no Hermes or network needed."""2+3+import asyncio4+import json5+import sys6+from pathlib import Path7+8+import pytest9+10+sys.path.insert(0, str(Path(__file__).parents[1]))11+12+import client as module13+from client import TinyRPC14+15+16+class FakeStdin:17+ def __init__(self, process):18+ self.process = process19+ self.lines = []20+21+ def write(self, data):22+ self.lines.append(json.loads(data))23+ self.process.on_request(self.lines[-1])24+25+ async def drain(self):26+ pass27+28+29+class FakeProcess:30+ """Answers every request immediately and dies on demand."""31+32+ def __init__(self):33+ self.stdout = asyncio.StreamReader()34+ self.stdin = FakeStdin(self)35+ self.returncode = None36+ self.terminated = False37+38+ def on_request(self, request):39+ if request["method"] == "hang":40+ return41+ reply = {"id": request["id"], "result": {"method": request["method"]}}42+ self.stdout.feed_data((json.dumps(reply) + "\n").encode())43+ if request["method"] == "subscribe":44+ self.stdout.feed_data(json.dumps({"event": "connected", "subscription": request["params"]["subscription"]}).encode() + b"\n")45+46+ def die(self):47+ self.returncode = 148+ self.stdout.feed_eof()49+50+ def terminate(self):51+ self.terminated = True52+ self.returncode = -1553+ self.stdout.feed_eof()54+55+ async def wait(self):56+ return self.returncode57+58+59+60+def spawn(monkeypatch):61+ spawned = []62+63+ async def create(*args, **kwargs):64+ process = FakeProcess()65+ spawned.append((args, kwargs, process))66+ return process67+68+ monkeypatch.setattr(module.asyncio, "create_subprocess_exec", create)69+ return spawned70+71+72+def test_helper_gets_a_scoped_environment(spawn, monkeypatch):73+ monkeypatch.setenv("TINY_PRIVATE_KEY", "secret")74+ monkeypatch.setenv("HTTPS_PROXY", "http://proxy")75+ monkeypatch.setenv("no_proxy", "localhost")76+ monkeypatch.setenv("OPENAI_API_KEY", "leak")77+ monkeypatch.setenv("HERMES_HOME", "/somewhere")78+79+ async def run():80+ rpc = TinyRPC("http://relay", cli_path="/bin/tinyagent")81+ await rpc.connect()82+ await rpc.close()83+84+ asyncio.run(run())85+ args, kwargs, _process = spawn[0]86+ assert args[:2] == ("/bin/tinyagent", "rpc")87+ env = kwargs["env"]88+ assert env["TINY_PRIVATE_KEY"] == "secret"89+ assert env["HTTPS_PROXY"] == "http://proxy" and env["no_proxy"] == "localhost"90+ assert "OPENAI_API_KEY" not in env and "HERMES_HOME" not in env91+ assert set(env) <= {"PATH", "HOME", "TMPDIR", "TINY_PRIVATE_KEY", "HTTP_PROXY", "HTTPS_PROXY", "NO_PROXY",92+ "http_proxy", "https_proxy", "no_proxy"}93+94+95+def test_helper_death_fails_pending_and_reports_once(spawn):96+ exits = []97+98+ async def run():99+ rpc = TinyRPC("http://relay", on_exit=exits.append)100+ assert (await rpc.call("identity")) == {"method": "identity"}101+ process = spawn[0][2]102+ hanging = asyncio.ensure_future(rpc.call("hang"))103+ await asyncio.sleep(0)104+ process.die()105+ with pytest.raises(RuntimeError, match="tiny rpc exited"):106+ await hanging107+ await asyncio.sleep(0)108+ assert exits == [rpc]109+ assert not rpc.alive110+ # No silent respawn: the owner rebuilds the session instead.111+ with pytest.raises(RuntimeError, match="tiny rpc exited"):112+ await rpc.call("identity")113+ assert len(spawn) == 1114+ await rpc.close()115+ assert exits == [rpc]116+117+ asyncio.run(run())118+119+120+def test_close_does_not_report_an_exit(spawn):121+ exits = []122+123+ async def run():124+ rpc = TinyRPC("http://relay", on_exit=exits.append)125+ await rpc.connect()126+ await rpc.close()127+ await asyncio.sleep(0)128+129+ asyncio.run(run())130+ assert exits == []131+ assert spawn[0][2].terminated132+133+134+def test_async_exit_handler_is_awaited(spawn):135+ seen = []136+137+ async def run():138+ done = asyncio.Event()139+140+ async def on_exit(rpc):141+ seen.append(rpc)142+ done.set()143+144+ rpc = TinyRPC("http://relay", on_exit=on_exit)145+ await rpc.connect()146+ spawn[0][2].die()147+ await asyncio.wait_for(done.wait(), 1)148+149+ asyncio.run(run())150+ assert len(seen) == 1151+152+153+def test_dead_helper_keeps_in_flight_callbacks(spawn):154+ async def run():155+ finished = asyncio.Event()156+157+ async def callback(event):158+ await asyncio.sleep(0.01)159+ finished.set()160+161+ rpc = TinyRPC("http://relay")162+ await rpc.subscribe("s", {"kinds": [9]}, callback)163+ process = spawn[0][2]164+ process.stdout.feed_data(json.dumps({"event": "event", "subscription": "s", "data": {"id": "x"}}).encode() + b"\n")165+ await asyncio.sleep(0)166+ await asyncio.sleep(0)167+ process.die()168+ await asyncio.sleep(0)169+ await rpc.close()170+ await asyncio.wait_for(finished.wait(), 1)171+172+ asyncio.run(run())@@ -1,11 +1,16 @@
11 import json22 import hashlib3+import subprocess34 import sys5+import textwrap46 from pathlib import Path57 8+import pytest9+610 sys.path.insert(0, str(Path(__file__).parents[1]))711 8-from ingress import EventTracker, event_thread, is_reply_to_own, reply_parent12+import ingress13+from ingress import EventTracker, IdentityLock, event_thread, is_reply_to_own, reply_parent914 1015 1116 def event(event_id, created=100, tags=None):@@ -32,7 +37,13 @@ def test_expiration_and_bounded_seen(tmp_path):
3237 assert not tracker.should_accept(event("expired", 100, [["expiration", "100"]]))3338 for index in range(3):3439 tracker.mark(event(str(index), 101 + index))35- assert len(json.loads(tracker.path.read_text())["seen"]) == 240+ # 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.43+ state = json.loads(tracker.path.read_text())44+ assert len(state["seen"]) == 245+ assert set(state["seen"].values()) == {102, 103}46+ assert state["floor"] == 1023647 3748 3849 def test_nip10_root_reply_and_own_parent():@@ -58,5 +69,138 @@ def test_overlap_accepts_out_of_order_events_and_survives_restart(tmp_path):
5869 def test_evicted_ids_cannot_become_commands_again(tmp_path):5970 tracker = EventTracker('r', 'a', ['lab'], state_dir=tmp_path, now=100, max_seen=2)6071 for i in range(3): tracker.mark(event(str(i), 101 + i))72+ # ID 0 (second 101) was retired by the overflow; the floor now rejects that second.6173 assert not tracker.should_accept(event('0', 101))74+ assert not tracker.should_accept(event('1', 102))6275 assert tracker.should_accept(event('next', 104))76+77+78+def test_overflow_keeps_same_second_and_replay_window_events(tmp_path):79+ # Overflowing the ID cache may retire old IDs, but never events the relay can80+ # still legitimately deliver: unseen events in the cursor's own second and81+ # out-of-order events inside the replay window.82+ tracker = EventTracker('r', 'a', ['lab'], state_dir=tmp_path, now=100, max_seen=3)83+ tracker.mark(event('a', 101))84+ tracker.mark(event('b', 102))85+ tracker.mark(event('c', 103))86+ tracker.mark(event('d', 103))87+ assert tracker.should_accept(event('same-second', 103))88+ assert tracker.should_accept(event('overlap', 102))89+ assert not tracker.should_accept(event('b', 102))90+ assert not tracker.should_accept(event('d', 103))91+ assert not tracker.should_accept(event('a', 101))92+93+94+def test_window_eviction_does_not_raise_floor(tmp_path):95+ # IDs older than the replay window are dropped without touching the floor, so a96+ # large cache never has to retire live seconds.97+ tracker = EventTracker('r', 'a', ['lab'], state_dir=tmp_path, now=100, max_seen=2)98+ tracker.mark(event('old-1', 101))99+ tracker.mark(event('old-2', 102))100+ tracker.mark(event('new', 1000))101+ assert tracker.floor == 100102+ assert set(tracker.seen.values()) == {1000}103+ assert tracker.filter_since()['since'] == 700104+ assert not tracker.should_accept(event('old-1', 101))105+106+107+def test_legacy_seen_list_loads(tmp_path):108+ tracker = EventTracker('r', 'a', ['lab'], state_dir=tmp_path, now=100)109+ legacy_id = event('legacy', 105)['id']110+ tracker.path.write_text(json.dumps({"floor": 100, "cursor": 105, "seen": [legacy_id]}))111+ restored = EventTracker('r', 'a', ['lab'], state_dir=tmp_path, now=200)112+ assert restored.cursor == 105113+ assert not restored.should_accept(event('legacy', 105))114+ assert restored.should_accept(event('fresh', 105))115+ state = json.loads(restored.path.read_text())116+ assert state["seen"] == {legacy_id: 105}117+ assert state["attempts"] == {}118+119+120+def test_failed_handoffs_retire_after_three_attempts(tmp_path):121+ tracker = EventTracker('r', 'a', ['lab'], state_dir=tmp_path, now=100)122+ poison = event('poison', 101)123+ assert not tracker.fail(poison)124+ assert tracker.should_accept(poison)125+ restored = EventTracker('r', 'a', ['lab'], state_dir=tmp_path, now=100)126+ assert restored.attempts == {poison["id"]: 1}127+ assert not restored.fail(poison)128+ assert restored.fail(poison)129+ assert not restored.should_accept(poison)130+ assert restored.attempts == {}131+ assert json.loads(restored.path.read_text())["attempts"] == {}132+133+134+def test_attempt_counts_are_bounded(tmp_path):135+ tracker = EventTracker('r', 'a', ['lab'], state_dir=tmp_path, now=100)136+ for index in range(ingress._MAX_ATTEMPT_ENTRIES + 5):137+ tracker.fail(event(f'fail-{index}', 101))138+ assert len(tracker.attempts) == ingress._MAX_ATTEMPT_ENTRIES139+ assert event('fail-0', 101)["id"] not in tracker.attempts140+141+142+def test_state_root_prefers_hermes_home_env(tmp_path, monkeypatch):143+ monkeypatch.setitem(sys.modules, "hermes_cli", None)144+ monkeypatch.setitem(sys.modules, "hermes_cli.config", None)145+ monkeypatch.setenv("HERMES_HOME", str(tmp_path / "home"))146+ assert ingress.state_root() == tmp_path / "home"147+ monkeypatch.delenv("HERMES_HOME")148+ assert ingress.state_root() == Path("~/.hermes").expanduser()149+ tracker = EventTracker('r', 'a', ['lab'], state_dir=tmp_path)150+ assert tracker.path.parent == tmp_path / "state" / "tinyagent"151+152+153+def test_state_root_uses_hermes_active_home(tmp_path, monkeypatch):154+ import types155+ fake = types.ModuleType("hermes_cli.config")156+ fake.get_hermes_home = lambda: tmp_path / "profile"157+ package = types.ModuleType("hermes_cli")158+ package.config = fake159+ monkeypatch.setitem(sys.modules, "hermes_cli", package)160+ monkeypatch.setitem(sys.modules, "hermes_cli.config", fake)161+ monkeypatch.setenv("HERMES_HOME", str(tmp_path / "ignored"))162+ assert ingress.state_root() == tmp_path / "profile"163+164+165+def test_identity_lock_is_exclusive_in_process(tmp_path):166+ first = IdentityLock("http://relay", "A" * 64, state_dir=tmp_path)167+ second = IdentityLock("http://relay", "a" * 64, state_dir=tmp_path)168+ first.acquire()169+ first.acquire()170+ assert first.held and first.path == second.path171+ with pytest.raises(RuntimeError, match="another Hermes gateway already runs this Tiny identity"):172+ second.acquire()173+ assert not second.held174+ other = IdentityLock("http://relay", "b" * 64, state_dir=tmp_path)175+ other.acquire()176+ first.release()177+ second.acquire()178+ assert second.held and not first.held179+ second.release()180+ other.release()181+182+183+def test_identity_lock_is_exclusive_across_processes(tmp_path):184+ lock = IdentityLock("http://relay", "a" * 64, state_dir=tmp_path)185+ script = textwrap.dedent(f"""186+ import sys187+ sys.path.insert(0, {str(Path(__file__).parents[1])!r})188+ from ingress import IdentityLock189+ lock = IdentityLock("http://relay", "a" * 64, state_dir={str(tmp_path)!r})190+ lock.acquire()191+ print("held", flush=True)192+ sys.stdin.readline()193+ lock.release()194+ """)195+ holder = subprocess.Popen([sys.executable, "-c", script], stdin=subprocess.PIPE, stdout=subprocess.PIPE, text=True)196+ try:197+ assert holder.stdout.readline().strip() == "held"198+ with pytest.raises(RuntimeError):199+ lock.acquire()200+ finally:201+ holder.stdin.write("done\n")202+ holder.stdin.close()203+ holder.wait(timeout=10)204+ lock.acquire()205+ assert lock.held206+ lock.release()