dotfiles/claude/governance/fool/reduce.py

#!/usr/bin/env python3
"""
Reduction arm — tile a document into taggable units, and gate every claim about it.

WHY THIS EXISTS
    Control Kernel v1.0 (frozen 2026-08-02, sha256 67c9b870…) requires that every
    sentence of a control document carry exactly one tag. "Every sentence" is only
    meaningful relative to a declared splitter, so the splitter is part of the
    record — §3.1 of the kernel says so explicitly.

    The reduction arm exists to FALSIFY the kernel, not to ratify it. It runs
    before the synthetic arm because a generated corpus can only confirm whatever
    the kernel already believes.

THE TILING INVARIANT
    Spans TILE the document: concatenating every span in order reproduces the
    source byte-for-byte. Nothing is dropped, nothing is silently normalised.
    This is what makes non-destructive quarantine checkable rather than promised —
    laundering a document means changing it, and a change that preserves the
    tiling must appear in the ledger.

    Assertive spans need a tag. Structural spans (blank lines, fences, table rows,
    list bullets) do not, and are marked so the distinction is visible rather than
    implicit.

USAGE
    ./reduce.py split  <doc.md>              → doc.units.jsonl  (+ tiling gate)
    ./reduce.py check  <doc.md> <tags.tsv>   → kernel §3 checks
"""

from __future__ import annotations

import argparse
import hashlib
import json
import re
import sys
from pathlib import Path

# Bumped whenever unit boundaries could change. A tag file is only valid against
# the splitter version that produced its units.
# 1.1.0 — three defects found by contact with a real jurist ruling, not by review:
#   (a) a `##` line INSIDE a fenced block was kinded `heading` and made taggable,
#       because heading was tested before code. Quoted content is not structure.
#   (b) `---` rules were `block` and taggable. A horizontal rule is not a sentence.
#   (c) a `?` inside a quotation split a sentence mid-clause, yielding a FRAGMENT
#       ("…asserts to be true?" | "alone — is less safe…"). Tagging a fragment is
#       meaningless, so it must not be produced.
# 1.2.0 — two further defects, again found by contact rather than review, this
# time on a package rather than a ruling:
#   (d) a numbered marker ("**1.", "2.") was read as a sentence end, orphaning the
#       marker as a fragment and decapitating the sentence after it.
#   (e) YAML frontmatter was treated as flowing prose and shredded mid-key
#       (`…quoted verbatim below." status: "DRAFT.`). Frontmatter is line-oriented.
#       It stays TAGGABLE — it carries real assertions about the document, and
#       excluding it would quietly shrink the quarantine in the author's favour.
SPLITTER_VERSION = "1.2.0"

# The kernel this tool enforces. It was left at v1.0's hash after v1.1 was
# frozen, so every run record printed a provenance line that named the wrong
# governing document — the tool asserting a fact about itself that the substrate
# contradicted, which is the class this whole programme exists to catch.
KERNEL_VERSION = "1.1"
KERNEL_SHA256 = "d4b48db23612b30ff66e26b6235065a3c2f3c9be19d750dafc97e80a1329974d"
KERNEL_FILE = "CONTROL-KERNEL-v1.1.md"

TAGS = {"D", "Q", "A", "N", "X"}

# `!` is NOT a kernel tag. It is the reduction's record that a unit cannot be
# typed under Kernel v1.0 and must therefore leave the sound remainder. Quarantine
# is non-destructive: the unit stays in the units file and in the census, with its
# reason, so what was removed is inspectable rather than silently absent.
QUARANTINE = "!"

QUARANTINE_REASONS = {
    # First-person report of an act performed outside the document. Cannot be
    # demonstrated in-document by construction ("I read ~/CLAUDE.md in full").
    "TESTIMONY",
    # A determination constituted by being uttered, not by being argued
    # ("Design gate PASSED", "AFFIRMED", "Keep both clauses").
    "PERFORMATIVE",
    # Faithfully derived from a §1 axiom source but not verbatim, so it cannot be
    # `Q`; and not argued in-document, so it cannot be `D`.
    "PARAPHRASE",
    # A factual claim about the world or another document, not traceable to a §1
    # source at all.
    "UNSOURCED-FACT",
    # Quotation of a party that is not a §1 axiom source.
    "UNSOURCED-QUOTE",
    # §2c — more than one primitive in a single sentence, unsplittable without
    # editing the source, which the reduction arm may not do silently.
    "BLEND",
    # Rests on a quarantined or `A` unit, so §2's transitivity clause forbids `D`.
    "INHERITED",
}

