path: root/experiments/audit_manuscript.py

#!/usr/bin/env python3
"""Bind publication-facing manuscript claims to audited result sources."""
import argparse
import hashlib
import json
import os
import re


ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
EXPECTED_SECTIONS = [
    "## Abstract",
    "## 1. Introduction",
    "## 2. Somato-dendritic innovation learning",
    "## 3. What residualization guarantees—and what it does not",
    "## 4. Experimental protocol",
    "## 5. Results",
    "## 6. Biological-signature test and outcome-surprise evidence",
    "## 7. Related work",
    "## 8. Limitations and discussion",
    "## 9. Reproducibility statement",
]


def absolute(path):
    return path if os.path.isabs(path) else os.path.join(ROOT, path)


def load_json(path):
    with open(path) as handle:
        return json.load(handle)


def sha256(path):
    digest = hashlib.sha256()
    with open(path, "rb") as handle:
        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def require(condition, message):
    if not condition:
        raise RuntimeError(message)


def pointer_get(value, pointer):
    require(pointer.startswith("/"), f"invalid JSON pointer: {pointer}")
    current = value
    for part in pointer[1:].split("/"):
        part = part.replace("~1", "/").replace("~0", "~")
        if isinstance(current, list):
            current = current[int(part)]
        else:
            require(part in current, f"missing JSON pointer component {part!r}")
            current = current[part]
    return current


def render(value, format_name):
    require(
        isinstance(value, (int, float)) and not isinstance(value, bool),
        f"claim source is not numeric: {value!r}",
    )
    value = float(value)
    if format_name == "fixed0_comma":
        return f"{value:,.0f}"
    if format_name == "fixed1":
        return f"{value:.1f}"
    if format_name == "fixed2":
        return f"{value:.2f}"
    if format_name == "fixed2_percent":
        return f"{value:.2f}%"
    if format_name == "fixed3":
        return f"{value:.3f}"
    if format_name == "fixed3_percent":
        return f"{value:.3f}%"
    if format_name == "fixed4":
        return f"{value:.4f}"
    if format_name == "fixed6":
        return f"{value:.6f}"
    if format_name == "percent3":
        return f"{100 * value:.3f}"
    if format_name == "percent2_percent":
        return f"{100 * value:.2f}%"
    if format_name == "percent3_percent":
        return f"{100 * value:.3f}%"
    if format_name == "abs_percent3":
        return f"{100 * abs(value):.3f}"
    raise RuntimeError(f"unknown claim format: {format_name}")


