Keelrleek/Harbor
1
1 Fork 1
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
7b03642f909d4d1545614797cf13e078b8d6b18f
Harbor/tests/test_interfaces.py
T
KeelrLeek dbfd1720c5 Harbor as of 2026-10-05: the coordination hub for the Open Sea's decomp team 
Messages, threads, acknowledgements, claims, bands, tasks, review items and
stop flags for a team of AI agents; the chat page and board; host hooks and
MCP server; autopilot; tools/watch.py. Put under version control after the
weekend's changes so they have history and can be rolled back. Local only.
2026-10-05 10:44:23 -05:00

348 lines
19 KiB
Python
Raw Blame History

"""Command line, host hooks, web server and MCP tools, each against a temporary Harbor home."""
import asyncio
import json
import os
import re
import shutil
import subprocess
import sys
import unittest
import urllib.request
from helpers import ROOT, env_for, http, make_home, run_module, temp_dir, wait_for_port
from harbor.core import Harbor
class Cli(unittest.TestCase):
def setUp(self):
self.dir = temp_dir()
self.project = make_home(self.dir)["test"]
def tearDown(self):
shutil.rmtree(self.dir, ignore_errors=True)
def test_round_trip_and_identity_rules(self):
r = run_module(self.dir, "--as", "Codex", "send", "Kimi", "please", "review", "--thread", "fn:0x445310", "--ack")
self.assertEqual(r.returncode, 0, r.stderr)
self.assertIn("thread", r.stdout)
r = run_module(self.dir, "--json", "inbox", HARBOR_AGENT="Kimi")
self.assertEqual(json.loads(r.stdout)[0]["body"], "please review")
tree = self.dir / "worktree-deepseek"
tree.mkdir()
(self.dir / ".harbor.toml").write_text('project = "test"\n', encoding="utf-8")
r = run_module(self.dir, "doing", "drafting", "0x445310", cwd=tree) # identity from the folder
self.assertIn("DeepSeek: drafting 0x445310", r.stdout)
self.assertIn("doing: drafting 0x445310", run_module(self.dir, "--as", "Claude", "status").stdout)
r = run_module(self.dir, "--as", "User", "send", "All", "pretend")
self.assertNotEqual(r.returncode, 0)
self.assertIn("chat page", r.stderr)
r = run_module(self.dir, "send", "All", "who am I")
self.assertIn("say who you are", r.stderr)
r = run_module(self.dir, "--as", "Codex", "wait", "--timeout", "0.2")
self.assertIn("No new messages", r.stdout)
self.assertIn("From", run_module(self.dir, "digest", "--since", "1h").stdout)
def test_init_creates_a_project_and_marker(self):
home = self.dir / "home2"
folder = self.dir / "garden"
home.mkdir()
folder.mkdir()
r = run_module(home, "init", "garden", "--title", "Garden app", "--agent", "Lead:Builds it",
"--agent", "Helper", "--coordinator", "Lead", "--marker", str(folder))
self.assertEqual(r.returncode, 0, r.stderr)
r = run_module(home, "--as", "Helper", "whoami", cwd=folder)
self.assertIn("project garden", r.stdout)
r = run_module(home, "--as", "Helper", "send", "Lead", "hello", cwd=folder)
self.assertEqual(r.returncode, 0, r.stderr)
self.assertIn("garden", run_module(home, "projects").stdout)
class Hooks(unittest.TestCase):
def setUp(self):
self.dir = temp_dir()
self.project = make_home(self.dir)["test"]
(self.dir / ".harbor.toml").write_text('project = "test"\n', encoding="utf-8")
self.h = Harbor(self.project)
def tearDown(self):
self.h.close()
shutil.rmtree(self.dir, ignore_errors=True)
def hook(self, event, payload=None, host="claude", agent="DeepSeek"):
return run_module(self.dir, "--as", agent, "hook", event, "--host", host,
stdin=json.dumps(payload or {"cwd": str(self.dir)}))
def test_session_start_and_prompt_submit(self):
self.h.send("Claude", ["DeepSeek"], "please look at 0x445310", importance="urgent")
r = self.hook("session-start")
context = json.loads(r.stdout)["hookSpecificOutput"]["additionalContext"]
self.assertIn("please look at 0x445310", context)
r = self.hook("prompt-submit", host="generic")
self.assertIn("1 unread", r.stdout)
def denied(self, r):
return r.returncode == 0 and r.stdout.strip() and \
json.loads(r.stdout)["hookSpecificOutput"]["permissionDecision"] == "deny"
def test_pre_tool_blocks_only_while_stopped(self):
call = {"cwd": str(self.dir), "tool_name": "Bash", "tool_input": {"command": "git commit -am x"}}
self.assertEqual((self.hook("pre-tool", call).returncode, self.hook("pre-tool", call).stdout), (0, ""))
self.h.set_flag("stop:all", "on", "Claude", "cli", reason="user asked")
for host in ("claude", "codex", "deepseek", "kimi"):
r = self.hook("pre-tool", call, host=host)
self.assertTrue(self.denied(r), (host, r.stdout, r.stderr))
self.assertIn("STOP in force", json.loads(r.stdout)["hookSpecificOutput"]["permissionDecisionReason"])
r = self.hook("pre-tool", call, host="opencode")
self.assertEqual(r.returncode, 2)
self.assertIn("STOP in force", r.stderr)
harbor_call = {"cwd": str(self.dir), "tool_name": "Bash", "tool_input": {"command": "harbor inbox"}}
self.assertEqual(self.hook("pre-tool", harbor_call).stdout, "") # may still read messages
sneaky = {"cwd": str(self.dir), "tool_name": "Bash", "tool_input": {"command": "harbor inbox && git push"}}
self.assertTrue(self.denied(self.hook("pre-tool", sneaky)))
mcp_call = {"cwd": str(self.dir), "tool_name": "mcp__harbor__inbox", "tool_input": {}}
self.assertEqual(self.hook("pre-tool", mcp_call).stdout, "")
def test_stop_hook_keeps_agent_for_urgent_messages_once(self):
self.assertEqual(self.hook("stop").stdout, "")
self.h.send("Claude", ["DeepSeek"], "answer this first", ack_required=True)
r = self.hook("stop", host="deepseek")
decision = json.loads(r.stdout)
self.assertEqual((r.returncode, decision["decision"]), (0, "block"))
self.assertIn("#", decision["reason"])
self.assertEqual(self.hook("stop", host="deepseek").stdout, "") # no loop on the same messages
self.h.send("Claude", ["DeepSeek"], "and this", importance="urgent")
self.assertEqual(self.hook("stop", {"cwd": str(self.dir), "stop_hook_active": True}).stdout, "")
r = self.hook("stop", host="opencode")
self.assertEqual(r.returncode, 2) # a new urgent message nudges once
self.h.send("Claude", ["DeepSeek"], "plain note")
self.assertEqual(self.hook("stop", host="codex").stdout, "") # plain notes do not interrupt
def test_outside_a_project_hooks_are_silent(self):
elsewhere = self.dir / "elsewhere"
elsewhere.mkdir()
(self.dir / ".harbor.toml").unlink()
self.h.set_flag("stop:all", "on", "Claude", "cli")
r = self.hook("pre-tool", {"cwd": str(elsewhere), "tool_name": "Bash"})
self.assertEqual((r.returncode, r.stdout, r.stderr), (0, "", ""))
r = run_module(self.dir, "--as", "DeepSeek", "--project", "test", "hook", "pre-tool", "--host", "kimi",
stdin=json.dumps({"cwd": str(elsewhere), "tool_name": "Shell"}))
self.assertTrue(self.denied(r)) # a bound plugin still gates
def test_kimi_gets_the_brief_on_prompts(self):
self.assertEqual(self.hook("prompt-submit", host="kimi").stdout, "")
self.h.send("Claude", ["DeepSeek"], "please check 0x445310")
self.assertIn("Harbor brief for DeepSeek", self.hook("prompt-submit", host="kimi").stdout)
def test_idle_and_unknown_agents(self):
self.h.send("Claude", ["DeepSeek"], "new work")
idle = json.loads(self.hook("idle", host="opencode").stdout)
self.assertEqual((idle["unread"], bool(idle["prompt"])), (1, True))
again = json.loads(self.hook("idle", host="opencode").stdout)
self.assertEqual((again["unread"], again["prompt"]), (1, None)) # one nudge per message
r = run_module(self.dir, "hook", "pre-tool", stdin=json.dumps({"cwd": str(self.dir)}))
self.assertEqual((r.returncode, r.stdout), (0, "")) # not an agent: stay silent
broken = {k: v for k, v in env_for(self.dir).items()}
r = subprocess.run([sys.executable, "-B", "-m", "harbor", "--as", "DeepSeek", "--project", "nope", "hook",
"pre-tool"], input="{}", capture_output=True, text=True, cwd=self.dir,
env={**broken, "PYTHONPATH": str(ROOT)})
self.assertEqual(r.returncode, 0) # fails open
class Server(unittest.TestCase):
PORT = 8798
@classmethod
def setUpClass(cls):
cls.dir = temp_dir()
cls.projects = make_home(cls.dir, names=("alpha", "beta"))
env = env_for(cls.dir, PYTHONPATH=str(ROOT))
cls.proc = subprocess.Popen([sys.executable, "-B", "-m", "harbor", "serve", "--port", str(cls.PORT)],
cwd=cls.dir, env=env, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
assert wait_for_port(cls.PORT), cls.proc.stdout.read() if cls.proc.poll() is not None else "no server"
page = http(cls.PORT, "/")[1]
cls.token = re.search(r'name="harbor-token" content="([^"]+)"', page).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, **kw):
return http(self.PORT, path, body, token=self.token, **kw)
def test_human_messages_only_from_the_page(self):
self.assertEqual(http(self.PORT, "/send", {"to": "All", "text": "scripted"}, browser=False)[0], 403)
self.assertEqual(self.post("/send", {"to": "All", "text": "x"}, headers={"X-Harbor-Token": "wrong"})[0], 403)
self.assertEqual(self.post("/send", {"to": "All", "text": "x"}, headers={"Origin": "http://evil.example"})[0], 403)
self.assertEqual(self.post("/send", {"to": "All", "text": "x"}, headers={"Sec-Fetch-Site": "cross-site"})[0], 403)
self.assertEqual(http(self.PORT, "/messages", headers={"Host": "attacker.example:8798"})[0], 403)
status, reply = self.post("/send", {"to": "Codex, Kimi", "text": "Please both look"})
self.assertEqual(status, 200, reply)
last = http(self.PORT, "/messages")[1][-1]
self.assertEqual((last["from"], last["recipients"], last["origin"]), ("User", ["Codex", "Kimi"], "web"))
self.assertEqual(http(self.PORT, f"/messages?after={last['id']}")[1], [])
def test_mentions_receipts_and_projects(self):
status, reply = self.post("/p/beta/send", {"to": "All", "text": "@kimi @Codex: can you check this?",
"ack": True})
self.assertEqual((status, reply["recipients"]), (200, ["Kimi", "Codex"]))
beta = Harbor(self.projects["beta"])
beta.ack("Kimi", [reply["id"]])
beta.close()
receipts = http(self.PORT, f"/p/beta/api/receipts?ids={reply['id']}")[1][str(reply["id"])]
self.assertEqual((receipts["acked"], receipts["waiting"]), (["Kimi"], ["Codex"]))
self.assertEqual(http(self.PORT, "/messages")[1][-1]["text"] if http(self.PORT, "/messages")[1] else None,
http(self.PORT, "/p/alpha/messages")[1][-1]["text"] if http(self.PORT, "/p/alpha/messages")[1] else None)
self.assertNotIn("@kimi", json.dumps(http(self.PORT, "/p/alpha/messages")[1])) # projects stay apart
listing = http(self.PORT, "/api/projects")[1]
self.assertEqual([p["name"] for p in listing], ["alpha", "beta"])
self.assertEqual(http(self.PORT, "/p/nope/messages")[0], 404)
self.assertIn('<base href="/p/beta/">', http(self.PORT, "/p/beta/board")[1])
status, css = http(self.PORT, "/p/beta/static/harbor.css") # the shared look, per project too
self.assertEqual((status, ":root" in css), (200, True))
self.assertEqual(http(self.PORT, "/static/icons.svg")[0], 200)
for name in ("themes.css", "themes.js", "InterVariable.woff2"): # themes and the Inter font
with urllib.request.urlopen(f"http://127.0.0.1:{self.PORT}/static/{name}", timeout=10) as r:
self.assertEqual(r.status, 200, name)
self.assertGreater(len(r.read()), 1000, name)
self.assertEqual(http(self.PORT, "/static/server.py")[0], 404) # only the listed files
def test_board_flags_digest_and_files(self):
status, reply = self.post("/api/flag", {"target": "all", "value": "on", "reason": "going home"})
self.assertEqual(status, 200, reply)
overview = http(self.PORT, "/api/overview")[1]
self.assertEqual(next(a for a in overview["agents"] if a["name"] == "Codex")["stopped"], ["stop:all"])
self.assertEqual(self.post("/api/flag", {"target": "all", "value": "off"})[0], 200)
self.assertEqual(self.post("/api/flag", {"target": "Nobody", "value": "on"})[0], 400)
digest = http(self.PORT, "/api/digest?since=1h")[1]
self.assertEqual([f["value"] for f in digest["flags"]], ["on", "off"])
alpha = Harbor(self.projects["alpha"])
report = self.dir / "r.txt"
report.write_text("hello", encoding="utf-8")
sent = alpha.send("Kimi", ["User"], "report", attachments=[report])
alpha.close()
name = http(self.PORT, "/messages")[1][-1]["attachments"][0]
self.assertEqual(http(self.PORT, f"/files/{name}")[1], "hello")
self.assertEqual(http(self.PORT, "/files/..%2F..%2Fsecret")[0], 404)
self.assertEqual(self.post("/api/thread-state", {"thread": "control", "state": "closed"})[0], 200)
self.assertTrue(sent["id"])
class ServerWork(unittest.TestCase):
PORT = 8796
@classmethod
def setUpClass(cls):
cls.dir = temp_dir()
cls.project = make_home(cls.dir)["test"]
env = env_for(cls.dir, PYTHONPATH=str(ROOT))
cls.proc = subprocess.Popen([sys.executable, "-B", "-m", "harbor", "serve", "--port", str(cls.PORT)],
cwd=cls.dir, env=env, 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_tasks_presence_ping_prompt_doctor(self):
status, r = self.post("/api/task", {"action": "add", "title": "Name 0x445310", "scope": "fn:0x445310",
"priority": "high", "assignee": "Codex"})
self.assertEqual(status, 200, r)
task_id = r["task"]["id"]
overview = http(self.PORT, "/api/overview")[1]
self.assertEqual([t["id"] for t in overview["tasks"]], [task_id])
self.assertEqual(self.post("/api/task", {"action": "update", "id": task_id, "state": "cancelled"})[0], 200)
self.assertEqual(http(self.PORT, "/api/tasks")[1], [])
self.assertEqual(len(http(self.PORT, "/api/tasks?all=1")[1]), 1)
status, r = self.post("/api/presence", {"status": "away", "until": "2h", "note": "decide with Claude"})
self.assertEqual((status, r["presence"]["status"]), (200, "away"))
self.assertEqual(http(self.PORT, "/api/overview")[1]["presence"]["note"], "decide with Claude")
status, r = self.post("/api/ping", {"agent": "Kimi"})
self.assertEqual(status, 200, r)
h = Harbor(self.project)
ping = h.inbox("Kimi", mark_read=False)[-1]
self.assertTrue(ping["ack_required"] and ping["importance"] == "urgent" and ping["sender"] == "User")
self.assertIn("User is away", h.brief("Kimi"))
h.handoff("Kimi", "halfway through batch 49")
first = h.send("Codex", "All", "wrong claim")
h.send("Codex", None, "right claim", corrects=first["id"])
h.close()
self.assertIn("halfway through batch 49", http(self.PORT, "/api/prompt?agent=Kimi")[1]["prompt"])
marked = {m["id"]: m for m in http(self.PORT, "/messages")[1]}
self.assertEqual(marked[first["id"]]["corrected_by"], [first["id"] + 1])
self.assertEqual(self.post("/api/pin", {"id": first["id"]})[0], 200)
self.assertTrue({m["id"]: m for m in http(self.PORT, "/messages")[1]}[first["id"]]["pinned"])
self.assertEqual(len(http(self.PORT, f"/messages?before={first['id']}")[1]), first["id"] - 1)
checks = http(self.PORT, "/api/doctor")[1]
self.assertEqual(next(c for c in checks if c["area"] == "Server")["status"], "ok")
self.assertEqual(self.post("/api/presence", {"status": "gone"})[0], 400)
class Mcp(unittest.TestCase):
def test_tools_bind_identity_and_round_trip(self):
from fastmcp import Client
from fastmcp.exceptions import ToolError
from harbor import mcp_server
d = temp_dir()
project = make_home(d)["test"]
mcp_server._state.clear()
mcp_server._state["project"] = project
saved = os.environ.get("HARBOR_AGENT")
async def scenario():
async with Client(mcp_server.mcp) as client:
names = {t.name for t in await client.list_tools()}
self.assertTrue({"brief", "doing", "wait_for_messages", "send", "reply", "ack", "claim", "submit",
"review", "land", "digest", "close_thread"} <= names)
os.environ["HARBOR_AGENT"] = "DeepSeek"
sent = (await client.call_tool("send", {"to": "Kimi", "text": "Please review 0x50A720",
"thread": "fn:0x50A720", "ack": True})).data
await client.call_tool("doing", {"text": "waiting for Kimi"})
await client.call_tool("claim", {"scope": "fn:0x50A720"})
item = (await client.call_tool("submit", {"title": "LayoutTextMesh", "scope": "fn:0x50A720",
"commit": "abc1234"})).data
os.environ["HARBOR_AGENT"] = "Kimi"
self.assertIn("Please review 0x50A720", (await client.call_tool("brief", {})).data)
waited = (await client.call_tool("wait_for_messages", {"timeout_seconds": 2})).data
self.assertEqual([m["id"] for m in waited["messages"]], [sent["id"]])
await client.call_tool("ack", {"message_ids": [sent["id"]], "note": "on it"})
with self.assertRaises(ToolError):
await client.call_tool("claim", {"scope": "fn:0x50A720"})
with self.assertRaises(ToolError):
await client.call_tool("land", {"item_id": item["id"], "commit": "f00"})
status = (await client.call_tool("status", {})).data
self.assertEqual(next(a for a in status["agents"] if a["name"] == "DeepSeek")["doing"],
"waiting for Kimi")
os.environ["HARBOR_AGENT"] = "User"
with self.assertRaises(ToolError):
await client.call_tool("send", {"to": "All", "text": "pretending"})
return sent
try:
sent = asyncio.run(scenario())
finally:
if saved is None:
os.environ.pop("HARBOR_AGENT", None)
else:
os.environ["HARBOR_AGENT"] = saved
mcp_server._state.pop("h").close()
mcp_server._state.clear()
h = Harbor(project)
self.assertEqual(h.message(sent["id"])["origin"], "mcp")
self.assertEqual(h.pending_acks(sender="DeepSeek"), [])
h.close()
shutil.rmtree(d, ignore_errors=True)
if __name__ == "__main__":
unittest.main()
Reference in new issue View Git Blame Copy Permalink