1"""Exercise real Hermes waiters through the Tiny adapter's public hooks."""23import asyncio4import json5import sys6import time7from pathlib import Path89import pytest1011sys.path.insert(0, str(Path(__file__).parents[1]))12from adapter import TinyAdapter13from gateway.config import PlatformConfig14from gateway.platform_registry import PlatformEntry, platform_registry15from tools import approval, clarify_gateway, slash_confirm16from tools.approval_gateway_wait import _ApprovalEntry1718OWNER = "b" * 6419BOT = "a" * 6420OTHER = "c" * 64212223def admit(message):24 """What Hermes's handle_message records once it starts or queues a turn."""25 message._gateway_accepted = True262728class FakeRPC:29 def __init__(self):30 self.events = []31 self.rows = []3233 async def query(self, filter):34 return [row for row in self.rows if row["id"] in filter.get("ids", [])]3536 async def publish(self, event):37 result = {"id": f"{len(self.events) + 1:064x}", "created_at": int(time.time()), "pubkey": BOT, **event}38 self.events.append(result)39 return result4041 async def query(self, _filter):42 return []43444546def adapter(monkeypatch, tmp_path):47 monkeypatch.setenv("HERMES_HOME", str(tmp_path))48 platform_registry.register(PlatformEntry(name="tiny", label="Tiny", adapter_factory=TinyAdapter,49 check_fn=lambda: True))50 instance = TinyAdapter(PlatformConfig(enabled=True, extra={"rooms": "room", "allowed_users": f"{OWNER},{OTHER}"}))51 instance.rpc = FakeRPC()52 instance.pubkey = BOT53 instance._session_assignees["session"] = OWNER54 yield instance55 clarify_gateway._entries.clear()56 clarify_gateway._session_index.clear()57 approval._gateway_queues.clear()58 slash_confirm._pending.clear()596061def response(request, content, author=OWNER):62 event_id = request["id"]63 return {"kind": 1111, "pubkey": author, "created_at": int(time.time()), "content": content,64 "tags": [["h", "room"], ["e", event_id], ["E", event_id], ["p", BOT], ["P", BOT],65 ["k", str(request["kind"])], ["K", str(request["kind"])]]}666768def test_real_inbound_source_tracks_human_session(adapter):69 async def run():70 received = []71 async def handle(message):72 received.append(message)73 adapter.handle_message = handle74 await adapter._on_event({"kind": 9, "id": "d" * 64, "pubkey": OWNER, "created_at": int(time.time()),75 "content": "hello", "tags": [["h", "room"], ["p", BOT]]})76 assert len(received) == 177 key = adapter._event_session_key(received[0])78 assert adapter._session_assignees[key] == OWNER79 assert received[0].source.platform.value == "tiny"80 asyncio.run(run())8182838485868788def test_clarify_resolves_exact_real_waiter(adapter, multiple, content, expected):89 async def run():90 entry = clarify_gateway.register("question-id", "session", "Pick", ["alpha", "beta"], multiple)91 untouched = clarify_gateway.register("other-question", "session", "Other", ["yes"])92 result = await adapter.send_clarify("room", "Pick", ["alpha", "beta"], "question-id", "session")93 assert result.success94 request = adapter.rpc.events[-1]95 assert ["p", OWNER] in request["tags"]96 assert not any(tag[0] == "mention" for tag in request["tags"])97 assert ["freeform", "true"] in request["tags"]98 assert all("session" not in str(tag) for tag in request["tags"])99 await adapter._on_answer(response(request, content))100 assert entry.event.is_set() and entry.response == expected101 assert not untouched.event.is_set()102 asyncio.run(run())103104105106def test_invalid_answers_do_not_resolve(adapter, mutation):107 async def run():108 entry = clarify_gateway.register("question-id", "session", "Pick", ["alpha", "beta"], True)109 await adapter.send_clarify("room", "Pick", ["alpha", "beta"], "question-id", "session")110 request = adapter.rpc.events[-1]111 answer = response(request, '["c0"]')112 if mutation == "actor": answer["pubkey"] = OTHER113 if mutation == "room": answer["tags"][0][1] = "elsewhere"114 if mutation == "root": answer["tags"][2][1] = "f" * 64115 if mutation == "kind": answer["kind"] = 7116 if mutation == "invalid": answer["content"] = '["unknown"]'117 if mutation == "duplicate": answer["content"] = '["c0","c0"]'118 if mutation == "expired": adapter._pending[request["id"]].expires = int(time.time()) - 1119 await adapter._on_answer(answer)120 assert not entry.event.is_set()121 asyncio.run(run())122123124def test_open_ended_question_is_structured(adapter):125 async def run():126 entry = clarify_gateway.register("question-id", "session", "Explain", None)127 await adapter.send_clarify("room", "Explain", None, "question-id", "session")128 request = adapter.rpc.events[-1]129 assert ["selection", "text"] in request["tags"]130 await adapter._on_answer(response(request, "Here is my answer"))131 assert entry.response == "Here is my answer"132 asyncio.run(run())133134135def test_approval_uses_exact_request_and_offered_scope(adapter):136 async def run():137 first = _ApprovalEntry({"command": "first", "request_id": "first-id"})138 second = _ApprovalEntry({"command": "second", "request_id": "second-id"})139 approval._gateway_queues["session"] = [first, second]140 assert (await adapter.send_exec_approval("room", "second", "session",141 allow_permanent=False, allow_session=False)).success142 request = adapter.rpc.events[-1]143 await adapter._on_answer(response(request, "always"))144 assert not first.event.is_set() and not second.event.is_set()145 await adapter._on_answer(response(request, "once"))146 assert not first.event.is_set() and second.event.is_set()147 assert second.result == "once"148 await adapter._on_answer(response(request, "once"))149 assert not first.event.is_set()150 asyncio.run(run())151152153def test_confirmation_calls_real_handler_once(adapter):154 async def run():155 choices = []156 async def handler(choice):157 choices.append(choice)158 return "Reloaded"159 slash_confirm.register("session", "confirm-id", "reload", handler)160 await adapter.send_slash_confirm("room", "Reload?", "Confirm", "session", "confirm-id")161 request = adapter.rpc.events[-1]162 await adapter._on_answer(response(request, "once"))163 await adapter._on_answer(response(request, "once"))164 assert choices == ["once"]165 assert adapter.rpc.events[-1]["content"] == "Reloaded"166 asyncio.run(run())167168169def test_explicit_mentions_and_edits(adapter):170 async def run():171 result = await adapter.send("room", "ordinary", reply_to="d" * 64)172 assert not any(tag[0] in ("p", "mention") for tag in adapter.rpc.events[-1]["tags"])173 edited = await adapter.edit_message("room", result.message_id, f"Attention @{OWNER}")174 assert edited.message_id == result.message_id175 event = adapter.rpc.events[-1]176 assert event["kind"] == 40003177 assert ["p", OWNER] in event["tags"] and ["mention", OWNER] in event["tags"]178 assert not await adapter.delete_message("room", result.message_id)179 asyncio.run(run())180181182def test_native_prompt_mentions_assignee_only_when_explicit(adapter):183 async def run():184 clarify_gateway.register("q", "session", "Pick", ["yes"])185 await adapter.send_clarify("room", f"@{OWNER} Pick", ["yes"], "q", "session")186 assert ["mention", OWNER] in adapter.rpc.events[-1]["tags"]187 asyncio.run(run())188189190def test_native_prompt_stays_in_current_thread(adapter):191 async def run():192 clarify_gateway.register("q", "session", "Pick", ["yes"])193 result = await adapter.send_clarify("room", "Pick", ["yes"], "q", "session",194 metadata={"thread_id": "r" * 64, "message_id": "m" * 64})195 assert result.success196 event = adapter.rpc.events[-1]197 assert event["kind"] == 12198 assert ["e", "r" * 64, "", "root"] in event["tags"]199 assert ["e", "m" * 64, "", "reply"] in event["tags"]200 asyncio.run(run())201202203204def test_real_inbound_reply_keeps_prompt_and_answer_in_existing_thread(adapter, kind):205 async def run():206 root, parent, incoming = "d" * 64, "e" * 64, "f" * 64207 adapter.rpc.rows = [{"id": parent, "kind": kind, "pubkey": BOT, "content": "Earlier reply",208 "tags": [["h", "room"], ["e", root, "", "root"]]}]209 async def handle(message):210 session = adapter._event_session_key(message)211 assert (message.source.thread_id is None) == (kind == 9)212 entry = clarify_gateway.register("thread-question", session, "Pick", ["Yes", "No"])213 # A newer queued message must not steal the active turn's assignee.214 adapter._session_assignees[session] = OTHER215 current_route = adapter._turn_routes[session]216 adapter._turn_routes[session] = {**current_route, "user_id": OTHER, "reply_to_message_id": "1" * 64}217 # Hermes forwards only thread_id for Tiny, and nothing for kind 9.218 metadata = {"thread_id": message.source.thread_id} if message.source.thread_id else None219 await adapter.send_clarify("room", "Pick", ["Yes", "No"], "thread-question", session, metadata)220 request = adapter.rpc.events[-1]221 assert ["p", OWNER] in request["tags"]222 assert request["kind"] == kind223 assert ["e", root, "", "root"] in request["tags"]224 assert ["e", incoming, "", "reply"] in request["tags"]225 await adapter._on_answer(response(request, '{"choices":["c1"],"text":"Keep this thread."}'))226 assert entry.event.is_set() and entry.response == "No\nKeep this thread."227 adapter._turn_routes[session] = current_route228 await adapter.send("room", "Acknowledged", metadata=metadata)229 assert ["e", root, "", "root"] in adapter.rpc.events[-1]["tags"]230 adapter.set_message_handler(handle)231 adapter.handle_message = adapter._message_handler232 await adapter._on_event({"kind": kind, "id": incoming, "pubkey": OWNER,233 "created_at": int(time.time()), "content": "Ask here", "tags": [234 ["h", "room"], ["p", BOT], ["e", root, "", "root"], ["e", parent, "", "reply"]]})235 # The final gateway send occurs after the handler completes.236 await adapter.send("room", "Final reply", reply_to=incoming)237 event = adapter.rpc.events[-1]238 assert event["kind"] == kind239 assert ["e", root, "", "root"] in event["tags"]240 assert ["e", incoming, "", "reply"] in event["tags"]241 assert not adapter._turn_routes and adapter._turn_route.get() is None242 asyncio.run(run())243244245def test_gateway_final_attachment_keeps_the_turn_route(adapter, tmp_path):246 async def run():247 import hashlib248 root, incoming = "d" * 64, "e" * 64249 note = tmp_path / "answer.txt"250 note.write_text("Details for this answer.")251 async def upload(room, filename, mime, data):252 return {"url": "https://tiny.test/answer.txt", "sha256": hashlib.sha256(data).hexdigest()}253 adapter.rpc.upload = upload254 async def handler(message):255 return f"Here is the answer.\nMEDIA:{note}"256 adapter.set_message_handler(handler)257 async def process(message):258 await adapter._process_message_background(message, adapter._event_session_key(message))259 adapter.handle_message = process260 await adapter._on_event({"kind": 9, "id": incoming, "pubkey": OWNER,261 "created_at": int(time.time()), "content": "Respond here", "tags": [262 ["h", "room"], ["p", BOT], ["e", root, "", "root"]]})263 replies = [event for event in adapter.rpc.events if event["kind"] == 9]264 assert len(replies) > 1265 assert all(["e", root, "", "root"] in event["tags"] for event in replies)266 assert not adapter._turn_routes and adapter._turn_route.get() is None267 asyncio.run(run())268269270def test_threaded_kind9_reply_keeps_root_session(adapter):271 async def run():272 received = []273 async def handle(message):274 received.append(message)275 adapter.handle_message = handle276 root = "r" * 64277 event_id = "e" * 64278 await adapter._on_event({"kind": 9, "id": event_id, "pubkey": OWNER,279 "created_at": int(time.time()), "content": "follow-up",280 "tags": [["h", "room"], ["e", root], ["p", BOT]]})281 assert len(received) == 1282 assert received[0].source.thread_id is None283 assert received[0].reply_to_message_id == root284 asyncio.run(run())285286287def test_threaded_question_callback_uses_kind12_correlation(adapter):288 async def run():289 entry = clarify_gateway.register("question-id", "session", "Pick", ["yes"])290 await adapter.send_clarify("room", "Pick", ["yes"], "question-id", "session",291 metadata={"thread_id": "r" * 64})292 request = adapter.rpc.events[-1]293 await adapter._on_answer(response(request, "c0"))294 assert entry.response == "yes"295 asyncio.run(run())296297298def test_outbound_document_preserves_thread_caption_and_bytes(adapter, tmp_path):299 import hashlib300 data = b'attachment bytes'301 path = tmp_path / 'note.txt'302 path.write_bytes(data)303 digest = hashlib.sha256(data).hexdigest()304 uploads = []305 async def upload(room, name, mime, body):306 uploads.append((room, name, mime, body))307 return {'url': f'http://tiny/media/{digest}', 'sha256': digest}308 adapter.rpc.upload = upload309 result = asyncio.run(adapter.send_document(chat_id='room', file_path=str(path),310 file_name='renamed.txt', caption=f'For @{OWNER}', metadata={'thread_id': 'd' * 64}, reply_to='e' * 64))311 assert result.success312 event = adapter.rpc.events[-1]313 assert uploads == [('room', 'renamed.txt', 'text/plain', data)]314 assert event['kind'] == 12315 assert ['e', 'd' * 64, '', 'root'] in event['tags']316 assert ['e', 'e' * 64, '', 'reply'] in event['tags']317 assert ['p', OWNER] in event['tags'] and ['mention', OWNER] in event['tags']318 assert any(t[0] == 'imeta' and f'x {digest}' in t and 'size 16' in t for t in event['tags'])319320321def test_standalone_delivery_uses_native_media(adapter, tmp_path, monkeypatch):322 import adapter as module323 import hashlib324 path = tmp_path / 'voice.mp3'325 path.write_bytes(b'audio bytes')326 rpc = adapter.rpc327 async def connect(): pass328 async def close(): pass329 async def upload(room, name, mime, body):330 digest = hashlib.sha256(body).hexdigest()331 return {'url': f'http://tiny/media/{digest}', 'sha256': digest}332 rpc.connect, rpc.close, rpc.upload = connect, close, upload333 monkeypatch.setattr(module, 'TinyRPC', lambda *a, **kw: rpc)334 result = asyncio.run(module._standalone_send(adapter.config, 'room', f'For @{OWNER}',335 thread_id='d' * 64, media_files=[(str(path), True)]))336 assert result['success'] and result['media_delivered']337 assert len(rpc.events) == 1338 event = rpc.events[0]339 assert event['kind'] == 12340 assert ['mention', OWNER] in event['tags']341 assert any(t[0] == 'imeta' and 'm audio/mpeg' in t for t in event['tags'])342343344def test_outbound_media_rejects_wrong_relay_hash(adapter, tmp_path):345 path = tmp_path / 'note.txt'346 path.write_text('local content')347 async def upload(*args): return {'url': 'http://tiny/media/' + '0' * 64, 'sha256': '0' * 64}348 adapter.rpc.upload = upload349 result = asyncio.run(adapter.send_document('room', str(path)))350 assert not result.success and adapter.rpc.events == []351352353def test_duplicate_callbacks_cannot_race_before_media_intake(adapter, monkeypatch):354 import adapter as module355 received = []356 async def media(*args):357 await asyncio.sleep(0.01)358 return [], []359 async def handle(message): received.append(message)360 monkeypatch.setattr(module, 'incoming_media', media)361 adapter.handle_message = handle362 event = {'kind': 9, 'id': 'e' * 64, 'pubkey': OWNER, 'created_at': int(time.time()),363 'content': 'hello', 'tags': [['h', 'room'], ['p', BOT]]}364 async def run(): await asyncio.gather(adapter._on_event(event), adapter._on_event(event))365 asyncio.run(run())366 assert len(received) == 1367368369370def test_room_message_trigger_respects_require_mention(adapter, setting, expected):371 instance = TinyAdapter(PlatformConfig(enabled=True, extra={372 'rooms': 'room', 'allowed_users': OWNER, 'require_mention': setting}))373 instance.pubkey = BOT374 instance.rpc = adapter.rpc375 received = []376 async def handle(message): received.append(message)377 instance.handle_message = handle378 async def run():379 for author in (OWNER, OTHER):380 await instance._on_event({'kind': 9, 'id': author, 'pubkey': author,381 'created_at': int(time.time()), 'content': 'hello', 'tags': [['h', 'room']]})382 asyncio.run(run())383 assert len(received) == expected384 assert all(message.source.user_id == OWNER for message in received)385386387388389390391392393394def test_public_approval_hook_preserves_scopes_on_older_hermes(adapter, monkeypatch, legacy, session, permanent, smart, expected):395 from gateway.platforms.base import BasePlatformAdapter396 if legacy:397 monkeypatch.delattr(BasePlatformAdapter, 'send_exec_approval', raising=False)398 async def run():399 entry = _ApprovalEntry({'command': 'safe preview', 'request_id': 'scoped-id'})400 approval._gateway_queues['session'] = [entry]401 result = await adapter.send_exec_approval('room', 'safe preview', 'session',402 allow_session=session, allow_permanent=permanent, smart_denied=smart)403 assert result.success404 request = adapter.rpc.events[-1]405 assert {t[1] for t in request['tags'] if t[0] == 'option'} == expected406 assert not any(t[0] == 'mention' for t in request['tags'])407 await adapter._on_answer(response(request, 'once'))408 assert entry.event.is_set() and entry.result == 'once'409 asyncio.run(run())410411412class FakeHelper:413 """Stands in for TinyRPC: answers identity, records filters and dies on demand."""414415 instances = []416 fail_identity = False417418 def __init__(self, relay_url, key_env="TINY_PRIVATE_KEY", cli_path="tinyagent", on_exit=None):419 self.relay_url, self.on_exit = relay_url, on_exit420 self.filters, self.subscriptions, self.events = [], {}, []421 self.closed = self.dead = False422 FakeHelper.instances.append(self)423424425 def alive(self):426 return not (self.closed or self.dead)427428 async def call(self, method, params=None, timeout=30):429 if FakeHelper.fail_identity:430 raise RuntimeError("helper unavailable")431 return {"pubkey": BOT} if method == "identity" else {}432433 async def subscribe(self, name, filter, callback):434 self.filters.append(filter)435 self.subscriptions[name] = callback436437 async def unsubscribe(self, name):438 self.subscriptions.pop(name, None)439440 async def query(self, _filter):441 return []442443 async def close(self):444 self.closed = True445446 def die(self):447 self.dead = True448 self.on_exit(self)449450451452def helper(monkeypatch):453 import adapter as module454 FakeHelper.instances.clear()455 FakeHelper.fail_identity = False456 monkeypatch.setattr(module, "TinyRPC", FakeHelper)457 yield FakeHelper458 FakeHelper.fail_identity = False459460461def live_adapter(marks=None):462 instance = TinyAdapter(PlatformConfig(enabled=True, extra={"relay_url": "http://relay", "rooms": "room",463 "allowed_users": OWNER}))464 instance.RECONNECT_MIN = 0.01465 instance.RECONNECT_MAX = 0.03466 if marks is not None:467 original = instance._mark_connected468 def marked(**kwargs):469 marks.append("connected")470 original(**kwargs)471 instance._mark_connected = marked472 return instance473474475def test_helper_death_reconnects_from_persisted_cursor(adapter, helper, tmp_path):476 from ingress import EventTracker477 marks = []478 instance = live_adapter(marks)479 now = int(time.time())480 # A tracker with history: the floor is old and the cursor moves with delivered events.481 instance._tracker = EventTracker("http://relay", BOT, ["room"], now=now - 1000)482 received = []483 async def handle(message):484 received.append(message)485 admit(message)486 instance.handle_message = handle487488 async def run():489 assert await instance.connect()490 first = helper.instances[-1]491 assert instance.is_connected and instance._lock.held492 assert first.filters == [{"kinds": [9, 11, 12, 1111, 43001, 43005], "since": now - 1000, "#h": ["room"]}]493 await first.subscriptions["tinyagent"]({"kind": 9, "id": "d" * 64, "pubkey": OWNER, "created_at": now,494 "content": "hello", "tags": [["h", "room"], ["p", BOT]]})495 assert len(received) == 1 and instance._tracker.cursor == now496 first.die()497 assert not instance.is_connected498 await asyncio.wait_for(instance._reconnect_task, 2)499 second = helper.instances[-1]500 assert second is not first and first.closed and instance.rpc is second501 assert second.filters == [{"kinds": [9, 11, 12, 1111, 43001, 43005], "since": now - 300, "#h": ["room"]}]502 assert instance.is_connected and instance._lock.held and marks == ["connected", "connected"]503 # The same tracker persists across the reconnect: the delivered event stays deduplicated.504 assert not instance._tracker.should_accept({"id": "d" * 64, "created_at": now, "tags": []})505 await instance.disconnect()506 assert not instance.is_connected and instance._lock is None and second.closed507 asyncio.run(run())508509510def test_reconnect_backs_off_and_disconnect_cancels_it(adapter, helper, monkeypatch):511 import adapter as module512 instance = live_adapter()513 delays = []514 real_sleep = asyncio.sleep515 async def sleep(delay, *args):516 delays.append(delay)517 await real_sleep(0)518 monkeypatch.setattr(module.asyncio, "sleep", sleep)519520 async def run():521 await instance.connect()522 first = helper.instances[-1]523 helper.fail_identity = True524 first.die()525 task = instance._reconnect_task526 for _ in range(12):527 await real_sleep(0)528 assert delays[:4] == [0.01, 0.02, 0.03, 0.03]529 assert not instance.is_connected530 await instance.disconnect()531 assert task.cancelled() and instance._reconnect_task is None532 spawned = len(helper.instances)533 helper.fail_identity = False534 for _ in range(5):535 await real_sleep(0)536 assert len(helper.instances) == spawned537 assert not instance.is_connected538 asyncio.run(run())539540541def test_watcher_reconnect_uses_the_same_path(adapter, helper):542 instance = live_adapter()543544 async def run():545 await instance.connect()546 first = helper.instances[-1]547 tracker = instance._tracker548 await instance.disconnect()549 assert first.closed and not instance.is_connected550 # Hermes's watcher calls connect(is_reconnect=True); the persisted cursor is reused.551 assert await instance.connect(is_reconnect=True)552 second = helper.instances[-1]553 assert second is not first and instance.is_connected and instance._lock.held554 assert second.filters[0]["since"] == tracker.filter_since()["since"]555 await instance.disconnect()556 asyncio.run(run())557558559def test_identity_lock_refuses_a_second_gateway(adapter, helper):560 first, second = live_adapter(), live_adapter()561562 async def run():563 await first.connect()564 with pytest.raises(RuntimeError, match="another Hermes gateway already runs this Tiny identity"):565 await second.connect()566 assert helper.instances[-1].closed and not second.is_connected and second._lock is None567 await first.disconnect()568 assert await second.connect()569 assert second.is_connected570 await second.disconnect()571 asyncio.run(run())572573574def test_event_is_marked_only_after_hermes_accepts_it(adapter, tmp_path):575 from ingress import EventTracker576 now = int(time.time())577 adapter._tracker = EventTracker("http://relay", BOT, ["room"], state_dir=tmp_path, now=now - 10)578 poison = {"kind": 9, "id": "d" * 64, "pubkey": OWNER, "created_at": now, "content": "boom",579 "tags": [["h", "room"], ["p", BOT]]}580 attempts = []581 async def failing(message):582 attempts.append(message.message_id)583 raise RuntimeError("Hermes rejected the turn")584 adapter.handle_message = failing585 for attempt in (1, 2):586 with pytest.raises(RuntimeError):587 asyncio.run(adapter._on_event(poison))588 assert adapter._tracker.should_accept(poison)589 assert adapter._tracker.attempts == {poison["id"]: attempt}590 with pytest.raises(RuntimeError):591 asyncio.run(adapter._on_event(poison))592 assert len(attempts) == 3593 assert not adapter._tracker.should_accept(poison) and adapter._tracker.attempts == {}594 asyncio.run(adapter._on_event(poison))595 assert len(attempts) == 3596 handled = []597 async def accept(message):598 handled.append(message.message_id)599 admit(message)600 adapter.handle_message = accept601 good = {**poison, "id": "e" * 64, "content": "hello"}602 asyncio.run(adapter._on_event(good))603 asyncio.run(adapter._on_event(good))604 assert handled == [good["id"]]605 assert not adapter._tracker.should_accept(good)606 assert json.loads(adapter._tracker.path.read_text())["seen"] == {poison["id"]: now, good["id"]: now}607608609def 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)647648649def 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}661662663class RacingProcess:664 """A helper that delivers subscription frames before acknowledging the subscribe call,665 which the Go helper's scheduling permits."""666667 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 = False674675676 def returncode(self):677 return self.inner.returncode678679 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())700701 def terminate(self):702 self.terminated = True703 self.inner.terminate()704705 async def wait(self):706 return self.inner.returncode707708709class Racing:710 """Plans the next helper process; the process itself is built inside the running loop."""711712 def __init__(self):713 self.kwargs = {}714 self.process = None715716 def __call__(self, **kwargs):717 self.kwargs = kwargs718 return self719720721 def terminated(self):722 return self.process is not None and self.process.terminated723724725 def returncode(self):726 return None if self.process is None else self.process.returncode727728729730def 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 plan738739740def 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 = guarded761762 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())773774775776def 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 = handle788789 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())802803804def test_reconnect_keeps_the_previous_helper_reference_until_the_new_one_subscribes(adapter, helper):805 instance = live_adapter()806807 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())822823824def 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"854855856def 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()860861 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())876877878def test_env_enablement_seeds_a_home_channel(adapter, monkeypatch):879 import adapter as module880 from gateway.config import GatewayConfig, HomeChannel, Platform881 from gateway.config_env import _enable_plugin_platform882 monkeypatch.setenv("TINY_RELAY_URL", "http://relay")883 monkeypatch.setenv("TINY_PRIVATE_KEY", "k" * 64)884 monkeypatch.setenv("TINY_HOME_CHANNEL", "home-room")885 assert module._env_enablement()["home_channel"] == {"chat_id": "home-room", "name": "Home"}886 class Ctx:887 def register_platform(self, **kwargs):888 platform_registry.register(PlatformEntry(**kwargs))889 module.register(Ctx())890 config = GatewayConfig()891 _enable_plugin_platform(config, platform_registry.get("tiny"))892 home = config.platforms[Platform("tiny")].home_channel893 assert isinstance(home, HomeChannel) and home.chat_id == "home-room" and home.name == "Home"894 assert "home_channel" not in config.platforms[Platform("tiny")].extra895 assert TinyAdapter(config.platforms[Platform("tiny")]).home_room == "home-room"896897898def test_answers_resume_typing_after_waits(adapter):899 async def run():900 entry = _ApprovalEntry({"command": "ls", "request_id": "approval-id"})901 approval._gateway_queues["session"] = [entry]902 adapter.pause_typing_for_chat("room")903 await adapter.send_exec_approval("room", "ls", "session")904 request = adapter.rpc.events[-1]905 assert "room" in adapter._typing_paused906 await adapter._on_answer(response(request, "once"))907 assert entry.event.is_set() and "room" not in adapter._typing_paused908 async def handler(choice): return None909 slash_confirm.register("session", "confirm-id", "reload", handler)910 adapter.pause_typing_for_chat("room")911 await adapter.send_slash_confirm("room", "Reload?", "Confirm", "session", "confirm-id")912 await adapter._on_answer(response(adapter.rpc.events[-1], "once"))913 assert "room" not in adapter._typing_paused914 asyncio.run(run())915916917def test_pending_requests_are_bounded_and_swept(adapter, monkeypatch):918 import interactions919 monkeypatch.setattr(interactions, "MAX_PENDING", 3)920 async def run():921 for index in range(5):922 clarify_gateway.register(f"q{index}", "session", "Pick", ["yes"])923 await adapter.send_clarify("room", "Pick", ["yes"], f"q{index}", "session")924 assert [entry.callback_id for entry in adapter._pending.values()] == ["q2", "q3", "q4"]925 stale = next(iter(adapter._pending.values()))926 stale.expires = int(time.time()) - 1927 adapter._sweep_pending()928 assert [entry.callback_id for entry in adapter._pending.values()] == ["q3", "q4"]929 # An answer sweeps too, so an expired card never lingers.930 list(adapter._pending.values())[0].expires = int(time.time()) - 1931 await adapter._on_answer({"kind": 1111, "tags": [["e", "0" * 64]], "pubkey": OWNER,932 "created_at": int(time.time()), "content": "c0"})933 assert [entry.callback_id for entry in adapter._pending.values()] == ["q4"]934 asyncio.run(run())935936937def test_receive_path_sweeps_pending_once_a_minute(adapter, monkeypatch):938 sweeps = []939 monkeypatch.setattr(adapter, "_sweep_pending", lambda now=None: sweeps.append(now))940 async def handle(message): pass941 adapter.handle_message = handle942 base = {"kind": 9, "pubkey": OWNER, "created_at": int(time.time()), "content": "hello",943 "tags": [["h", "room"], ["p", BOT]]}944 async def run():945 await adapter._on_event({**base, "id": "d" * 64})946 await adapter._on_event({**base, "id": "e" * 64})947 asyncio.run(run())948 assert len(sweeps) == 1949 adapter._last_sweep = time.monotonic() - adapter.SWEEP_INTERVAL950 asyncio.run(adapter._on_event({**base, "id": "f" * 64}))951 assert len(sweeps) == 2