# Kernel §3.3 — a control document may not collect its caveats into a section.
# Widened after a FALSE PASS on a real package: "Part VII — Disconfirming
# evidence, which the steward specifically asked to be carried" is a collected
# limitations section in §2a's sense — trial 03 showed the model skipping exactly
# that section wholesale — and the original pattern did not match it because it
# never says "limitations".
#
# THIS CHECK IS A SCREEN, NOT A DECISION. No pattern can decide whether a section
# collects the author's own caveats; a section titled "Part VII" alone would defeat
# any wordlist. §2a compliance therefore belongs in Kernel §4's judgement residue,
# and a pass here means only that the obvious namings were absent.
FORBIDDEN_HEADING_RE = re.compile(
    r"limitation|caveat|assumption|what this does not|open question"
    r"|disconfirming|evidence against|weakness|objection|counter-?argument"
    r"|self-?critique|known (?:issue|gap|problem)|shortcoming|scope boundary",
    re.IGNORECASE,
)

# Abbreviations after which a period does NOT end a sentence. Deliberately short:
# every entry is a judgement about English, and this list is part of the trusted
# base in the same way the tags are.
ABBREVIATIONS = {
    "e.g", "i.e", "cf", "vs", "etc", "al", "no", "vol", "pp", "ch",
    "Mr", "Mrs", "Ms", "Dr", "St", "Prof", "Fig", "approx",
}

_SENT_END = re.compile(r"([.!?])([\"'’”\)\]]*)(\s+)")


def sha256(text: str) -> str:
    return hashlib.sha256(text.encode("utf-8")).hexdigest()


def _is_abbrev(text: str, dot_index: int) -> bool:
    """True if the period at dot_index closes a known abbreviation."""
    start = dot_index
    while start > 0 and (text[start - 1].isalnum() or text[start - 1] == "."):
        start -= 1
    return text[start:dot_index].rstrip(".") in ABBREVIATIONS


def _is_enumerator(text: str, dot_index: int) -> bool:
    """
    True if the period at dot_index closes a numbered marker such as `1.` or
    `**2.` rather than a sentence.

    Deliberately narrow: at most two digits, and nothing before them on the line
    except markdown emphasis or whitespace. A bare numeric token is NOT enough —
    "…formalized in 2026. The next…" is a real boundary and must stay one.
    """
    start = dot_index
    while start > 0 and text[start - 1].isdigit():
        start -= 1
    digits = text[start:dot_index]
    if not (1 <= len(digits) <= 2):
        return False
    line_start = text.rfind("\n", 0, start) + 1
    return text[line_start:start].strip(" \t*_>#") == ""


def split_prose(block: str, offset: int) -> list[tuple[int, int]]:
    """
    Split a prose block into sentence spans as (start, end) absolute offsets.

    Spans are CONTIGUOUS and cover the block exactly — trailing whitespace stays
    attached to the sentence it follows, so the tiling invariant holds without a
    separate whitespace span per gap.
    """
    spans: list[tuple[int, int]] = []
    cursor = 0
    for m in _SENT_END.finditer(block):
        dot = m.start(1)
        if block[dot] == "." and (_is_abbrev(block, dot) or _is_enumerator(block, dot)):
            continue
        end = m.end()  # include the closing punctuation and the following space
        # A sentence-ending mark inside a quotation is usually not the end of the
        # sentence: `collapse to "does it change what X asserts?" alone — is less
        # safe` is one sentence, and splitting it produced a fragment. English
        # sentences do not open in lower case, so the following character decides.
        if end < len(block) and block[end].islower():
            continue
        spans.append((offset + cursor, offset + end))
        cursor = end
    if cursor < len(block):
        spans.append((offset + cursor, offset + len(block)))
    return spans


