#!/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 inspect import json import os import re import subprocess import sys import time HOME = os.path.expanduser("~") D = os.path.join(HOME, "dotfiles") SCRIPTS = os.path.join(D, "scripts") ENGINE = os.path.join(HOME, "_Dev", "studium-engine") 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-95 AMENDMENT 1, 2026-08-24. Same shape as the PENDING-86 (a) precedent # below: the jurist is asked to rule options (a)-(d) on the grounding gate, and the # two artifacts the ruling GOVERNS are the two it could not read. Ruling from the # record alone would make the subject the executor's description of the hooks rather # than the hooks -- the defect recorded inside REVIEWED-125 ("the ruling's subject # was the pasted text, not the filed artifact"). Read-only, keyed, enum unchanged. "grounding-hook-pretool": (os.path.join(HOME, ".claude", "hooks", "verify-before-compose.sh"), "PreToolUse grounding gate (Write|Edit route)"), "grounding-hook-commit": (os.path.join(HOME, "_Dev", "chamber-library", ".githooks", "pre-commit"), "pre-commit grounding gate (tool-agnostic route)"), # 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"), # PENDING-86 class, FOURTH instance — steward-requested 2026-08-10. Same shape # as the two above and the same remedy: the jurist design-gates a V0-lane # matter (§7.4(i) governs what Tier-1 must REFUSE) whose governing text it # could not open, and said so at the top of its own ruling — "every quotation # in Part I is your testimony, not something I can reach. That is exactly the # material two of today's three reversals turned on." Read-only, keyed, no # path argument; the enum design is unchanged. "v2-harness-design": (os.path.join(ENGINE, "docs", "v2-validation-harness-design-2026-07-09.md"), "the V2 validation-harness design — the ratified text behind the " "nested-voice arc. §5 = failure-mode table (F1–F10, each with a " "Direction and an 'Exercised by' column); §6.2 = the PRE-REGISTERED " "difficulty strata; §7.4 = whole-for-part attribution, sub-type (i) " "nested-voice. ⚠ Long (~65 KB); page to the § you need rather than " "reading from the top"), # ⚠ DESCRIPTIONS REFRESHED 2026-08-13. They are what the jurist reads to decide # which key to open, and both went stale the same day the dispositions landed — # a description standing in for the thing, which is the class this whole arc is # about. Refreshed BEFORE the client restart that first serves these keys, so # the jurist's first read is not of a stale index. "v2-stratum-tags": (os.path.join(ENGINE, "corpus", "v2-stratum-tags.yaml"), "P7 — the stratum tagging of the INHERITED gold (fr Mauss; en March " "chavruta). ⚠ Its `en` block is `taggable: false`. ⚠ The fr ratio is " "now VOID, not 1:9 — re-derived ONCE after PENDING-134 lands and all " "dispositions are recorded (REVIEWED-116 pt 5). Cell as of 2026-08-13: " "11 listed spans / 8 grounded (1 A + 7 B) / 2 reclassified out " "(Havámál L856 at P7; instance 8 L1551, PENDING-135) / 1 retracted " "(L926, REVIEWED-118) / 10 grounded bound instances. Count fields now " "carry their populations — bare `distinct_spans` was retired " "(PENDING-136)"), "mauss-fixture-spans": (os.path.join(ENGINE, "corpus", "mauss-phase2-spans.yaml"), "the fr fixture — 15 BOUND instances / 11 distinct spans, with " "per-instance Tier-1 verdicts. Locations and verdicts only. ⚠ " "BINDING ≠ ADMISSIBILITY: all 15 still bind and the counts here do " "not move for a disposition, but 3 instances (6/12/16) are " "retracted and 1 (8) reclassified — see the " "`retracted_from_grounded_gold` and `reclassified_from_grounded_gold` " "keys in the file, and `v2-stratum-tags` for the grounded set"), "mauss-fixture-citations": (os.path.join(ENGINE, "corpus", "mauss-phase2-reanchored.yaml"), "the fr fixture's CITATION TEXTS — the quoted strings " "themselves. This is the file that answers 'whose " "proposition does the claim assert?'"), "chamber-v1-diff": (os.path.expanduser( "~/dotfiles/claude/governance/chamber-v1-formation-diff-2026-08-25.md"), "PENDING-151 STEP 1 — the mechanical formation diff over the 9 v1 Chamber pairs. " "Counts and word lists only. ⚠ It renders NO verdict: whether a divergence is " "content or register is step 2, and the executor is barred from it."), } 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) SEARCH_KEYS = ("pending", "pending-archive", "reviewed") def t_search(args): """Keyword search across the authorization corpus — PENDING-86 option (d). The failure this exists for is NOT 'the jurist cannot read item X'. It is 'the jurist cannot DISCOVER item X whose id it does not already know' — the 2026-07-29 case where a ruling demanded an outcome REVIEWED-74 had settled four days earlier, in a file the jurist could read but had no reason to open. Keyed retrieval cannot fix that; only search can. Three deliberate properties: * The result unit is the ITEM, because the jurist's next move is governance_item(id). Boundaries come from wd.item_spans — no second definition of 'an item' (the 2026-07-28 bug). * Terms are ANDed, and that is disclosed, because a silently conjunctive matcher is exactly how recall dies as a question lengthens (the engine's PENDING-97, found 2026-08-04). A zero-result search says what it searched and how, so silence is legible rather than bare. * A raw-text pass runs BESIDE the item pass and reports matches lying outside every span item_spans can see — an indented header, a region the parser drops. Search reveals the hole instead of papering it. """ q = (args.get("query") or "").strip() if not q: return "ERROR: query is required, e.g. 'order attestation' or 'two-column'." terms = [t for t in q.lower().split() if t] lim = max(1, min(int(args.get("limit") or 12), 50)) hits, scanned, orphan_lines = [], 0, [] for key in SEARCH_KEYS: text = read(FILES[key][0]) if text is None: continue fname = os.path.basename(FILES[key][0]) lines = text.split("\n") covered = set() for head, start, end in wd.item_spans(text): scanned += 1 covered.update(range(start, end)) body = "\n".join(lines[start - 1:end - 1]) low = body.lower() if not all(t in low for t in terms): continue n = sum(low.count(t) for t in terms) phrase = q.lower() in low snippet = "" for ln in body.split("\n"): if any(t in ln.lower() for t in terms): snippet = " ".join(ln.split())[:200] break hits.append((phrase, n, head, fname, start, snippet)) # Structural health pass, deliberately INDEPENDENT of the query: an item header # hidden by leading whitespace is invisible to item_spans, so it can never appear # in results and its absence would look like a genuine miss. Query-independent # because the jurist needs to know the index is incomplete whatever it searched. # Narrow by construction — an earlier draft flagged any uncovered matching line # and drowned the signal in each file's preamble, which is exactly how a warning # stops being read. infence = False for i, ln in enumerate(lines, 1): if ln.lstrip().startswith("```"): infence = not infence continue if infence or i in covered: continue m = re.match(r"^\s+## ((?:PENDING|REVIEWED|COMPLETED)-\S+)", ln) if m: orphan_lines.append((fname, i, " ".join(ln.split())[:160])) hits.sort(key=lambda h: (not h[0], -h[1])) mode = ("exact phrase, else all-terms-present" if len(terms) > 1 else "single term, substring") o = [f"[governance_search {q!r} — {len(hits)} matching item(s) of {scanned} scanned " f"across {', '.join(SEARCH_KEYS)}]", f"match mode: ALL {len(terms)} term(s) must appear in the same item " f"({mode}); ranked by exact-phrase first, then raw term-count — " f"term-count is a computed field, not a relevance score.", ""] if not hits: o.append("NO ITEM MATCHED. This is a legible empty, not a claim that the corpus " "is silent on the subject: the matcher is CONJUNCTIVE, so a longer query " "narrows fast. Retry with fewer or more common terms before concluding " "nothing is on file.") for phrase, n, head, fname, start, snip in hits[:lim]: o.append(f"{'★ ' if phrase else ' '}{head} [{fname}:{start}] {n} hit(s)") if snip: o.append(f" … {snip}") if len(hits) > lim: o.append(f"\n[… {len(hits) - lim} more — raise limit (max 50)]") if orphan_lines: o.append("\n⚠ MATCHES OUTSIDE ANY ITEM THE PARSER CAN SEE — a header hidden by " "leading whitespace, or a region item_spans drops. governance_item() " "cannot retrieve these by id; they are reported here so the gap is " "visible rather than silent:") for fname, i, ln in orphan_lines[:10]: o.append(f" {fname}:{i} {ln}") o.append("\nNext: governance_item(id) for the verbatim body of any hit above.") return "\n".join(o) 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}" # ── PENDING-151 step 2 ──────────────────────────────────────────────────────────── # The step-2 judge reads a PAIR, not a file: the question is whether two arms diverge in # content or only in register, and neither arm answers it alone. So the unit served here # is the unit of the work — derived from the consumer, not from how the files happen to # sit on disk. # # ⚠ STATICALLY ENUMERATED, and deliberately. `t_read`'s design property is "no path # argument", and the point of it is that the reachable set is REVIEWED rather than # matched. Walking the directory at import would preserve the letter (the caller still # passes no path) and lose the substance: a file dropped into the archive would become # jurist-readable with nobody having looked at it. The 2025 archive is a closed record. # # ⚠ The three filename defects are reproduced EXACTLY and are not repaired: a leading # space in one claude arm, a doubled extension in another, a trailing space in a # directory name. PENDING-151: they are the 2025 record and renaming is a separate [FIX] # the steward owns. A reader who sees them here is seeing the archive, not a tidied copy. # # ⚠ This is `chamber-sessions-private`. Steward-authorized 2026-08-25 to reach the jurist # for step 2; noted because it is the first non-governance, non-public material on this # surface, and the surface exists to be bounded. V1_ROOT = os.path.expanduser( "~/_Dev/animal-davidglidden-eu/chamber-sessions-private/2025") PAIRS = { "06-14-owl-emblem/shadow": ("2025-06-14-owl-emblem/[shadow]gpt-raw.txt", "2025-06-14-owl-emblem/ [shadow]claude-raw.txt"), "06-14-owl-emblem/standard": ("2025-06-14-owl-emblem/[standard]gpt-raw.txt", "2025-06-14-owl-emblem/[standard]claude-raw.txt"), "06-16-first-submission-first-light/first-light": ("2025-06-16-first-submission-first-light/[first-light]gpt-raw.txt", "2025-06-16-first-submission-first-light/[first-light]claude-raw.txt"), "06-17-the-ethics-of-the-reply/shadow": ("2025-06-17-The Ethics of the Reply /[shadow]gpt-raw.txt", "2025-06-17-The Ethics of the Reply /[shadow]claude-raw.txt"), "06-17-the-ethics-of-the-reply/standard": ("2025-06-17-The Ethics of the Reply /[standard]gpt-raw.txt", "2025-06-17-The Ethics of the Reply /[standard]claude-raw.txt"), "06-19-savall-prometheus-21/standard": ("2025-06-19-Savall-Prometheus-21/[standard]gpt-raw.txt", "2025-06-19-Savall-Prometheus-21/[standard]claude-raw.txt.txt"), "07-01-marginalia/standard": ("2025-07-01-marginalia/[standard]gpt-raw.txt", "2025-07-01-marginalia/[standard]claude-raw.txt"), "07-11-the-ethics-of-the-reply-part-ii/shadow": ("2025-07-11-the-ethics-of-the-reply-part-ii/[shadow]gpt-raw.txt", "2025-07-11-the-ethics-of-the-reply-part-ii/[shadow]claude-raw.txt"), "07-11-the-ethics-of-the-reply-part-ii/standard": ("2025-07-11-the-ethics-of-the-reply-part-ii/[standard]gpt-raw.txt", "2025-07-11-the-ethics-of-the-reply-part-ii/[standard]claude-raw.txt"), } def t_pair(args): """Both arms of one enumerated v1 Chamber pair, verbatim. No path argument.""" key = (args.get("pair") or "").strip() if key not in PAIRS: return ("ERROR: unknown pair %r. Allowed: %s" % (key, ", ".join(sorted(PAIRS)))) g_rel, c_rel = PAIRS[key] out = [f"[v1 Chamber pair — {key} — verbatim, both arms]", "", "⚠ STEP 2 IS THE QUESTION THIS SERVES: does a divergence carry DIFFERENT", "CONTENT, or the SAME CONTENT IN A DIFFERENT REGISTER? PENDING-151 reserves it", "to the jurist or steward: the executor is one of the two formations compared.", "⚠ The Claude arm is longer in 9 of 9 pairs (1.41x-3.40x). Distinctive-term", "counts rise with length by construction; see chamber-v1-diff for the", "length-normalised columns and their declared limits.", ""] for arm, rel in (("GPT", g_rel), ("CLAUDE", c_rel)): text = read(os.path.join(V1_ROOT, rel)) out += [f"───────── {arm} arm ── {rel}", ""] out += [text if text is not None else f"UNREADABLE: {rel} is absent or unreadable at this time.", ""] return "\n".join(out) 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"]}), ("governance_pair", t_pair, "Both arms of one v1 Chamber formation pair (2025 GPT and 2025 Claude reading the " "same submitted text), verbatim, chosen by key. This is PENDING-151 step 2's " "material: the executor may run the mechanical diff but MAY NOT judge whether a " "divergence is content or register, being one of the two formations compared. " "Read chamber-v1-diff first for the counts and their length confound.", {"type": "object", "properties": { "pair": {"type": "string", "enum": sorted(PAIRS), "description": "Which pair, as session/protocol."}}, "required": ["pair"]}), ("governance_search", t_search, "Keyword search across PENDING.md, PENDING-archive.md and REVIEWED.md. Use this to " "DISCOVER a relevant prior item or ruling whose id you do not already know — the case " "governance_item cannot serve, because it needs the id up front. Terms are ANDed " "within a single item; results name ids to pass to governance_item.", {"type": "object", "properties": { "query": {"type": "string", "description": "Words that must all appear in the same item, e.g. " "'order attestation' or 'two-column'. Not a regex."}, "limit": {"type": "integer", "description": "Max items to list, max 50 (default 12)."}}, "required": ["query"]}), ("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 4th instance — the V0-lane texts behind the nested-voice arc:") chk("v2-harness-design key reads", t_read({"file": "v2-harness-design"}).startswith("[v2-harness-design —")) chk("...and carries §6.2's PRE-REGISTERED strata [the clause the ruling turns on]", "Difficulty strata (pre-registered" in t_read({"file": "v2-harness-design"})) chk("...and carries §7.4's nested-voice sub-type", "nested-voice" in t_read({"file": "v2-harness-design"})) chk("...and carries §5's F4 row, whose omission from §6.2 is the counter-argument", "Irony / persona distance / unasserted content" in t_read({"file": "v2-harness-design"})) chk("v2-stratum-tags (P7) key reads, and its en cell is the untaggable one", "taggable: false" in t_read({"file": "v2-stratum-tags"})) chk("mauss-fixture-spans key reads", t_read({"file": "mauss-fixture-spans"}).startswith("[mauss-fixture-spans —")) chk("mauss-fixture-citations carries the CITATION TEXT, not just the locations" " [the file that answers whose-proposition]", "Le charpentier dit à Arthur" in t_read({"file": "mauss-fixture-citations"})) chk("...and the L926 testimony the whole arc turns on", "Le hau n'est pas le vent qui souffle" in t_read({"file": "mauss-fixture-citations"})) 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})) print("\nPENDING-86 (d) — discovery without knowing the id:") _s = t_search({"query": "added-side fabrication"}) chk("search finds REVIEWED-74 from CONTENT alone [the 2026-07-29 failure: the jurist " "demanded an outcome this ruling had settled four days earlier, and could not find it]", "REVIEWED-74" in _s) chk("search spans the archive too, not just open items", "PENDING-archive.md" in _s) chk("results carry ids to hand to governance_item [the two tools compose]", "governance_item(id)" in _s) _e = t_search({"query": "zzzq nonexistent phrase"}) chk("a miss is a LEGIBLE empty, not a bare one [it discloses the conjunctive matcher, " "the failure shape that killed engine recall — PENDING-97]", "0 matching item(s)" in _e and "CONJUNCTIVE" in _e) chk("the match mode is stated on EVERY result, hit or miss [positive control for the above]", "match mode:" in _s and "match mode:" in _e) chk("empty query refused rather than matching everything", t_search({"query": " "}).startswith("ERROR: query is required")) chk("search takes no path and no file key — corpus is a fixed tuple", set(SEARCH_KEYS).issubset(FILES) and len(SEARCH_KEYS) == 3) chk("no second definition of 'an item' — search reuses wd.item_spans [the 07-28 bug]", "wd.item_spans" in inspect.getsource(t_search)) chk("orphan pass reports NOTHING today [the three indented headers were unhidden " "2026-08-05; this control goes red if a header is ever hidden again]", "MATCHES OUTSIDE ANY ITEM" not in t_search({"query": "the"})) 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