#!/usr/bin/env python3 """governance-mcp — read-only MCP server giving the jurist eyes on the substrate. WHY THIS EXISTS The three-party model asks Claude.app (jurist) to rule on governance items it cannot read. Its reader has no filesystem, so until now its picture of PENDING / REVIEWED / drift could only be *cached* — pasted in by the steward and stale from the moment it landed. `wake-digest.py --brief` narrows that gap; it does not close it, because a snapshot cannot answer a question nobody anticipated. This closes it for Claude.app's **chat** surface, where local MCP servers have always been exposed (steward-confirmed 2026-07-28). Cowork gets local MCP only while its agent loop still runs on-device — the mode being phased out as default — so a Cowork-dependent design would have been the fragile one. WHAT IT REFUSES TO BE - **Read-only.** No tool writes, moves, or deletes. The jurist proposes; the steward authorizes; the executor acts. An MCP write path would collapse three parties into one. - **Not an agent.** Every tool returns *data*, verbatim where possible. The alternative considered and rejected was a second Claude with filesystem access reporting back: that yields an agent's testimony about the substrate, not the substrate. A tool returns data; an agent returns a claim. - **No second parser.** "Open item" is defined ONCE, in wake-digest.py, imported here. On 2026-07-28 a splitter's private definition of "item" hid twenty items, ten of them open, and every check inherited the blind spot. Two implementations of one field is that failure waiting to recur across a boundary nobody watches. - **No path arguments.** Every tool takes a KEY from a fixed enum, never a path. There is no traversal to defend against because there is no path to traverse — and the reachable domain is enumerable rather than defined by the instrument. governance-mcp.py --selftest exercise every tool, presence AND absence governance-mcp.py --stdio serve (how the app launches it; also the default) Provenance: 2026-07-28, on PENDING-81's MCP leg. Sibling of wake-digest.py and governance-drift-check.py; imports the first. """ import importlib.util import json import os import subprocess import sys import time HOME = os.path.expanduser("~") D = os.path.join(HOME, "dotfiles") SCRIPTS = os.path.join(D, "scripts") NAME, VERSION = "governance", "1.0.0" # Newest protocol we speak; we echo the client's version when it sends a known one. PROTOCOLS = ("2025-06-18", "2025-03-26", "2024-11-05") def _load(mod_name, filename): """Import a sibling script whose filename is not a valid module name.""" spec = importlib.util.spec_from_file_location(mod_name, os.path.join(SCRIPTS, filename)) m = importlib.util.module_from_spec(spec) spec.loader.exec_module(m) return m wd = _load("wake_digest", "wake-digest.py") # the single definition of "open item" # ---- the enumerated domain: keys, never paths ------------------------------- FILES = { "pending": (os.path.join(D, "PENDING.md"), "open authorization items"), "pending-archive": (os.path.join(D, "PENDING-archive.md"), "closed authorization items"), "reviewed": (os.path.join(D, "REVIEWED.md"), "steward/jurist decisions"), "claude-md": (os.path.join(D, "CLAUDE.md"), "the executor's governing document"), "memory-index": (os.path.join(wd.MEM, "MEMORY.md"), "wake-loaded memory index"), "app-brief": (wd.BRIEF_PATH, "last generated .app Standing Context block"), } REPOS = wd.REPOS def read(path): try: return open(path, encoding="utf-8").read() except Exception as e: return None if isinstance(e, FileNotFoundError) else None # ---- tool implementations (pure-ish; selftest exercises these directly) ----- def t_state(_args): """The whole picture, computed now: open items with tags, recent rulings, drift, repos.""" wd.warn.clear() items, revs, drift, repos = wd.sec_pending(), wd.sec_reviewed(6), wd.sec_drift(), wd.sec_repos() o = [f"GOVERNANCE STATE — computed {time.strftime('%Y-%m-%d %H:%M')} local, not cached.", "", f"OPEN AUTHORIZATION ITEMS ({len(items)}) — full body via governance_item(id=…)"] for h, ln, tag in items: o.append(f" {tag:<14} {h} [{FILES['pending'][0].rsplit('/', 1)[1]}:{ln}]") o.append("\nRECENT RULINGS") for h, d in revs: o.append(f" {d:<11} {h}") o.append(f"\nGOVERNANCE DRIFT — CLAUDE.md: {drift} substrate-contradicted claim(s)." " Detection only; correcting doctrine requires [ESCALATE]. Detail via drift_report().") o.append("\nREPOS (branch · uncommitted files · last subject)") for r, head, dirty, subj in repos: o.append(f" {r:<24} {head:<32} {dirty:>3} dirty {subj}") if wd.warn: o.append("\n⚠ DEGRADED — these could not be computed (treat as unknown, not as clean):") o += [f" - {w}" for w in wd.warn] return "\n".join(o) def t_item(args): """Verbatim body of one PENDING/REVIEWED/COMPLETED item, wherever it lives. Item boundaries use the same rule as wake-digest: an item starts at any '## ' header and ends at the next one. No family regex — that is the 2026-07-28 bug.""" ident = (args.get("id") or "").strip() if not ident: return "ERROR: id is required, e.g. 'PENDING-81' or 'REVIEWED-80'." for key in ("pending", "pending-archive", "reviewed"): text = read(FILES[key][0]) if text is None: continue lines = text.split("\n") for head, start, end in wd.item_spans(text): if head == ident or head.startswith(ident + " "): return (f"[{ident} — verbatim from {os.path.basename(FILES[key][0])}, " f"line {start}]\n\n" + "\n".join(lines[start - 1:end - 1]).rstrip()) return (f"NOT FOUND: no '## {ident}' header in PENDING.md, PENDING-archive.md, or " f"REVIEWED.md. Use governance_state() to list open items by id.") def t_read(args): """Verbatim paged read of one enumerated file. No path argument by design.""" key = (args.get("file") or "").strip() if key not in FILES: return ("ERROR: unknown file key %r. Allowed: %s" % (key, ", ".join(f"{k} ({d})" for k, (_p, d) in FILES.items()))) path, desc = FILES[key] text = read(path) if text is None: return f"UNREADABLE: {key} ({desc}) is absent or unreadable at this time." lines = text.split("\n") off = max(0, int(args.get("offset") or 0)) lim = max(1, min(int(args.get("limit") or 400), 2000)) chunk = lines[off:off + lim] more = ("" if off + lim >= len(lines) else f"\n\n[… {len(lines) - off - lim} more lines — call again with offset={off + lim}]") return (f"[{key} — {desc} — lines {off + 1}–{off + len(chunk)} of {len(lines)}]\n\n" + "\n".join(chunk) + more) def t_drift(_args): """Full drift-check output. Reports its own non-verification rather than a clean bill.""" out = wd.sh([sys.executable, wd.DRIFT], timeout=30) if out is None: return ("UNESTABLISHED: governance-drift-check.py did not run. This is not a clean " "result — the absence of findings from an instrument that failed to run " "carries no information.") return out def t_repo(args): """Recent commits for one enumerated repo. Repo name from a fixed list, never a path.""" name = (args.get("repo") or "").strip() if name not in REPOS: return f"ERROR: unknown repo %r. Allowed: %s" % (name, ", ".join(REPOS)) path = os.path.join(HOME, "_Dev", name) if not os.path.isdir(os.path.join(path, ".git")): return f"NOT A REPO: {name} has no .git at {path}." n = max(1, min(int(args.get("count") or 15), 100)) log = wd.sh(["git", "-C", path, "log", f"-{n}", "--format=%h %ad %s", "--date=short"]) status = wd.sh(["git", "-C", path, "status", "-sb"]) if log is None: return f"UNREADABLE: git log failed in {name}." return f"[{name}]\n\n{status}\n\nLAST {n} COMMITS\n{log}" TOOLS = [ ("governance_state", t_state, "Current governance state, computed live: every open authorization item with its " "[TAG], recent REVIEWED rulings, the CLAUDE.md drift count, and per-repo status. " "Start here, then drill in with governance_item.", {"type": "object", "properties": {}}), ("governance_item", t_item, "Verbatim text of one authorization item or ruling by id (e.g. PENDING-81, " "PENDING-S4, REVIEWED-80). Searches PENDING.md, PENDING-archive.md and REVIEWED.md. " "Never summarised.", {"type": "object", "properties": {"id": {"type": "string", "description": "Item id exactly as it appears after '## ', e.g. 'PENDING-81'."}}, "required": ["id"]}), ("governance_read", t_read, "Verbatim paged read of a governance document, chosen by key (not by path).", {"type": "object", "properties": { "file": {"type": "string", "enum": sorted(FILES), "description": "Which document to read."}, "offset": {"type": "integer", "description": "0-based first line (default 0)."}, "limit": {"type": "integer", "description": "Lines to return, max 2000 (default 400)."}}, "required": ["file"]}), ("drift_report", t_drift, "Full governance-drift-check output: claims in CLAUDE.md the substrate contradicts. " "Detection only — correcting doctrine requires steward authorization.", {"type": "object", "properties": {}}), ("repo_activity", t_repo, "Branch, uncommitted-file status and recent commits for one of the active repos.", {"type": "object", "properties": { "repo": {"type": "string", "enum": REPOS, "description": "Repo name."}, "count": {"type": "integer", "description": "Commits to list, max 100 (default 15)."}}, "required": ["repo"]}), ] IMPL = {n: f for n, f, _d, _s in TOOLS} # ---- JSON-RPC over stdio ---------------------------------------------------- def handle(msg): """-> response dict, or None for notifications (which must never be answered).""" mid, method, params = msg.get("id"), msg.get("method"), msg.get("params") or {} def ok(result): return {"jsonrpc": "2.0", "id": mid, "result": result} if mid is None: # notification return None if method == "initialize": want = params.get("protocolVersion") return ok({"protocolVersion": want if want in PROTOCOLS else PROTOCOLS[0], "capabilities": {"tools": {}}, "serverInfo": {"name": NAME, "version": VERSION}, "instructions": "Read-only view of the CapableMind governance substrate. " "Nothing here writes. Call governance_state first."}) if method == "ping": return ok({}) if method == "tools/list": return ok({"tools": [{"name": n, "description": d, "inputSchema": s} for n, _f, d, s in TOOLS]}) if method == "tools/call": name = params.get("name") if name not in IMPL: return {"jsonrpc": "2.0", "id": mid, "error": {"code": -32602, "message": f"unknown tool: {name}"}} try: text = IMPL[name](params.get("arguments") or {}) return ok({"content": [{"type": "text", "text": text}]}) except Exception as e: # a crash must not kill the session return ok({"content": [{"type": "text", "text": f"TOOL ERROR {type(e).__name__}: {e}"}], "isError": True}) return {"jsonrpc": "2.0", "id": mid, "error": {"code": -32601, "message": f"method not found: {method}"}} def serve(): for line in sys.stdin: line = line.strip() if not line: continue try: msg = json.loads(line) except Exception as e: print(json.dumps({"jsonrpc": "2.0", "id": None, "error": {"code": -32700, "message": f"parse error: {e}"}}), flush=True) continue resp = handle(msg) if resp is not None: print(json.dumps(resp), flush=True) # ---- selftest --------------------------------------------------------------- def write_calls(src): """-> list of filesystem-mutating calls in `src`, found by AST rather than text. A grep for a forbidden-token list cannot audit the file that CONTAINS the list — the first version of this check searched for `"w"`, `os.remove`, `shutil.` and duly found all of them, in its own definition. That is the day's recurring shape: an instrument whose domain includes itself. The AST sees calls, not characters.""" import ast MUTATORS = {"remove", "unlink", "rename", "replace", "rmdir", "mkdir", "makedirs", "chmod", "truncate", "write", "writelines", "write_text", "write_bytes"} out = [] for n in ast.walk(ast.parse(src)): if not isinstance(n, ast.Call): continue f = n.func if isinstance(f, ast.Name) and f.id == "open": mode = next((a.value for a in n.args[1:2] if isinstance(a, ast.Constant)), None) mode = next((k.value.value for k in n.keywords if k.arg == "mode" and isinstance(k.value, ast.Constant)), mode) if mode and any(c in str(mode) for c in "wax+"): out.append(f"open(mode={mode!r}) at line {n.lineno}") elif isinstance(f, ast.Attribute) and f.attr in MUTATORS: out.append(f"{f.attr}() at line {n.lineno}") return out def git_subcommands(src): """-> set of git subcommands invoked in `src`, read from AST argv literals. The third self-inclusion failure of 2026-07-28, and the one that finally made the pattern obvious: after moving the write-primitive audit to the AST, the git half of the same check was left as `'"commit"' not in src` — and the source it reads now contains `"commit"` and `"push"` as the literals of the check itself. A text search for forbidden words can never clear a file that must name those words. Read argv lists, not characters: `["git", "-C", path, "log", …]` -> {"log"}.""" import ast out = set() for n in ast.walk(ast.parse(src)): if isinstance(n, ast.List) and n.elts: first = n.elts[0] if isinstance(first, ast.Constant) and first.value == "git": for e in n.elts[1:]: if not (isinstance(e, ast.Constant) and isinstance(e.value, str)): continue # a variable (path, count) — skip if e.value.startswith("-"): continue # a flag, not the subcommand out.add(e.value) break return out def selftest(): ok = True def chk(name, cond): nonlocal ok ok = ok and bool(cond) print(f" [{'ok ' if cond else 'FAIL'}] {name}") print("protocol:") init = handle({"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {"protocolVersion": "2024-11-05"}}) chk("initialize echoes a known client protocol", init["result"]["protocolVersion"] == "2024-11-05") chk("initialize falls back for an unknown protocol", handle({"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {"protocolVersion": "1999-01-01"}})["result"]["protocolVersion"] == PROTOCOLS[0]) chk("notifications get no response", handle({"jsonrpc": "2.0", "method": "notifications/initialized"}) is None) chk("tools/list returns all tools", len(handle({"jsonrpc": "2.0", "id": 2, "method": "tools/list"})["result"]["tools"]) == len(TOOLS)) chk("unknown method -> -32601", handle({"jsonrpc": "2.0", "id": 3, "method": "nope"})["error"]["code"] == -32601) chk("unknown tool -> -32602", handle({"jsonrpc": "2.0", "id": 4, "method": "tools/call", "params": {"name": "rm_rf"}})["error"]["code"] == -32602) print("\ndomain is an enum, and the refusal is detectable:") chk("governance_read refuses an unlisted key", t_read({"file": "secrets"}).startswith("ERROR: unknown file key")) chk("governance_read refuses a path traversal attempt (no path is accepted at all)", t_read({"file": "../../.ssh/id_rsa"}).startswith("ERROR: unknown file key")) chk("governance_read ACCEPTS a listed key [positive control for the refusal above]", t_read({"file": "pending"}).startswith("[pending —")) chk("repo_activity refuses an unlisted repo", t_repo({"repo": "/etc"}).startswith("ERROR: unknown repo")) chk("repo_activity ACCEPTS a listed repo [positive control]", t_repo({"repo": REPOS[0], "count": 3}).startswith(f"[{REPOS[0]}]")) print("\nitem lookup — presence, absence, and the 07-28 blind-spot family:") chk("finds a numeric item", t_item({"id": "PENDING-81"}).startswith("[PENDING-81 — verbatim")) chk("finds an S-series item [the family a regex-on-digits missed]", t_item({"id": "PENDING-S4"}).startswith("[PENDING-S4 — verbatim")) chk("finds a ruling in REVIEWED.md", t_item({"id": "REVIEWED-80"}).startswith("[REVIEWED-80 — verbatim")) chk("finds an item that lives only in the archive", t_item({"id": "PENDING-72"}).startswith("[PENDING-72 — verbatim")) chk("absent id -> NOT FOUND, not a silent empty", t_item({"id": "PENDING-99999"}).startswith("NOT FOUND")) chk("missing id -> explicit error", t_item({}).startswith("ERROR")) chk("a shorter id does not match a longer one (PENDING-8 ≠ PENDING-81)", t_item({"id": "PENDING-8"}).startswith("NOT FOUND")) body = t_item({"id": "PENDING-81"}) # The body carries exactly ONE '## ' header — its own. The first version of this # check asserted ZERO and failed against correct code: the returned text is # prefix + the item INCLUDING its header. A self-test written before reading real # output tests the author's model of the output (07-28T09:35, same shape). chk("item body contains exactly its own header, no neighbour's", len([l for l in body.split("\n") if l.startswith("## ")]) == 1) chk("item body ends before the next item begins", "PENDING-82" not in body and "PENDING-80 —" not in body.split("\n", 3)[2]) print("\nno second definition of 'open item':") state = t_state({}) n_digest = len(wd.sec_pending()) chk(f"governance_state item count == wake-digest sec_pending() ({n_digest})", f"OPEN AUTHORIZATION ITEMS ({n_digest})" in state) chk("state names its own degradation when something cannot be computed", "DEGRADED" in state or not wd.warn) print("\nlive substrate:") chk("PENDING.md readable", read(FILES["pending"][0]) is not None) chk("REVIEWED.md readable", read(FILES["reviewed"][0]) is not None) d = t_drift({}) chk("drift_report runs and does not fake a clean result", d and not d.startswith("UNESTABLISHED")) print("\nread-only guarantee (structural, via AST):") src = open(__file__, encoding="utf-8").read() chk("no filesystem-mutating call in this file", write_calls(src) == []) chk("the checker DOES flag writes when present [positive control — a text search " "here would match its own token list, which is how the first version of this " "check failed]", len(write_calls("open('f','w')\nimport os\nos.remove('g')\np.write_text('h')")) == 3) subs = git_subcommands(src) chk(f"git is invoked read-only — subcommands present: {sorted(subs) or 'none'}", subs and subs <= {"log", "status", "show", "rev-parse", "ls-files", "diff"}) chk("the checker DOES flag a mutating subcommand [positive control]", git_subcommands('subprocess.run(["git", "-C", p, "push", "origin", "main"])') == {"push"}) print("\nSELFTEST", "PASS" if ok else "FAIL") return 0 if ok else 2 if __name__ == "__main__": if "--selftest" in sys.argv: sys.exit(selftest()) try: serve() except (BrokenPipeError, KeyboardInterrupt): pass