"""Exercise radio.py: the radio and post channels, the shared store, radio silence, the hooks and the listener.""" from __future__ import annotations import fcntl import json import os import re import subprocess import sys import tempfile import time import unittest from pathlib import Path REPO_ROOT = Path(__file__).resolve().parents[1] RADIO = REPO_ROOT / "skills" / "kvb-org-radio" / "scripts" / "radio.py" LINE = re.compile(r"^(?:RADIO|MAIL) (CALL|BROADCAST|SECURITY|PAN-PAN|MAYDAY|POST|NEWSLETTER) \d\d:\d\d (.*)$", re.DOTALL) LISTENER = ( "import importlib.util, sys\n" "spec = importlib.util.spec_from_file_location('radio', sys.argv[1])\n" "radio = importlib.util.module_from_spec(spec)\n" "spec.loader.exec_module(radio)\n" "radio.POLL_SECONDS = 0.05\n" "sys.argv = [sys.argv[1], 'listen']\n" "sys.exit(radio.main())\n" ) def strip(line: str) -> str: match = LINE.match(line) return f"{match[1]} {match[2]}" if match else line class TestRadio(unittest.TestCase): def setUp(self): tmp = tempfile.TemporaryDirectory() self.addCleanup(tmp.cleanup) self.home = Path(tmp.name) / "home" self.root = Path(tmp.name) / "council" self.config = Path(tmp.name) / "config.json" sessions = self.home / ".claude" / "sessions" sessions.mkdir(parents=True) for sid, name in (("s1", "One"), ("s2", "Two"), ("s3", "Three")): (sessions / f"{sid}.json").write_text(json.dumps({"sessionId": sid, "name": name}), encoding="utf-8") def _env(self, sid: str | None) -> dict[str, str]: env = { "PATH": "/usr/bin:/bin", "HOME": str(self.home), "AGENTS_COUNCIL_ROOT": str(self.root), "AGENTS_COUNCIL_CONFIG": str(self.config), } if sid: env["CLAUDE_CODE_SESSION_ID"] = sid return env def _exec(self, sid: str | None, stdin: str, args: tuple[str, ...]) -> subprocess.CompletedProcess: return subprocess.run( [sys.executable, str(RADIO), *args], input=stdin, capture_output=True, text=True, env=self._env(sid), check=False, timeout=30, ) def _cli(self, sid: str | None, *args: str) -> subprocess.CompletedProcess: return self._exec(sid, "", args) def _send(self, sid: str, *args: str) -> list[str]: result = self._cli(sid, *args) self.assertEqual(result.returncode, 0, result.stderr) return result.stdout.splitlines() def _hook(self, command: str, sid: str, **extra: str) -> str: result = self._exec(None, json.dumps({"session_id": sid, **extra}), (command,)) self.assertEqual(result.returncode, 0, result.stderr) return result.stdout def _prompt(self, sid: str) -> list[str]: return [strip(line) for line in self._hook("hook", sid).splitlines()] def _tool(self, sid: str) -> list[str]: out = self._hook("tool", sid) if not out: return [] output = json.loads(out)["hookSpecificOutput"] self.assertEqual(output["hookEventName"], "PostToolBatch") return [strip(line) for line in output["additionalContext"].splitlines()] def _mailbox(self, sid: str, **extra: str) -> tuple[int, list[str]]: result = self._exec(None, json.dumps({"session_id": sid, **extra}), ("mailbox",)) self.assertIn(result.returncode, (0, 2), result.stderr) self.assertEqual(result.stdout, "") return result.returncode, [strip(line) for line in result.stderr.splitlines()] def _block(self, sid: str, tool: str, command: str | None = None, **extra: str) -> int: tool_input = {"command": command} if tool == "Bash" else {"file_path": "/tmp/x"} payload = {"session_id": sid, "tool_name": tool, "tool_input": tool_input, **extra} result = self._exec(None, json.dumps(payload), ("block",)) self.assertEqual(result.stdout, "") return result.returncode def _rows(self) -> list[dict]: return [json.loads(line) for line in (self.root / "news.jsonl").read_text(encoding="utf-8").splitlines()] def _pidfile(self, sid: str) -> Path: return self.root / "radio-listen" / f"{sid}.pid" def _listen(self, sid: str) -> subprocess.Popen: proc = subprocess.Popen( [sys.executable, "-c", LISTENER, str(RADIO)], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, env=self._env(None), ) self.addCleanup(self._stop, proc) proc.stdin.write(json.dumps({"session_id": sid})) proc.stdin.close() deadline = time.monotonic() + 10 while time.monotonic() < deadline and proc.poll() is None: pidfile = self._pidfile(sid) if pidfile.is_file() and pidfile.read_text(encoding="utf-8") == str(proc.pid): return proc time.sleep(0.02) self.fail(f"listener for {sid} did not start") def _stop(self, proc: subprocess.Popen) -> None: if proc.poll() is None: proc.kill() proc.communicate() def _finish(self, proc: subprocess.Popen) -> tuple[int, list[str]]: _, err = proc.communicate(timeout=10) return proc.returncode, [strip(line) for line in err.splitlines()] def test_send_takes_only_radio_classes_and_post_takes_the_mail(self): for args in ( ("send", "post", "x", "--to", "Two"), ("send", "newsletter", "x"), ("send", "receipt", "x", "--to", "Two"), ("send", "broadcast", "x", "--to", "Two"), ("send", "call", "x"), ("send", "direct", "x"), ("send", "mayday", "x", "--to", "Two"), ("send", "security", "x", "--wake"), ("send", "call", "x", "--to", "Two", "--wake"), ("send", "bogus", "x"), ("send", "pan-pan", " "), ("post", " "), ): with self.subTest(args=args): self.assertEqual(self._cli("s1", *args).returncode, 2) self.assertEqual(self._cli(None, "send", "broadcast", "x").returncode, 2) self.assertEqual(self._send("s1", "send", "call", "a b", "--to", "Two", "--to", "Three")[0], "sent call to Two, Three") self.assertEqual(self._send("s1", "post", "board"), ["sent newsletter to all"]) self.assertEqual(self._send("s1", "post", "one peer", "--to", "Two"), ["sent post to Two"]) self.assertEqual(self._send("s1", "send", "direct", "old mail", "--to", "Two"), ["sent post to Two"]) self.assertEqual(self._send("s1", "send", "direct", "old wake", "--to", "Two", "--wake")[0], "sent call to Two") self.assertEqual( [(r["class"], r["to"], r["text"], r["v"], "wake" in r) for r in self._rows()], [ ("call", ["Two", "Three"], "a b", 2, False), ("newsletter", [], "board", 2, False), ("post", ["Two"], "one peer", 2, False), ("post", ["Two"], "old mail", 2, False), ("call", ["Two"], "old wake", 2, False), ], ) def test_rows_without_the_version_mark_keep_their_old_meaning(self): self.root.mkdir() now = time.time() old = [ {"ts": now - 50, "from": "Three", "to": [], "text": "old board"}, {"ts": now - 40, "from": "Three", "to": ["Two"], "text": "old direct"}, {"ts": now - 30, "from": "Three", "to": [], "text": "old broadcast", "class": "broadcast"}, {"ts": now - 20, "from": "Three", "to": ["Two"], "text": "old wake", "class": "direct", "wake": True}, {"ts": now - 10, "from": "Three", "to": [], "text": "old security", "class": "security"}, ] (self.root / "news.jsonl").write_text("".join(json.dumps(r) + "\n" for r in old), encoding="utf-8") self.assertEqual( [strip(line) for line in self._send("s2", "show", "--all")], [ "NEWSLETTER Three: old board", "POST Three to Two: old direct", "NEWSLETTER Three: old broadcast", "CALL Three to Two: old wake", "SECURITY Three: old security", ], ) self.assertEqual(self._prompt("s2"), ["SECURITY Three: old security", "CALL Three to Two: old wake"]) self.assertEqual( self._mailbox("s2"), (2, ["NEWSLETTER Three: old board", "POST Three to Two: old direct", "NEWSLETTER Three: old broadcast"]), ) self._send("s1", "send", "security", "new") rows = self._rows() self.assertEqual(rows[:5], old) for row in rows: self.assertIsInstance(row["ts"], float) self.assertIsInstance(row["from"], str) self.assertIsInstance(row["to"], list) self.assertIsInstance(row["text"], str) self.assertEqual((rows[-1]["class"], rows[-1]["v"]), ("security", 2)) def test_ts_strictly_increases_under_the_lock(self): self.root.mkdir() row = {"ts": time.time() + 1000, "from": "Three", "to": [], "text": "clock ahead"} (self.root / "news.jsonl").write_text(json.dumps(row) + "\n", encoding="utf-8") self._send("s1", "post", "first") self._send("s1", "post", "second") stamps = [r["ts"] for r in self._rows()] self.assertGreater(stamps[1], stamps[0]) self.assertGreater(stamps[2], stamps[1]) def test_mayday_sets_silence_and_only_its_sender_clears_it(self): self.assertEqual(self._send("s1", "clear"), ["no radio silence active"]) self.assertEqual(self._send("s2", "assist"), ["no radio silence active"]) self.assertEqual(self._send("s1", "send", "mayday", "disk full")[0], "sent mayday to all") silence = self.root / "radio-silence.json" state = json.loads(silence.read_text(encoding="utf-8")) self.assertEqual(sorted(state), ["by", "since", "text", "until"]) self.assertEqual((state["by"], state["text"], state["until"] - state["since"]), ("One", "disk full", 7200)) self.assertEqual(state["since"], self._rows()[-1]["ts"]) held = self._cli("s2", "clear") self.assertEqual(held.returncode, 3) self.assertIn("held by One", held.stdout) self.assertTrue(silence.exists()) self.assertEqual(self._send("s1", "clear"), ["cleared radio silence by One"]) self.assertFalse(silence.exists()) self._send("s1", "send", "mayday", "again") self.assertEqual(self._send("s2", "clear", "--force"), ["cleared radio silence by One, forced by Two"]) self.assertFalse(silence.exists()) def test_the_radio_names_every_addressee_without_a_live_listener_and_the_post_none(self): self._pidfile("s2").parent.mkdir(parents=True) self._pidfile("s2").write_text(str(os.getpid()), encoding="utf-8") for cls in ("mayday", "broadcast", "security", "pan-pan"): with self.subTest(cls=cls): self.assertEqual(self._send("s1", "send", cls, "stop"), [f"sent {cls} to all", "wake: SendMessage to Three"]) self.assertEqual( self._send("s1", "send", "call", "answer", "--to", "Two", "--to", "Three"), ["sent call to Two, Three", "wake: SendMessage to Three"], ) dead = subprocess.Popen([sys.executable, "-c", "pass"]) dead.wait() self._pidfile("s2").write_text(str(dead.pid), encoding="utf-8") self.assertEqual(self._send("s1", "send", "call", "answer", "--to", "Two"), ["sent call to Two", "wake: SendMessage to Two"]) self.assertEqual(self._send("s1", "post", "no wake", "--to", "Three"), ["sent post to Three"]) self.assertEqual(self._send("s1", "post", "no wake"), ["sent newsletter to all"]) def test_prompt_hook_delivers_only_the_radio_alarms_first(self): self._send("s1", "post", "board") self._send("s1", "post", "for two", "--to", "Two") self._send("s1", "send", "broadcast", "all") self._send("s1", "send", "call", "ring", "--to", "Two") self._send("s1", "send", "security", "sec") self._send("s1", "send", "pan-pan", "pan") self._send("s1", "send", "mayday", "may") lines = self._prompt("s2") self.assertRegex( lines[0], r"^MAYDAY ACTIVE since \d\d:\d\d by One: may; writing actions wait until radio\.py clear unless you help " r"with radio\.py assist$", ) self.assertEqual( lines[1:], ["MAYDAY One: may", "PAN-PAN One: pan", "SECURITY One: sec", "CALL One to Two: ring", "BROADCAST One: all"], ) self.assertEqual(len(self._prompt("s2")), 1) self.assertNotIn("CALL One to Two: ring", self._prompt("s3")) self.assertEqual(len(self._prompt("s1")), 1) def test_prompt_hook_shows_ten_and_counts_the_rest(self): self._send("s1", "send", "security", "early warning") for n in range(12): self._send("s1", "send", "broadcast", f"item {n}") lines = self._prompt("s2") self.assertEqual(lines[0], "SECURITY One: early warning") self.assertEqual(lines[1:-1], [f"BROADCAST One: item {n}" for n in range(3, 12)]) self.assertEqual(lines[-1], "RADIO 3 more items: radio.py show --all") self.assertEqual(self._prompt("s2"), []) def test_the_mailbox_delivers_post_and_newsletters_once_when_the_turn_ends(self): self._send("s1", "post", "board") self._send("s1", "post", "for two", "--to", "Two") self._send("s1", "send", "security", "sec") self.assertEqual(self._mailbox("s2"), (2, ["NEWSLETTER One: board", "POST One to Two: for two"])) self.assertEqual(self._mailbox("s2"), (0, [])) self.assertEqual(self._mailbox("s3"), (2, ["NEWSLETTER One: board"])) self.assertEqual(self._mailbox("s1"), (0, [])) self.assertEqual(self._prompt("s2"), ["SECURITY One: sec"]) def test_the_mailbox_shows_ten_and_counts_the_rest(self): for n in range(12): self._send("s1", "post", f"item {n}") code, lines = self._mailbox("s2") self.assertEqual(code, 2) self.assertEqual(lines[:-1], [f"NEWSLETTER One: item {n}" for n in range(2, 12)]) self.assertEqual(lines[-1], "MAIL 2 more items: radio.py show --all") self.assertEqual(self._mailbox("s2"), (0, [])) def test_a_read_call_puts_one_receipt_in_the_callers_mailbox_without_waking_it(self): caller = self._listen("s1") self._send("s1", "send", "call", "ring", "--to", "Two") self.assertEqual(self._tool("s2"), ["CALL One to Two: ring"]) self.assertEqual(self._tool("s2"), []) self.assertEqual(self._prompt("s2"), []) rows = self._rows() receipts = [r for r in rows if r["class"] == "receipt"] self.assertEqual([(r["from"], r["to"], r["text"], r["v"], r["call"]) for r in receipts], [("Two", ["One"], "ring", 2, rows[0]["ts"])]) time.sleep(0.5) self.assertIsNone(caller.poll()) code, lines = self._mailbox("s1") self.assertEqual(code, 2) self.assertEqual(len(lines), 1) self.assertRegex(lines[0], r"^MAIL CALL Two read \d\d:\d\d: ring$") self.assertEqual(self._mailbox("s1"), (0, [])) self.assertEqual(self._mailbox("s2"), (0, [])) def test_tool_hook_delivers_radio_rows_once_and_leaves_the_mail_for_the_stop(self): self._send("s1", "post", "board") self.assertEqual(self._tool("s2"), []) self._send("s1", "send", "pan-pan", "pan") self.assertEqual(self._tool("s2"), ["PAN-PAN One: pan"]) self.assertEqual(self._tool("s2"), []) self.assertEqual(self._prompt("s2"), []) self.assertEqual(self._mailbox("s2"), (2, ["NEWSLETTER One: board"])) self.assertEqual(self._tool("s3"), ["PAN-PAN One: pan"]) self.assertEqual(self._tool("s1"), []) def test_tool_hook_skips_parsing_while_the_store_is_unchanged(self): self._send("s1", "post", "x") self.assertEqual(self._tool("s2"), []) news = self.root / "news.jsonl" before = news.stat() row = {"ts": time.time(), "from": "Three", "to": [], "text": "hidden", "class": "security"} with news.open("a", encoding="utf-8") as handle: handle.write(json.dumps(row) + "\n") os.utime(news, ns=(before.st_atime_ns, before.st_mtime_ns)) self.assertEqual(self._tool("s2"), []) os.utime(news, ns=(before.st_atime_ns, before.st_mtime_ns + 10**9)) self.assertEqual(self._tool("s2"), ["SECURITY Three: hidden"]) def test_hooks_skip_subagents_and_never_fail(self): self._send("s1", "send", "pan-pan", "pan") self._send("s1", "post", "board") self.assertEqual(self._hook("hook", "s2", agent_id="a1"), "") self.assertEqual(self._hook("tool", "s2", agent_id="a1"), "") self.assertEqual(self._mailbox("s2", agent_id="a1"), (0, [])) listen = subprocess.run( [sys.executable, "-c", LISTENER, str(RADIO)], input=json.dumps({"session_id": "s2", "agent_id": "a1"}), capture_output=True, text=True, env=self._env(None), check=False, timeout=30, ) self.assertEqual((listen.returncode, listen.stderr), (0, "")) self.assertFalse(self._pidfile("s2").exists()) self.assertFalse((self.root / "radio-seen" / "s2.json").exists()) for command in ("hook", "tool", "mailbox", "block"): for stdin in ("not json", "{}", "[]"): with self.subTest(command=command, stdin=stdin): result = self._exec(None, stdin, (command,)) self.assertEqual((result.returncode, result.stdout), (0, "")) def test_the_first_hook_notifies_the_operator_once(self): log = self.root.parent / "notified.log" script = "import sys; open(sys.argv[1], 'a').write(sys.argv[2] + '\\n')" self.config.write_text(json.dumps({"notify": [sys.executable, "-c", script, str(log)]}), encoding="utf-8") self._send("s1", "send", "mayday", "smoke") self._prompt("s2") self._prompt("s3") self.assertEqual(log.read_text(encoding="utf-8"), "Agents radio MAYDAY by One: smoke\n") def test_listener_wakes_on_mayday(self): proc = self._listen("s2") self.assertEqual(self._send("s1", "send", "mayday", "host down"), ["sent mayday to all", "wake: SendMessage to Three"]) self.assertEqual(self._finish(proc), (2, ["MAYDAY One: host down"])) self.assertEqual(self._tool("s2"), []) self.assertEqual(len(self._prompt("s2")), 1) def test_listener_ignores_the_mail_and_wakes_on_a_broadcast(self): proc = self._listen("s2") self._send("s1", "post", "board") self._send("s1", "post", "plain", "--to", "Two") self._send("s1", "send", "call", "elsewhere", "--to", "Three") time.sleep(0.5) self.assertIsNone(proc.poll()) self._send("s1", "send", "broadcast", "all hands") self.assertEqual(self._finish(proc), (2, ["BROADCAST One: all hands"])) self.assertEqual(self._prompt("s2"), []) self.assertEqual(self._mailbox("s2"), (2, ["NEWSLETTER One: board", "POST One to Two: plain"])) def test_listener_wakes_on_a_call_and_on_pan_pan(self): proc = self._listen("s2") self._send("s1", "send", "call", "answer me", "--to", "Two") self.assertEqual(self._finish(proc), (2, ["CALL One to Two: answer me"])) proc = self._listen("s2") self._send("s1", "send", "pan-pan", "pan") self.assertEqual(self._finish(proc), (2, ["PAN-PAN One: pan"])) def test_listener_yields_to_a_newer_one(self): first = self._listen("s2") second = self._listen("s2") self.assertEqual(self._finish(first), (0, [])) self.assertIsNone(second.poll()) self._pidfile("s2").write_text("0", encoding="utf-8") self.assertEqual(self._finish(second), (0, [])) def test_a_mayday_blocks_the_writing_actions_of_everyone_but_its_sender_and_helpers_until_clear(self): self.assertEqual(self._block("s2", "Edit"), 0) self._send("s1", "send", "mayday", "disk full") for tool in ("Edit", "Write", "MultiEdit", "NotebookEdit"): with self.subTest(tool=tool): self.assertEqual(self._block("s2", tool), 2) self.assertEqual(self._block("s2", "Edit", agent_id="a1"), 2) for command in ( "rm -rf build", "git status", f"python3 {RADIO} show; rm -rf build", f"python3 {RADIO} show && touch x", f"echo $(python3 {RADIO} show)", "python3 /elsewhere/radio.py show", f"python3 -W {RADIO} /tmp/other.py", f"python3 -X {RADIO} /tmp/other.py", f"python3 -c {RADIO}", f"./python-other {RADIO} show", f"/tmp/elsewhere/python3 {RADIO} show", f"python3 -i {RADIO} show", f'python3 {RADIO} show "$(touch x)"', f"python3 {RADIO} show \\';touch x;\\'", f"python3 {RADIO} \"it's\" ; touch x ; echo 'z'", f"python3 {RADIO} show <", shown) self.assertNotIn("", shown) def _victims(self) -> tuple[Path, Path]: victim = self.root.parent / "victim" victim.write_text("keep", encoding="utf-8") victims = self.root.parent / "victims" victims.mkdir() return victim, victims def test_the_mailbox_writes_no_session_state_through_a_planted_name(self): self._send("s1", "post", "board") victim, _ = self._victims() (self.root / "radio-seen").mkdir() (self.root / "radio-seen" / "s2.json").symlink_to(victim) self._exec(None, json.dumps({"session_id": "s2"}), ("mailbox",)) self.assertEqual(victim.read_text(encoding="utf-8"), "keep") def test_the_mailbox_writes_no_session_state_through_a_planted_directory(self): self._send("s1", "post", "board") _, victims = self._victims() (self.root / "radio-seen").symlink_to(victims) self._exec(None, json.dumps({"session_id": "s2"}), ("mailbox",)) self.assertEqual(os.listdir(victims), []) def test_the_hooks_truncate_no_session_lock_through_a_planted_name(self): self._send("s1", "send", "security", "sec") victim, _ = self._victims() (self.root / "radio-seen").mkdir() (self.root / "radio-seen" / "s2.lock").symlink_to(victim) self._hook("hook", "s2") self._hook("tool", "s2") self._exec(None, json.dumps({"session_id": "s2"}), ("mailbox",)) self.assertEqual(victim.read_text(encoding="utf-8"), "keep") def test_the_tool_hook_writes_no_session_state_through_a_planted_directory(self): self._send("s1", "send", "pan-pan", "pan") _, victims = self._victims() (self.root / "radio-seen").symlink_to(victims) self._hook("tool", "s2") self.assertEqual(os.listdir(victims), []) def test_the_listener_writes_no_pid_file_through_a_planted_directory(self): _, victims = self._victims() self.root.mkdir() (self.root / "radio-listen").symlink_to(victims) proc = subprocess.Popen( [sys.executable, "-c", LISTENER, str(RADIO)], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, env=self._env(None), ) self.addCleanup(self._stop, proc) proc.stdin.write(json.dumps({"session_id": "s2"})) proc.stdin.close() try: code = proc.wait(timeout=10) except subprocess.TimeoutExpired: code = None self.assertEqual(os.listdir(victims), []) self.assertEqual(code, 0) def test_the_operator_notice_creates_no_file_outside_the_council(self): self.config.write_text(json.dumps({"notify": [sys.executable, "-c", "pass"]}), encoding="utf-8") self._send("s1", "send", "mayday", "fire") escaped = self.root.parent / "escaped" silence = {"since": f"../../{escaped.name}", "until": time.time() + 600, "by": "One", "text": "fire"} (self.root / "radio-silence.json").write_text(json.dumps(silence), encoding="utf-8") self.assertEqual(self._tool("s2"), ["MAYDAY One: fire"]) self.assertFalse(escaped.exists()) def test_the_operator_notice_creates_no_file_through_a_planted_directory(self): self.config.write_text(json.dumps({"notify": [sys.executable, "-c", "pass"]}), encoding="utf-8") self._send("s1", "send", "mayday", "fire") _, victims = self._victims() (self.root / "radio-notified").symlink_to(victims) self.assertEqual(self._tool("s2"), ["MAYDAY One: fire"]) self.assertEqual(os.listdir(victims), []) def test_the_operator_notice_runs_no_binary_planted_on_the_path(self): planted = self.root.parent / "planted" planted.mkdir() marker = self.root.parent / "ran" fake = planted / "council-probe" fake.write_text(f"#!/bin/sh\n/usr/bin/touch '{marker}'\n", encoding="utf-8") fake.chmod(0o755) self.config.write_text(json.dumps({"notify": ["council-probe"]}), encoding="utf-8") self._send("s1", "send", "mayday", "fire") result = subprocess.run( [sys.executable, str(RADIO), "tool"], input=json.dumps({"session_id": "s2"}), capture_output=True, text=True, env={**self._env(None), "PATH": str(planted)}, check=False, timeout=30, ) self.assertEqual(result.returncode, 0, result.stderr) self.assertIn("MAYDAY", result.stdout) self.assertFalse(marker.exists()) def test_the_hooks_skip_a_session_lock_another_process_holds(self): self._send("s1", "post", "board") lock = self.root / "radio-seen" / "s2.lock" lock.parent.mkdir(parents=True, exist_ok=True) with lock.open("w") as handle: fcntl.flock(handle, fcntl.LOCK_EX) start = time.monotonic() self._hook("hook", "s2") self._hook("tool", "s2") self.assertEqual(self._mailbox("s2"), (0, [])) self.assertLess(time.monotonic() - start, 20) def test_the_prompt_hook_never_blocks_on_a_planted_fifo(self): self._send("s1", "send", "security", "sec") os.mkfifo(self.root / "radio-silence.json", 0o600) self.assertTrue(any("sec" in line for line in self._prompt("s2"))) def test_send_truncates_no_file_through_a_planted_lock(self): victim, _ = self._victims() self.root.mkdir() (self.root / "news.lock").symlink_to(victim) result = self._cli("s1", "post", "board") self.assertEqual(victim.read_text(encoding="utf-8"), "keep") self.assertFalse((self.root / "news.jsonl").exists()) self.assertNotEqual(result.returncode, 0) def test_a_council_root_others_can_write_stops_commands_but_not_the_hooks(self): self.root.mkdir() self.root.chmod(0o777) refused = self._cli("s1", "show") self.assertEqual(refused.returncode, 2) self.assertIn("cannot write", refused.stderr) quiet = self._exec(None, json.dumps({"session_id": "s2"}), ("hook",)) self.assertEqual(quiet.returncode, 0) self.assertIn("council hooks stay off", quiet.stdout) if __name__ == "__main__": unittest.main()