def split_spans(text: str) -> list[dict]:
    """
    Tile `text` into spans. Guarantees sum(spans) == text, byte for byte.

    Line-oriented, because Markdown structure is line-oriented: headings, list
    items, table rows, blank lines and fenced code are decided per line, and only
    paragraph prose is split into sentences.
    """
    spans: list[dict] = []
    pos = 0
    in_fence = False
    lines = text.splitlines(keepends=True)

    # YAML frontmatter: a `---` on the very first line opens it, the next `---`
    # closes it. Line-oriented, so it must not flow into the prose splitter.
    fm_end = -1
    if lines and lines[0].strip() == "---":
        for i in range(1, len(lines)):
            if lines[i].strip() == "---":
                fm_end = i
                break

    para: list[str] = []
    para_start = 0

    def flush_para() -> None:
        nonlocal para, para_start
        if not para:
            return
        block = "".join(para)
        for s, e in split_prose(block, para_start):
            spans.append({"kind": "prose", "start": s, "end": e})
        para = []

    for lineno, line in enumerate(lines):
        stripped = line.strip()

        if 0 < lineno < fm_end:
            flush_para()
            spans.append({"kind": "frontmatter", "start": pos, "end": pos + len(line)})
            pos += len(line)
            continue

        fence = stripped.startswith("```")
        structural = (
            fence
            or in_fence
            or not stripped
            or stripped.startswith("#")
            or stripped.startswith("|")
            or stripped.startswith(">")
            or re.match(r"^\s*([-*+]|\d+\.)\s", line) is not None
            or stripped.startswith("---")
            or stripped.startswith("<!--")
        )
        if structural:
            flush_para()
            # ORDER MATTERS. `code` is tested before `heading`: a `##` line inside
            # a fenced block is quoted content, not a heading of this document.
            # The reverse order made the paste-ready REVIEWED-85 draft's own
            # heading a taggable assertion of the ruling that merely quotes it.
            kind = (
                "code" if (fence or in_fence)
                else "blank" if not stripped
                else "heading" if stripped.startswith("#")
                # A horizontal rule is formatting, not a sentence.
                else "rule" if set(stripped) <= set("-*_") and len(stripped) >= 3
                else "block"
            )
            spans.append({"kind": kind, "start": pos, "end": pos + len(line)})
            if fence:
                in_fence = not in_fence
        else:
            if not para:
                para_start = pos
            para.append(line)
        pos += len(line)
    flush_para()

    spans.sort(key=lambda s: s["start"])
    return spans


def verify_tiling(spans: list[dict], text: str) -> list[str]:
    """
    The gate. Any failure here invalidates every downstream claim about the
    document, so it is reported in full rather than as a boolean.
    """
    problems: list[str] = []
    cursor = 0
    for i, sp in enumerate(spans):
        if sp["start"] != cursor:
            problems.append(
                f"span {i}: gap or overlap — expected start {cursor}, got {sp['start']}"
            )
        cursor = sp["end"]
    if cursor != len(text):
        problems.append(f"tiling ends at {cursor}, document is {len(text)} bytes")
    rebuilt = "".join(text[s["start"]:s["end"]] for s in spans)
    if rebuilt != text:
        problems.append("RECONSTRUCTION FAILED: spans do not reproduce the source")
    return problems


# Spans that carry an assertion and therefore require a tag. Headings are
# INCLUDED: kernel §4 rules that "Why the current approach fails" asserts that it
# fails, so a heading is X only if declarative conversion yields no claim.
TAGGABLE = {"prose", "heading", "block", "frontmatter"}


def load_tags(path: Path) -> dict[int, tuple[str, str]]:
    """Parse `idx<TAB>TAG<TAB>note`, skipping blanks and # comments."""
    out: dict[int, tuple[str, str]] = {}
    for lineno, raw in enumerate(path.read_text(encoding="utf-8").splitlines(), 1):
        if not raw.strip() or raw.lstrip().startswith("#"):
            continue
        parts = raw.split("\t")
        if len(parts) < 2:
            sys.exit(f"FATAL: {path}:{lineno}: expected 'idx<TAB>TAG[<TAB>note]'")
        try:
            idx = int(parts[0])
        except ValueError:
            sys.exit(f"FATAL: {path}:{lineno}: index is not an integer: {parts[0]!r}")
        tag = parts[1].strip()
        if tag != QUARANTINE:
            tag = tag.upper()
        note = parts[2].strip() if len(parts) > 2 else ""
        if tag == QUARANTINE:
            # A quarantine with no reason is an unexplained deletion, which is
            # exactly what non-destructive quarantine exists to prevent.
            reason = note.split(":", 1)[0].strip().upper()
            if reason not in QUARANTINE_REASONS:
                sys.exit(
                    f"FATAL: {path}:{lineno}: quarantine needs a reason from "
                    f"{sorted(QUARANTINE_REASONS)}, got {reason!r}"
                )
        elif tag not in TAGS:
            sys.exit(f"FATAL: {path}:{lineno}: unknown tag {tag!r} (expected {sorted(TAGS)})")
        if idx in out:
            sys.exit(f"FATAL: {path}:{lineno}: duplicate index {idx}")
        out[idx] = (tag, note)
    return out


