fix(kvb-org-organize): the resource watch never acts on a stale stage

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 <noreply@anthropic.com>

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.

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
 
 

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)