383 lines
21 KiB
Python
383 lines
21 KiB
Python
"""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_shared_server_takes_the_calling_session_from_the_hook_ticket(self):
|
|
"""One app's server shared by two sessions: each call runs as the agent its hook named in a ticket."""
|
|
from fastmcp import Client
|
|
from harbor import mcp_server, sessions
|
|
d = temp_dir()
|
|
project = make_home(d)["test"]
|
|
mcp_server._state.clear()
|
|
mcp_server._state["project"] = project
|
|
saved = {k: os.environ.get(k) for k in ("HARBOR_AGENT", "HARBOR_HOST")}
|
|
os.environ.update(HARBOR_AGENT="Codex", HARBOR_HOST="codex")
|
|
|
|
async def scenario():
|
|
async with Client(mcp_server.mcp) as client:
|
|
sessions.leave(mcp_server.harbor(), "codex", "doing", {"text": "second session"}, "DeepSeek")
|
|
await client.call_tool("doing", {"text": "second session"}) # the ticket names DeepSeek
|
|
await client.call_tool("doing", {"text": "first session"}) # no ticket: the server's own
|
|
status = (await client.call_tool("status", {})).data
|
|
doing = {a["name"]: a["doing"] for a in status["agents"]}
|
|
self.assertEqual(doing["DeepSeek"], "second session")
|
|
self.assertEqual(doing["Codex"], "first session")
|
|
sessions.leave(mcp_server.harbor(), "codex", "i_am", {"agent": "DeepSeek"}, "DeepSeek")
|
|
self.assertTrue((await client.call_tool("i_am", {"agent": "DeepSeek"})).data["matches"])
|
|
self.assertFalse((await client.call_tool("i_am", {"agent": "DeepSeek"})).data["matches"]) # used up
|
|
|
|
try:
|
|
asyncio.run(scenario())
|
|
finally:
|
|
for k, v in saved.items():
|
|
if v is None:
|
|
os.environ.pop(k, None)
|
|
else:
|
|
os.environ[k] = v
|
|
mcp_server._state.pop("h").close()
|
|
mcp_server._state.clear()
|
|
|
|
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()
|