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)