"""Questions with answer buttons, team memory, collision radar, pulse, metrics, wins, export, handoff reminders.""" import json import re import shutil import subprocess import sys import unittest from helpers import ROOT, HarborCase, env_for, http, make_home, run_module, temp_dir, wait_for_port from harbor import hooks, projects from harbor.core import Harbor, HarborError class Questions(HarborCase): def test_ask_answer_and_fallback(self): q = self.h.ask("Codex", "Rename 0x445310 to RouteCellCost?", ["Yes", "No", "Later"], due="2h", fallback="Claude") self.assertEqual((q["to_name"], q["options"]), ("User", ["Yes", "No", "Later"])) self.assertEqual(self.h.questions_by_message([q["message_id"]])[q["message_id"]]["id"], q["id"]) self.assertTrue(any(a["kind"] == "questions" for a in self.h.attention())) with self.assertRaises(HarborError): self.h.answer("Kimi", q["id"], "Yes") # not for Kimi with self.assertRaises(HarborError): self.h.answer("Claude", q["id"], "Yes") # the human is here with self.assertRaises(HarborError): self.h.answer("User", q["id"], "Yes", origin="cli") # the human answers on the page with self.assertRaises(HarborError): self.h.answer("User", q["id"], "Maybe", origin="web") # not an option self.h.set_presence("User", "web", "away", until="8h") done = self.h.answer("Claude", q["id"], "Yes") # the fallback while away self.assertEqual((done["answer"], done["answered_by"]), ("Yes", "Claude")) reply = self.h.inbox("Codex")[-1] self.assertEqual((reply["kind"], reply["reply_to"]), ("answer", q["message_id"])) self.assertIn("Yes", reply["body"]) with self.assertRaises(HarborError): self.h.answer("User", q["id"], "No", origin="web") # answered already def test_agent_to_agent_and_free_text(self): q = self.h.ask("Codex", "Which header owns g_audio_manager?", to="Kimi") self.assertIn("Questions for you", self.h.brief("Kimi")) self.assertIn("Your open questions", self.h.brief("Codex")) self.h.answer("Kimi", q["id"], "audio_manager.h, declared extern") self.assertEqual(self.h.questions(open_only=True), []) with self.assertRaises(HarborError): self.h.ask("Codex", "x", fallback="Kimi") # fallback must coordinate class Memory(HarborCase): def test_notes_and_context(self): self.h.note("Kimi", "fn:0x52BD10", "ABI: stdcall, ret 4; the third push is its argument") old = self.h.note("Codex", "0x445310", "uses a signed read") self.h.note("Codex", "fn:0x445310", "uses an unsigned read (movzx)", replaces=old["id"]) self.h.send("DeepSeek", "All", "0x445310 RouteCellCost draft is ready") self.h.add_task("Claude", "make 0x445310 readable", scope="fn:0x445310") self.h.submit("DeepSeek", "RouteCellCost readable", scope="fn:0x445310", commit="abc") text = self.h.context_text("fn:0x445310") for part in ("unsigned read", "review item", "task #", "RouteCellCost draft"): self.assertIn(part, text) self.assertNotIn("uses a signed read", text) # replaced notes drop out self.assertEqual(len(self.h.notes("range:0x440000-0x47FFFF")), 1) self.assertIn("nothing yet", self.h.context_text("fn:0x400000")) task = self.h.next_task("Codex") self.assertEqual(task["scope"], "fn:0x445310") class Collisions(unittest.TestCase): def setUp(self): self.dir = temp_dir() data = {"name": "radar", "coordinators": ["Lead"], "database": str(self.dir / "r.db"), "agents": [{"name": "Lead"}, {"name": "Ann", "workdir": str(self.dir / "wt-ann")}, {"name": "Bob", "workdir": str(self.dir / "wt-bob")}]} self.h = Harbor(projects.project_from_dict(data)) def tearDown(self): self.h.close() shutil.rmtree(self.dir, ignore_errors=True) def test_same_file_in_two_worktrees(self): paths = self.h.edited_paths("Edit", {"file_path": str(self.dir / "wt-ann" / "annotations" / "names.csv")}) self.assertEqual(self.h.record_touches("Ann", paths, "Edit"), []) patch = "*** Begin Patch\n*** Update File: annotations/names.csv\n@@\n-a\n+b\n*** End Patch" paths = self.h.edited_paths("apply_patch", {"input": patch}) self.assertEqual(paths, ["annotations/names.csv"]) hits = self.h.record_touches("Bob", paths, "apply_patch", cwd=str(self.dir / "wt-bob")) self.assertEqual([(c["path"], c["other"]) for c in hits], [("annotations/names.csv", "Ann")]) alert = self.h.inbox("Ann")[-1] self.assertEqual((alert["sender"], alert["kind"], alert["topic"]), ("Bob", "finding", "file:annotations/names.csv")) self.assertEqual(self.h.inbox("Lead")[-1]["id"], alert["id"]) self.h.record_touches("Bob", paths, "apply_patch") self.assertEqual(len([m for m in self.h.inbox("Ann", unread_only=False) if m["kind"] == "finding"]), 1) self.assertIn("Files you share", self.h.brief("Ann")) self.assertTrue(any(a["kind"] == "collision" for a in self.h.attention())) self.assertEqual(self.h.edited_paths("Read", {"file_path": "x.c"}), []) def test_hook_records_edits(self): (self.dir / ".harbor.toml").write_text(f"project = 'radar'\n", encoding="utf-8") home = self.dir / "home" (home / "projects").mkdir(parents=True) data = (f"name = 'radar'\ncoordinators = ['Lead']\ndatabase = '{self.dir / 'r.db'}'\n" f"[[agents]]\nname = 'Lead'\n[[agents]]\nname = 'Ann'\nworkdir = '{self.dir / 'wt-ann'}'\n" f"[[agents]]\nname = 'Bob'\nworkdir = '{self.dir / 'wt-bob'}'\n") (home / "projects" / "radar.toml").write_text(data, encoding="utf-8") for agent, wt in (("Ann", "wt-ann"), ("Bob", "wt-bob")): payload = {"cwd": str(self.dir), "tool_name": "Write", "tool_input": {"file_path": str(self.dir / wt / "src" / "x.c")}} r = run_module(home, "--as", agent, "--project", "radar", "hook", "pre-tool", "--host", "claude", stdin=json.dumps(payload)) self.assertEqual((r.returncode, r.stdout), (0, ""), r.stderr) self.assertTrue(any(m["kind"] == "finding" for m in self.h.inbox("Ann"))) class PulseScore(HarborCase): def test_pulse_metrics_wins_export(self): for _ in range(3): self.h.activity("Codex", "shell", "codex") self.h.send("Codex", ["Claude"], "hi") rows = self.h.pulse(hours=1)["agents"]["Codex"] self.assertEqual((sum(r["tools"] for r in rows), sum(r["messages"] for r in rows)), (3, 1)) self.h.metric("Claude", "Readable", 778) self.h.metric("Claude", "readable", 779, note="b4b44f0") m = self.h.metrics()["readable"] self.assertEqual(([p["value"] for p in m["series"]], m["latest"]["note"]), ([778, 779], "b4b44f0")) with self.assertRaises(HarborError): self.h.metric("Claude", "readable", "lots") item = self.h.submit("DeepSeek", "RouteCellCost readable", scope="fn:0x445310", commit="abc") self.h.transition("Claude", item["id"], "approved") self.h.transition("Claude", item["id"], "landed", commit="b4b44f0") self.assertEqual([w["title"] for w in self.h.wins()], ["RouteCellCost readable"]) win = self.h.inbox("User")[-1] self.assertEqual(win["topic"], "wins") md = self.h.export_thread(f"item:{item['id']}") self.assertTrue(md.startswith("# Review: RouteCellCost readable")) self.assertIn("approved -> landed", md) def test_handoff_reminder_once(self): for _ in range(31): self.h.activity("Kimi", "shell") text = hooks.handoff_reminder(self.h, "Kimi") self.assertIn("handoff is missing", text) self.assertIsNone(hooks.handoff_reminder(self.h, "Kimi")) self.h.handoff("Codex", "x") self.assertIsNone(hooks.handoff_reminder(self.h, "Codex")) # not active enough anyway def test_stop_message_asks_for_a_handoff(self): self.h.set_flag("stop:Kimi", "on", "Claude", "cli") self.assertIn("write a handoff", self.h.inbox("Kimi")[-1]["body"]) class ServerExtras(unittest.TestCase): PORT = 8794 @classmethod def setUpClass(cls): cls.dir = temp_dir() cls.project = make_home(cls.dir)["test"] cls.proc = subprocess.Popen([sys.executable, "-B", "-m", "harbor", "serve", "--port", str(cls.PORT)], cwd=cls.dir, env=env_for(cls.dir, PYTHONPATH=str(ROOT)), stdout=subprocess.PIPE, stderr=subprocess.STDOUT) assert wait_for_port(cls.PORT) cls.token = re.search(r'name="harbor-token" content="([^"]+)"', http(cls.PORT, "/")[1]).group(1) @classmethod def tearDownClass(cls): cls.proc.terminate() cls.proc.wait(timeout=10) cls.proc.stdout.close() shutil.rmtree(cls.dir, ignore_errors=True) def test_answer_from_the_page_and_new_endpoints(self): h = Harbor(self.project) q = h.ask("Codex", "Ship it?", ["Yes", "No"]) h.note("Kimi", "fn:0x445310", "unsigned read") h.activity("Codex", "shell") h.close() shown = {m["id"]: m for m in http(self.PORT, "/messages")[1]}[q["message_id"]] self.assertEqual((shown["question"]["options"], shown["question"]["answer"]), (["Yes", "No"], None)) self.assertEqual(len(http(self.PORT, "/api/overview")[1]["questions"]), 1) status, r = http(self.PORT, "/api/answer", {"question": q["id"], "answer": "Yes"}, token=self.token) self.assertEqual((status, r["question"]["answer"]), (200, "Yes")) shown = {m["id"]: m for m in http(self.PORT, "/messages")[1]}[q["message_id"]] self.assertEqual(shown["question"]["answered_by"], "User") self.assertIn("Codex", http(self.PORT, "/api/pulse?hours=2")[1]["agents"]) self.assertIn("unsigned read", http(self.PORT, "/api/context?scope=fn:0x445310")[1]["text"]) status, md = http(self.PORT, "/api/export?thread=questions") self.assertEqual(status, 200) self.assertIn("Ship it?", md) class CliExtras(unittest.TestCase): def test_commands(self): d = temp_dir() make_home(d) try: r = run_module(d, "--as", "Codex", "ask", "Merge", "now?", "--options", "Yes|No") self.assertIn("question #1", r.stdout, r.stderr) self.assertIn("note #1", run_module(d, "--as", "Kimi", "note", "fn:0x445310", "unsigned", "read").stdout) self.assertIn("unsigned read", run_module(d, "--as", "Kimi", "context", "0x445310").stdout) self.assertIn("readable = 779", run_module(d, "--as", "Claude", "metric", "readable", "779").stdout) self.assertIn("Activity per hour", run_module(d, "--as", "Claude", "pulse").stdout) self.assertIn("# Questions", run_module(d, "export", "questions").stdout) finally: shutil.rmtree(d, ignore_errors=True) if __name__ == "__main__": unittest.main()