diff --git a/skills/kvb-org-queue/scripts/queue.py b/skills/kvb-org-queue/scripts/queue.py index 62b9f6e..cb95200 100755 --- a/skills/kvb-org-queue/scripts/queue.py +++ b/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 /queue.json, every read-modify-write holds /queue.lock. + Each hold keeps an exclusive flock on /holds/.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 /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: diff --git a/tests/test_queue.py b/tests/test_queue.py index 89bc901..5322cc3 100644 --- a/tests/test_queue.py +++ b/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()