1"""Long-task cards through the real Hermes processing pipeline."""23import asyncio4import hashlib5import 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 gateway.platforms.event import ProcessingOutcome16from ingress import EventTracker1718OWNER = "b" * 6419BOT = "a" * 6420OTHER = "c" * 6421STRANGER = "f" * 6422REQUEST = "d" * 64232425def tag(event, name):26 return next((row[1] for row in event.get("tags", []) if len(row) > 1 and row[0] == name), None)272829class FakeRPC:30 def __init__(self):31 self.events = []32 self.rows = []33 self.terminal = []34 self.fail_kinds = set()35 # kind -> number of publishes of that kind still to refuse36 self.fail_times = {}37 self.attempts = []3839 async def query(self, filter):40 if "ids" in filter:41 return [row for row in self.rows if row["id"] in filter["ids"]]42 if "#e" in filter:43 return [row for row in self.terminal44 if row["kind"] in filter.get("kinds", []) and tag(row, "e") in filter["#e"]]45 return []4647 async def publish(self, event):48 self.attempts.append(event["kind"])49 if event["kind"] in self.fail_kinds or self.fail_times.get(event["kind"], 0) > 0:50 self.fail_times[event["kind"]] = self.fail_times.get(event["kind"], 0) - 151 raise RuntimeError("relay refused the event")52 result = {"id": f"{len(self.events) + 1:064x}", "created_at": int(time.time()), "pubkey": BOT, **event}53 self.events.append(result)54 return result5556 async def upload(self, room, filename, mime, data):57 return {"url": f"https://tiny.test/media/{filename}", "sha256": hashlib.sha256(data).hexdigest()}5859 def kinds(self, kind):60 return [event for event in self.events if event["kind"] == kind]61626364def adapter(monkeypatch, tmp_path):65 monkeypatch.setenv("HERMES_HOME", str(tmp_path))66 platform_registry.register(PlatformEntry(name="tiny", label="Tiny", adapter_factory=TinyAdapter,67 check_fn=lambda: True))68 instance = TinyAdapter(PlatformConfig(enabled=True, extra={"rooms": "room", "allowed_users": f"{OWNER},{OTHER}"}))69 instance.rpc = FakeRPC()70 instance.pubkey = BOT71 yield instance727374def request(request_id=REQUEST, *, author=OWNER, assignee=BOT, room="room", subject=None, content="Run the build",75 root=None):76 tags = [["h", room], ["p", assignee]]77 if subject:78 tags.append(["subject", subject])79 if root:80 tags.append(["e", root, "", "root"])81 return {"kind": 43001, "id": request_id, "pubkey": author, "created_at": int(time.time()),82 "content": content, "tags": tags}838485def cancel(request_id=REQUEST, *, author=OWNER, event_id="9" * 64):86 return {"kind": 43005, "id": event_id, "pubkey": author, "created_at": int(time.time()), "content": "",87 "tags": [["e", request_id], ["h", "room"]]}888990def bang_cancel(*, author=OTHER, kind=9, event_id="8" * 64, extra=None):91 return {"kind": kind, "id": event_id, "pubkey": author, "created_at": int(time.time()), "content": " !cancel \n",92 "tags": [["h", "room"], ["p", BOT], *(extra or [])]}939495def answer_tags(event):96 return [row for row in event["tags"] if row[0] in ("e", "p", "h")]979899async def settle(rounds=5):100 for _ in range(rounds):101 await asyncio.sleep(0)102103104def admit(message):105 """What Hermes's handle_message records once it starts or queues a turn."""106 message._gateway_accepted = True107108109def capture(adapter):110 received = []111 async def handle(message):112 received.append(message)113 admit(message)114 adapter.handle_message = handle115 return received116117118def test_request_becomes_a_thread_turn_anchored_at_the_request(adapter):119 received = capture(adapter)120 asyncio.run(adapter._on_event(request(subject="Nightly build", content="Run it and report.")))121 message = received[0]122 assert message.text == "Nightly build\n\nRun it and report."123 assert message.message_id == REQUEST and message.source.thread_id == REQUEST124 assert message.source.chat_id == "room" and message.source.user_id == OWNER125 task = adapter._tasks[REQUEST]126 assert task.requester == OWNER and task.thread_root == REQUEST and not task.accepted127 assert adapter._session_assignees[task.session_key] == OWNER128 assert adapter.rpc.events == []129 # A request inside an existing thread keeps its own session, named by the request,130 # while its replies sit in that thread under the request.131 asyncio.run(adapter._on_event(request("e" * 64, root="7" * 64, content="Continue here")))132 assert received[1].text == "Continue here" and received[1].source.thread_id == "e" * 64133 assert adapter._tasks["e" * 64].thread_root == "7" * 64134 assert adapter._tasks["e" * 64].session_key != task.session_key135 reply = asyncio.run(adapter.send("room", "On it", reply_to="e" * 64, metadata={"thread_id": "e" * 64}))136 assert reply.success and adapter.rpc.events[-1]["kind"] == 12137 assert ["e", "7" * 64, "", "root"] in adapter.rpc.events[-1]["tags"]138 assert ["e", "e" * 64, "", "reply"] in adapter.rpc.events[-1]["tags"]139 assert adapter._tasks["e" * 64].classes == {reply.message_id: "chrome"}140141142143def test_requests_need_an_allowed_author_our_key_and_a_watched_room(adapter, mutation):144 received = capture(adapter)145 changes = {"author": {"author": STRANGER}, "assignee": {"assignee": OTHER}, "room": {"room": "elsewhere"}}146 asyncio.run(adapter._on_event(request(**changes[mutation])))147 assert received == [] and adapter._tasks == {} and adapter.rpc.events == []148149150151def test_finished_requests_are_not_run_again(adapter, tmp_path, kind):152 received = capture(adapter)153 adapter._tracker = EventTracker("http://relay", BOT, ["room"], state_dir=tmp_path, now=int(time.time()) - 10)154 adapter.rpc.terminal = [{"kind": kind, "tags": [["e", REQUEST], ["h", "room"]]}]155 event = request()156 asyncio.run(adapter._on_event(event))157 assert received == [] and adapter._tasks == {} and adapter.rpc.events == []158 assert not adapter._tracker.should_accept(event)159160161def test_accepted_is_published_when_processing_starts(adapter):162 received = capture(adapter)163 async def run():164 await adapter._on_event(request())165 assert adapter.rpc.events == []166 await adapter.on_processing_start(received[0])167 await adapter.on_processing_start(received[0])168 accepted = adapter.rpc.kinds(43002)169 assert len(accepted) == 1 and accepted[0]["content"] == ""170 assert answer_tags(accepted[0]) == [["h", "room"], ["e", REQUEST], ["p", OWNER]]171 assert adapter._tasks[REQUEST].accepted172 asyncio.run(run())173174175def test_progress_is_coalesced_and_throttled(adapter, monkeypatch):176 received = capture(adapter)177 clock = [1000.0]178 monkeypatch.setattr(adapter, "_clock", lambda: clock[0])179 thread = {"thread_id": REQUEST}180 async def run():181 await adapter._on_event(request())182 await adapter.send("room", "Before accept", metadata=thread)183 await settle()184 assert adapter.rpc.kinds(43003) == []185 await adapter.on_processing_start(received[0])186 chrome = await adapter.send("room", "\U0001f527 terminal: \"ls\"", metadata=thread)187 await settle()188 progress = adapter.rpc.kinds(43003)189 assert [event["content"] for event in progress] == ["\U0001f527 terminal: \"ls\""]190 assert answer_tags(progress[0]) == [["h", "room"], ["e", REQUEST], ["p", OWNER]]191 await adapter.edit_message("room", chrome.message_id, "\U0001f527 terminal: \"ls\"\n\U0001f50d web_search: \"docs\"")192 draft = await adapter.send("room", "Reading the docs ▉", metadata={**thread, "expect_edits": True})193 await adapter.edit_message("room", draft.message_id, "Reading the docs\nComparing versions ▉")194 await settle()195 assert len(adapter.rpc.kinds(43003)) == 1196 task = adapter._tasks[REQUEST]197 assert task.timer is not None and task.has_progress198 clock[0] += adapter.PROGRESS_INTERVAL199 await adapter._flush_progress(REQUEST)200 progress = adapter.rpc.kinds(43003)201 assert len(progress) == 2202 assert progress[-1]["content"] == "Comparing versions\n\U0001f50d web_search: \"docs\""203 clock[0] += adapter.PROGRESS_INTERVAL204 await adapter._flush_progress(REQUEST)205 assert len(adapter.rpc.kinds(43003)) == 2206 assert [event["kind"] for event in adapter.rpc.events if event["kind"] in (9, 12)] == [12, 12, 12]207 assert all(["e", REQUEST, "", "root"] in event["tags"] for event in adapter.rpc.events if event["kind"] == 12)208 asyncio.run(run())209210211def pipeline(adapter, handler):212 seen = []213 async def wrapped(message):214 seen.append(message)215 return await handler(message)216 adapter.set_message_handler(wrapped)217 async def process(message):218 admit(message)219 await adapter._process_message_background(message, adapter._event_session_key(message))220 adapter.handle_message = process221 return seen222223224def grade_like_0_21_1(adapter):225 """Hermes 0.21.1 does not count attachment delivery: a file-only turn grades as failure."""226 original = adapter.on_processing_complete227 async def complete(event, outcome):228 posted_text = any(event["kind"] == 12 and not any(row[0] == "imeta" for row in event["tags"])229 for event in adapter.rpc.events)230 if outcome == ProcessingOutcome.SUCCESS and not posted_text:231 outcome = ProcessingOutcome.FAILURE232 await original(event, outcome)233 adapter.on_processing_complete = complete234235236async def finish_turns(adapter, timeout=5):237 tasks = list(adapter._session_tasks.values())238 if tasks:239 await asyncio.wait_for(asyncio.gather(*tasks, return_exceptions=True), timeout)240 await settle()241242243def test_success_publishes_one_result_linking_the_thread_reply(adapter):244 async def handler(message):245 return "Build passed on main.\nAll 42 tests green."246 seen = pipeline(adapter, handler)247 asyncio.run(adapter._on_event(request()))248 events = adapter.rpc.events249 replies = [event for event in events if event["kind"] == 12]250 assert len(replies) == 1 and ["e", REQUEST, "", "root"] in replies[0]["tags"]251 results = adapter.rpc.kinds(43004)252 assert len(results) == 1253 assert results[0]["content"] == "Build passed on main."254 assert answer_tags(results[0]) == [["h", "room"], ["e", REQUEST], ["p", OWNER], ["e", replies[0]["id"]]]255 order = [event["kind"] for event in events if event["kind"] in (43002, 12, 43004)]256 assert order == [43002, 12, 43004]257 assert adapter.rpc.kinds(43006) == [] and adapter.rpc.kinds(43003) == [] and REQUEST not in adapter._tasks258 asyncio.run(adapter.on_processing_complete(seen[0], ProcessingOutcome.SUCCESS))259 assert len(adapter.rpc.kinds(43004)) == 1260261262263def test_file_only_result_names_and_links_the_files(adapter, tmp_path, hermes):264 note = tmp_path / "report.txt"265 note.write_text("build log")266 async def handler(message):267 return f"MEDIA:{note}"268 pipeline(adapter, handler)269 if hermes == "0.21.1":270 grade_like_0_21_1(adapter)271 asyncio.run(adapter._on_event(request()))272 files = [event for event in adapter.rpc.events if any(row[0] == "imeta" for row in event["tags"])]273 assert len(files) == 1 and files[0]["kind"] == 12274 results = adapter.rpc.kinds(43004)275 assert len(results) == 1 and results[0]["content"] == "Posted report.txt"276 assert ["e", files[0]["id"]] in results[0]["tags"]277 assert adapter.rpc.kinds(43006) == [] and REQUEST not in adapter._tasks278279280281def test_true_failures_still_publish_an_error(adapter, shape):282 async def handler(message):283 return "Build passed."284 seen = pipeline(adapter, handler)285 if shape == "nothing-delivered":286 # Hermes graded the turn as failed and the adapter saw no reply reach the room.287 original = adapter.on_processing_complete288 async def complete(event, outcome):289 adapter._tasks[REQUEST].classes.clear()290 await original(event, ProcessingOutcome.FAILURE)291 adapter.on_processing_complete = complete292 else:293 adapter.rpc.fail_kinds = {12}294 asyncio.run(adapter._on_event(request()))295 errors = adapter.rpc.kinds(43006)296 assert len(errors) == 1 and adapter.rpc.kinds(43004) == [] and REQUEST not in adapter._tasks297 expected = "The task failed" if shape == "nothing-delivered" else "relay refused the event"298 assert errors[0]["content"] == expected299300301def test_failure_publishes_one_error_with_the_exposed_text(adapter):302 async def handler(message):303 raise RuntimeError("model exploded")304 pipeline(adapter, handler)305 asyncio.run(adapter._on_event(request()))306 errors = adapter.rpc.kinds(43006)307 assert len(errors) == 1 and errors[0]["content"] == "model exploded"308 assert answer_tags(errors[0]) == [["h", "room"], ["e", REQUEST], ["p", OWNER]]309 notice = [event for event in adapter.rpc.events if event["kind"] == 12]310 assert len(notice) == 1 and "model exploded" in notice[0]["content"]311 assert adapter.rpc.kinds(43004) == [] and REQUEST not in adapter._tasks312313314def test_terminal_publication_is_retried_and_happens_once(adapter):315 async def handler(message):316 return "Done"317 seen = pipeline(adapter, handler)318 adapter.TERMINAL_RETRY_DELAYS = (0, 0)319 adapter.rpc.fail_times = {43004: 2}320 asyncio.run(adapter._on_event(request()))321 assert adapter.rpc.attempts.count(43004) == 3322 assert len(adapter.rpc.kinds(43004)) == 1 and adapter.rpc.kinds(43006) == []323 assert REQUEST not in adapter._tasks324 asyncio.run(adapter.on_processing_complete(seen[0], ProcessingOutcome.SUCCESS))325 asyncio.run(adapter.on_processing_complete(seen[0], ProcessingOutcome.FAILURE))326 assert len(adapter.rpc.kinds(43004)) == 1 and adapter.rpc.kinds(43006) == []327328329def test_terminal_publication_gives_up_after_three_attempts(adapter):330 async def handler(message):331 return "Done"332 seen = pipeline(adapter, handler)333 adapter.TERMINAL_RETRY_DELAYS = (0, 0)334 adapter.rpc.fail_kinds = {43004}335 forgotten = []336 original = adapter._forget_task337 def forget(request_id):338 forgotten.append(request_id in adapter._tasks)339 original(request_id)340 adapter._forget_task = forget341 asyncio.run(adapter._on_event(request()))342 assert adapter.rpc.attempts.count(43004) == 3343 # The task stayed known until the last attempt, then was dropped exactly once.344 assert forgotten[0] and REQUEST not in adapter._tasks345 assert adapter.rpc.kinds(43004) == [] and adapter.rpc.kinds(43006) == []346 adapter.rpc.fail_kinds = set()347 asyncio.run(adapter.on_processing_complete(seen[0], ProcessingOutcome.SUCCESS))348 assert adapter.rpc.kinds(43004) == []349350351def test_requests_sharing_a_thread_run_as_separate_turns(adapter):352 # Three requests in one thread arrive while the first still runs. Each gets its353 # own session, so Hermes never folds two requests into one turn: three handler354 # calls, three accepted events, three results, no orphaned card.355 adapter._busy_text_mode = "queue"356 adapter._busy_text_debounce_seconds = 0357 release = asyncio.Event()358 seen = []359 async def handler(message):360 seen.append(message)361 if len(seen) == 1:362 await release.wait()363 return f"Finished {message.text}"364 adapter.set_message_handler(handler)365 common = "7" * 64366 ids = [f"{index + 10:064x}" for index in range(3)]367 async def run():368 for index, request_id in enumerate(ids):369 await adapter._on_event(request(request_id, root=common, content=f"job {index}"))370 await settle()371 assert [message.text for message in seen] == ["job 0", "job 1", "job 2"]372 assert adapter._pending_messages == {}373 release.set()374 await finish_turns(adapter)375 asyncio.run(run())376 assert sorted(tag(event, "e") for event in adapter.rpc.kinds(43002)) == ids377 results = adapter.rpc.kinds(43004)378 assert sorted(tag(event, "e") for event in results) == ids379 assert adapter.rpc.kinds(43006) == [] and adapter._tasks == {}380 replies = [event for event in adapter.rpc.events if event["kind"] == 12]381 assert len(replies) == 3382 assert all(["e", common, "", "root"] in event["tags"] for event in replies)383 assert sorted(row[1] for event in replies for row in event["tags"] if len(row) > 3 and row[3] == "reply") == ids384 for result in results:385 reply = next(event for event in replies if ["e", tag(result, "e"), "", "reply"] in event["tags"])386 assert ["e", reply["id"]] in result["tags"] and result["content"] == reply["content"]387388389def test_requests_merged_by_hermes_end_with_an_error_naming_the_carrier(adapter):390 # Safety net for a Hermes that still coalesces two requests into one turn: the391 # request that lost its turn gets an error card naming the turn that carried it.392 received = capture(adapter)393 ids = [f"{index + 10:064x}" for index in range(3)]394 async def run():395 for index, request_id in enumerate(ids):396 await adapter._on_event(request(request_id, content=f"job {index}"))397 first, second, third = (adapter._tasks[request_id] for request_id in ids)398 second.session_key = third.session_key = first.session_key399 await adapter.on_processing_start(received[0])400 # The third request is still queued under the session: it keeps its card.401 adapter._pending_messages[first.session_key] = received[2]402 await adapter.on_processing_complete(received[0], ProcessingOutcome.SUCCESS)403 errors = adapter.rpc.kinds(43006)404 assert [tag(event, "e") for event in errors] == [ids[1]]405 assert errors[0]["content"] == f"Merged into task {ids[2]}"406 assert [tag(event, "e") for event in adapter.rpc.kinds(43004)] == [ids[0]]407 assert ids[0] not in adapter._tasks and ids[1] not in adapter._tasks and ids[2] in adapter._tasks408 adapter._pending_messages.clear()409 await adapter.on_processing_start(received[2])410 await adapter.on_processing_complete(received[2], ProcessingOutcome.SUCCESS)411 assert adapter._tasks == {} and len(adapter.rpc.kinds(43004)) == 2412 asyncio.run(run())413414415def slow_turn(adapter):416 """Install a handler that never finishes on its own; the real handle_message spawns it."""417 seen = []418 async def handler(message):419 seen.append(message)420 await asyncio.sleep(30)421 return "never"422 adapter.set_message_handler(handler)423 return seen424425426def test_cancelled_turn_publishes_cancelled(adapter):427 seen = slow_turn(adapter)428 async def run():429 await adapter._on_event(request())430 await settle()431 assert len(seen) == 1 and adapter._tasks[REQUEST].accepted432 await adapter.cancel_session_processing(adapter._tasks[REQUEST].session_key)433 await settle()434 errors = adapter.rpc.kinds(43006)435 assert len(errors) == 1 and errors[0]["content"] == "Cancelled"436 assert adapter.rpc.kinds(43004) == [] and REQUEST not in adapter._tasks437 asyncio.run(run())438439440def test_requester_cancel_ends_the_task_silently(adapter):441 seen = slow_turn(adapter)442 async def run():443 await adapter._on_event(request())444 await settle()445 session = adapter._tasks[REQUEST].session_key446 await adapter._on_event(cancel(author=OTHER))447 assert REQUEST in adapter._tasks and not adapter._session_tasks[session].done()448 await adapter._on_event(cancel())449 await settle()450 assert REQUEST not in adapter._tasks451 assert session not in adapter._session_tasks452 assert adapter.rpc.kinds(43006) == [] and adapter.rpc.kinds(43004) == []453 assert len(seen) == 1454 asyncio.run(run())455456457458def test_bang_cancel_stops_a_task_and_names_the_canceller(adapter, shape):459 seen = slow_turn(adapter)460 async def run():461 await adapter._on_event(request())462 await settle()463 session = adapter._tasks[REQUEST].session_key464 if shape == "thread":465 event = bang_cancel(kind=12, extra=[["e", REQUEST, "", "root"]])466 elif shape == "room-reference":467 event = bang_cancel(extra=[["e", REQUEST]])468 else:469 event = bang_cancel()470 await adapter._on_event(event)471 await settle()472 errors = adapter.rpc.kinds(43006)473 assert len(errors) == 1 and errors[0]["content"] == f"Cancelled by {OTHER}"474 assert REQUEST not in adapter._tasks and session not in adapter._session_tasks475 assert len(seen) == 1476 asyncio.run(run())477478479def test_bang_cancel_stops_an_ordinary_turn_without_forwarding(adapter):480 seen = slow_turn(adapter)481 cancelled = []482 async def cancel_session(session_key, **kwargs):483 cancelled.append(session_key)484 adapter.cancel_session_processing = cancel_session485 async def run():486 plain = {"kind": 9, "id": "5" * 64, "pubkey": OWNER, "created_at": int(time.time()),487 "content": "hello", "tags": [["h", "room"], ["p", BOT]]}488 await adapter._on_event(plain)489 await settle()490 assert len(seen) == 1491 await adapter._on_event(bang_cancel(author=OWNER))492 assert cancelled == [adapter._event_session_key(seen[0])]493 assert len(seen) == 1 and {event["kind"] for event in adapter.rpc.events} <= {20002}494 # Without our p tag the text is an ordinary message and follows the mention rule.495 await adapter._on_event({**bang_cancel(author=OWNER, event_id="6" * 64), "tags": [["h", "room"]]})496 assert len(seen) == 1 and cancelled == [adapter._event_session_key(seen[0])]497 for task in list(adapter._session_tasks.values()):498 task.cancel()499 await asyncio.gather(*adapter._session_tasks.values(), return_exceptions=True)500 asyncio.run(run())501502503def test_create_handoff_thread_opens_a_real_thread(adapter):504 async def run():505 thread_id = await adapter.create_handoff_thread("room", "Hermes - review")506 root = adapter.rpc.events[-1]507 assert thread_id == root["id"] and root["kind"] == 11 and root["content"] == "Hermes - review"508 assert root["tags"] == [["h", "room"]]509 result = await adapter.send("room", "Continuing here", metadata={"thread_id": thread_id})510 reply = adapter.rpc.events[-1]511 assert result.success and reply["kind"] == 12 and ["e", thread_id, "", "root"] in reply["tags"]512 assert await adapter.create_handoff_thread("room", "") is not None513 assert adapter.rpc.events[-1]["content"] == "Handoff"514 adapter.rpc.fail_kinds = {11}515 assert await adapter.create_handoff_thread("room", "Broken") is None516 asyncio.run(run())