David F Glidden
4c3758e559
Jurist ruling accepted in full on all six questions. Q1: the variable is right, the derivation is not. §6.2 is PRE-REGISTERED, and a test that changes stratum-B membership, derived after reading the spans it reclassifies and entering by interpretation, voids that guarantee whether or not the test is right. Filed as PENDING-134 — new doctrine, dated, with §7.4(ii) as SUPPORTING ARGUMENT rather than derivation and §6.2's double omission recorded as the counter-argument heard and overruled. The decisive form of that objection is the jurist's: §6.2 admits F5 as 'qualified span (F5)', a construction that would have admitted 'reported-speech span (F4)' and was in use one item away. Q3: ran the fused-claim test on instance 8's fragments. It goes against retention — fragment 2 opens on the tail of the carpenter's speech with NO attributing clause before reaching Mauss's conclusion. The B4 shape. PENDING-132 amended: the retention is split out, and the retraction re-grounded on two convergent bases so it is authorizable regardless of how 134 resolves. 133: rescoped from two F4-carrying spans to every fr grounded span, because the bound assumed P7's tagging was complete and the item's own diagnosis says it had no claim-side step at all. And the access gap the ruling opened with: governance_read gains v2-harness-design, v2-stratum-tags, mauss-fixture-spans, mauss-fixture-citations — PENDING-86's fourth instance, same shape and same remedy as chamber-spec. The jurist can now verify Part I rather than take it as testimony. Eight controls including that the served text actually carries §6.2's pre-registration clause, §5's F4 row and the L926 citation strings. Self-test 54 checks, 0 failures.
644 lines
34 KiB
Python
644 lines
34 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 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-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"),
|
||
"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`, and its fr ratio "
|
||
"1:9 is under challenge — see PENDING-132/133"),
|
||
"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"),
|
||
"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?'"),
|
||
}
|
||
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}"
|
||
|
||
|
||
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_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
|