clone

a self-hosted multitenant nostr relay with sites, file storage, git hosting, and inbox and outbox delivery

main
tinyrelay / tinyagent / hermes / tests / test_tasks.py
test_tasks.py 3.7K
1"""Hermes-free rules for long-task cards: intake text, progress coalescing and results."""23import sys4from pathlib import Path56sys.path.insert(0, str(Path(__file__).parents[1]))78from tasks import CHROME, FILE, TEXT, TaskState, answer_tags, clip, last_line, request_root, request_text910OWNER = "b" * 64111213def task(**overrides):14 fields = dict(request_id="1" * 64, room="room", requester=OWNER, thread_root="1" * 64, session_key="s")15 fields.update(overrides)16 return TaskState(**fields)171819def test_request_text_prefixes_subject_and_falls_back():20 assert request_text({"content": "Do the thing", "tags": [["subject", "Nightly build"]]}) == "Nightly build\n\nDo the thing"21 assert request_text({"content": "", "tags": [["subject", "Nightly build"]]}) == "Nightly build"22 assert request_text({"content": "Do the thing", "tags": []}) == "Do the thing"232425def test_request_root_prefers_marked_root():26 assert request_root({"id": "1" * 64, "tags": [["e", "2" * 64, "", "root"]]}) == "2" * 6427 assert request_root({"id": "1" * 64, "tags": [["e", "2" * 64]]}) == "1" * 64282930def test_progress_coalesces_latest_text_and_tool_line():31 state = task(accepted=True)32 state.observe("m1", "Looking at the repo ▉", TEXT)33 state.observe("m2", "\U0001f527 terminal: \"ls\"", CHROME)34 state.observe("m2", "\U0001f527 terminal: \"ls\"\n\U0001f50d web_search: \"docs\"", CHROME)35 state.observe("m1", "Looking at the repo\nStill reading the docs ▉", CHROME)36 assert state.has_progress37 assert state.take_progress(now=100.0) == "Still reading the docs\n\U0001f50d web_search: \"docs\""38 assert not state.has_progress and state.last_progress == 100.039 assert state.progress_due(now=110.0, interval=15.0) == 5.040 assert state.progress_due(now=115.0, interval=15.0) == 0.041 assert task().progress_due(now=0.0, interval=15.0) == 0.0424344def test_result_links_final_message_with_summary_line():45 state = task()46 state.observe("draft", "Working on it", TEXT)47 state.observe("final", " First line of the answer.\nMore detail below.", TEXT, final=True)48 assert not state.has_progress49 content, tags = state.result()50 assert content == "First line of the answer."51 assert tags == [["e", "final"]]525354def test_result_uses_last_text_message_without_a_final_mark():55 state = task()56 state.observe("first", "One", TEXT)57 state.observe("chrome", "\U0001f527 tool", CHROME)58 state.observe("second", "Two " * 100, TEXT)59 content, tags = state.result()60 assert content.endswith("...") and len(content) == 20061 assert tags == [["e", "second"]]626364def test_result_names_posted_files():65 state = task()66 state.observe("f1", "", FILE, file_name="report.pdf")67 state.observe("f2", "caption", FILE, file_name="chart.png")68 assert state.result() == ("Posted report.pdf, chart.png", [["e", "f1"], ["e", "f2"]])69 state.observe("final", "Here you go", TEXT, final=True)70 assert state.result() == ("Here you go\nPosted report.pdf, chart.png", [["e", "final"], ["e", "f1"], ["e", "f2"]])717273def test_result_without_posts_uses_returned_text_or_done():74 state = task()75 assert state.result() == ("Done", [])76 state.final_text = "Quiet answer"77 assert state.result() == ("Quiet answer", [])787980def test_terminal_guard_and_failure_text():81 state = task()82 assert state.failure() == "The task failed"83 state.note_error(" boom\nline two ")84 assert state.failure() == "boom line two"85 assert state.end() and not state.end()86 state.observe("late", "ignored", TEXT)87 assert state.messages == {} and not state.has_progress888990def test_helpers():91 assert answer_tags(task()) == [["e", "1" * 64], ["p", OWNER]]92 assert clip("a b\n c", 5) == "a b c" and clip("abcdef", 5) == "ab..."93 assert last_line("one\ntwo ▉\n\n") == "two"