feat(democracy): claim offices through a council script and gate unclaimed edits

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

README.md

@@ -26,7 +26,7 @@ Copy the skills into a specific project (`<repo>/.agents/skills` and `<repo>/.cl
 make project TARGET=/path/to/repo
 ```
 
-Both commands also patch the target's `.claude/settings.local.json` — the machine-local, untracked file, so a project's reviewed `.claude/settings.json` is never rewritten by an install: the caveman and ponytail plugins are enabled so their modes auto-activate on session start, and two reminder hooks are registered — `autotune` points you at `/autotune` at most once an hour, `dream` points you at `/dream` at most once every twelve hours per repository (stamped in that repository's git dir). Existing settings are preserved.
+Both commands also patch the target's `.claude/settings.local.json` — the machine-local, untracked file, so a project's reviewed `.claude/settings.json` is never rewritten by an install: the caveman and ponytail plugins are enabled so their modes auto-activate on session start, and two reminder hooks are registered — `autotune` points you at `/autotune` at most once an hour, `dream` points you at `/dream` at most once every twelve hours per repository (stamped in that repository's git dir). A PreToolUse gate from the `democracy` skill blocks edits to files in a git worktree until the session holds a claim (`skills/democracy/scripts/council.py`). Existing settings are preserved.
 
 Restart your agent afterwards so it loads the new skills.
 

scripts/enable-plugins.js

@@ -1,7 +1,7 @@
 #!/usr/bin/env node
 // Enable the caveman and ponytail plugins in a Claude Code settings.json so
-// their SessionStart hooks auto-activate both modes, and register the autotune
-// and dream reminder hooks. Merges non-destructively: existing marketplaces,
+// their SessionStart hooks auto-activate both modes, register the autotune
+// and dream reminder hooks and the democracy claim gate. Merges non-destructively: existing marketplaces,
 // plugins and hooks are preserved, unparseable files are left untouched.
 "use strict";
 
@@ -51,5 +51,14 @@ for (const skill of REMINDERS) {
   });
 }
 
+data.hooks.PreToolUse = data.hooks.PreToolUse || [];
+if (!JSON.stringify(data.hooks.PreToolUse).includes("democracy/scripts/council.py")) {
+  const council = path.join(path.dirname(settingsPath), "skills", "democracy", "scripts", "council.py");
+  data.hooks.PreToolUse.push({
+    matcher: "Edit|Write|MultiEdit|NotebookEdit",
+    hooks: [{ type: "command", command: `python3 ${JSON.stringify(council)} gate` }],
+  });
+}
+
 fs.writeFileSync(settingsPath, JSON.stringify(data, null, 2) + "\n");
-console.log(`skills: enabled caveman + ponytail plugins and the ${REMINDERS.join(" + ")} reminders in ${settingsPath}`);
+console.log(`skills: enabled caveman + ponytail plugins, the ${REMINDERS.join(" + ")} reminders and the democracy claim gate in ${settingsPath}`);

skills/democracy/SKILL.md

@@ -46,23 +46,25 @@ appointment outranks the election.
 ## Election
 
 The first to claim wins, and the claim is atomic, so two agents can never both
-be first. Use the literal path `/tmp/agents-council`, never `$TMPDIR`, which a
-sandbox may make session-private.
+be first. `${CLAUDE_SKILL_DIR}/scripts/council.py` keeps the claims in
+`/tmp/agents-council`; other runtimes call the same script from their skills
+directory with `--session <id>`.
 
 1. **Name the office**: `work/<slug>`, `cluster/<slug>`,
    `<dimension>/<line>` for a representative, or `chair`. The slug names the
    work concretely: the issue number, the failing line, the cost's log line.
-2. **Look first**: list `/tmp/agents-council` and read the claims. A claim for
-   the same work under another slug counts as first.
-3. **Claim**: `mkdir -p` the parent directory, then `mkdir` the office itself
-   without `-p`, as two separate commands. Success means elected; `File
-   exists` means someone was first.
-4. **Elected**: write `claim` into the office directory (session name, ISO
-   time, one line of what the work is, its evidence, branch or worktree), then
-   announce office, holder and work to every peer on the system (in Claude
-   Code: `ListAgents`, then `SendMessage`).
-5. **Not elected**: read the holder's `claim`, send the holder your findings,
-   and stop on that work.
+2. **Look first**: `council.py list`. A claim for the same work under another
+   slug counts as first.
+3. **Claim**: `council.py claim <office> --work "<what, evidence, branch or
+   worktree>"`. Exit 0: elected, or already yours. Exit 3: a live agent holds
+   it, and its claim is printed. Exit 4: the holder's session is dead (see
+   Succession).
+4. **Elected**: announce office, holder and work to every peer on the system
+   (in Claude Code: `ListAgents`, then `SendMessage`).
+5. **Not elected**: send the holder your findings and stop on that work.
+
+A Claude Code session cannot edit a file in a git worktree before it holds a
+claim: a PreToolUse hook (`council.py gate`) blocks the edit.
 
 Peers on other machines cannot see this host's claims. Between machines the
 earliest announcement by message time wins, and each host's claim directory
