"""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()