David F GliddenandClaude Opus 5
3df09228c0
The three-party model asks Claude.app to rule on items it cannot read. Steward
confirmed 2026-07-28 that local MCP servers are exposed to the app's *chat*
surface — and always have been, predating Cowork by about a year. My earlier
framing ("chat, not only Cowork") had the relationship backwards: it is "chat,
always; Cowork, only while its loop still runs locally," and local Cowork is the
mode being phased out as default. The jurist chat is therefore the sturdy target.
Five read-only tools. The one a pasted cache can never provide is
governance_item(id): the verbatim body of any item or ruling, across PENDING.md,
PENDING-archive.md and REVIEWED.md. Four refusals are designed in, each with a
control proving the refusal is detectable — no writes (AST-audited), no path
arguments (keys from a fixed enum, so there is no traversal to defend), no second
parser (item_spans is imported, not reimplemented), and not an agent (tools
return data; an agent would return testimony about the substrate instead).
[FIX] to the shared definition while here: item_spans() is now fence-aware. A
'## ' header inside a fenced block is neither an item nor a boundary. Zero such
headers exist today — 17 open items before and after — but governance drafts are
written as fenced markdown carrying '## REVIEWED-N' headers, which is the
steward's own practice, so the next draft would have created a phantom item and
truncated the item containing it. PENDING-82's own fenced JSON block confirms the
fix within the hour.
Not installed. The mcpServers key edits the steward's desktop-app config; the
snippet is in PENDING-82 and the server is inert until someone loads it.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01WuMjg3ipEVa3n8CoSzoyvc
423 lines
20 KiB
Python
423 lines
20 KiB
Python
#!/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
|