def cmd_split(doc: Path) -> None:
    text = doc.read_text(encoding="utf-8")
    spans = split_spans(text)
    problems = verify_tiling(spans, text)

    out = doc.with_suffix(".units.jsonl")
    with out.open("w", encoding="utf-8") as fh:
        for i, sp in enumerate(spans):
            rec = {
                "idx": i,
                "kind": sp["kind"],
                "taggable": sp["kind"] in TAGGABLE,
                "start": sp["start"],
                "end": sp["end"],
                "text": text[sp["start"]:sp["end"]],
            }
            fh.write(json.dumps(rec, ensure_ascii=False) + "\n")

    taggable = sum(1 for s in spans if s["kind"] in TAGGABLE)
    print(f"document      {doc.name}  sha256 {sha256(text)[:16]}…")
    print(f"splitter      v{SPLITTER_VERSION}")
    print(f"spans         {len(spans)}  ({taggable} taggable)")
    for kind in ("heading", "prose", "block", "code", "blank"):
        n = sum(1 for s in spans if s["kind"] == kind)
        if n:
            print(f"  {kind:<10} {n}")
    print(f"units         {out.name}")

    if problems:
        print("\nTILING GATE FAILED — every downstream claim is void:")
        for p in problems:
            print(f"  - {p}")
        sys.exit(1)
    print("\nTILING GATE PASSED — spans reproduce the source byte-for-byte.")


_MD_NOISE = re.compile(r"[*_`>]+")
_WS = re.compile(r"\s+")


def normalise_quote(s: str) -> str:
    """
    Normalise for §3.2 containment.

    'Verbatim' is operationalised as: identical after removing markdown emphasis
    and collapsing whitespace. This is WEAKER than byte-identity and is declared
    as such — a blockquote re-wraps its source's lines, and bolding a phrase for
    emphasis is a presentational act, not a change of words. What it does NOT
    tolerate is a changed, added or dropped word, which is the failure §3.2 exists
    to catch.
    """
    s = _MD_NOISE.sub("", s)
    s = s.replace("…", "...").replace("—", "-").replace("–", "-")
    s = s.replace("“", '"').replace("”", '"').replace("’", "'").replace("‘", "'")
    return _WS.sub(" ", s).strip()


def check_q_resolution(
    spans: list[dict], text: str, tags: dict[int, tuple[str, str]], sources: dict[str, Path]
) -> list[str]:
    """
    §3.2 — every `Q` must appear verbatim in a declared §1 source.

    A `Q` whose note names no source, or names one not in the axiom set, fails:
    an unlocatable quotation is exactly the 'quoted but not traced' defect.
    """
    problems: list[str] = []
    cache = {k: normalise_quote(p.read_text(encoding="utf-8")) for k, p in sources.items()}
    for idx, (tag, note) in sorted(tags.items()):
        if tag != "Q":
            continue
        key = note.split(":", 1)[0].strip()
        if key not in cache:
            problems.append(f"§3.2 span {idx}: Q names source {key!r}, not in the axiom set")
            continue
        quoted = normalise_quote(text[spans[idx]["start"]:spans[idx]["end"]].lstrip("> "))
        if quoted and quoted not in cache[key]:
            problems.append(
                f"§3.2 span {idx}: NOT FOUND verbatim in {key} — {quoted[:70]!r}…"
            )
    return problems


# Axiom sources per kernel §1, plus documents a given package names in its header.
AXIOM_SOURCES: dict[str, Path] = {
    "CLAUDE.md": Path.home() / "CLAUDE.md",
    "REVIEWED.md": Path.home() / "REVIEWED.md",
    "contamination-problem.md": Path.home()
    / "_Dev/CapableMind-AI/docs/thinking/David/methodology/contamination-problem.md",
    "central-path.md": Path.home()
    / ".claude/projects/-Users-davidglidden/memory/feedback-central-path-answerability-not-purity.md",
}


