fix(kvb-org-queue): hold liveness by flock, not pids across PID namespaces

Every sandboxed tool call runs in its own PID namespace, so alive() saw every other hold as dead, pruned it and granted parallel heavy jobs. Each hold now keeps an exclusive flock on <root>/holds/<tag>.lock, shared with its command via pass_fds, from before its entry exists until it leaves; list and evaluate drop an entry only when nobody holds its lock. Entries without a lock fall back to the pid check. held() accepts only <pid>-<ns> tags, so an entry cannot unlink a path outside holds/.

MEDUSA: queue.py scores 25 % against 75 % for HEAD's queue.py alone, because MEDUSA-ACP-SCAN-028 flags two cfg = watch.load_config() lines that are byte-identical at HEAD; the Popen high (SAST-PATH-001) also pre-exists. None is accepted: kvb-think-criticize found the config agent-writable. The operator approved the commit after this comparison.

🎯 Fix: 95 %
💥 Break: 3 %
💸 Cost: 99 %
⏱️ Time: 100 %
🛡️ Security: 25 %
🏆 Score: 23
🚦 Gate: ⚠️ security 25 % < 100: ask the operator before the commit

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

skills/kvb-org-queue/scripts/queue.py

@@ -12,6 +12,10 @@ Environment:
     AGENTS_COUNCIL_ROOT: shared state directory; /tmp/agents-council unless set,
         because every agent on the host must agree on it. The queue is
         <root>/queue.json, every read-modify-write holds <root>/queue.lock.
+        Each hold keeps an exclusive flock on <root>/holds/<tag>.lock, shared
+        with its command, from before its entry exists until it leaves; an
+        entry whose lock nobody holds is dead. Pids cannot tell, because every
+        sandboxed tool call has its own PID namespace.
         Every finished job appends a row (history.py) to
         <root>/queue-history.jsonl; the watch hook of watch.py keeps a copy in
         ~/.local/state/agents-council/queue-history.jsonl, and reading merges
@@ -119,6 +123,8 @@ watch = sibling("watch")
 history = sibling("history")
 STORE = watch.ROOT / "queue.json"
 LOCK = watch.ROOT / "queue.lock"
+HOLDS = watch.ROOT / "holds"
+TAG = re.compile(r"\d+-\d+")
 SILENCE = watch.ROOT / "radio-silence.json"
 HISTORIES = (watch.HISTORY, watch.KEPT_HISTORY)
 
@@ -178,11 +184,46 @@ def alive(pid: int | None, start: str | None) -> bool:
     return pid is not None and start is not None and identity(pid) == start
 
 
+def held(tag: object) -> bool:
+    if not isinstance(tag, str) or not TAG.fullmatch(tag):
+        return False
+    path = HOLDS / f"{tag}.lock"
+    try:
+        handle = path.open(encoding="utf-8")
+    except FileNotFoundError:
+        return False
+    with handle:
+        try:
+            fcntl.flock(handle, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        except BlockingIOError:
+            return True
+    path.unlink(missing_ok=True)
+    return False
+
+
+@contextlib.contextmanager
+def holding() -> Iterator[tuple[str, int]]:
+    HOLDS.mkdir(parents=True, exist_ok=True)
+    tag = f"{os.getpid()}-{time.time_ns()}"
+    path = HOLDS / f"{tag}.lock"
+    with path.open("w", encoding="utf-8") as handle:
+        fcntl.flock(handle, fcntl.LOCK_EX)
+        try:
+            yield tag, handle.fileno()
+        finally:
+            path.unlink(missing_ok=True)
+
+
+def living(entry: dict) -> bool:
+    if "lock" in entry:
+        return held(entry["lock"])
+    return alive(entry["pid"], entry["pid_start"]) or (
+        entry["status"] == "active" and alive(entry["child"], entry["child_start"])
+    )
+
+
 def prune(entries: list[dict]) -> list[dict]:
-    return [
-        e for e in entries
-        if alive(e["pid"], e["pid_start"]) or (e["status"] == "active" and alive(e["child"], e["child_start"]))
-    ]
+    return [e for e in entries if living(e)]
 
 
 def read_store() -> dict:
@@ -447,36 +488,38 @@ def cmd_hold(args: argparse.Namespace) -> int:
 
     for signum in FORWARDED:
         signal.signal(signum, forward)
-    entry = enqueue({
-        "kind": args.kind, "task": args.task, "need_gb": need, "hotfix": args.hotfix,
-        "workdir": workdir(), "session": session, "name": council.live_sessions().get(session) or session,
-        "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),
-    })
-    try:
-        code, concurrent = wait(cfg, entry["id"], args.kind, caught)
-        if code is not None:
-            return code
-        if caught:
-            return 128 + caught[0]
-        samples = [measure(cfg)]
-        start = time.time()
+    with holding() as (tag, lock_fd):
+        entry = enqueue({
+            "kind": args.kind, "task": args.task, "need_gb": need, "hotfix": args.hotfix,
+            "workdir": workdir(), "session": session, "name": council.live_sessions().get(session) or session,
+            "lock": tag, "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),
+        })
         try:
-            child = subprocess.Popen(args.command, start_new_session=True)
-        except OSError as error:
-            print(f"queue: cannot start {args.command[0]}: {error.strerror}", file=sys.stderr)
-            return 2
-        for signum in caught:
-            send(child, signum)
-        update(entry["id"], child=child.pid, child_start=identity(child.pid))
-        code = supervise(cfg, child, samples)
-        code = code if code >= 0 else 128 - code
-        job = {"id": f"{entry['id']}-{time.time_ns()}", "kind": args.kind, "signature": signature,
-               "workdir": entry["workdir"]}
-        keep(history.row(job, start, time.time(), code, samples, concurrent))
-        return code
-    finally:
-        leave(entry["id"])
+            code, concurrent = wait(cfg, entry["id"], args.kind, caught)
+            if code is not None:
+                return code
+            if caught:
+                return 128 + caught[0]
+            samples = [measure(cfg)]
+            start = time.time()
+            try:
+                child = subprocess.Popen(args.command, start_new_session=True, pass_fds=(lock_fd,))
+            except OSError as error:
+                print(f"queue: cannot start {args.command[0]}: {error.strerror}", file=sys.stderr)
+                return 2
+            for signum in caught:
+                send(child, signum)
+            update(entry["id"], child=child.pid, child_start=identity(child.pid))
+            code = supervise(cfg, child, samples)
+            code = code if code >= 0 else 128 - code
+            job = {"id": f"{entry['id']}-{time.time_ns()}", "kind": args.kind, "signature": signature,
+                   "workdir": entry["workdir"]}
+            keep(history.row(job, start, time.time(), code, samples, concurrent))
+            return code
+        finally:
+            leave(entry["id"])
 
 
 def cmd_list(_: argparse.Namespace) -> int:

tests/test_queue.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import fcntl
 import importlib.util
 import json
 import os
@@ -400,6 +401,54 @@ class TestQueueCommands(unittest.TestCase):
             [["run", "test-4", "active"], ["1", "test-5", "waiting"]],
         )
 
