KeelrLeek
dbfd1720c5
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.
123 lines
6.4 KiB
Python
123 lines
6.4 KiB
Python
"""Regression tests for the second independent review of 2026-10-02."""
|
|
import json
|
|
import threading
|
|
import unittest
|
|
|
|
from helpers import HarborCase
|
|
from harbor import hooks
|
|
from harbor.core import Harbor, HarborError
|
|
|
|
|
|
def gate(command):
|
|
return hooks._tool_is_harbor({"tool_name": "shell", "tool_input": {"command": command}})
|
|
|
|
|
|
class Findings2(HarborCase):
|
|
def test_01_stop_gate_tricks(self):
|
|
for c in ["harbor brief (Remove-Item -Recurse -Force R:/x/build)",
|
|
"py -3.12 -B R:/Workspace/Tools/Harbor/run.py inbox (Set-Content src/main.c 'x')",
|
|
"harbor brief @(git push --force)", "harbor brief {x}",
|
|
["bash", "-c", "rm -rf build", "harbor inbox"],
|
|
["powershell.exe", "-Command", "Remove-Item -Recurse build #", "harbor inbox"],
|
|
["cmd.exe", "/c", "del /q build &", "harbor inbox"],
|
|
"py R:/Workspace/Tools/backup-2026-10-02/harbor-v1/test_harbor.py brief",
|
|
"py -3.12 -B R:/elsewhere/run.py inbox"]:
|
|
self.assertFalse(gate(c), c)
|
|
|
|
def test_09_plain_harbor_commands_still_work_while_stopped(self):
|
|
for c in ["harbor --json inbox", 'py -3.12 -B "R:/Workspace/Tools/Harbor/run.py" --as Codex inbox',
|
|
'harbor handoff "parser done, lexer in flight"', 'harbor send Claude "half done"',
|
|
"C:/Users/Administrator/AppData/Local/Programs/Python/Python312/python.exe -B "
|
|
"R:/Workspace/Tools/Harbor/run.py brief", "py -3.12 -B R:\\Workspace\\Tools\\harbor.py brief",
|
|
["bash", "-lc", "harbor inbox"], ["powershell", "-Command", "harbor brief"]]:
|
|
self.assertTrue(gate(c), c)
|
|
|
|
def test_02_answers_cannot_be_relayed(self):
|
|
self.h.set_flag("stop:Codex", "on", "User", "web")
|
|
q = self.h.ask("Kimi", "Should Codex resume the parser work now? stop Codex?", ["No", "Yes"])
|
|
self.h.answer("User", q["id"], "No", origin="web")
|
|
answer_id = self.h.question(q["id"])["answer_message_id"]
|
|
self.assertEqual(self.h.message(answer_id)["origin"], "system")
|
|
with self.assertRaises(HarborError):
|
|
self.h.set_flag("stop:Codex", "off", "Kimi", "cli", relay_of=answer_id)
|
|
with self.assertRaises(HarborError):
|
|
self.h.set_flag("stop:DeepSeek", "on", "Kimi", "cli", relay_of=answer_id) # not a typed note
|
|
|
|
def test_03_negations_do_not_resume(self):
|
|
self.h.set_flag("stop:all", "on", "User", "web")
|
|
m = self.h.send("User", "All", "Before anyone continues, I need to look at the build", origin="web")
|
|
with self.assertRaises(HarborError):
|
|
self.h.set_flag("stop:all", "off", "Codex", "cli", relay_of=m["id"])
|
|
|
|
def test_04_metric_must_be_finite(self):
|
|
for bad in ("inf", "nan", "-inf"):
|
|
with self.assertRaises(HarborError):
|
|
self.h.metric("Claude", "x", bad)
|
|
|
|
def test_05_stopped_coordinator_does_not_block_assigned_work(self):
|
|
task = self.h.add_task("Claude", "assigned", scope="fn:0x445310", assignee="Codex")
|
|
self.h.set_flag("stop:Claude", "on", "Claude", "cli")
|
|
self.assertEqual(self.h.next_task("Codex")["id"], task["id"])
|
|
|
|
def test_06_one_answer_wins(self):
|
|
q = self.h.ask("Codex", "Ship?", ["Yes", "No"], fallback="Claude")
|
|
self.h.set_presence("User", "web", "away")
|
|
other = Harbor(self.project, clock=self.clock)
|
|
results = []
|
|
|
|
def answer(h, who, value, origin):
|
|
try:
|
|
results.append(h.answer(who, q["id"], value, origin=origin)["answer"])
|
|
except HarborError as error:
|
|
results.append(str(error))
|
|
t = threading.Thread(target=answer, args=(other, "Claude", "Yes", "cli"))
|
|
t.start()
|
|
answer(self.h, "User", "No", "web")
|
|
t.join()
|
|
other.close()
|
|
final = self.h.question(q["id"])["answer"]
|
|
self.assertEqual(sum(1 for r in results if r in ("Yes", "No")), 1, results)
|
|
self.assertEqual(len([m for m in self.h.inbox("Codex", unread_only=False) if m["kind"] == "answer"]), 1)
|
|
self.assertIn(final, results)
|
|
|
|
def test_07_relayed_stops_are_not_the_coordinators_to_lift(self):
|
|
ask = self.h.send("User", ["Claude"], "stop everyone please", origin="web")
|
|
self.h.set_flag("stop:all", "on", "Claude", "cli", relay_of=ask["id"])
|
|
with self.assertRaises(HarborError):
|
|
self.h.set_flag("stop:all", "off", "Claude", "cli")
|
|
self.h.set_flag("stop:Codex", "on", "Codex", "cli", relay_of=ask["id"]) # self-stop: relay unused
|
|
self.assertIsNone(self.h.flags()["stop:Codex"]["relay_of"])
|
|
|
|
def test_08_fallback_is_not_the_asker_and_the_human_hears_the_answer(self):
|
|
with self.assertRaises(HarborError):
|
|
self.h.ask("Claude", "May I force-push main?", ["Yes", "No"], fallback="Claude")
|
|
q = self.h.ask("Codex", "Rename it?", ["Yes", "No"], fallback="Claude")
|
|
self.h.set_presence("User", "web", "dnd")
|
|
self.h.answer("Claude", q["id"], "Yes")
|
|
told = [m for m in self.h.inbox("User") if m["kind"] == "answer"]
|
|
self.assertEqual(len(told), 1)
|
|
self.assertIn("answered by Claude while User was away", told[0]["body"])
|
|
|
|
def test_10_options_with_odd_spacing(self):
|
|
q = self.h.ask("Codex", "Which?", ["Keep both", "Drop\nlast"])
|
|
self.assertEqual(self.h.question(q["id"])["options"], ["Keep both", "Drop last"])
|
|
self.h.answer("User", q["id"], "Keep both", origin="web")
|
|
|
|
def test_11_notes_are_replaced_only_by_their_author(self):
|
|
kimi = self.h.note("Kimi", "fn:0x52BD10", "stdcall, ret 4")
|
|
with self.assertRaises(HarborError):
|
|
self.h.note("Codex", "file:unrelated.c", "x", replaces=kimi["id"])
|
|
self.assertEqual(len(self.h.notes("fn:0x52BD10")), 1)
|
|
self.h.note("Claude", "fn:0x52BD10", "stdcall, ret 4 (checked)", replaces=kimi["id"])
|
|
|
|
def test_12_paths(self):
|
|
self.assertEqual(self.h._relative("Codex", ".github/workflows/ci.yml"), ".github/workflows/ci.yml")
|
|
self.assertEqual(self.h._relative("Codex", ".env"), ".env")
|
|
self.assertEqual(self.h._relative("Codex", "./src/a.c"), "src/a.c")
|
|
patch = {"input": "*** Begin Patch\n*** Update File: R:\\ws\\new\\a.c\n*** Delete File: src/b.c\n*** End Patch"}
|
|
self.assertEqual(self.h.edited_paths("shell", patch), ["R:\\ws\\new\\a.c", "src/b.c"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|