KeelrLeek
f1241c19a3
DeepSeek's inbox was buried under 78 automatic baseline notices read oldest first, so it missed direct answers.
169 lines
8.6 KiB
Python
169 lines
8.6 KiB
Python
"""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()
|