@@ -89,19 +91,20 @@ records the result.
 
 - A piece of work or a cluster ends with its result, a proposal, or an
   abandoned attempt, reported to the peers who heard the claim. The holder
-  then removes its office directory, and the next agent that starts the work
-  is elected afresh.
+  then runs `council.py release <office>`, and the next agent that starts the
+  work is elected afresh.
 - A representative or the chair that ends hands its office to the next-oldest
   live representative (by claim time) and announces it.
-- A claim is stale when its holder no longer appears in the agent listing and
-  does not answer a message. A stale work or cluster claim goes to the next
+- A claim is stale when its holder's session is gone: in Claude Code its
+  session id is no longer in `~/.claude/sessions`; other runtimes run
+  `council.py beat` at least hourly and go stale after 60 minutes without
+  one. `claim` then exits 4. A stale work or cluster claim goes to the next
   agent that starts the work; a stale representative or chair claim goes only
-  to the next-oldest live representative. The successor removes the stale
-  claim, claims anew and announces the succession; everyone else leaves it
-  alone.
-- After a restart or reboot, check every office you believe you hold: `/tmp`
-  may have been wiped, and your session name may have changed. Re-claim and
-  update the claim file instead of assuming the office is still yours.
+  to the next-oldest live representative. The successor runs
+  `council.py takeover <office> --work "..."` and announces the succession;
+  everyone else leaves it alone.
+- After a restart or reboot, run `council.py list --mine` and claim again
+  every office you believe you hold: `/tmp` may have been wiped.
 
 ## Rules
 
@@ -115,7 +118,7 @@ records the result.
 
 ## Output
 
-On /democracy, one table of the current offices, read from the claim
-directory and the agent listing, vacant and stale claims marked:
+On /democracy, one table of the current offices from `council.py list`, stale
+claims marked:
 
 | Office | Holder | Since | Work | Grants or queue |

skills/democracy/scripts/council.py

