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 hashlib10import json11import os12import re13import tempfile14import time15from pathlib import Path16from typing import Iterable1718_EVENT_ID = re.compile(r"^[0-9a-f]{64}$", re.I)19_REPLAY_WINDOW = 30020_FUTURE_SKEW = 300212223def _tag_values(event: dict, name: str) -> list[list[str]]:24 return [tag for tag in event.get("tags", []) if isinstance(tag, list) and len(tag) >= 2 and tag[0] == name]252627def _expiration(event: dict) -> int | None:28 values = _tag_values(event, "expiration")29 if not values:30 return None31 try:32 return int(values[0][1])33 except (TypeError, ValueError):34 return None353637class EventTracker:38 """Bounded persistent event cursor and ID set for one Tiny subscription."""3940 def __init__(self, relay: str, identity: str, rooms: Iterable[str], *, state_dir: str | Path | None = None,41 now: int | None = None, max_seen: int = 10000):42 self.relay, self.identity = str(relay), str(identity).lower()43 self.rooms = tuple(sorted(str(room) for room in rooms))44 root = Path(state_dir or os.environ.get("HERMES_HOME", "~/.hermes")).expanduser()45 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"47 self.max_seen = max(1, int(max_seen))48 self.now = int(time.time() if now is None else now)49 self.floor = self.now50 self.cursor = self.now51 self.seen: list[str] = []52 self._seen: set[str] = set()53 self._load()54 # Establish the first-start floor immediately, including a clean restart before55 # any event arrives. This prevents a later process from treating its own startup56 # time as an advancing cursor.57 self._save()5859 def _load(self) -> None:60 try:61 data = json.loads(self.path.read_text())62 self.floor = int(data.get("floor", data.get("cursor", self.floor)))63 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):67 return6869 def _save(self) -> None:70 self.path.parent.mkdir(parents=True, exist_ok=True)71 os.chmod(self.path.parent, 0o700)72 payload = json.dumps({"floor": self.floor, "cursor": self.cursor, "seen": self.seen[-self.max_seen:]}, separators=(",", ":"))73 fd, temporary = tempfile.mkstemp(prefix="tinyagent-", dir=self.path.parent)74 try:75 with os.fdopen(fd, "w") as stream:76 stream.write(payload)77 stream.flush()78 os.fsync(stream.fileno())79 os.replace(temporary, self.path)80 finally:81 try:82 os.unlink(temporary)83 except FileNotFoundError:84 pass8586 def should_accept(self, event: dict, *, now: int | None = None) -> bool:87 """Return true only for unseen, non-expired events at/after the cursor."""88 event_id = str(event.get("id", ""))89 if not _EVENT_ID.fullmatch(event_id) or event_id in self._seen:90 return False91 try:92 created = int(event.get("created_at"))93 except (TypeError, ValueError):94 return False95 if created < max(self.floor, self.cursor - _REPLAY_WINDOW):96 return False97 if created > int(time.time() if now is None else now) + _FUTURE_SKEW:98 return False99 expires = _expiration(event)100 if expires is None and any(len(tag) >= 2 and tag[0] == "expiration" for tag in event.get("tags", [])):101 return False102 if expires is not None and expires <= int(time.time() if now is None else now):103 return False104 return True105106 def mark(self, event: dict) -> None:107 """Record a delivered event and atomically persist the cursor."""108 event_id = str(event.get("id", ""))109 if not _EVENT_ID.fullmatch(event_id):110 return111 try:112 self.cursor = max(self.cursor, int(event.get("created_at", self.cursor)))113 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:]123 self._save()124125 def filter_since(self) -> dict:126 """Return an overlapping subscription filter; same-timestamp IDs are retained."""127 # Match should_accept's bounded overlap. A one-second overlap is insufficient128 # when the helper returns events out of order or the gateway was offline.129 result = {"since": max(self.floor, self.cursor - _REPLAY_WINDOW)}130 if self.rooms:131 result["#h"] = list(self.rooms)132 return result133134135def event_thread(event: dict) -> str | None:136 """Return canonical NIP-10 root thread ID, including marked e tags."""137 marked_root = [tag[1] for tag in _tag_values(event, "e") if len(tag) >= 4 and tag[3] == "root"]138 if marked_root:139 return marked_root[0]140 marked_reply = [tag[1] for tag in _tag_values(event, "e") if len(tag) >= 4 and tag[3] == "reply"]141 if marked_reply:142 return marked_reply[0]143 e_tags = _tag_values(event, "e")144 return e_tags[0][1] if e_tags else None145146147def reply_parent(event: dict) -> str | None:148 """Return the direct NIP-10 parent, preferring the marked reply e tag."""149 marked = [tag[1] for tag in _tag_values(event, "e") if len(tag) >= 4 and tag[3] == "reply"]150 return marked[0] if marked else event_thread(event)151152153def is_reply_to_own(event: dict, own_ids: set[str]) -> bool:154 """True when the event's canonical parent or root is one of our event IDs."""155 return any(tag[1] in own_ids for tag in _tag_values(event, "e"))156157158__all__ = ["EventTracker", "event_thread", "reply_parent", "is_reply_to_own"]