feat(kvb-git-hotfix): share hotfixes in the register and sync before every action

The Maintainer messaged every agent about every hotfix, waking each for a full turn. Hotfixes are now register rows hotfix/<slug> (repository identity by git common dir and normalized origin URL, branch, sha, PR, failing-line signature, targets, stage), written with hotfix.py add/update and queried only when an agent acts, never by cron: pending lists merged hotfixes missing from HEAD (filtered by targets) and open ones; match finds hotfixes whose signature fits a failing line, evaluated in one worker under a shared deadline; void tells gate.py run, before every target, when a merged hotfix missing from HEAD makes the rest of the run worthless (exit 3), the operator's one exception to never cutting a job short; sync fetches first, then under the worktree locks merges origin/<default> and every base up the funnel chain recorded with rasci.py set --base, refusing a detached or foreign branch, a diverged base or a busy worktree, aborting cleanly on a conflict.

gate.py gains hold, which runs any heavy command, deploys included, under the worktree lock, forwards signals, returns the command's code and exits 75 when busy; run holds the same lock and takes the sync lock shared. kvb-check-gate, kvb-ci-triage, kvb-org-organize (Peer protocol, capacity), kvb-org-coordinate and kvb-org-split follow; rasci gaps, wsjf and artifacts skip hotfix and operator rows. Three independent skeptic rounds found 34 defects (three of them high), all fixed with tests; negative controls fail against the earlier code.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Kevin Veen-BirkenbachandClaude Opus 5.5 committed 2026-11-04 17:07:23 +01:00
1 parent 713267dd15
commit b4c4cefadd
17 files changed
+2497 -115

No files matched your search

+31
View File
@@ -151,6 +151,15 @@ class TestRasci(unittest.TestCase):
self._run("set", "work/a", "--status", "running")
self.assertFalse(any(line.startswith("end:") for line in self._out("get", "work/a")))
def test_base_names_the_upstream_branch_after_the_branch(self):
self._run("set", "work/side", "--branch", "sidetree", "--base", "line")
lines = self._out("get", "work/side")
self.assertIn("base: line", lines)
self.assertEqual(lines.index("base: line"), lines.index("branch: sidetree") + 1)
self.assertEqual(json.loads((self.root / "rasci.json").read_text(encoding="utf-8"))["work/side"]["base"], "line")
self._run("set", "work/side", "--base", "")
self.assertFalse(any(line.startswith("base:") for line in self._out("get", "work/side")))
def test_done_tasks_show_only_with_all(self):
self._run("set", "work/a", "--status", "done", "--end", "2026-11-01T12:00:00+00:00")
self._run("role", "work/a", "Skills", "RA")
@@ -279,6 +288,28 @@ class TestRasci(unittest.TestCase):
self._run("estimate", "work/b", "One", "--drop")
self.assertFalse(any(line.startswith("estimate ") for line in self._out("get", "work/b")))
def test_hotfix_and_operator_rows_are_no_work_tasks_for_gaps_and_wsjf(self):
self._run("set", "work/a")
self._run("role", "work/a", "Skills", "RA")
self._run("kr", "work/a", "add", "released")
self._run("set", "hotfix/db", "--status", "running")
self._run("set", "operator/push", "--status", "running")
shown = self._out("show")
self.assertEqual([line.split(" | ")[2] for line in shown[2:]], ["hotfix/db", "operator/push", "work/a"])
self.assertFalse(any(line.startswith("gaps:") for line in shown))
self.assertEqual(
self._out("wsjf")[2:], ["| 1 | work/a | - (0/1 KR) | - | - | - | - | queued |", "unestimated: work/a"]
)
self._run("set", "work/b", "--title", "Beta")
self.assertEqual(self._out("show")[-1], "gaps: work/b has 0 A; work/b has no R; work/b has no key result")
self._run("set", "hotfix/db", "--status", "done", "--branch", "fix/db", "--workdir", "/w/db")
self._run("set", "operator/push", "--branch", "main", "--artifact", "dir:/tmp/push")
self._run("set", "work/a", "--branch", "feat/a")
for args in (("artifacts",), ("artifacts", "--ended"), ("artifacts", "--task", "hotfix/db")):
with self.subTest(args=args):
self.assertEqual([line.split("\t")[1] for line in self._out(*args)],
["work/a"] if args == ("artifacts",) else [])
def test_order_pins_the_agents_in_the_operators_order(self):
self._run("set", "work/a1", "--worker", "Alpha", "--prio", "1")
self._run("set", "work/a2", "--worker", "Alpha", "--prio", "2")