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" * 64212223class FakeRPC:24 def __init__(self):25 self.events = []26 self.rows = []2728 async def query(self, filter):29 return [row for row in self.rows if row["id"] in filter.get("ids", [])]3031 async def publish(self, event):32 result = {"id": f"{len(self.events) + 1:064x}", "created_at": int(time.time()), "pubkey": BOT, **event}33 self.events.append(result)34 return result3536 async def query(self, _filter):37 return []38394041def adapter(monkeypatch, tmp_path):42 monkeypatch.setenv("HERMES_HOME", str(tmp_path))43 platform_registry.register(PlatformEntry(name="tiny", label="Tiny", adapter_factory=TinyAdapter,44 check_fn=lambda: True))45 instance = TinyAdapter(PlatformConfig(enabled=True, extra={"rooms": "room", "allowed_users": f"{OWNER},{OTHER}"}))46 instance.rpc = FakeRPC()47 instance.pubkey = BOT48 instance._session_assignees["session"] = OWNER49 yield instance50 clarify_gateway._entries.clear()51 clarify_gateway._session_index.clear()52 approval._gateway_queues.clear()53 slash_confirm._pending.clear()545556def response(request, content, author=OWNER):57 event_id = request["id"]58 return {"kind": 1111, "pubkey": author, "created_at": int(time.time()), "content": content,59 "tags": [["h", "room"], ["e", event_id], ["E", event_id], ["p", BOT], ["P", BOT],60 ["k", str(request["kind"])], ["K", str(request["kind"])]]}616263def test_real_inbound_source_tracks_human_session(adapter):64 async def run():65 received = []66 async def handle(message):67 received.append(message)68 adapter.handle_message = handle69 await adapter._on_event({"kind": 9, "id": "d" * 64, "pubkey": OWNER, "created_at": int(time.time()),70 "content": "hello", "tags": [["h", "room"], ["p", BOT]]})71 assert len(received) == 172 key = adapter._event_session_key(received[0])73 assert adapter._session_assignees[key] == OWNER74 assert received[0].source.platform.value == "tiny"75 asyncio.run(run())7677787980818283def test_clarify_resolves_exact_real_waiter(adapter, multiple, content, expected):84 async def run():85 entry = clarify_gateway.register("question-id", "session", "Pick", ["alpha", "beta"], multiple)86 untouched = clarify_gateway.register("other-question", "session", "Other", ["yes"])87 result = await adapter.send_clarify("room", "Pick", ["alpha", "beta"], "question-id", "session")88 assert result.success89 request = adapter.rpc.events[-1]90 assert ["p", OWNER] in request["tags"]91 assert not any(tag[0] == "mention" for tag in request["tags"])92 assert ["freeform", "true"] in request["tags"]93 assert all("session" not in str(tag) for tag in request["tags"])94 await adapter._on_answer(response(request, content))95 assert entry.event.is_set() and entry.response == expected96 assert not untouched.event.is_set()97 asyncio.run(run())9899100101def test_invalid_answers_do_not_resolve(adapter, mutation):102 async def run():103 entry = clarify_gateway.register("question-id", "session", "Pick", ["alpha", "beta"], True)104 await adapter.send_clarify("room", "Pick", ["alpha", "beta"], "question-id", "session")105 request = adapter.rpc.events[-1]106 answer = response(request, '["c0"]')107 if mutation == "actor": answer["pubkey"] = OTHER108 if mutation == "room": answer["tags"][0][1] = "elsewhere"109 if mutation == "root": answer["tags"][2][1] = "f" * 64110 if mutation == "kind": answer["kind"] = 7111 if mutation == "invalid": answer["content"] = '["unknown"]'112 if mutation == "duplicate": answer["content"] = '["c0","c0"]'113 if mutation == "expired": adapter._pending[request["id"]].expires = int(time.time()) - 1114 await adapter._on_answer(answer)115 assert not entry.event.is_set()116 asyncio.run(run())117118119def test_open_ended_question_is_structured(adapter):120 async def run():121 entry = clarify_gateway.register("question-id", "session", "Explain", None)122 await adapter.send_clarify("room", "Explain", None, "question-id", "session")123 request = adapter.rpc.events[-1]124 assert ["selection", "text"] in request["tags"]125 await adapter._on_answer(response(request, "Here is my answer"))126 assert entry.response == "Here is my answer"127 asyncio.run(run())128129130def test_approval_uses_exact_request_and_offered_scope(adapter):131 async def run():132 first = _ApprovalEntry({"command": "first", "request_id": "first-id"})133 second = _ApprovalEntry({"command": "second", "request_id": "second-id"})134 approval._gateway_queues["session"] = [first, second]135 assert (await adapter.send_exec_approval("room", "second", "session",136 allow_permanent=False, allow_session=False)).success137 request = adapter.rpc.events[-1]138 await adapter._on_answer(response(request, "always"))139 assert not first.event.is_set() and not second.event.is_set()140 await adapter._on_answer(response(request, "once"))141 assert not first.event.is_set() and second.event.is_set()142 assert second.result == "once"143 await adapter._on_answer(response(request, "once"))144 assert not first.event.is_set()145 asyncio.run(run())146147148def test_confirmation_calls_real_handler_once(adapter):149 async def run():150 choices = []151 async def handler(choice):152 choices.append(choice)153 return "Reloaded"154 slash_confirm.register("session", "confirm-id", "reload", handler)155 await adapter.send_slash_confirm("room", "Reload?", "Confirm", "session", "confirm-id")156 request = adapter.rpc.events[-1]157 await adapter._on_answer(response(request, "once"))158 await adapter._on_answer(response(request, "once"))159 assert choices == ["once"]160 assert adapter.rpc.events[-1]["content"] == "Reloaded"161 asyncio.run(run())162163164def test_explicit_mentions_and_edits(adapter):165 async def run():166 result = await adapter.send("room", "ordinary", reply_to="d" * 64)167 assert not any(tag[0] in ("p", "mention") for tag in adapter.rpc.events[-1]["tags"])168 edited = await adapter.edit_message("room", result.message_id, f"Attention @{OWNER}")169 assert edited.message_id == result.message_id170 event = adapter.rpc.events[-1]171 assert event["kind"] == 40003172 assert ["p", OWNER] in event["tags"] and ["mention", OWNER] in event["tags"]173 assert not await adapter.delete_message("room", result.message_id)174 asyncio.run(run())175176177def test_native_prompt_mentions_assignee_only_when_explicit(adapter):178 async def run():179 clarify_gateway.register("q", "session", "Pick", ["yes"])180 await adapter.send_clarify("room", f"@{OWNER} Pick", ["yes"], "q", "session")181 assert ["mention", OWNER] in adapter.rpc.events[-1]["tags"]182 asyncio.run(run())183184185def test_native_prompt_stays_in_current_thread(adapter):186 async def run():187 clarify_gateway.register("q", "session", "Pick", ["yes"])188 result = await adapter.send_clarify("room", "Pick", ["yes"], "q", "session",189 metadata={"thread_id": "r" * 64, "message_id": "m" * 64})190 assert result.success191 event = adapter.rpc.events[-1]192 assert event["kind"] == 12193 assert ["e", "r" * 64, "", "root"] in event["tags"]194 assert ["e", "m" * 64, "", "reply"] in event["tags"]195 asyncio.run(run())196197198199def test_real_inbound_reply_keeps_prompt_and_answer_in_existing_thread(adapter, kind):200 async def run():201 root, parent, incoming = "d" * 64, "e" * 64, "f" * 64202 adapter.rpc.rows = [{"id": parent, "kind": kind, "pubkey": BOT, "content": "Earlier reply",203 "tags": [["h", "room"], ["e", root, "", "root"]]}]204 async def handle(message):205 session = adapter._event_session_key(message)206 assert (message.source.thread_id is None) == (kind == 9)207 entry = clarify_gateway.register("thread-question", session, "Pick", ["Yes", "No"])208 # A newer queued message must not steal the active turn's assignee.209 adapter._session_assignees[session] = OTHER210 current_route = adapter._turn_routes[session]211 adapter._turn_routes[session] = {**current_route, "user_id": OTHER, "reply_to_message_id": "1" * 64}212 # Hermes forwards only thread_id for Tiny, and nothing for kind 9.213 metadata = {"thread_id": message.source.thread_id} if message.source.thread_id else None214 await adapter.send_clarify("room", "Pick", ["Yes", "No"], "thread-question", session, metadata)215 request = adapter.rpc.events[-1]216 assert ["p", OWNER] in request["tags"]217 assert request["kind"] == kind218 assert ["e", root, "", "root"] in request["tags"]219 assert ["e", incoming, "", "reply"] in request["tags"]220 await adapter._on_answer(response(request, '{"choices":["c1"],"text":"Keep this thread."}'))221 assert entry.event.is_set() and entry.response == "No\nKeep this thread."222 adapter._turn_routes[session] = current_route223 await adapter.send("room", "Acknowledged", metadata=metadata)224 assert ["e", root, "", "root"] in adapter.rpc.events[-1]["tags"]225 adapter.set_message_handler(handle)226 adapter.handle_message = adapter._message_handler227 await adapter._on_event({"kind": kind, "id": incoming, "pubkey": OWNER,228 "created_at": int(time.time()), "content": "Ask here", "tags": [229 ["h", "room"], ["p", BOT], ["e", root, "", "root"], ["e", parent, "", "reply"]]})230 # The final gateway send occurs after the handler completes.231 await adapter.send("room", "Final reply", reply_to=incoming)232 event = adapter.rpc.events[-1]233 assert event["kind"] == kind234 assert ["e", root, "", "root"] in event["tags"]235 assert ["e", incoming, "", "reply"] in event["tags"]236 assert not adapter._turn_routes and adapter._turn_route.get() is None237 asyncio.run(run())238239240def test_gateway_final_attachment_keeps_the_turn_route(adapter, tmp_path):241 async def run():242 import hashlib243 root, incoming = "d" * 64, "e" * 64244 note = tmp_path / "answer.txt"245 note.write_text("Details for this answer.")246 async def upload(room, filename, mime, data):247 return {"url": "https://tiny.test/answer.txt", "sha256": hashlib.sha256(data).hexdigest()}248 adapter.rpc.upload = upload249 async def handler(message):250 return f"Here is the answer.\nMEDIA:{note}"251 adapter.set_message_handler(handler)252 async def process(message):253 await adapter._process_message_background(message, adapter._event_session_key(message))254 adapter.handle_message = process255 await adapter._on_event({"kind": 9, "id": incoming, "pubkey": OWNER,256 "created_at": int(time.time()), "content": "Respond here", "tags": [257 ["h", "room"], ["p", BOT], ["e", root, "", "root"]]})258 replies = [event for event in adapter.rpc.events if event["kind"] == 9]259 assert len(replies) > 1260 assert all(["e", root, "", "root"] in event["tags"] for event in replies)261 assert not adapter._turn_routes and adapter._turn_route.get() is None262 asyncio.run(run())263264265def test_threaded_kind9_reply_keeps_root_session(adapter):266 async def run():267 received = []268 async def handle(message):269 received.append(message)270 adapter.handle_message = handle271 root = "r" * 64272 event_id = "e" * 64273 await adapter._on_event({"kind": 9, "id": event_id, "pubkey": OWNER,274 "created_at": int(time.time()), "content": "follow-up",275 "tags": [["h", "room"], ["e", root], ["p", BOT]]})276 assert len(received) == 1277 assert received[0].source.thread_id is None278 assert received[0].reply_to_message_id == root279 asyncio.run(run())280281282def test_threaded_question_callback_uses_kind12_correlation(adapter):283 async def run():284 entry = clarify_gateway.register("question-id", "session", "Pick", ["yes"])285 await adapter.send_clarify("room", "Pick", ["yes"], "question-id", "session",286 metadata={"thread_id": "r" * 64})287 request = adapter.rpc.events[-1]288 await adapter._on_answer(response(request, "c0"))289 assert entry.response == "yes"290 asyncio.run(run())291292293def test_outbound_document_preserves_thread_caption_and_bytes(adapter, tmp_path):294 import hashlib295 data = b'attachment bytes'296 path = tmp_path / 'note.txt'297 path.write_bytes(data)298 digest = hashlib.sha256(data).hexdigest()299 uploads = []300 async def upload(room, name, mime, body):301 uploads.append((room, name, mime, body))302 return {'url': f'http://tiny/media/{digest}', 'sha256': digest}303 adapter.rpc.upload = upload304 result = asyncio.run(adapter.send_document(chat_id='room', file_path=str(path),305 file_name='renamed.txt', caption=f'For @{OWNER}', metadata={'thread_id': 'd' * 64}, reply_to='e' * 64))306 assert result.success307 event = adapter.rpc.events[-1]308 assert uploads == [('room', 'renamed.txt', 'text/plain', data)]309 assert event['kind'] == 12310 assert ['e', 'd' * 64, '', 'root'] in event['tags']311 assert ['e', 'e' * 64, '', 'reply'] in event['tags']312 assert ['p', OWNER] in event['tags'] and ['mention', OWNER] in event['tags']313 assert any(t[0] == 'imeta' and f'x {digest}' in t and 'size 16' in t for t in event['tags'])314315316def test_standalone_delivery_uses_native_media(adapter, tmp_path, monkeypatch):317 import adapter as module318 import hashlib319 path = tmp_path / 'voice.mp3'320 path.write_bytes(b'audio bytes')321 rpc = adapter.rpc322 async def connect(): pass323 async def close(): pass324 async def upload(room, name, mime, body):325 digest = hashlib.sha256(body).hexdigest()326 return {'url': f'http://tiny/media/{digest}', 'sha256': digest}327 rpc.connect, rpc.close, rpc.upload = connect, close, upload328 monkeypatch.setattr(module, 'TinyRPC', lambda *a, **kw: rpc)329 result = asyncio.run(module._standalone_send(adapter.config, 'room', f'For @{OWNER}',330 thread_id='d' * 64, media_files=[(str(path), True)]))331 assert result['success'] and result['media_delivered']332 assert len(rpc.events) == 1333 event = rpc.events[0]334 assert event['kind'] == 12335 assert ['mention', OWNER] in event['tags']336 assert any(t[0] == 'imeta' and 'm audio/mpeg' in t for t in event['tags'])337338339def test_outbound_media_rejects_wrong_relay_hash(adapter, tmp_path):340 path = tmp_path / 'note.txt'341 path.write_text('local content')342 async def upload(*args): return {'url': 'http://tiny/media/' + '0' * 64, 'sha256': '0' * 64}343 adapter.rpc.upload = upload344 result = asyncio.run(adapter.send_document('room', str(path)))345 assert not result.success and adapter.rpc.events == []346347348def test_duplicate_callbacks_cannot_race_before_media_intake(adapter, monkeypatch):349 import adapter as module350 received = []351 async def media(*args):352 await asyncio.sleep(0.01)353 return [], []354 async def handle(message): received.append(message)355 monkeypatch.setattr(module, 'incoming_media', media)356 adapter.handle_message = handle357 event = {'kind': 9, 'id': 'e' * 64, 'pubkey': OWNER, 'created_at': int(time.time()),358 'content': 'hello', 'tags': [['h', 'room'], ['p', BOT]]}359 async def run(): await asyncio.gather(adapter._on_event(event), adapter._on_event(event))360 asyncio.run(run())361 assert len(received) == 1362363364365def test_room_message_trigger_respects_require_mention(adapter, setting, expected):366 instance = TinyAdapter(PlatformConfig(enabled=True, extra={367 'rooms': 'room', 'allowed_users': OWNER, 'require_mention': setting}))368 instance.pubkey = BOT369 instance.rpc = adapter.rpc370 received = []371 async def handle(message): received.append(message)372 instance.handle_message = handle373 async def run():374 for author in (OWNER, OTHER):375 await instance._on_event({'kind': 9, 'id': author, 'pubkey': author,376 'created_at': int(time.time()), 'content': 'hello', 'tags': [['h', 'room']]})377 asyncio.run(run())378 assert len(received) == expected379 assert all(message.source.user_id == OWNER for message in received)380381382383384385386387388389def test_public_approval_hook_preserves_scopes_on_older_hermes(adapter, monkeypatch, legacy, session, permanent, smart, expected):390 from gateway.platforms.base import BasePlatformAdapter391 if legacy:392 monkeypatch.delattr(BasePlatformAdapter, 'send_exec_approval', raising=False)393 async def run():394 entry = _ApprovalEntry({'command': 'safe preview', 'request_id': 'scoped-id'})395 approval._gateway_queues['session'] = [entry]396 result = await adapter.send_exec_approval('room', 'safe preview', 'session',397 allow_session=session, allow_permanent=permanent, smart_denied=smart)398 assert result.success399 request = adapter.rpc.events[-1]400 assert {t[1] for t in request['tags'] if t[0] == 'option'} == expected401 assert not any(t[0] == 'mention' for t in request['tags'])402 await adapter._on_answer(response(request, 'once'))403 assert entry.event.is_set() and entry.result == 'once'404 asyncio.run(run())