def false_check_paths(value, prefix="checks"):
    paths = []
    if isinstance(value, dict):
        for key, child in value.items():
            paths.extend(false_check_paths(child, f"{prefix}.{key}"))
    elif value is False:
        paths.append(prefix)
    return paths


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--ledger", default="paper/CLAIM_LEDGER.json")
    parser.add_argument("--out", default="paper/manuscript_audit.json")
    args = parser.parse_args()

    ledger_path = absolute(args.ledger)
    ledger = load_json(ledger_path)
    require(ledger.get("version") == 1, "claim-ledger version drift")
    manuscript_path = absolute(ledger["manuscript"])
    require(os.path.isfile(manuscript_path), "manuscript is missing")
    with open(manuscript_path) as handle:
        manuscript = handle.read()

    require(
        manuscript.startswith(
            "# Learning from the Unexpected: Somato-Dendritic Innovations "
            "for Local Credit Assignment\n"
        ),
        "working title drift",
    )
    for section in EXPECTED_SECTIONS:
        require(manuscript.count(section) == 1, f"section drift: {section}")
    require("TODO" not in manuscript and "TBD" not in manuscript,
            "unresolved manuscript placeholder")

    manuscript_dir = os.path.dirname(manuscript_path)
    expected_figures = {
        os.path.normpath(os.path.join(manuscript_dir, item))
        for item in ledger["figures"]
    }
    linked_figures = {
        os.path.normpath(os.path.join(manuscript_dir, item))
        for item in re.findall(r"!\[[^\]]*\]\(([^)]+)\)", manuscript)
        if not item.startswith(("http://", "https://"))
    }
    require(
        linked_figures == expected_figures,
        "manuscript figure set disagrees with claim ledger",
    )
    for path in sorted(expected_figures):
        require(os.path.isfile(path), f"missing manuscript figure: {path}")

    source_cache = {}
    audited_claims = []
    for claim in ledger["numeric_claims"]:
        source_path = absolute(claim["source"])
        if source_path not in source_cache:
            source_cache[source_path] = load_json(source_path)
        value = pointer_get(source_cache[source_path], claim["pointer"])
        rendered = render(value, claim["format"])
        require(
            rendered == claim["token"],
            f"{claim['id']}: ledger token {claim['token']!r} "
            f"does not match source rendering {rendered!r}",
        )
        require(
            claim["token"] in manuscript,
            f"{claim['id']}: audited number is absent from manuscript",
        )
        audited_claims.append(
            {
                "id": claim["id"],
                "pointer": claim["pointer"],
                "rendered": rendered,
                "source": claim["source"],
            }
        )

    gate_audits = []
    for gate_spec in ledger["gate_statuses"]:
        gate_path = absolute(gate_spec["source"])
        gate = load_json(gate_path)
        require(
            gate.get("status") == gate_spec["expected"],
            f"{gate_spec['source']}: gate status drift",
        )
        false_paths = false_check_paths(gate.get("checks", {}))
        if gate_spec["expected"] == "passed":
            require(not false_paths, f"{gate_spec['source']}: passed gate has false check")
        else:
            require(false_paths, f"{gate_spec['source']}: failed gate has no false check")
        gate_audits.append(
            {
                "false_checks": false_paths,
                "source": gate_spec["source"],
                "status": gate["status"],
            }
        )

    manuscript_flat = re.sub(r"\s+", " ", manuscript)
    for sentence in ledger["required_boundary_text"]:
        require(
            re.sub(r"\s+", " ", sentence) in manuscript_flat,
            f"required claim boundary absent: {sentence}",
        )
    for url in ledger["required_reference_urls"]:
        require(url in manuscript, f"required primary reference absent: {url}")

    source_paths = sorted(
        {absolute(item["source"]) for item in ledger["numeric_claims"]}
        | {absolute(item["source"]) for item in ledger["gate_statuses"]}
    )
    output = {
        "audited_numeric_claims": audited_claims,
        "figures": [
            {
                "path": os.path.relpath(path, ROOT),
                "sha256": sha256(path),
            }
            for path in sorted(expected_figures)
        ],
        "gates": gate_audits,
        "ledger": {
            "path": os.path.relpath(ledger_path, ROOT),
            "sha256": sha256(ledger_path),
        },
        "manuscript": {
            "path": os.path.relpath(manuscript_path, ROOT),
            "sha256": sha256(manuscript_path),
            "word_count": len(re.findall(r"\b[\w'-]+\b", manuscript)),
        },
        "sources": [
            {
                "path": os.path.relpath(path, ROOT),
                "sha256": sha256(path),
            }
            for path in source_paths
        ],
        "status": "passed",
        "strict": True,
    }
    out_path = absolute(args.out)
    os.makedirs(os.path.dirname(out_path), exist_ok=True)
    with open(out_path, "w") as handle:
        json.dump(output, handle, indent=2, sort_keys=True)
        handle.write("\n")
    print(
        json.dumps(
            {
                "false_gate_checks_retained": sum(
                    len(item["false_checks"]) for item in gate_audits
                ),
                "numeric_claims": len(audited_claims),
                "status": output["status"],
                "word_count": output["manuscript"]["word_count"],
            },
            indent=2,
            sort_keys=True,
        )
    )


if __name__ == "__main__":
    main()