@@ -0,0 +1,224 @@
+#!/usr/bin/env python3
+"""Claim, release and list democracy offices in the shared council directory.
+
+Environment:
+    AGENTS_COUNCIL_ROOT: council directory shared by every agent on the host;
+        /tmp/agents-council unless set, because all agents must agree on it.
+    CLAUDE_CODE_SESSION_ID: the caller's session; other runtimes pass --session.
+
+Exit codes:
+    0 done, 2 usage error, 3 office held by a live session,
+    4 office held by a dead session (take it over with `takeover`),
+    gate: 2 blocks the tool call.
+"""
+
+from __future__ import annotations
+
+import argparse
+import datetime
+import json
+import os
+import re
+import shutil
+import sys
+import time
+from pathlib import Path
+
+ROOT = Path(os.environ.get("AGENTS_COUNCIL_ROOT", "/tmp/agents-council"))
+HEARTBEAT_LIMIT = 3600
+OFFICE = re.compile(r"^[a-z0-9][a-z0-9._-]*(/[a-z0-9][a-z0-9._-]*)*$")
+CLAIM = "claim.json"
+
+
+def live_sessions() -> dict[str, str | None]:
+    found: dict[str, str | None] = {}
+    for path in (Path.home() / ".claude" / "sessions").glob("*.json"):
+        try:
+            data = json.loads(path.read_text(encoding="utf-8"))
+        except (OSError, ValueError):
+            continue
+        if data.get("sessionId"):
+            found[data["sessionId"]] = data.get("name")
+    return found
+
+
+def caller(session: str | None) -> tuple[str, str]:
+    if session:
+        return session, "other"
+    env = os.environ.get("CLAUDE_CODE_SESSION_ID")
+    if env:
+        return env, "claude-code"
+    print("council: no session id; pass --session", file=sys.stderr)
+    sys.exit(2)
+
+
+def office_dir(office: str) -> Path:
+    if not OFFICE.match(office):
+        print(f"council: invalid office {office!r}", file=sys.stderr)
+        sys.exit(2)
+    return ROOT / office
+
+
+def read_claim(path: Path) -> dict | None:
+    try:
+        return json.loads((path / CLAIM).read_text(encoding="utf-8"))
+    except (FileNotFoundError, ValueError):
+        return None
+
+
+def is_stale(claim: dict | None, live: dict[str, str | None]) -> bool:
+    if claim is None:
+        return False
+    if claim["runtime"] == "claude-code":
+        return claim["session"] not in live
+    return time.time() - claim["beat"] > HEARTBEAT_LIMIT
+
+
+def claims() -> list[tuple[str, dict]]:
+    found = []
+    for path in sorted(ROOT.glob(f"**/{CLAIM}")):
+        claim = read_claim(path.parent)
+        if claim is not None:
+            found.append((path.parent.relative_to(ROOT).as_posix(), claim))
+    return found
+
+
+def write_claim(path: Path, office: str, sid: str, runtime: str, work: str) -> None:
+    claim = {
+        "office": office,
+        "session": sid,
+        "runtime": runtime,
+        "name": live_sessions().get(sid),
+        "since": datetime.datetime.now(datetime.timezone.utc).isoformat(timespec="seconds"),
+        "beat": time.time(),
+        "work": work,
+    }
+    (path / CLAIM).write_text(json.dumps(claim), encoding="utf-8")
+
+
+def cmd_claim(args: argparse.Namespace) -> int:
+    sid, runtime = caller(args.session)
+    path = office_dir(args.office)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    try:
+        path.mkdir()
+    except FileExistsError:
+        claim = read_claim(path)
+        if claim is not None and claim["session"] == sid:
+            write_claim(path, args.office, sid, runtime, args.work)
+            print(f"held {args.office}")
+            return 0
+        print(json.dumps(claim) if claim else f"{args.office}: claim in progress")
+        return 4 if is_stale(claim, live_sessions()) else 3
+    write_claim(path, args.office, sid, runtime, args.work)
+    print(f"elected {args.office}")
+    return 0
+
+
+def cmd_takeover(args: argparse.Namespace) -> int:
+    path = office_dir(args.office)
+    claim = read_claim(path)
+    if path.exists() and not is_stale(claim, live_sessions()):
+        print(json.dumps(claim) if claim else f"{args.office}: claim in progress")
+        return 3
+    # Two successors racing between this check and the rename can both win;
+    # democracy names exactly one successor per office to keep that window closed.
+    if path.exists():
+        tomb = path.with_name(f"{path.name}.stale-{os.getpid()}-{time.time_ns()}")
+        try:
+            path.rename(tomb)
+        except FileNotFoundError:
+            pass
+        else:
+            shutil.rmtree(tomb)
+    return cmd_claim(args)
+
+
+def cmd_release(args: argparse.Namespace) -> int:
+    sid, _ = caller(args.session)
+    path = office_dir(args.office)
+    claim = read_claim(path)
+    if claim is None:
+        print(f"{args.office}: not claimed")
+        return 0
+    if claim["session"] != sid and not is_stale(claim, live_sessions()):
+        print(json.dumps(claim))
+        return 3
+    shutil.rmtree(path)
+    print(f"released {args.office}")
+    return 0
+
+
+def cmd_beat(args: argparse.Namespace) -> int:
+    sid, _ = caller(args.session)
+    for office, claim in claims():
+        if claim["session"] == sid:
+            claim["beat"] = time.time()
+            (ROOT / office / CLAIM).write_text(json.dumps(claim), encoding="utf-8")
+    return 0
+
+
+def cmd_list(args: argparse.Namespace) -> int:
+    live = live_sessions()
+    sid = None
+    if args.mine:
+        sid, _ = caller(args.session)
+    for office, claim in claims():
+        if sid is not None and claim["session"] != sid:
+            continue
+        state = "stale" if is_stale(claim, live) else "live"
+        holder = claim["name"] or claim["session"]
+        print(f"{office}\t{holder}\t{claim['since']}\t{state}\t{claim['work']}")
+    return 0
+
+
+def in_git_worktree(target: Path) -> bool:
+    return any((path / ".git").exists() for path in (target, *target.parents))
+
+
+def cmd_gate(_: argparse.Namespace) -> int:
+    payload = json.load(sys.stdin)
+    tool_input = payload.get("tool_input") or {}
+    target = tool_input.get("file_path") or tool_input.get("notebook_path")
+    if not target or not in_git_worktree(Path(target).resolve()):
+        return 0
+    if any(claim["session"] == payload.get("session_id") for _, claim in claims()):
+        return 0
+    script = Path(__file__).resolve()
+    print(
+        "democracy: this session holds no claim, so it may not edit files in a git "
+        f"worktree. Claim the work first: python3 {script} claim work/<slug> "
+        "--work '<one line>'. Exit 3 or 4 means another agent holds it; read the "
+        "democracy skill.",
+        file=sys.stderr,
+    )
+    return 2
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(prog="council")
+    sub = parser.add_subparsers(dest="cmd", required=True)
+    for name, func in (("claim", cmd_claim), ("takeover", cmd_takeover)):
+        cmd = sub.add_parser(name)
+        cmd.add_argument("office")
+        cmd.add_argument("--work", required=True)
+        cmd.add_argument("--session")
+        cmd.set_defaults(func=func)
+    release = sub.add_parser("release")
+    release.add_argument("office")
+    release.add_argument("--session")
+    release.set_defaults(func=cmd_release)
+    beat = sub.add_parser("beat")
+    beat.add_argument("--session")
+    beat.set_defaults(func=cmd_beat)
+    listing = sub.add_parser("list")
+    listing.add_argument("--mine", action="store_true")
+    listing.add_argument("--session")
+    listing.set_defaults(func=cmd_list)
+    sub.add_parser("gate").set_defaults(func=cmd_gate)
+    args = parser.parse_args()
+    return args.func(args)
+
+
+if __name__ == "__main__":
+    sys.exit(main())

