1"""Replay-safe ingress bookkeeping for Tiny's Nostr event stream.23This module has no Hermes or network dependencies. State is scoped to the relay,4agent identity and subscribed rooms, so reconnects cannot replay old commands.5"""67from __future__ import annotations89import fcntl10import hashlib11import json12import logging13import os14import re15import tempfile16import time17from pathlib import Path18from typing import Iterable1920_EVENT_ID = re.compile(r"^[0-9a-f]{64}$", re.I)21_REPLAY_WINDOW = 30022_FUTURE_SKEW = 30023# Hard ceiling on remembered IDs, far above anything the replay window holds in practice.24MAX_SEEN = 10000025# A poison event is retired after this many failed hand-offs to Hermes.26MAX_ATTEMPTS = 327_MAX_ATTEMPT_ENTRIES = 2562829logger = logging.getLogger(__name__)303132def state_root(state_dir: str | Path | None = None) -> Path:33 """Return the Hermes home used for adapter state.3435 The active Hermes home wins when the CLI package is importable (profiles may36 override the environment); otherwise HERMES_HOME, then ~/.hermes.37 """38 if state_dir:39 return Path(state_dir).expanduser()40 try:41 from hermes_cli.config import get_hermes_home42 return Path(get_hermes_home()).expanduser()43 except Exception:44 return Path(os.environ.get("HERMES_HOME", "").strip() or "~/.hermes").expanduser()454647def adapter_state_dir(root: str | Path | None = None) -> Path:48 """Return the directory holding tracker state for the active Hermes home."""49 return state_root(root) / "state" / "tinyagent"505152def 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 home616263def lock_dir(root: str | Path | None = None) -> Path:64 """Return the directory holding identity locks.6566 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 fallback848586def _tag_values(event: dict, name: str) -> list[list[str]]:87 return [tag for tag in event.get("tags", []) if isinstance(tag, list) and len(tag) >= 2 and tag[0] == name]888990def _expiration(event: dict) -> int | None:91 values = _tag_values(event, "expiration")92 if not values:93 return None94 try:95 return int(values[0][1])96 except (TypeError, ValueError):97 return None9899100class EventTracker:101 """Persistent event cursor and ID set for one Tiny subscription.102103 State file (JSON): ``floor`` and ``cursor`` are Unix seconds; ``seen`` maps a104 delivered event ID to its ``created_at``; ``attempts`` maps an event ID to the105 number of failed hand-offs. Older files stored ``seen`` as a list of IDs and106 load with every listed ID pinned to the cursor.107108 ``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.112 """113114 def __init__(self, relay: str, identity: str, rooms: Iterable[str], *, state_dir: str | Path | None = None,115 now: int | None = None, max_seen: int = MAX_SEEN):116 self.relay, self.identity = str(relay), str(identity).lower()117 self.rooms = tuple(sorted(str(room) for room in rooms))118 key = hashlib.sha256((self.relay + "\0" + self.identity + "\0" + "\0".join(self.rooms)).encode()).hexdigest()[:24]119 self.path = adapter_state_dir(state_dir) / f"{key}.json"120 self.max_seen = max(1, int(max_seen))121 self.now = int(time.time() if now is None else now)122 self.floor = self.now123 self.cursor = self.now124 self.seen: dict[str, int] = {}125 self.attempts: dict[str, int] = {}126 self._load()127 # Establish the first-start floor immediately, including a clean restart before128 # any event arrives. This prevents a later process from treating its own startup129 # time as an advancing cursor.130 self._save()131132 def _load(self) -> None:133 try:134 data = json.loads(self.path.read_text())135 self.floor = int(data.get("floor", data.get("cursor", self.floor)))136 self.cursor = int(data.get("cursor", self.cursor))137 seen = data.get("seen", {})138 if isinstance(seen, list):139 # Legacy format: IDs without timestamps stay in the replay window.140 self.seen = {str(value): self.cursor for value in seen}141 else:142 self.seen = {str(key): int(value) for key, value in dict(seen).items()}143 attempts = data.get("attempts", {})144 self.attempts = {str(key): int(value) for key, value in dict(attempts).items()}145 except (OSError, ValueError, TypeError, AttributeError):146 return147 self._prune()148149 def _save(self) -> None:150 self.path.parent.mkdir(parents=True, exist_ok=True)151 os.chmod(self.path.parent, 0o700)152 payload = json.dumps({"floor": self.floor, "cursor": self.cursor, "seen": self.seen,153 "attempts": self.attempts}, separators=(",", ":"))154 fd, temporary = tempfile.mkstemp(prefix="tinyagent-", dir=self.path.parent)155 try:156 with os.fdopen(fd, "w") as stream:157 stream.write(payload)158 stream.flush()159 os.fsync(stream.fileno())160 os.replace(temporary, self.path)161 finally:162 try:163 os.unlink(temporary)164 except FileNotFoundError:165 pass166167 def _window_start(self) -> int:168 return max(self.floor, self.cursor - _REPLAY_WINDOW)169170 def _prune(self) -> None:171 """Drop IDs the window already rejects; the hard ceiling alone may raise the floor."""172 start = self._window_start()173 self.seen = {key: stamp for key, stamp in self.seen.items() if stamp >= start}174 if len(self.seen) > self.max_seen:175 stamps = sorted(self.seen.values(), reverse=True)176 threshold = stamps[self.max_seen - 1]177 if sum(1 for stamp in stamps if stamp >= threshold) > self.max_seen:178 # More IDs share the oldest retained second than fit; retire that second.179 threshold += 1180 self.seen = {key: stamp for key, stamp in self.seen.items() if stamp >= threshold}181 self.floor = max(self.floor, threshold)182 logger.warning("Tiny replay cache hit its hard ceiling of %d IDs; events before %d are now ignored",183 self.max_seen, self.floor)184 self.attempts = {key: count for key, count in self.attempts.items() if key not in self.seen}185 while len(self.attempts) > _MAX_ATTEMPT_ENTRIES:186 self.attempts.pop(next(iter(self.attempts)))187188 def should_accept(self, event: dict, *, now: int | None = None) -> bool:189 """Return true only for unseen, non-expired events at/after the cursor."""190 event_id = str(event.get("id", ""))191 if not _EVENT_ID.fullmatch(event_id) or event_id in self.seen:192 return False193 try:194 created = int(event.get("created_at"))195 except (TypeError, ValueError):196 return False197 if created < self._window_start():198 return False199 if created > int(time.time() if now is None else now) + _FUTURE_SKEW:200 return False201 expires = _expiration(event)202 if expires is None and any(len(tag) >= 2 and tag[0] == "expiration" for tag in event.get("tags", [])):203 return False204 if expires is not None and expires <= int(time.time() if now is None else now):205 return False206 return True207208 def mark(self, event: dict) -> None:209 """Record a delivered event and atomically persist the cursor."""210 event_id = str(event.get("id", ""))211 if not _EVENT_ID.fullmatch(event_id):212 return213 try:214 created = int(event.get("created_at", self.cursor))215 except (TypeError, ValueError):216 created = self.cursor217 self.cursor = max(self.cursor, created)218 self.seen.setdefault(event_id, created)219 self.attempts.pop(event_id, None)220 self._prune()221 self._save()222223 def fail(self, event: dict) -> bool:224 """Count a failed hand-off; retire the event after MAX_ATTEMPTS and return True."""225 event_id = str(event.get("id", ""))226 if not _EVENT_ID.fullmatch(event_id):227 return False228 count = self.attempts.pop(event_id, 0) + 1229 if count >= MAX_ATTEMPTS:230 self.mark(event)231 return True232 self.attempts[event_id] = count233 self._prune()234 self._save()235 return False236237 def filter_since(self) -> dict:238 """Return an overlapping subscription filter; same-timestamp IDs are retained."""239 # Match should_accept's bounded overlap. A one-second overlap is insufficient240 # when the helper returns events out of order or the gateway was offline.241 result = {"since": self._window_start()}242 if self.rooms:243 result["#h"] = list(self.rooms)244 return result245246247class IdentityLock:248 """Exclusive per-identity lock so one gateway consumes a Tiny key at a time.249250 The lock file is keyed by relay and public key and lives in ``lock_dir()``, which251 every profile of one Hermes installation shares.252 """253254 def __init__(self, relay: str, pubkey: str, *, state_dir: str | Path | None = None):255 key = hashlib.sha256((str(relay) + "\0" + str(pubkey).lower()).encode()).hexdigest()[:24]256 self.path = lock_dir(state_dir) / f"{key}.lock"257 self._fd: int | None = None258259260 def held(self) -> bool:261 return self._fd is not None262263 def acquire(self) -> None:264 if self._fd is not None:265 return266 self.path.parent.mkdir(parents=True, exist_ok=True)267 os.chmod(self.path.parent, 0o700)268 fd = os.open(self.path, os.O_RDWR | os.O_CREAT, 0o600)269 try:270 fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)271 except OSError:272 os.close(fd)273 raise RuntimeError("another Hermes gateway already runs this Tiny identity") from None274 try:275 os.ftruncate(fd, 0)276 os.write(fd, str(os.getpid()).encode())277 except OSError:278 pass279 self._fd = fd280281 def release(self) -> None:282 if self._fd is None:283 return284 fd, self._fd = self._fd, None285 try:286 fcntl.flock(fd, fcntl.LOCK_UN)287 finally:288 os.close(fd)289290291def event_thread(event: dict) -> str | None:292 """Return canonical NIP-10 root thread ID, including marked e tags."""293 marked_root = [tag[1] for tag in _tag_values(event, "e") if len(tag) >= 4 and tag[3] == "root"]294 if marked_root:295 return marked_root[0]296 marked_reply = [tag[1] for tag in _tag_values(event, "e") if len(tag) >= 4 and tag[3] == "reply"]297 if marked_reply:298 return marked_reply[0]299 e_tags = _tag_values(event, "e")300 return e_tags[0][1] if e_tags else None301302303def reply_parent(event: dict) -> str | None:304 """Return the direct NIP-10 parent, preferring the marked reply e tag."""305 marked = [tag[1] for tag in _tag_values(event, "e") if len(tag) >= 4 and tag[3] == "reply"]306 return marked[0] if marked else event_thread(event)307308309def is_reply_to_own(event: dict, own_ids: set[str]) -> bool:310 """True when the event's canonical parent or root is one of our event IDs."""311 return any(tag[1] in own_ids for tag in _tag_values(event, "e"))312313314__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"]