[PROPOSAL] governance-mcp: read-only substrate access for the jurist (PENDING-82)
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
This commit is contained in:
David F Glidden
co-authored by
Claude Opus 5
Claude Opus 5
parent
d12feb5ba9
commit
3df09228c0
@@ -0,0 +1,422 @@
|
||||
#!/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
|
||||
+40
-9
@@ -47,16 +47,39 @@ def read(p):
|
||||
|
||||
# ---------- extractors (pure functions of text, so --selftest can exercise them) ----------
|
||||
|
||||
def item_spans(text):
|
||||
"""-> [(header, start, end)] for every item; 1-based start, end exclusive.
|
||||
|
||||
THE single definition of "an item" for every consumer. Two lessons are welded in:
|
||||
|
||||
- **Any `## ` header is an item.** A family-specific regex (`^## PENDING-<digits>`)
|
||||
hid twenty items on 2026-07-28, ten of them open. Never match on family.
|
||||
- **Fenced blocks are not the document's structure.** Governance drafts are
|
||||
written as plain fenced markdown carrying their own `## REVIEWED-N` headers
|
||||
(the steward's copy-paste-clean practice), so a fenced header would otherwise
|
||||
register as a phantom item AND truncate the real item containing it. Zero such
|
||||
headers exist in the substrate today; the trigger is a drafting habit already
|
||||
in use, so this is a defect waiting on the next draft rather than a hypothetical.
|
||||
"""
|
||||
lines = text.split("\n")
|
||||
heads, fence = [], False
|
||||
for i, l in enumerate(lines):
|
||||
s = l.lstrip()
|
||||
if s.startswith("```") or s.startswith("~~~"):
|
||||
fence = not fence
|
||||
continue
|
||||
if not fence and l.startswith("## "):
|
||||
heads.append((l[3:].strip(), i + 1))
|
||||
return [(h, s, heads[k + 1][1] if k + 1 < len(heads) else len(lines) + 1)
|
||||
for k, (h, s) in enumerate(heads)]
|
||||
|
||||
|
||||
def open_items(text):
|
||||
"""-> [(header, line_no)] for every '## ' item lacking closure evidence.
|
||||
Closure evidence = a REVIEWED-N (resolved by the caller) or CLOSED/COMPLETED
|
||||
in the header. Any '## ' line is an item — assuming one naming family missed
|
||||
twenty items on 2026-07-28."""
|
||||
out = []
|
||||
for i, l in enumerate(text.split("\n"), 1):
|
||||
if l.startswith("## ") and "CLOSED" not in l and not l.startswith("## COMPLETED"):
|
||||
out.append((l[3:].strip(), i))
|
||||
return out
|
||||
"""-> [(header, line_no)] for every item lacking closure evidence. Closure
|
||||
evidence = a REVIEWED-N (resolved by the caller) or CLOSED/COMPLETED in the
|
||||
header."""
|
||||
return [(h, s) for h, s, _e in item_spans(text)
|
||||
if "CLOSED" not in h and not h.startswith("COMPLETED")]
|
||||
|
||||
|
||||
def tag_of(text, header):
|
||||
@@ -290,6 +313,14 @@ def selftest():
|
||||
chk("open_items excludes CLOSED / COMPLETED",
|
||||
open_items("## PENDING-1 — x (CLOSED)\n## COMPLETED — y") == [])
|
||||
chk("open_items returns empty on empty input", open_items("") == [])
|
||||
FENCED = "## PENDING-9 — real\nbody\n```markdown\n## REVIEWED-9 — a draft\n```\ntail"
|
||||
chk("open_items ignores a header inside a fenced block [drafting-habit trigger]",
|
||||
[h for h, _ in open_items(FENCED)] == ["PENDING-9 — real"])
|
||||
chk("open_items DOES find that same header unfenced [positive control for the line above]",
|
||||
[h for h, _ in open_items(FENCED.replace("```markdown\n", "").replace("```\n", ""))]
|
||||
== ["PENDING-9 — real", "REVIEWED-9 — a draft"])
|
||||
chk("item_spans runs a fenced item to the tail, not to the fenced header",
|
||||
item_spans(FENCED)[0][2] == 7)
|
||||
chk("tag_of reads a tag",
|
||||
tag_of("## PENDING-9 — t\n**Date:** d\n**Tag:** [ESCALATE]\n", "PENDING-9 — t") == "[ESCALATE]")
|
||||
chk("tag_of returns '' when absent", tag_of("## PENDING-9 — t\n", "PENDING-9 — t") == "")
|
||||
|
||||
Reference in New Issue
Block a user