Kevin Veen-BirkenbachandClaude Opus 5.5
4540494c30
The Agents views gain a Hierarchie view: an org chart of the chair and the line offices read from the council claims. The operator appoints, hands over and recalls an office there; the page signs each change with the passkey and a host unit (client/offices.py, offices path unit) verifies the signature, removes a live holder itself and runs a plain council.py claim for the successor. council.py loses the --operator flag, so no agent can override a live holder. The agent order is now separate from task priority: rasci.py agents order stores it in agent-order.json beside the register through the nofollow helpers and never writes a --prio pin. It only breaks ties of equal task rank, also among the queue grants. The Hierarchie view edits it as a drag list. Host units run the installed skill scripts, so orchestrator.sh keeps the offices unit off while the installed council.py, nofollow.py or rasci.py differ from the tree. Job files, council files and hub records are read with O_NOFOLLOW and O_NONBLOCK and size caps; nofollow.rewrite replaces a claim without creating directories, so a beat never revives a released office. 🎯 Fix: 92 % 💥 Break: 4 % 💸 Cost: 100 % ⏱️ Time: 79 % 🛡️ Security: 100 % 🏆 Score: 100 🚦 Gate: ✅ 90/5, cost and time <= 100, security 100 % Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
212 lines
10 KiB
Python
212 lines
10 KiB
Python
"""Exercise the kvb-org-organize council script: election, staleness, release, list, gate."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[1]
|
|
COUNCIL = REPO_ROOT / "skills" / "kvb-org-organize" / "scripts" / "council.py"
|
|
|
|
|
|
class TestCouncil(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.tmp.cleanup)
|
|
self.home = Path(self.tmp.name) / "home"
|
|
self.root = Path(self.tmp.name) / "council"
|
|
self.sessions = self.home / ".claude" / "sessions"
|
|
self.sessions.mkdir(parents=True)
|
|
self._live("s1", "One")
|
|
self._live("s2", "Two")
|
|
|
|
def _live(self, sid: str, name: str) -> None:
|
|
(self.sessions / f"{sid}.json").write_text(
|
|
json.dumps({"sessionId": sid, "name": name}), encoding="utf-8"
|
|
)
|
|
|
|
def _run(self, sid: str, *args: str, stdin: str = "", cwd: str | None = None) -> subprocess.CompletedProcess:
|
|
env = {
|
|
"PATH": "/usr/bin:/bin",
|
|
"HOME": str(self.home),
|
|
"AGENTS_COUNCIL_ROOT": str(self.root),
|
|
"CLAUDE_CODE_SESSION_ID": sid,
|
|
}
|
|
return subprocess.run(
|
|
[sys.executable, str(COUNCIL), *args],
|
|
capture_output=True,
|
|
text=True,
|
|
env=env,
|
|
input=stdin,
|
|
cwd=cwd,
|
|
check=False,
|
|
)
|
|
|
|
def test_first_claim_wins_and_second_is_refused(self):
|
|
self.assertEqual(self._run("s1", "claim", "work/x", "--work", "w").returncode, 0)
|
|
second = self._run("s2", "claim", "work/x", "--work", "w")
|
|
self.assertEqual(second.returncode, 3)
|
|
self.assertIn('"session": "s1"', second.stdout)
|
|
|
|
def test_dead_holder_is_stale_and_can_be_taken_over(self):
|
|
self._run("s1", "claim", "work/x", "--work", "w")
|
|
(self.sessions / "s1.json").unlink()
|
|
self.assertEqual(self._run("s2", "claim", "work/x", "--work", "w").returncode, 4)
|
|
self.assertEqual(self._run("s2", "takeover", "work/x", "--work", "w").returncode, 0)
|
|
self.assertIn('"session": "s2"', (self.root / "work/x/claim.json").read_text())
|
|
|
|
def test_takeover_of_a_live_holder_is_refused(self):
|
|
self._run("s1", "claim", "work/x", "--work", "w")
|
|
self.assertEqual(self._run("s2", "takeover", "work/x", "--work", "w").returncode, 3)
|
|
|
|
def test_session_flag_cannot_hide_the_claude_code_session(self):
|
|
self._run("s1", "claim", "work/x", "--work", "w", "--session", "One")
|
|
self.assertIn('"session": "s1"', (self.root / "work/x/claim.json").read_text())
|
|
self.assertEqual(self._run("s1", "release", "work/x", "--session", "One").returncode, 0)
|
|
|
|
def test_only_the_holder_releases(self):
|
|
self._run("s1", "claim", "work/x", "--work", "w")
|
|
self.assertEqual(self._run("s2", "release", "work/x").returncode, 3)
|
|
self.assertEqual(self._run("s1", "release", "work/x").returncode, 0)
|
|
self.assertFalse((self.root / "work/x").exists())
|
|
|
|
def test_no_flag_lets_an_agent_take_over_or_release_a_live_holder(self):
|
|
self._run("s1", "claim", "security/gui", "--work", "w")
|
|
before = (self.root / "security/gui/claim.json").read_text(encoding="utf-8")
|
|
for args in (("takeover", "security/gui", "--work", "w2"), ("takeover", "security/gui", "--work", "w2", "--operator"), ("release", "security/gui"), ("release", "security/gui", "--operator")):
|
|
with self.subTest(args=args):
|
|
self.assertIn(self._run("s2", *args).returncode, (2, 3))
|
|
self.assertEqual((self.root / "security/gui/claim.json").read_text(encoding="utf-8"), before)
|
|
|
|
def test_no_office_operation_follows_a_planted_link(self):
|
|
outside = Path(self.tmp.name) / "outside"
|
|
(outside / "gui").mkdir(parents=True)
|
|
(outside / "gui" / "keep.txt").write_text("x", encoding="utf-8")
|
|
(outside / "gui" / "claim.json").write_text(json.dumps({"session": "s1", "runtime": "claude-code", "beat": 1.0}), encoding="utf-8")
|
|
self.root.mkdir()
|
|
(self.root / "security").symlink_to(outside)
|
|
for args in (
|
|
("takeover", "security/gui", "--work", "w"),
|
|
("release", "security/gui"),
|
|
("claim", "security/new", "--work", "w"),
|
|
("claim", "security/gui", "--work", "w"),
|
|
):
|
|
with self.subTest(args=args):
|
|
self.assertNotEqual(self._run("s2", *args).stdout.split(" ")[0], "elected")
|
|
self.assertEqual(sorted(path.name for path in outside.rglob("*")), ["claim.json", "gui", "keep.txt"])
|
|
(self.root / "work").mkdir()
|
|
(self.root / "work" / "x").symlink_to(outside / "gui")
|
|
self.assertEqual(self._run("s2", "takeover", "work/x", "--work", "w").returncode, 3)
|
|
self.assertTrue((outside / "gui" / "keep.txt").exists())
|
|
|
|
def test_a_claim_from_the_root_directory_records_no_workdir_or_branch(self):
|
|
self.assertEqual(self._run("s2", "claim", "work/y", "--work", "appointed", cwd="/").returncode, 0)
|
|
row = json.loads((self.root / "rasci.json").read_text(encoding="utf-8"))["work/y"]
|
|
self.assertEqual((row["worker"], row["description"], row["status"]), ("Two", "appointed", "running"))
|
|
self.assertNotIn("workdir", row)
|
|
self.assertNotIn("branch", row)
|
|
|
|
def test_a_vacant_office_is_taken_over_without_a_council_root_or_group(self):
|
|
self.assertEqual(self._run("s1", "takeover", "work/x", "--work", "w").stdout, "elected work/x\n")
|
|
self.assertEqual(self._run("s1", "takeover", "security/gui", "--work", "w").stdout, "elected security/gui\n")
|
|
|
|
def test_a_beat_racing_a_release_never_brings_the_office_back(self):
|
|
spec = importlib.util.spec_from_file_location("council_beat_race", COUNCIL)
|
|
council = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(council)
|
|
self.root.mkdir()
|
|
council.ROOT = self.root
|
|
released = {"session": "s1", "runtime": "claude-code", "beat": 1.0}
|
|
council.claims = lambda: [("work/gone", dict(released))]
|
|
with mock.patch.dict(os.environ, {"CLAUDE_CODE_SESSION_ID": "s1"}):
|
|
self.assertEqual(council.cmd_beat(argparse.Namespace(session=None)), 0)
|
|
self.assertEqual(sorted(self.root.rglob("*")), [])
|
|
|
|
def test_gate_blocks_unclaimed_edits_in_a_git_worktree_only(self):
|
|
repo = Path(self.tmp.name) / "repo"
|
|
(repo / ".git").mkdir(parents=True)
|
|
inside = json.dumps({"session_id": "s1", "tool_input": {"file_path": str(repo / "f")}})
|
|
outside = json.dumps(
|
|
{"session_id": "s1", "tool_input": {"file_path": str(self.home / "f")}}
|
|
)
|
|
self.assertEqual(self._run("s1", "gate", stdin=inside).returncode, 2)
|
|
self.assertEqual(self._run("s1", "gate", stdin=outside).returncode, 0)
|
|
self._run("s1", "claim", "work/x", "--work", "w")
|
|
self.assertEqual(self._run("s1", "gate", stdin=inside).returncode, 0)
|
|
|
|
def test_gate_still_blocks_past_a_planted_fifo_or_malformed_claim(self):
|
|
repo = Path(self.tmp.name) / "repo"
|
|
(repo / ".git").mkdir(parents=True)
|
|
inside = json.dumps({"session_id": "s1", "tool_input": {"file_path": str(repo / "f")}})
|
|
(self.root / "work" / "fifo").mkdir(parents=True)
|
|
os.mkfifo(self.root / "work" / "fifo" / "claim.json", 0o600)
|
|
for claim in ({}, [], {"session": "s2", "runtime": "claude-code"}):
|
|
with self.subTest(claim=claim):
|
|
(self.root / "work" / "bad").mkdir(exist_ok=True)
|
|
(self.root / "work" / "bad" / "claim.json").write_text(json.dumps(claim), encoding="utf-8")
|
|
env = {"PATH": "/usr/bin:/bin", "HOME": str(self.home), "AGENTS_COUNCIL_ROOT": str(self.root)}
|
|
result = subprocess.run(
|
|
[sys.executable, str(COUNCIL), "gate"],
|
|
capture_output=True, text=True, env=env, input=inside, check=False, timeout=30,
|
|
)
|
|
self.assertEqual(result.returncode, 2, result.stderr)
|
|
|
|
def _two_claims(self) -> str:
|
|
work = "fix the failing deploy of web-app-x on branch fix/web-app-x/deploy"
|
|
self._run("s1", "claim", "work/x", "--work", work)
|
|
self._run("s2", "claim", "work/y", "--work", "short work")
|
|
(self.sessions / "s2.json").unlink()
|
|
return work
|
|
|
|
def _since(self, office: str) -> str:
|
|
return json.loads((self.root / office / "claim.json").read_text(encoding="utf-8"))["since"]
|
|
|
|
def test_list_prints_office_holder_state_and_a_short_work_cell(self):
|
|
self._two_claims()
|
|
result = self._run("s1", "list")
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertEqual(
|
|
result.stdout.splitlines(),
|
|
["work/x\tOne\tlive\tfix the failing deploy of web-app-x on …", "work/y\tTwo\tstale\tshort work"],
|
|
)
|
|
|
|
def test_list_collapses_whitespace_in_the_short_work_cell(self):
|
|
self._run("s1", "claim", "work/z", "--work", "a\tb\nc")
|
|
self.assertEqual(self._run("s1", "list").stdout.splitlines(), ["work/z\tOne\tlive\ta b c"])
|
|
|
|
def test_list_long_prints_since_and_the_whole_work_line(self):
|
|
work = self._two_claims()
|
|
self.assertEqual(
|
|
self._run("s1", "list", "--long").stdout.splitlines(),
|
|
[
|
|
f"work/x\tOne\t{self._since('work/x')}\tlive\t{work}",
|
|
f"work/y\tTwo\t{self._since('work/y')}\tstale\tshort work",
|
|
],
|
|
)
|
|
|
|
def test_list_mine_keeps_only_the_callers_claims_in_both_forms(self):
|
|
work = self._two_claims()
|
|
self.assertEqual(
|
|
self._run("s1", "list", "--mine").stdout.splitlines(),
|
|
["work/x\tOne\tlive\tfix the failing deploy of web-app-x on …"],
|
|
)
|
|
self.assertEqual(
|
|
self._run("s1", "list", "--mine", "--long").stdout.splitlines(),
|
|
[f"work/x\tOne\t{self._since('work/x')}\tlive\t{work}"],
|
|
)
|
|
|
|
def test_invalid_office_is_a_usage_error(self):
|
|
self.assertEqual(self._run("s1", "claim", "../x", "--work", "w").returncode, 2)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|