"""Trial runs, roll calls, launch status, the live activity feed, experts and thread summaries.""" import json import os import re import shutil import subprocess import sys import unittest os.environ["HARBOR_NO_TOASTS"] = "1" from helpers import ROOT, HarborCase, env_for, http, make_home, run_module, temp_dir, wait_for_port from harbor.core import Harbor, HarborError from harbor.launch import KEEP_ACTIVITY, describe_target class TrialRuns(HarborCase): def test_trial_run_ends_by_itself(self): self.h.set_flag("stop:Codex", "on", "User", "web") with self.assertRaises(HarborError): self.h.set_flag("stop:Codex", "off", "Claude", "cli", until="30m") # only the human self.h.set_flag("stop:Codex", "off", "User", "web", until="30m") self.assertEqual(self.h.stopped("Codex"), []) self.assertIsNotNone(self.h.trial("Codex")) self.assertIn("Trial run: you may work until", self.h.brief("Codex")) self.assertIn("trial run until", self.h.inbox("Codex")[0]["body"]) self.clock.t += 31 * 60 stops = self.h.stopped("Codex") self.assertEqual(len(stops), 1) self.assertIn("trial run ended", stops[0]["reason"]) self.assertIsNone(self.h.trial("Codex")) with self.assertRaises(HarborError): self.h.claim("Codex", "fn:0x445310") # stopped again self.h.set_flag("stop:Codex", "off", "User", "web") # a real resume self.assertEqual(self.h.stopped("Codex"), []) def test_trial_for_everyone_and_a_personal_stop(self): self.h.set_flag("stop:all", "on", "User", "web") self.h.set_flag("stop:all", "off", "User", "web", until="15m") self.h.set_flag("stop:Kimi", "on", "User", "web") self.assertEqual(self.h.stopped("Codex"), []) self.assertTrue(self.h.stopped("Kimi")) # its own stop holds self.clock.t += 16 * 60 self.assertTrue(self.h.stopped("Codex")) class RollCall(HarborCase): def test_rollcall_works_while_stopped(self): self.h.set_flag("stop:all", "on", "User", "web") with self.assertRaises(HarborError): self.h.rollcall("Claude", "cli") status = self.h.rollcall("User", "web", agents=["Codex", "Kimi"]) self.assertEqual([a["agent"] for a in status["agents"]], ["Codex", "Kimi"]) q = next(q for q in self.h.questions(to="Codex")) self.h.inbox("Codex") self.clock.t += 42 self.h.answer("Codex", q["id"], "Here") # stopped, still answers rows = {a["agent"]: a for a in self.h.rollcall_status()["agents"]} self.assertEqual((rows["Codex"]["answer"], rows["Codex"]["seconds"]), ("Here", 42)) self.assertIsNone(rows["Kimi"]["answer"]) self.assertTrue(any(m["kind"] == "answer" for m in self.h.inbox("User"))) launch = {a["agent"]: a for a in self.h.launch_status()} self.assertEqual(launch["Codex"]["state"], "stopped") self.assertEqual(launch["Codex"]["rollcall"]["answer"], "Here") self.h.brief("Kimi") self.assertIsNotNone({a["agent"]: a for a in self.h.launch_status()}["Kimi"]["brief_at"]) class Activity(HarborCase): def test_feed_targets_and_pruning(self): self.assertEqual(describe_target("Edit", {"file_path": "x.c"}, ["x.c"]), "x.c") masked = describe_target("Bash", {"command": "curl -H 'Authorization: Bearer abc123' --token=s3cret x"}, []) self.assertNotIn("abc123", masked) self.assertNotIn("s3cret", masked) self.assertEqual(describe_target("shell", {"command": ["bash", "-lc", "make all"]}, []), "make all") for i in range(KEEP_ACTIVITY + 25): self.h.activity("Codex", "shell", "codex", target=f"step {i}") feed = self.h.activity_feed(limit=5, agent="Codex") self.assertEqual(feed[0]["target"], f"step {KEEP_ACTIVITY + 24}") self.assertEqual(self.h.conn.execute("SELECT COUNT(*) FROM activity_log WHERE agent='Codex'").fetchone()[0], KEEP_ACTIVITY) def test_hook_records_targets(self): (self.dir / ".harbor.toml").write_text('project = "test"\n', encoding="utf-8") payload = {"cwd": str(self.dir), "tool_name": "Edit", "tool_input": {"file_path": "annotations/names.csv"}} run_module(self.dir, "--as", "DeepSeek", "hook", "pre-tool", "--host", "claude", stdin=json.dumps(payload)) self.assertEqual(self.h.activity_feed(limit=1)[0]["target"], "annotations/names.csv") class Knowledge(HarborCase): def test_experts_and_summaries(self): self.h.note("Kimi", "fn:0x52BD10", "stdcall, ret 4") self.h.claim("DeepSeek", "fn:0x52BD10") item = self.h.submit("DeepSeek", "AudioManagerTick", scope="fn:0x52BD10", commit="abc") self.h.transition("Kimi", item["id"], "approved") self.h.send("Codex", "All", "0x52BD10 takes the manager in ECX") ranked = [e["agent"] for e in self.h.experts("fn:0x52BD10")] self.assertEqual(sorted(ranked[:2]), ["DeepSeek", "Kimi"]) self.assertIn("who knows this", self.h.context_text("fn:0x52BD10")) for i in range(21): self.h.send("Codex" if i % 2 else "Kimi", ["Claude"], f"point {i}", thread="fn:0x52BD10") self.assertIn("Long threads without a recent summary", self.h.brief("Kimi")) self.h.summarize("Kimi", "fn:0x52BD10", "Decided: stdcall ret 4; DeepSeek fixes the header.") self.assertNotIn("Long threads without", self.h.brief("Kimi")) self.assertIn("summary of fn:0x52BD10", self.h.context_text("fn:0x52BD10")) with self.assertRaises(HarborError): self.h.summarize("Kimi", "fn:0x52BD10", "x" * 4001) class ServerLaunch(unittest.TestCase): PORT = 8792 @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 post(self, path, body): return http(self.PORT, path, body, token=self.token) def test_launch_endpoints(self): self.assertEqual(self.post("/api/flag", {"target": "Codex", "value": "on"})[0], 200) status, r = self.post("/api/flag", {"target": "Codex", "value": "off", "until": "30m", "reason": "trial run"}) self.assertEqual(status, 200, r) launch = {a["agent"]: a for a in http(self.PORT, "/api/launch")[1]["agents"]} self.assertEqual(launch["Codex"]["state"], "trial") status, r = self.post("/api/rollcall", {"agents": ["Kimi"]}) self.assertEqual((status, [a["agent"] for a in r["rollcall"]["agents"]]), (200, ["Kimi"])) h = Harbor(self.project) h.activity("Kimi", "Edit", "kimi", target="src/a.c") h.summarize("Kimi", "rollcall", "Kimi is here.") h.close() self.assertEqual(http(self.PORT, "/api/feed")[1][0]["target"], "src/a.c") overview = http(self.PORT, "/api/overview")[1] kimi = next(a for a in overview["agents"] if a["name"] == "Kimi") self.assertEqual(kimi["last_target"], "src/a.c") self.assertIsNotNone(next(a for a in overview["agents"] if a["name"] == "Codex")["trial_until"]) thread = http(self.PORT, "/api/thread?id=rollcall")[1] self.assertEqual(thread["summary"]["body"], "Kimi is here.") class CliLaunch(unittest.TestCase): def test_commands(self): d = temp_dir() make_home(d) try: run_module(d, "--as", "Kimi", "note", "fn:0x52BD10", "stdcall") self.assertIn("Kimi", run_module(d, "--as", "Codex", "who", "fn:0x52BD10").stdout) self.assertIn("harbor note", run_module(d, "--as", "Codex", "feed").stdout) # CLI calls are activity run_module(d, "--as", "Kimi", "send", "Claude", "x", "--thread", "t1") self.assertIn("pinned to t1", run_module(d, "--as", "Kimi", "summarize", "t1", "all", "settled").stdout) finally: shutil.rmtree(d, ignore_errors=True) if __name__ == "__main__": unittest.main()