feat(democracy): admit test slots only below the configured CPU and RAM limits

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

skills/democracy/SKILL.md

@@ -76,7 +76,8 @@ records the result.
   sends the chair what, where, for how long and its expected disk need, waits
   for the grant, and reports the release. The chair keeps one queue, answers
   every request with its position, enforces the limits the operator set,
-  checks every grant with `watch.py admit --need-gb <N>` (or `--kind <slot>`),
+  checks every grant with `watch.py admit --need-gb <N>` (or `--kind <slot>`;
+  test kinds also need CPU and RAM below the configured limits),
   and never cancels a running grant to free capacity.
 - Handover is explicit: the outgoing chair sends every peer the new holder,
   the queue and the running grants, and keeps allocating until the incoming

skills/democracy/scripts/watch.py

@@ -26,7 +26,9 @@ never_block (regexes), cleanup (shell commands), worktree_cleanup (shell
 commands with {worktree}), worktrees_command (shell command printing the
 running stacks' worktree paths, one per line), worktree_marker (path relative
 to a worktree that must exist), cache_dirs, cache_budget_percent,
-slot_need_gb {kind: GB}, notify (argv the alert text is appended to).
+slot_need_gb {kind: GB}, notify (argv the alert text is appended to),
+load_limits {cpu_percent, mem_percent, sample_seconds, kinds}: admit for a
+kind listed in kinds also needs CPU busy and RAM used below the limits.
 """
 
 from __future__ import annotations
@@ -223,13 +225,49 @@ def cmd_block(cfg: dict) -> int:
     return 0
 
 
+def cpu_times() -> tuple[int, int]:
+    fields = [int(v) for v in Path("/proc/stat").read_text(encoding="utf-8").split("\n", 1)[0].split()[1:]]
+    return sum(fields), fields[3] + fields[4]
+
+
+def cpu_busy_percent(before: tuple[int, int], after: tuple[int, int]) -> float:
+    total = after[0] - before[0]
+    idle = after[1] - before[1]
+    return 100.0 * (total - idle) / total
+
+
+def mem_used_percent(meminfo: str) -> float:
+    values = {}
+    for line in meminfo.splitlines():
+        key, _, rest = line.partition(":")
+        if rest:
+            values[key] = int(rest.split()[0])
+    return 100.0 * (1 - values["MemAvailable"] / values["MemTotal"])
+
+
+def load(cfg: dict) -> tuple[float, float]:
+    before = cpu_times()
+    time.sleep(cfg["load_limits"]["sample_seconds"])
+    cpu = cpu_busy_percent(before, cpu_times())
+    return cpu, mem_used_percent(Path("/proc/meminfo").read_text(encoding="utf-8"))
+
+
 def cmd_admit(cfg: dict, args: argparse.Namespace) -> int:
     need = args.need_gb if args.need_gb is not None else cfg["slot_need_gb"][args.kind]
     free, _ = disk(cfg)
     left = free - need
     floor = cfg["thresholds_gb"]["stop"]
-    verdict = "grant" if left >= floor else "deny"
-    print(f"{verdict}: free {free:.1f} GB, need {need} GB, leaves {left:.1f} GB (floor {floor} GB)")
+    passed = [left >= floor]
+    report = [f"free {free:.1f} GB, need {need} GB, leaves {left:.1f} GB (floor {floor} GB)"]
+    limits = cfg["load_limits"]
+    if args.kind in limits["kinds"]:
+        cpu, mem = load(cfg)
+        passed += [cpu < limits["cpu_percent"], mem < limits["mem_percent"]]
+        report.append(
+            f"cpu {cpu:.0f} % (limit {limits['cpu_percent']}), ram {mem:.0f} % (limit {limits['mem_percent']})"
+        )
+    verdict = "grant" if all(passed) else "deny"
+    print(f"{verdict}: {'; '.join(report)}")
     return 0 if verdict == "grant" else 3
 
 

tests/test_watch.py

@@ -34,6 +34,11 @@ class TestStage(unittest.TestCase):
         for free, stage in ((50, "ok"), (30, "warn"), (14, "stop"), (5, "critical")):
             self.assertEqual(self.watch.stage_for(free, [], self.cfg, now), stage)
 
+    def test_cpu_and_ram_percentages(self):
+        self.assertEqual(self.watch.cpu_busy_percent((100, 40), (200, 60)), 80.0)
+        meminfo = "MemTotal:       1000 kB\nMemFree:  100 kB\nMemAvailable:    250 kB\n"
+        self.assertEqual(self.watch.mem_used_percent(meminfo), 75.0)
+
     def test_fast_fall_escalates_one_stage(self):
         now = time.time()
         self.assertEqual(self.watch.stage_for(32, [[now - 600, 48]], self.cfg, now), "stop")
@@ -63,8 +68,9 @@ class TestWatchCommands(unittest.TestCase):
             "worktree_marker": "Makefile",
             "cache_dirs": [str(self.cache)],
             "cache_budget_percent": 3,
-            "slot_need_gb": {"swarm": 30},
+            "slot_need_gb": {"swarm": 30, "test": 0.001},
             "notify": ["true"],
+            "load_limits": {"cpu_percent": 101, "mem_percent": 101, "sample_seconds": 0.1, "kinds": ["test"]},
         }
         self._write_config()
 
@@ -104,6 +110,17 @@ class TestWatchCommands(unittest.TestCase):
         self.assertEqual(self._run("admit", "--need-gb", "0.001").returncode, 0)
         self.assertEqual(self._run("admit", "--need-gb", "1e9").returncode, 3)
 
+    def test_admit_for_a_load_kind_also_needs_cpu_and_ram_below_the_limits(self):
+        self.cfg["thresholds_gb"] = {"warn": 0, "stop": 0, "critical": 0}
+        self._write_config()
+        self.assertEqual(self._run("admit", "--kind", "test").returncode, 0)
+        self.cfg["load_limits"]["cpu_percent"] = 0
+        self._write_config()
+        denied = self._run("admit", "--kind", "test")
+        self.assertEqual(denied.returncode, 3)
+        self.assertIn("cpu", denied.stdout)
+        self.assertEqual(self._run("admit", "--need-gb", "0.001").returncode, 0)
+
     def test_warn_alert_reaches_each_session_once(self):
         self.cfg["thresholds_gb"] = {"warn": 1e9, "stop": -1, "critical": -2}
         self._write_config()