def cmd_check(doc: Path, tags_path: Path) -> None:
    text = doc.read_text(encoding="utf-8")
    spans = split_spans(text)
    tiling = verify_tiling(spans, text)
    tags = load_tags(tags_path)

    failures: list[str] = []
    if tiling:
        failures.extend(tiling)

    taggable_idx = {i for i, s in enumerate(spans) if s["kind"] in TAGGABLE}

    # §3.1 — every assertive unit carries exactly one tag.
    untagged = sorted(taggable_idx - set(tags))
    if untagged:
        failures.append(
            f"§3.1 UNTAGGED: {len(untagged)} assertive unit(s) carry no tag: "
            f"{untagged[:12]}{'…' if len(untagged) > 12 else ''}"
        )
    stray = sorted(set(tags) - taggable_idx)
    if stray:
        failures.append(f"§3.1 STRAY TAGS on non-assertive spans: {stray[:12]}")

    # v1.1 §2a/§3.1 — `A` is a diagnostic, not a tag. A control document that
    # carries one is not kernel-sound: needing an assumption means the claim is
    # not derivable from §1, and naming it is precisely what v1.1 forbids.
    # Reduction runs may legitimately show `A`; a CONTROL may not.
    a_tagged = sorted(i for i, (t, _) in tags.items() if t == "A")
    if a_tagged:
        failures.append(
            f"§2a A-FREE VIOLATION: {len(a_tagged)} unit(s) tagged A: {a_tagged[:12]}. "
            "A control document must derive or quote the claim, or §1 must widen."
        )

    # §3.2 — every Q resolves verbatim in a declared axiom source.
    available = {k: p for k, p in AXIOM_SOURCES.items() if p.is_file()}
    missing = sorted(set(AXIOM_SOURCES) - set(available))
    if missing:
        failures.append(f"§1 SOURCE UNRESOLVABLE: {missing}")
    failures.extend(check_q_resolution(spans, text, tags, available))

    # §3.3 — no collected limitations section.
    for i in sorted(taggable_idx):
        if spans[i]["kind"] != "heading":
            continue
        head = text[spans[i]["start"]:spans[i]["end"]]
        if FORBIDDEN_HEADING_RE.search(head):
            failures.append(f"§3.3 FORBIDDEN HEADING at span {i}: {head.strip()!r}")

    counts = {t: sum(1 for t2, _ in tags.values() if t2 == t) for t in sorted(TAGS)}
    quarantined = {i: n for i, (t, n) in tags.items() if t == QUARANTINE}
    sound = len(tags) - len(quarantined)

    reasons: dict[str, int] = {}
    for note in quarantined.values():
        r = note.split(":", 1)[0].strip().upper()
        reasons[r] = reasons.get(r, 0) + 1

    print(f"document   {doc.name}")
    print(f"kernel     v{KERNEL_VERSION} ({KERNEL_FILE}) sha256 {KERNEL_SHA256[:16]}…")
    print(f"splitter   v{SPLITTER_VERSION}")
    print(f"tagged     {len(tags)} of {len(taggable_idx)} assertive units")
    print("counts     " + "  ".join(f"{t}={counts[t]}" for t in sorted(TAGS)))
    print(f"\nsound remainder   {sound}/{len(taggable_idx)} units "
          f"({100 * sound / max(len(taggable_idx), 1):.1f}%)")
    print(f"quarantined       {len(quarantined)}/{len(taggable_idx)} units "
          f"({100 * len(quarantined) / max(len(taggable_idx), 1):.1f}%)")
    if reasons:
        print("\nquarantine census — what real prose does that the kernel cannot type:")
        for r, n in sorted(reasons.items(), key=lambda kv: -kv[1]):
            print(f"  {n:>3}  {r}")

    if failures:
        print("\nKERNEL CHECKS FAILED:")
        for f in failures:
            print(f"  - {f}")
        sys.exit(1)

    print("\nMechanical checks passed: tiling · §3.1 tagging completeness ·")
    print("§3.2 Q-resolution against the declared axiom sources · §3.3 heading screen.")
    print("NOT checked, and NOT checkable: the whole of §4 — whether a D demonstrates,")
    print("an N is inert, an X asserts nothing, a Q sits within its source's scope, a")
    print("sentence carries one primitive. §3.3 is a SCREEN over obvious namings, not a")
    print("decision on §2a. This is not a soundness verdict.")
    if quarantined:
        print(f"\nThe document is NOT kernel-sound as written: {len(quarantined)} units")
        print("cannot be typed under Kernel v1.0. The census above is the finding.")


def main() -> None:
    ap = argparse.ArgumentParser(description="Kernel reduction tooling.")
    sub = ap.add_subparsers(dest="cmd", required=True)
    sp = sub.add_parser("split")
    sp.add_argument("doc", type=Path)
    ck = sub.add_parser("check")
    ck.add_argument("doc", type=Path)
    ck.add_argument("tags", type=Path)
    args = ap.parse_args()

    if args.cmd == "split":
        cmd_split(args.doc)
    else:
        cmd_check(args.doc, args.tags)


if __name__ == "__main__":
    main()