+    def test_list_keeps_a_lock_holder_whose_pid_another_namespace_hides(self):
+        gone = subprocess.Popen(["true"])
+        gone.wait()
+        holds = self.root / "holds"
+        holds.mkdir(parents=True)
+        script = (
+            "import fcntl, sys, time; h = open(sys.argv[1], 'w'); fcntl.flock(h, fcntl.LOCK_EX); "
+            "print('locked', flush=True); time.sleep(60)"
+        )
+        holder = subprocess.Popen(
+            [sys.executable, "-c", script, str(holds / "5-6.lock")], stdout=subprocess.PIPE, text=True
+        )
+        self.addCleanup(holder.wait)
+        self.addCleanup(holder.kill)
+        self.addCleanup(holder.stdout.close)
+        self.assertEqual(holder.stdout.readline().strip(), "locked")
+        (holds / "1-2.lock").write_text("", encoding="utf-8")
+        outside = self.root / "outside.lock"
+        outside.write_text("", encoding="utf-8")
+        self.write_store([
+            {**stored(1, "waiting", gone.pid, "1", None, None), "lock": "5-6"},
+            {**stored(2, "waiting", gone.pid, "1", None, None), "lock": "1-2"},
+            {**stored(3, "active", gone.pid, "1", gone.pid, "1"), "lock": "3-4"},
+            {**stored(4, "waiting", gone.pid, "1", None, None), "lock": "../outside"},
+        ])
+        listing = self.run_queue("list")
+        self.assertEqual(listing.returncode, 0, listing.stderr)
+        self.assertEqual([e["id"] for e in self.entries()], ["test-1"])
+        self.assertFalse((holds / "1-2.lock").exists())
+        self.assertTrue(outside.exists())
+        holder.kill()
+        holder.wait()
+        self.assertEqual(self.run_queue("list").returncode, 0)
+        self.assertEqual(self.entries(), [])
+
+    def test_hold_locks_while_waiting_and_cleans_up(self):
+        self.cfg["slot_limits"]["deploy"] = 0
+        self.write_config()
+        hold, _ = self.blocked_hold("true")
+        entry = self.entries()[0]
+        lock = self.root / "holds" / f"{entry['lock']}.lock"
+        with lock.open(encoding="utf-8") as handle, self.assertRaises(BlockingIOError):
+            fcntl.flock(handle, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        hold.send_signal(signal.SIGTERM)
+        self.assertEqual(hold.wait(timeout=30), 128 + signal.SIGTERM)
+        self.assertFalse(lock.exists())
+        self.assertEqual(self.entries(), [])
+
     def test_hold_takes_the_need_learned_for_its_command_from_both_history_files(self):
         self.cfg["slot_limits"]["deploy"] = 0
         self.write_config()