1import json2import hashlib3import os4import subprocess5import sys6import textwrap7from pathlib import Path89import pytest1011sys.path.insert(0, str(Path(__file__).parents[1]))1213import ingress14from ingress import EventTracker, IdentityLock, event_thread, is_reply_to_own, reply_parent151617def event(event_id, created=100, tags=None):18 return {"id": hashlib.sha256(event_id.encode()).hexdigest(), "created_at": created, "tags": tags or []}192021def test_first_tracker_does_not_replay_history(tmp_path):22 tracker = EventTracker("http://relay", "a" * 64, ["lab"], state_dir=tmp_path, now=100)23 assert not tracker.should_accept(event("old", 99))24 assert tracker.should_accept(event("same", 100))252627def test_persisted_cursor_overlaps_timestamp_and_deduplicates(tmp_path):28 first = EventTracker("r", "a", ["lab"], state_dir=tmp_path, now=100)29 first.mark(event("one", 101))30 second = EventTracker("r", "a", ["lab"], state_dir=tmp_path, now=100)31 assert second.filter_since()["since"] == 10032 assert second.should_accept(event("two", 101))33 assert not second.should_accept(event("one", 101))343536def test_expiration_and_the_hard_ceiling(tmp_path, caplog):37 tracker = EventTracker("r", "a", ["lab"], state_dir=tmp_path, now=100, max_seen=2)38 assert not tracker.should_accept(event("expired", 100, [["expiration", "100"]]))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.45 state = json.loads(tracker.path.read_text())46 assert len(state["seen"]) == 247 assert set(state["seen"].values()) == {102, 103}48 assert state["floor"] == 10249 assert any("hard ceiling of 2" in record.getMessage() for record in caplog.records)505152def 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)656667def test_nip10_root_reply_and_own_parent():68 root = "r" * 6469 parent = "p" * 6470 child = event("c", tags=[["e", root, "", "root"], ["e", parent, "", "reply"]])71 assert event_thread(child) == root72 assert reply_parent(child) == parent73 assert is_reply_to_own(child, {parent})747576def test_overlap_accepts_out_of_order_events_and_survives_restart(tmp_path):77 tracker = EventTracker('r', 'a', ['lab'], state_dir=tmp_path, now=100)78 tracker.mark(event('newer', 400))79 assert tracker.should_accept(event('earlier', 200))80 tracker.mark(event('earlier', 200))81 restored = EventTracker('r', 'a', ['lab'], state_dir=tmp_path, now=500)82 assert restored.filter_since()['since'] == 10083 assert not restored.should_accept(event('earlier', 200))84 assert restored.should_accept(event('same-time', 400))858687def test_evicted_ids_cannot_become_commands_again(tmp_path):88 tracker = EventTracker('r', 'a', ['lab'], state_dir=tmp_path, now=100, max_seen=2)89 for i in range(3): tracker.mark(event(str(i), 101 + i))90 # ID 0 (second 101) was retired by the overflow; the floor now rejects that second.91 assert not tracker.should_accept(event('0', 101))92 assert not tracker.should_accept(event('1', 102))93 assert tracker.should_accept(event('next', 104))949596def test_overflow_keeps_same_second_and_replay_window_events(tmp_path):97 # Overflowing the ID cache may retire old IDs, but never events the relay can98 # still legitimately deliver: unseen events in the cursor's own second and99 # out-of-order events inside the replay window.100 tracker = EventTracker('r', 'a', ['lab'], state_dir=tmp_path, now=100, max_seen=3)101 tracker.mark(event('a', 101))102 tracker.mark(event('b', 102))103 tracker.mark(event('c', 103))104 tracker.mark(event('d', 103))105 assert tracker.should_accept(event('same-second', 103))106 assert tracker.should_accept(event('overlap', 102))107 assert not tracker.should_accept(event('b', 102))108 assert not tracker.should_accept(event('d', 103))109 assert not tracker.should_accept(event('a', 101))110111112def test_window_eviction_does_not_raise_floor(tmp_path):113 # IDs older than the replay window are dropped without touching the floor, so a114 # large cache never has to retire live seconds.115 tracker = EventTracker('r', 'a', ['lab'], state_dir=tmp_path, now=100, max_seen=2)116 tracker.mark(event('old-1', 101))117 tracker.mark(event('old-2', 102))118 tracker.mark(event('new', 1000))119 assert tracker.floor == 100120 assert set(tracker.seen.values()) == {1000}121 assert tracker.filter_since()['since'] == 700122 assert not tracker.should_accept(event('old-1', 101))123124125def test_legacy_seen_list_loads(tmp_path):126 tracker = EventTracker('r', 'a', ['lab'], state_dir=tmp_path, now=100)127 legacy_id = event('legacy', 105)['id']128 tracker.path.write_text(json.dumps({"floor": 100, "cursor": 105, "seen": [legacy_id]}))129 restored = EventTracker('r', 'a', ['lab'], state_dir=tmp_path, now=200)130 assert restored.cursor == 105131 assert not restored.should_accept(event('legacy', 105))132 assert restored.should_accept(event('fresh', 105))133 state = json.loads(restored.path.read_text())134 assert state["seen"] == {legacy_id: 105}135 assert state["attempts"] == {}136137138def test_failed_handoffs_retire_after_three_attempts(tmp_path):139 tracker = EventTracker('r', 'a', ['lab'], state_dir=tmp_path, now=100)140 poison = event('poison', 101)141 assert not tracker.fail(poison)142 assert tracker.should_accept(poison)143 restored = EventTracker('r', 'a', ['lab'], state_dir=tmp_path, now=100)144 assert restored.attempts == {poison["id"]: 1}145 assert not restored.fail(poison)146 assert restored.fail(poison)147 assert not restored.should_accept(poison)148 assert restored.attempts == {}149 assert json.loads(restored.path.read_text())["attempts"] == {}150151152def test_attempt_counts_are_bounded(tmp_path):153 tracker = EventTracker('r', 'a', ['lab'], state_dir=tmp_path, now=100)154 for index in range(ingress._MAX_ATTEMPT_ENTRIES + 5):155 tracker.fail(event(f'fail-{index}', 101))156 assert len(tracker.attempts) == ingress._MAX_ATTEMPT_ENTRIES157 assert event('fail-0', 101)["id"] not in tracker.attempts158159160def test_state_root_prefers_hermes_home_env(tmp_path, monkeypatch):161 monkeypatch.setitem(sys.modules, "hermes_cli", None)162 monkeypatch.setitem(sys.modules, "hermes_cli.config", None)163 monkeypatch.setenv("HERMES_HOME", str(tmp_path / "home"))164 assert ingress.state_root() == tmp_path / "home"165 monkeypatch.delenv("HERMES_HOME")166 assert ingress.state_root() == Path("~/.hermes").expanduser()167 tracker = EventTracker('r', 'a', ['lab'], state_dir=tmp_path)168 assert tracker.path.parent == tmp_path / "state" / "tinyagent"169170171def test_state_root_uses_hermes_active_home(tmp_path, monkeypatch):172 import types173 fake = types.ModuleType("hermes_cli.config")174 fake.get_hermes_home = lambda: tmp_path / "profile"175 package = types.ModuleType("hermes_cli")176 package.config = fake177 monkeypatch.setitem(sys.modules, "hermes_cli", package)178 monkeypatch.setitem(sys.modules, "hermes_cli.config", fake)179 monkeypatch.setenv("HERMES_HOME", str(tmp_path / "ignored"))180 assert ingress.state_root() == tmp_path / "profile"181182183def 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.path205206207def 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)223224225def test_identity_lock_is_exclusive_in_process(tmp_path):226 first = IdentityLock("http://relay", "A" * 64, state_dir=tmp_path)227 second = IdentityLock("http://relay", "a" * 64, state_dir=tmp_path)228 first.acquire()229 first.acquire()230 assert first.held and first.path == second.path231 with pytest.raises(RuntimeError, match="another Hermes gateway already runs this Tiny identity"):232 second.acquire()233 assert not second.held234 other = IdentityLock("http://relay", "b" * 64, state_dir=tmp_path)235 other.acquire()236 first.release()237 second.acquire()238 assert second.held and not first.held239 second.release()240 other.release()241242243def test_identity_lock_is_exclusive_across_processes(tmp_path):244 lock = IdentityLock("http://relay", "a" * 64, state_dir=tmp_path)245 script = textwrap.dedent(f"""246 import sys247 sys.path.insert(0, {str(Path(__file__).parents[1])!r})248 from ingress import IdentityLock249 lock = IdentityLock("http://relay", "a" * 64, state_dir={str(tmp_path)!r})250 lock.acquire()251 print("held", flush=True)252 sys.stdin.readline()253 lock.release()254 """)255 holder = subprocess.Popen([sys.executable, "-c", script], stdin=subprocess.PIPE, stdout=subprocess.PIPE, text=True)256 try:257 assert holder.stdout.readline().strip() == "held"258 with pytest.raises(RuntimeError):259 lock.acquire()260 finally:261 holder.stdin.write("done\n")262 holder.stdin.close()263 holder.wait(timeout=10)264 lock.acquire()265 assert lock.held266 lock.release()