tests/test_council.py

@@ -0,0 +1,88 @@
+"""Exercise the democracy council script: election, staleness, release, gate."""
+
+from __future__ import annotations
+
+import json
+import subprocess
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+COUNCIL = REPO_ROOT / "skills" / "democracy" / "scripts" / "council.py"
+
+
+class TestCouncil(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.tmp.cleanup)
+        self.home = Path(self.tmp.name) / "home"
+        self.root = Path(self.tmp.name) / "council"
+        self.sessions = self.home / ".claude" / "sessions"
+        self.sessions.mkdir(parents=True)
+        self._live("s1", "One")
+        self._live("s2", "Two")
+
+    def _live(self, sid: str, name: str) -> None:
+        (self.sessions / f"{sid}.json").write_text(
+            json.dumps({"sessionId": sid, "name": name}), encoding="utf-8"
+        )
+
+    def _run(self, sid: str, *args: str, stdin: str = "") -> subprocess.CompletedProcess:
+        env = {
+            "PATH": "/usr/bin:/bin",
+            "HOME": str(self.home),
+            "AGENTS_COUNCIL_ROOT": str(self.root),
+            "CLAUDE_CODE_SESSION_ID": sid,
+        }
+        return subprocess.run(
+            [sys.executable, str(COUNCIL), *args],
+            capture_output=True,
+            text=True,
+            env=env,
+            input=stdin,
+            check=False,
+        )
+
+    def test_first_claim_wins_and_second_is_refused(self):
+        self.assertEqual(self._run("s1", "claim", "work/x", "--work", "w").returncode, 0)
+        second = self._run("s2", "claim", "work/x", "--work", "w")
+        self.assertEqual(second.returncode, 3)
+        self.assertIn('"session": "s1"', second.stdout)
+
+    def test_dead_holder_is_stale_and_can_be_taken_over(self):
+        self._run("s1", "claim", "work/x", "--work", "w")
+        (self.sessions / "s1.json").unlink()
+        self.assertEqual(self._run("s2", "claim", "work/x", "--work", "w").returncode, 4)
+        self.assertEqual(self._run("s2", "takeover", "work/x", "--work", "w").returncode, 0)
+        self.assertIn('"session": "s2"', (self.root / "work/x/claim.json").read_text())
+
+    def test_takeover_of_a_live_holder_is_refused(self):
+        self._run("s1", "claim", "work/x", "--work", "w")
+        self.assertEqual(self._run("s2", "takeover", "work/x", "--work", "w").returncode, 3)
+
+    def test_only_the_holder_releases(self):
+        self._run("s1", "claim", "work/x", "--work", "w")
+        self.assertEqual(self._run("s2", "release", "work/x").returncode, 3)
+        self.assertEqual(self._run("s1", "release", "work/x").returncode, 0)
+        self.assertFalse((self.root / "work/x").exists())
+
+    def test_gate_blocks_unclaimed_edits_in_a_git_worktree_only(self):
+        repo = Path(self.tmp.name) / "repo"
+        (repo / ".git").mkdir(parents=True)
+        inside = json.dumps({"session_id": "s1", "tool_input": {"file_path": str(repo / "f")}})
+        outside = json.dumps(
+            {"session_id": "s1", "tool_input": {"file_path": str(self.home / "f")}}
+        )
+        self.assertEqual(self._run("s1", "gate", stdin=inside).returncode, 2)
+        self.assertEqual(self._run("s1", "gate", stdin=outside).returncode, 0)
+        self._run("s1", "claim", "work/x", "--work", "w")
+        self.assertEqual(self._run("s1", "gate", stdin=inside).returncode, 0)
+
+    def test_invalid_office_is_a_usage_error(self):
+        self.assertEqual(self._run("s1", "claim", "../x", "--work", "w").returncode, 2)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_dream.py

@@ -126,6 +126,8 @@ class TestSettingsRegistration(unittest.TestCase):
         entries = json.dumps(data["hooks"]["UserPromptSubmit"])
         self.assertEqual(entries.count("dream/hooks/reminder.sh"), 1)
         self.assertEqual(entries.count("autotune/hooks/reminder.sh"), 1)
+        gate = json.dumps(data["hooks"]["PreToolUse"])
+        self.assertEqual(gate.count("democracy/scripts/council.py"), 1)
 
 
 if __name__ == "__main__":