fix(kvb-org-queue): restart holders on stale queue code; link host skills into projects

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 <noreply@anthropic.com>
This commit is contained in:
Kevin Veen-BirkenbachandClaude Opus 5.5 committed 2026-11-11 20:33:54 +00:00
1 parent e805b07c27
commit ca54420f4d
4 files changed
+135 -21

No files matched your search

+14 -2
View File
@@ -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 <TARGET>/.agents/skills and are
# mirrored to <TARGET>/.claude/skills.
# mirrored to <TARGET>/.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"
+55 -7
View File
@@ -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__":
+30 -12
View File
@@ -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()
+36
View File
@@ -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]