feat(democracy): find the running stacks' worktrees live for the resource cleanup

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

skills/democracy/scripts/watch.py

@@ -23,14 +23,15 @@ Environment:
 Config keys, all required: disk, thresholds_gb {warn, stop, critical},
 projection_minutes, interval_seconds, cache_interval_seconds, heavy_commands,
 never_block (regexes), cleanup (shell commands), worktree_cleanup (shell
-commands with {worktree}), worktrees (globs), cache_dirs, cache_budget_percent,
+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).
 """
 
 from __future__ import annotations
 
 import argparse
-import glob
 import json
 import os
 import re
@@ -263,11 +264,20 @@ def trim_caches(cfg: dict) -> float:
     return freed / GB
 
 
+def running_worktrees(cfg: dict) -> list[str]:
+    try:
+        out = subprocess.run(
+            cfg["worktrees_command"], shell=True, capture_output=True, text=True, timeout=60, check=False
+        ).stdout
+    except subprocess.TimeoutExpired:
+        return []
+    trees = sorted({line.strip() for line in out.splitlines() if line.strip()})
+    return [t for t in trees if os.path.exists(os.path.join(t, cfg["worktree_marker"]))]
+
+
 def run_cleanup(cfg: dict) -> list[str]:
     commands = list(cfg["cleanup"])
-    for pattern in cfg["worktrees"]:
-        for tree in sorted(glob.glob(os.path.expanduser(pattern))):
-            if os.path.isdir(tree):
+    for tree in running_worktrees(cfg):
         commands += [t.format(worktree=shlex.quote(tree)) for t in cfg["worktree_cleanup"]]
     results = []
     for command in commands:

tests/test_watch.py

@@ -59,7 +59,8 @@ class TestWatchCommands(unittest.TestCase):
             "never_block": ["^make swarm-down"],
             "cleanup": [],
             "worktree_cleanup": [],
-            "worktrees": [],
+            "worktrees_command": "true",
+            "worktree_marker": "Makefile",
             "cache_dirs": [str(self.cache)],
             "cache_budget_percent": 3,
             "slot_need_gb": {"swarm": 30},
@@ -128,6 +129,18 @@ class TestWatchCommands(unittest.TestCase):
         self.assertFalse(old.exists())
         self.assertTrue(new.exists())
 
+    def test_full_cleanup_reaches_only_marked_running_worktrees(self):
+        marked, foreign = self.base / "marked", self.base / "foreign"
+        for tree in (marked, foreign):
+            tree.mkdir()
+        (marked / "Makefile").write_text("", encoding="utf-8")
+        self.cfg["worktrees_command"] = f"echo {marked}; echo {foreign}; echo {marked}"
+        self.cfg["worktree_cleanup"] = ["echo x >> {worktree}/cleaned"]
+        self._write_config()
+        self.assertEqual(self._run("cleanup", "--full").returncode, 0)
+        self.assertEqual((marked / "cleaned").read_text(encoding="utf-8"), "x\n")
+        self.assertFalse((foreign / "cleaned").exists())
+
     def test_without_config_every_command_is_a_no_op(self):
         self.config.unlink()
         result = self._run("status")