From 94542e7b924edcae3e96645cd451e1a0229f19e4 Mon Sep 17 00:00:00 2001 From: Kevin Veen-Birkenbach Date: Fri, 6 Nov 2026 12:41:58 +0000 Subject: [PATCH] fix(kvb-org-organize): the resource watch never acts on a stale stage MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit block and status compute the stage from the live free space once the stored measurement is older than interval_seconds, so a disk that filled while no hook ran no longer passes heavy jobs as ok. watch writes the stage before it mirrors the history, and the mirror takes its lock with LOCK_NB: an agent holding a flock on the kept file no longer stalls every hook, the round is skipped and the next one catches up. 🎯 Fix: 92 % 💥 Break: 4 % 💸 Cost: 100 % ⏱️ Time: 100 % 🛡️ Security: 100 % 💰 Value: ? 🏆 Score: 88 🚦 Gate: ✅ 90/5, cost and time <= 100, security 100 % Co-Authored-By: Claude Opus 5.5 --- skills/kvb-org-organize/capacity.md | 3 +- skills/kvb-org-organize/scripts/watch.py | 24 ++++++-- tests/test_watch.py | 74 ++++++++++++++++++++++-- 3 files changed, 89 insertions(+), 12 deletions(-) diff --git a/skills/kvb-org-organize/capacity.md b/skills/kvb-org-organize/capacity.md index 7f96cac..67e673b 100644 --- a/skills/kvb-org-organize/capacity.md +++ b/skills/kvb-org-organize/capacity.md @@ -58,4 +58,5 @@ and stays silent below the thresholds. projection window raises the stage by one at once. - Caches together stay under their budget; the least recently used go first. -`watch.py status` shows the last measurement. +`watch.py status` shows the stage computed from the live free space and the +stored samples, and the age of the stored measurement. diff --git a/skills/kvb-org-organize/scripts/watch.py b/skills/kvb-org-organize/scripts/watch.py index 44f0b82..842384a 100755 --- a/skills/kvb-org-organize/scripts/watch.py +++ b/skills/kvb-org-organize/scripts/watch.py @@ -17,7 +17,8 @@ Usage: watch.py cleanup (--caches | --full) trim caches to their budget; --full also runs the configured low-risk cleanup commands - watch.py status print the last measurement + watch.py status print the stage computed from the live free space and the + stored samples, and the stored measurement's age in seconds Environment: AGENTS_COUNCIL_ROOT: shared state directory; /tmp/agents-council unless set, @@ -250,7 +251,7 @@ def keep_rows(origin: Path, target: Path) -> None: return target.parent.mkdir(parents=True, exist_ok=True) with target.open("a+", encoding="utf-8") as kept: - fcntl.flock(kept, fcntl.LOCK_EX) + fcntl.flock(kept, fcntl.LOCK_EX | fcntl.LOCK_NB) kept.seek(0) known = {row_id for row_id, _ in history_rows(kept.read())} for row_id, line in history_rows(nofollow.read_text(ROOT, origin)): @@ -268,12 +269,15 @@ def keep_history() -> None: keep_rows(origin, target) +def stale(state: dict, cfg: dict, now: float) -> bool: + return now - state.get("checked", 0) >= cfg["interval_seconds"] + + def cmd_watch(cfg: dict) -> int: payload = json.load(sys.stdin) - keep_history() now = time.time() state = read_json(STATE) - if now - state.get("checked", 0) >= cfg["interval_seconds"]: + if stale(state, cfg, now): free, _ = disk(cfg) samples = [s for s in state.get("samples", []) if now - s[0] <= cfg["projection_minutes"] * 60] stage = stage_for(free, samples, cfg, now) @@ -288,6 +292,7 @@ def cmd_watch(cfg: dict) -> int: act(cfg, stage, previous, free) elif due: spawn_cleanup("--caches") + keep_history() texts = unseen_alerts(payload["session_id"]) if texts: print(json.dumps({"hookSpecificOutput": { @@ -313,7 +318,11 @@ def silence() -> dict | None: def cmd_block(cfg: dict) -> int: payload = json.load(sys.stdin) command = executable_text((payload.get("tool_input") or {}).get("command", "")) - stage = read_json(STATE).get("stage", "ok") + state = read_json(STATE) + now = time.time() + stage = state.get("stage", "ok") + if stale(state, cfg, now): + stage = stage_for(disk(cfg)[0], state.get("samples", []), cfg, now) mayday = silence() if stage not in ("stop", "critical") and not mayday: return 0 @@ -492,7 +501,10 @@ def cmd_cleanup(cfg: dict, args: argparse.Namespace) -> int: def cmd_status(cfg: dict) -> int: state = read_json(STATE) free, total = disk(cfg) - print(f"stage={state.get('stage', 'unmeasured')} free={free:.1f}GB total={total:.1f}GB samples={len(state.get('samples', []))}") + now = time.time() + samples = state.get("samples", []) + age = f"{now - state['checked']:.0f}s" if "checked" in state else "unmeasured" + print(f"stage={stage_for(free, samples, cfg, now)} free={free:.1f}GB total={total:.1f}GB samples={len(samples)} age={age}") return 0 diff --git a/tests/test_watch.py b/tests/test_watch.py index 863e313..a8b437e 100644 --- a/tests/test_watch.py +++ b/tests/test_watch.py @@ -2,9 +2,11 @@ from __future__ import annotations +import fcntl import importlib.util import json import os +import re import subprocess import sys import tempfile @@ -58,7 +60,7 @@ class TestWatchCommands(unittest.TestCase): "disk": str(self.base), "thresholds_gb": LIMITS, "projection_minutes": 30, - "interval_seconds": 0, + "interval_seconds": 300, "cache_interval_seconds": 86400, "heavy_commands": ["^make compose-deploy", "^make swarm-"], "never_block": ["^make swarm-down"], @@ -78,21 +80,23 @@ class TestWatchCommands(unittest.TestCase): def _write_config(self): self.config.write_text(json.dumps(self.cfg), encoding="utf-8") - def _run(self, *args: str, stdin: str = "") -> subprocess.CompletedProcess: - env = { + def _env(self) -> dict[str, str]: + return { "PATH": str(self.base), "HOME": str(self.base), "AGENTS_COUNCIL_ROOT": str(self.root), "AGENTS_COUNCIL_CONFIG": str(self.config), } + + def _run(self, *args: str, stdin: str = "") -> subprocess.CompletedProcess: return subprocess.run( [sys.executable, str(WATCH), *args], - capture_output=True, text=True, env=env, input=stdin, check=False, + capture_output=True, text=True, env=self._env(), input=stdin, check=False, ) def _stage(self, stage: str) -> None: self.root.mkdir(exist_ok=True) - (self.root / "watch.json").write_text(json.dumps({"stage": stage}), encoding="utf-8") + (self.root / "watch.json").write_text(json.dumps({"stage": stage, "checked": time.time()}), encoding="utf-8") def _bash(self, command: str) -> str: return json.dumps({"tool_input": {"command": command}}) @@ -186,6 +190,66 @@ class TestWatchCommands(unittest.TestCase): handle.write('{"id": "b"}\n') self.assertEqual(kept_ids(), ["old", "a", "b"]) + def _stale_ok_below_stop(self) -> Path: + self.cfg["thresholds_gb"] = {"warn": 1e9, "stop": 1e9, "critical": -1} + self._write_config() + self.root.mkdir() + state = self.root / "watch.json" + state.write_text(json.dumps({"stage": "ok", "checked": time.time() - 9000, "samples": []}), encoding="utf-8") + return state + + def test_status_prints_the_live_stage_and_the_age_of_the_stored_one(self): + self._stale_ok_below_stop() + out = self._run("status").stdout + self.assertIn("stage=stop", out) + age = re.search(r"age=(\d+)s", out) + self.assertIsNotNone(age, out) + self.assertTrue(9000 <= int(age.group(1)) < 9060, out) + + def test_block_measures_a_stale_stage_live_without_writing_it(self): + state = self._stale_ok_below_stop() + stale = state.read_text(encoding="utf-8") + self.assertEqual(self._run("block", stdin=self._bash("make compose-deploy")).returncode, 2) + self.assertEqual(state.read_text(encoding="utf-8"), stale) + self._stage("ok") + self.assertEqual(self._run("block", stdin=self._bash("make compose-deploy")).returncode, 0) + + def _quiet_disk(self) -> Path: + self.cfg["thresholds_gb"] = {"warn": 0, "stop": 0, "critical": 0} + self._write_config() + self.root.mkdir() + return self.root / "queue-history.jsonl" + + def _hook_within(self, seconds: float) -> int | None: + payload = json.dumps({"session_id": "s1", "hook_event_name": "PostToolUse"}) + try: + return subprocess.run( + [sys.executable, str(WATCH), "watch"], + capture_output=True, text=True, env=self._env(), input=payload, timeout=seconds, check=False, + ).returncode + except subprocess.TimeoutExpired: + return None + + def test_watch_writes_the_stage_before_a_hanging_mirror_can_stop_it(self): + os.mkfifo(self._quiet_disk()) + self._hook_within(5) + state = self.root / "watch.json" + self.assertTrue(state.exists(), "the hook hung in the mirror before it wrote the stage") + self.assertEqual(json.loads(state.read_text(encoding="utf-8"))["stage"], "ok") + + def test_watch_skips_the_mirror_round_while_its_lock_is_held(self): + self._quiet_disk().write_text('{"id": "a"}\n', encoding="utf-8") + kept = self.base / ".local" / "state" / "agents-council" / "queue-history.jsonl" + kept.parent.mkdir(parents=True) + kept.write_text("", encoding="utf-8") + os.utime(kept, (1, 1)) + with kept.open("a+", encoding="utf-8") as holder: + fcntl.flock(holder, fcntl.LOCK_EX) + self.assertEqual(self._hook_within(5), 0, "the hook waited on the mirror lock") + self.assertEqual(kept.read_text(encoding="utf-8"), "") + self.assertEqual(self._hook_within(5), 0) + self.assertEqual(kept.read_text(encoding="utf-8"), '{"id": "a"}\n') + def test_cache_trim_evicts_least_recently_used_first(self): old, new = self.cache / "old", self.cache / "new" old.write_bytes(b"x" * 4096)