From ca54420f4d915c99d7f5a4fd52aa9443f2d615ac Mon Sep 17 00:00:00 2001 From: Kevin Veen-Birkenbach Date: Wed, 11 Nov 2026 20:33:54 +0000 Subject: [PATCH] fix(kvb-org-queue): restart holders on stale queue code; link host skills into projects MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Every holder judges the shared queue.json with its own loaded code, so the project copies from 2026-11-07 (no RAM/CPU budget) granted heavy jobs on the slot limit alone and overloaded the host at 18:24. Each entry now stamps a hash of queue.py and the sibling scripts it loads; a poll that finds other code installed in ~/.agents/skills leaves the queue ungranted and re-execs the same hold with the installed queue.py, on the realpath interpreter, with signals reset and a caught signal ending the hold instead. queue.py list prefixes entries on other or unstamped code. A project install now symlinks every skill the global install holds instead of copying it, so the project hooks and ../ references stay valid and one host runs one version. 🎯 Fix: 92 % 💥 Break: 4 % 💸 Cost: 100 % ⏱️ Time: 100 % 🛡️ Security: 100 % 🏆 Score: 88 🚦 Gate: ✅ 90/5, cost and time <= 100, security 100 % Co-Authored-By: Claude Opus 5.5 --- scripts/install.sh | 16 ++++++- skills/kvb-org-queue/scripts/queue.py | 62 ++++++++++++++++++++++++--- tests/test_install.py | 42 ++++++++++++------ tests/test_queue.py | 36 ++++++++++++++++ 4 files changed, 135 insertions(+), 21 deletions(-) diff --git a/scripts/install.sh b/scripts/install.sh index df05983..9e3db73 100644 --- a/scripts/install.sh +++ b/scripts/install.sh @@ -2,7 +2,8 @@ # Render the skills from skills-lock.json and copy them into TARGET. # TARGET defaults to $HOME (global install); pass a project root to # equip that project. Skills land in /.agents/skills and are -# mirrored to /.claude/skills. +# mirrored to /.claude/skills. A project install links every skill +# the global install in $HOME already holds to that copy instead. set -euo pipefail SCRIPT_DIR="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)" @@ -38,11 +39,22 @@ if [[ -d "${REPO_ROOT}/skills" ]]; then cp -a "${REPO_ROOT}/skills/." "${src}/" fi -for dst in "${TARGET}/.agents/skills" "${TARGET}/.claude/skills"; do +for root in .agents/skills .claude/skills; do + dst="${TARGET}/${root}" log "skills: copying -> ${dst}" mkdir -p "${dst}" find "${dst}" -mindepth 1 -delete cp -a "${src}/." "${dst}/" + if [[ ! "${TARGET}" -ef "${HOME}" ]]; then + log "skills: linking in ${dst} what ${HOME}/${root} already holds" + for skill in "${dst}"/*; do + global="${HOME}/${root}/${skill##*/}" + if [[ -e "${global}" ]]; then + rm -rf "${skill}" + ln -s "${global}" "${skill}" + fi + done + fi done settings="${TARGET}/.claude/settings.local.json" diff --git a/skills/kvb-org-queue/scripts/queue.py b/skills/kvb-org-queue/scripts/queue.py index 337b334..3c99f12 100755 --- a/skills/kvb-org-queue/scripts/queue.py +++ b/skills/kvb-org-queue/scripts/queue.py @@ -33,7 +33,10 @@ Args: CMD in the current directory with inherited stdio in a session of its own, forward SIGINT and SIGTERM to its process group, measure every POLL_SECONDS and at its end, record its history row and leave the - queue. + queue. The entry carries its code (a hash of this file and the + SIBLINGS); a poll that finds other code installed in + ~/.agents/skills leaves the queue ungranted and restarts the same hold + with the installed queue.py. --kind: a key of slot_need_gb in the config. --task: an open task of the kvb-org-rasci register; its rank orders the queue, and the operator's agent order (`rasci.py agents @@ -47,8 +50,9 @@ Args: id, status, kind, task, need GB, minutes waited, ETA minutes (until a running job ends or a waiting one starts, from learned medians; - while one is unknown), workdir, blocking reason (the one the holder saw at - its last poll, host load included). Entries of dead holders are - dropped first. + its last poll, host load included; prefixed when the entry runs other + code than the installed one). Entries of dead holders are dropped + first. history: a markdown table per command signature and one per kind of what the last 30 days of history teach, with each slot limit's source. cancel ID: remove the caller's waiting entry ID; its hold exits 3. @@ -80,8 +84,9 @@ Exit codes: a command exiting 75 (EX_TEMPFAIL: it did no work, such as a declined `operator.py granted`) leaves no history row; before the command started 2 on a usage or config error or when it - cannot start, 3 when cancel removed the entry, 128+N when signal N - reached hold. + cannot start or cannot restart with the installed code, 3 when cancel + removed the entry, 128+N when signal N reached hold; after a restart, + the installed hold's exit code. list, history: 0 done, 2 config error. cancel: 0 removed, 2 no waiting entry of the caller with that id. block: 0 allow, or ask when it prints an ask decision, or a council root @@ -97,6 +102,7 @@ import argparse import contextlib import fcntl import functools +import hashlib import importlib.util import json import os @@ -120,6 +126,9 @@ SIBLINGS = { "history": Path(__file__).resolve().parent / "history.py", "nofollow": SKILLS / "kvb-org-organize" / "scripts" / "nofollow.py", } +HOST_SKILLS = Path.home() / ".agents" / "skills" +EXE = os.path.realpath(sys.executable) +CODE_FILES = tuple(path.relative_to(SKILLS) for path in (Path(__file__).resolve(), *SIBLINGS.values())) POLL_SECONDS = 20 CLAIM_SECONDS = 3 * POLL_SECONDS TICK_SECONDS = 0.2 @@ -157,6 +166,20 @@ SILENCE = watch.ROOT / "radio-silence.json" HISTORIES = (watch.HISTORY, watch.KEPT_HISTORY) +def code_of(root: Path) -> str | None: + try: + return hashlib.sha256(b"".join((root / name).read_bytes() for name in CODE_FILES)).hexdigest()[:12] + except OSError: + return None + + +CODE = code_of(SKILLS) + + +class Stale(Exception): + pass + + def fail(message: str) -> NoReturn: print(f"queue: {message}", file=sys.stderr) sys.exit(2) @@ -603,6 +626,9 @@ def wait(cfg: dict, entry_id: str, caught: list[int]) -> tuple[int | None, int]: """ shown = None while not caught: + host = code_of(HOST_SKILLS) + if host not in (None, CODE): + raise Stale(host, caught) cpu, mem = watch.load(cfg) free, _ = watch.disk(cfg) now = time.time() @@ -710,7 +736,7 @@ def cmd_hold(args: argparse.Namespace) -> int: entry = enqueue({ "kind": args.kind, "task": args.task, "need_gb": need, "hotfix": args.hotfix, "workdir": here, "session": session, "name": council.live_sessions().get(session) or session, - "lock": tag, "pid": os.getpid(), "pid_start": identity(os.getpid()), "child": None, + "lock": tag, "code": CODE, "pid": os.getpid(), "pid_start": identity(os.getpid()), "child": None, "child_start": None, "status": "waiting", "enqueued": time.time(), "granted": None, "cmd": shlex.join(args.command), }) @@ -761,8 +787,11 @@ def cmd_list(_: argparse.Namespace) -> int: polled = reason if silent(e, now) else e.get("reason") or reason rows.append((str(n), e, history.eta(signatures, running, ahead, now), polled or "granted on its next poll")) ahead.append(signature_of(e)) + host = code_of(HOST_SKILLS) or CODE print("position\tid\tstatus\tkind\ttask\tneed_gb\twaited_min\teta_min\tworkdir\treason") for position, e, eta, reason in rows: + if e.get("code") != host: + reason = f"queue code {e.get('code') or 'without a stamp'}, the host runs {host}: {reason}" waited = ((e["granted"] or now) - e["enqueued"]) / 60 print(tab_line( [position, e["id"], e["status"], e["kind"], e["task"], f"{e['need_gb']:g}", f"{waited:.0f}", @@ -919,7 +948,26 @@ def main() -> int: except (OSError, ValueError, KeyError, TypeError, AttributeError, re.error): return 0 commands = {"hold": cmd_hold, "list": cmd_list, "history": cmd_history, "cancel": cmd_cancel} - return commands[args.cmd](args) + try: + return commands[args.cmd](args) + except Stale as stale: + host, caught = stale.args + for signum in FORWARDED: + signal.signal(signum, signal.SIG_DFL) + if caught: + return 128 + caught[0] + script = HOST_SKILLS / CODE_FILES[0] + print( + f"queue: this hold ran queue code {CODE} from {SKILLS}, but the queue code installed on this host is " + f"{host} in {HOST_SKILLS}. Holders on different code each apply their own rules to the shared queue " + f"and together overrun the RAM and CPU budget, so this hold grants nothing on its own code. " + f"Restarting this hold with {script}; run every queue.py command from that path.", + file=sys.stderr, flush=True, + ) + try: + os.execv(EXE, [EXE, str(script), *sys.argv[1:]]) + except OSError as error: + fail(f"cannot restart with {script}: {error.strerror}; run the same hold with {script}") if __name__ == "__main__": diff --git a/tests/test_install.py b/tests/test_install.py index f47371b..b96935f 100644 --- a/tests/test_install.py +++ b/tests/test_install.py @@ -14,29 +14,47 @@ NPX = "#!/bin/sh\nmkdir -p .agents/skills/locked\necho locked > .agents/skills/l class TestInstall(unittest.TestCase): + def install(self, root: Path, target: Path) -> None: + repo, stubs = root / "repo", root / "bin" + (repo / "scripts").mkdir(parents=True) + shutil.copy(INSTALL, repo / "scripts" / "install.sh") + (repo / "scripts" / "enable-plugins.js").write_text("", encoding="utf-8") + (repo / "skills" / "own").mkdir(parents=True) + (repo / "skills" / "own" / "SKILL.md").write_text("own", encoding="utf-8") + stubs.mkdir() + (stubs / "npx").write_text(NPX, encoding="utf-8") + (stubs / "npx").chmod(0o755) + env = {**os.environ, "PATH": f"{stubs}:/usr/bin:/bin", "TARGET": str(target), "HOME": str(root / "home")} + done = subprocess.run(["bash", str(repo / "scripts" / "install.sh")], env=env, capture_output=True, text=True, check=False) + self.assertEqual(done.returncode, 0, done.stderr) + def test_reinstall_empties_the_skill_directories_in_place(self): with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) - repo, target, stubs = root / "repo", root / "target", root / "bin" - (repo / "scripts").mkdir(parents=True) - shutil.copy(INSTALL, repo / "scripts" / "install.sh") - (repo / "scripts" / "enable-plugins.js").write_text("", encoding="utf-8") - (repo / "skills" / "own").mkdir(parents=True) - (repo / "skills" / "own" / "SKILL.md").write_text("own", encoding="utf-8") - stubs.mkdir() - (stubs / "npx").write_text(NPX, encoding="utf-8") - (stubs / "npx").chmod(0o755) + target = root / "target" installed = [target / ".agents" / "skills", target / ".claude" / "skills"] for folder in installed: (folder / "stale").mkdir(parents=True) before = [folder.stat().st_ino for folder in installed] - env = {**os.environ, "PATH": f"{stubs}:/usr/bin:/bin", "TARGET": str(target)} - done = subprocess.run(["bash", str(repo / "scripts" / "install.sh")], env=env, capture_output=True, text=True, check=False) - self.assertEqual(done.returncode, 0, done.stderr) + self.install(root, target) self.assertEqual([folder.stat().st_ino for folder in installed], before) for folder in installed: self.assertEqual(sorted(path.name for path in folder.iterdir()), ["locked", "own"]) + def test_a_project_install_links_every_skill_the_global_install_holds(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + target = root / "target" + target.mkdir() + linked = (".agents/skills/locked", ".claude/skills/own") + for folder in linked: + (root / "home" / folder).mkdir(parents=True) + self.install(root, target) + for folder in (".agents/skills", ".claude/skills"): + installed = {path.name: path.readlink() if path.is_symlink() else None for path in (target / folder).iterdir()} + expected = {name: root / "home" / folder / name if f"{folder}/{name}" in linked else None for name in ("locked", "own")} + self.assertEqual(installed, expected) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_queue.py b/tests/test_queue.py index 47f04e3..76fe125 100644 --- a/tests/test_queue.py +++ b/tests/test_queue.py @@ -6,6 +6,7 @@ import fcntl import importlib.util import json import os +import shutil import signal import subprocess import sys @@ -532,6 +533,41 @@ class TestQueueCommands(unittest.TestCase): reason = self.run_queue("list").stdout.splitlines()[1].split("\t")[-1] self.assertRegex(reason, r"^RAM \d+ % used \+ \d+ % still to come reach the limit 0 %$") + def install_host_skills(self) -> Path: + skills = self.base / "home" / ".agents" / "skills" + for name in ("kvb-org-queue", "kvb-org-organize", "kvb-org-rasci"): + shutil.copytree(QUEUE.parents[2] / name, skills / name) + return skills / "kvb-org-queue" / "scripts" / "queue.py" + + def test_a_hold_on_other_code_than_the_host_install_restarts_with_it_before_any_grant(self): + host = self.install_host_skills() + ran = self.base / "ran" + args = ("hold", "--kind", "test", "--task", "work/a", "--session", "s1", "--", "touch", str(ran)) + same = self.run_queue(*args) + self.assertEqual((same.returncode, same.stderr), (0, "")) + self.assertTrue(ran.exists()) + ran.unlink() + seen = self.base / "argv.json" + host.write_text(f"import json, sys\njson.dump(sys.argv[1:], open({str(seen)!r}, 'w'))\nsys.exit(9)\n", encoding="utf-8") + other = self.run_queue(*args) + self.assertEqual(other.returncode, 9, other.stderr) + self.assertIn(f"grants nothing on its own code. Restarting this hold with {host};", other.stderr) + self.assertEqual(json.loads(seen.read_text(encoding="utf-8")), list(args)) + self.assertFalse(ran.exists()) + self.assertEqual(self.entries(), []) + + def test_list_marks_an_entry_on_other_code_than_the_host_install(self): + queue = _load(QUEUE) + me, mine = os.getpid(), queue.identity(os.getpid()) + self.install_host_skills() + self.write_store([ + {**stored(1, "waiting", me, mine, None, None), "code": queue.CODE}, + stored(2, "waiting", me, mine, None, None), + ]) + reasons = [line.split("\t")[-1] for line in self.run_queue("list").stdout.splitlines()[1:]] + self.assertEqual(reasons[0], "granted on its next poll") + self.assertTrue(reasons[1].startswith(f"queue code without a stamp, the host runs {queue.CODE}: "), reasons[1]) + def test_hold_refuses_bad_input_with_exit_2_and_enqueues_nothing(self): def hold(*args: str) -> list[str]: return ["hold", "--session", "s1", *args]