Files
dotfiles/scripts/governance-mcp.py
T
David F GliddenandClaude Opus 5 5cd5faf66e [HARDENING] governance-mcp: the jurist can now read the constitution it design-gates (PENDING-86 a)
Steward-authorized 2026-08-05, on the jurist's own request while unable to close
PENDING-99's Q2 — a question that turns on the chamber constitution's vocabulary
(§II.3's "inline anchor marker", §V's marker exclusion), which governance_read
did not expose. Third recorded instance on PENDING-86: the constitution, the
skill files, contamination-problem.md.

Adds two keys to the existing enum: `chamber-spec`, `graduation-spec`. No new
tool, no path argument, no traversal surface — the domain stays enumerable and
every refusal control still passes.

⚠ THE NON-OBVIOUS PART. Reachability of the KEY is not reachability of the
CLAUSE. This file's operative sections begin around line 354; the ~330 lines
above them are SUPERSEDED version headers kept as the amendment trail. A jurist
reading with the default limit=400 would land squarely in obsoleted text and
could rule on superseded clauses — the new access CAUSING the misruling it
exists to prevent. So the trap is disclosed on the key's own description, at the
point of use, and two controls pin it:

  - the §V inline-anchor clause and the §II.3 marker constraint are both
    reachable in ONE paged call (offset=350, limit=2000) — the actual Q2 text
  - NEGATIVE CONTROL: a first-page read does land in the "(obsoleted)" region,
    proving the trap is real rather than hypothetical

Selftest 29 → 35 controls, 0 fail. Live stdio round-trip confirms the §V clause
arrives verbatim through governance_read.

⚠ Requires a Claude.app restart: the running server process carries the old
code and will not show the new keys until respawned.

Option (d) — keyword search across PENDING/PENDING-archive/REVIEWED — is NOT in
this change and remains open on PENDING-86. It is a new tool surface, not two
enum entries, and the jurist asked for (a)+(d) together.

Refs PENDING-86, PENDING-99 Q2, PENDING-82.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01AB3Kryoy6b1pm2Nz1DYdLh
2026-08-05 21:17:33 +02:00

458 lines
23 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/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"),
# PENDING-86 option (a), steward-authorized 2026-08-05. The jurist design-gates
# constitutional supersessions of documents it could not read; three distinct
# instances are recorded on that item. These two are the ones the loop actually
# rules on. Read-only, keyed, no path argument — the enum design is unchanged.
"chamber-spec": (os.path.join(HOME, "_Dev", "chamber-library", "docs",
"chamber-library-specification.md"),
"the chamber library constitution — ⚠ the OPERATIVE version header is "
"at the top, but roughly the next 330 lines are SUPERSEDED version "
"headers kept as the amendment trail; the operative sections start at "
"'## 0. What this document is'. Page past them (offset≈350) or you will "
"be reading obsoleted text"),
"graduation-spec": (os.path.join(HOME, "_Dev", "chamber-library", "_curation",
"graduation-spec.yaml"),
"the constitution's machine-readable convention-data + gate-list"),
}
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 —"))
print("\nPENDING-86 (a) — the documents the jurist design-gates are now reachable:")
chk("chamber-spec key reads",
t_read({"file": "chamber-spec"}).startswith("[chamber-spec —"))
chk("graduation-spec key reads",
t_read({"file": "graduation-spec"}).startswith("[graduation-spec —"))
# Reachability of the KEY is not reachability of the CLAUSE. These two controls
# test the actual thing the item exists for: the §II.3 / §V text a Q2-shaped
# ruling turns on. A key that opens onto 330 lines of superseded headers would
# pass the two checks above and still leave the gap wide open.
_spec_deep = t_read({"file": "chamber-spec", "offset": 350, "limit": 2000})
chk("§V's inline-anchor clause is reachable in one paged call [the Q2 clause]",
"is content-for-the-reader but is **not a prose word**" in _spec_deep)
chk("§II.3's marker constraint is reachable in the same call [the Q2 clause]",
"must not corrupt the" in _spec_deep and "prose-word-identity check under §V" in _spec_deep)
chk("the superseded-header trap is disclosed on the key itself [it would cause the "
"misruling this access exists to prevent]",
"SUPERSEDED version" in FILES["chamber-spec"][1])
chk("a first-page read lands in the SUPERSEDED region [negative control: proves the "
"trap is real, not hypothetical]",
"(obsoleted)" in t_read({"file": "chamber-spec", "limit": 300}))
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