check_deliverable_contract.py

#!/usr/bin/env python3
r"""Delivery gate: the full-text `.docx` the venue asked for actually exists.

Why this exists
---------------
`assemble_manuscript_docx.py` *writes* the Word deliverable. This checker is the
read-only counterpart, and the split is deliberate for the same reason
`check_table_style.py` is separate from `make_three_line_tables.py`: a gate that
only asks "did our own writer run?" proves nothing about a file someone
regenerated by hand, converted in Word, or copied in from a co-author.

So everything here is re-derived from the shipped artifact:

  - the file exists at the path the run declared;
  - it carries **at least as many tables and figures as the manuscript includes**
    — the silent-drop failure mode, where a conversion loses an exhibit and the
    paper ships a result nobody can see;
  - it contains real prose, not just floating exhibits;
  - it has no leftover `\input{}`, `??` cross-reference, citation placeholder or
    assembler `[UNRESOLVED …]` marker;
  - and the state file's claims about it are not greener than the file on disk.

The last one matters most. `manuscript.docx_status = verified` is a claim, and
until this checker exists nothing stopped a run from writing it beside a `.docx`
that was never built.

Not every run owes a Word file. A venue that takes LaTeX only sets
`manuscript.docx_status = not-required`, which is honoured — but it must be
*declared*, not inferred from the file's absence.

Usage:
    python3 scripts/check_deliverable_contract.py <workspace>
    python3 scripts/check_deliverable_contract.py <workspace> --json
    python3 scripts/check_deliverable_contract.py <workspace> --strict
    python3 scripts/check_deliverable_contract.py --selftest

Exit code is non-zero iff a hard violation is found.
"""

from __future__ import annotations

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

HERE = Path(__file__).resolve().parent
sys.path.insert(0, str(HERE))

from assemble_manuscript_docx import (  # noqa: E402  (path set above by design)
    CHAIN,
    count_docx_exhibits,
    docx_text,
    find_source,
    parse_body,
    scan_unresolved,
    assembly_inputs,
    declared_format,
)

FAIL = "FAIL"
WARN = "WARN"
INFO = "INFO"
OKAY = "OK"

# The prose floor is expressed in *English-equivalent* characters. A CJK codepoint
# carries roughly the information of an English word, so a genuine Chinese section
# that would clear this bar in English is only ~a quarter as many characters. A
# floor calibrated on English alone rejects real 中文 manuscripts, which is the
# opposite of what this check is for.
MIN_PROSE_CHARS = 400
CJK_WEIGHT = 3.5
MIN_PROSE_LINE_CHARS = 20
NOT_REQUIRED = "not-required"
VALID_STATUS = {"pending", "assembled", "verified", NOT_REQUIRED}


class Report:
    def __init__(self) -> None:
        self.rows: list[tuple[str, str, str]] = []

    def add(self, level: str, check: str, detail: str) -> None:
        self.rows.append((level, check, detail))

    @property
    def failures(self) -> list[tuple[str, str, str]]:
        return [r for r in self.rows if r[0] == FAIL]

    def to_dict(self) -> dict:
        return {"ok": not self.failures,
                "checks": [{"level": l, "check": c, "detail": d} for l, c, d in self.rows]}

    def render(self) -> str:
        width = max((len(c) for _, c, _ in self.rows), default=4)
        lines = ["", "Paper-WorkFlow deliverable contract", "=" * 64]
        for level, check, detail in self.rows:
            lines.append(f"[{level:<4}] {check:<{width}}  {detail}")
        lines.append("=" * 64)
        if self.failures:
            lines.append(f"RESULT: {len(self.failures)} hard violation(s) -> deliverable NOT verified")
        else:
            lines.append("RESULT: the deliverable the venue asked for is on disk and complete")
        return "\n".join(lines)


# --------------------------------------------------------------------------- #
# helpers                                                                      #
# --------------------------------------------------------------------------- #
def load_state(workspace: Path) -> dict:
    for rel in ("00_meta/workflow_state.json", "workflow_state.json"):
        path = workspace / rel
        if path.is_file():
            try:
                return json.loads(path.read_text(encoding="utf-8"))
            except json.JSONDecodeError:
                return {}
    return {}


def expected_exhibits(workspace: Path) -> tuple[int, int, str]:
    """(tables, figures, source) the manuscript body asks for."""
    source = find_source(workspace, declared_format(load_state(workspace)))
    if source is None:
        return 0, 0, ""
    blocks = parse_body(source)
    tables = sum(1 for b in blocks if b.kind in {"table", "include_table"})
    figures = sum(1 for b in blocks if b.kind == "include_figure")
    return tables, figures, str(source.relative_to(workspace))


def _is_cjk(ch: str) -> bool:
    code = ord(ch)
    return (0x3400 <= code <= 0x9FFF          # CJK unified ideographs (+ ext A)
            or 0xF900 <= code <= 0xFAFF       # compatibility ideographs
            or 0x3040 <= code <= 0x30FF       # kana
            or 0xAC00 <= code <= 0xD7AF)      # hangul syllables


def prose_chars(text: str) -> int:
    """English-equivalent characters of running prose.

    Short lines are dropped because table cells and captions arrive as their own
    paragraphs; CJK codepoints are weighted up so the same floor means the same
    amount of writing in Chinese as in English.
    """
    total = 0.0
    for raw in text.splitlines():
        line = raw.strip()
        if len(line) < MIN_PROSE_LINE_CHARS:
            continue
        total += sum(CJK_WEIGHT if _is_cjk(ch) else 1 for ch in line)
    return int(total)


# --------------------------------------------------------------------------- #
# checks                                                                       #
# --------------------------------------------------------------------------- #
def evaluate(workspace: Path, *, strict: bool = False) -> Report:
    rep = Report()
    if not workspace.is_dir():
        rep.add(FAIL, "workspace", f"workspace not found: {workspace}")
        return rep

    state = load_state(workspace)
    manuscript = state.get("manuscript") if isinstance(state.get("manuscript"), dict) else {}
    if not manuscript:
        rep.add(FAIL if strict else INFO, "state",
                "no manuscript block in workflow_state.json (schema < 14) — "
                "deliverable contract not yet declared")
        return rep

    status = str(manuscript.get("docx_status") or "pending")
    if status not in VALID_STATUS:
        rep.add(FAIL, "state:status",
                f"manuscript.docx_status={status!r} is not one of {sorted(VALID_STATUS)}")
        return rep

    fmt = str(manuscript.get("format") or "")
    if fmt not in {"latex", "markdown"}:
        rep.add(WARN, "state:format",
                f"manuscript.format={fmt!r} is undeclared; Stage 0 should pin it to "
                "'latex' or 'markdown' (Word-bound runs should author in markdown)")

    if status == NOT_REQUIRED:
        rep.add(INFO, "scope",
                "manuscript.docx_status=not-required — this venue takes LaTeX only; "
                "no Word deliverable is owed")
        if str(manuscript.get("deliverable") or "") == "docx":
            rep.add(FAIL, "scope:contradiction",
                    "manuscript.deliverable=docx but docx_status=not-required — the run "
                    "both owes and disclaims the Word file")
        return rep

    rel = str(manuscript.get("deliverable_docx") or "09_submission/main.docx")
    path = workspace / rel
    if not path.resolve().is_relative_to(workspace.resolve()):
        rep.add(FAIL, "deliverable:path", "deliverable must be inside the workspace")
        return rep

    want_tables, want_figures, source = expected_exhibits(workspace)
    if source:
        rep.add(INFO, "source", f"{source}: {want_tables} exhibit include(s), {want_figures} figure(s)")

    if not path.is_file():
        level = FAIL if (strict or status in {"assembled", "verified"}) else WARN
        rep.add(level, "deliverable",
                f"{rel} does not exist — build it with "
                f"`python3 scripts/assemble_manuscript_docx.py {workspace}`"
                + ("" if level == FAIL else " (advisory until --strict or docx_status advances)"))
        return rep

    try:
        with zipfile.ZipFile(path) as archive:
            if "word/document.xml" not in archive.namelist():
                rep.add(FAIL, "deliverable", f"{rel} is a zip but not a Word document")
                return rep
    except (OSError, zipfile.BadZipFile):
        rep.add(FAIL, "deliverable", f"{rel} is not a readable .docx")
        return rep

    tables, figures = count_docx_exhibits(path)
    text = docx_text(path)

    actual_source = find_source(workspace, declared_format(state))
    if actual_source is None:
        rep.add(FAIL if strict else WARN, "source", "no manuscript source available to verify completeness")
    inputs = manuscript.get("assembly_inputs")
    if isinstance(inputs, dict) and inputs and actual_source:
        try:
            current = assembly_inputs(workspace, actual_source, manuscript)
        except (OSError, ValueError) as exc:
            rep.add(FAIL, "freshness:inputs", str(exc))
        else:
            if current != inputs:
                rep.add(FAIL, "freshness:inputs", "source, exhibit, bibliography or style changed; reassemble the DOCX")
        digest = hashlib.sha256(path.read_bytes()).hexdigest()
        if digest != manuscript.get("assembled_docx_sha256"):
            rep.add(FAIL, "freshness:output", "DOCX changed since assembly; reconcile edits with the source and reassemble")
    else:
        rep.add(FAIL if strict else WARN, "freshness:inputs", "no conversion input manifest; reassemble before verification")
    declared_markers = manuscript.get("unresolved_markers")
    if isinstance(declared_markers, list) and declared_markers:
        rep.add(FAIL, "conversion:unresolved", "; ".join(map(str, declared_markers)))

    rep.add(OKAY, "deliverable", f"{rel}: {tables} table(s), {figures} figure(s), {len(text)} chars of text")

    if want_tables and tables < want_tables:
        rep.add(FAIL, "exhibits:tables",
                f"{rel} carries {tables} table(s) but the manuscript includes {want_tables} — "
                "exhibit(s) were lost in conversion; re-run the assembler and check its "
                "unresolved-marker report")
    elif want_tables:
        rep.add(OKAY, "exhibits:tables", f"all {want_tables} manuscript exhibit(s) present in the .docx")

    if want_figures and figures < want_figures:
        rep.add(FAIL, "exhibits:figures",
                f"{rel} carries {figures} figure(s) but the manuscript includes {want_figures} — "
                "figure(s) were lost in conversion (a .pdf-only figure cannot be embedded; "
                "export a .png alongside it)")
    elif want_figures:
        rep.add(OKAY, "exhibits:figures", f"all {want_figures} manuscript figure(s) present in the .docx")

    body = prose_chars(text)
    if body < MIN_PROSE_CHARS:
        rep.add(FAIL if strict else WARN, "prose",
                f"{rel} holds only {body} chars of running prose — a deliverable that is "
                "exhibits with no paper around them is a conversion failure, not a manuscript"
                + ("" if strict else " (advisory until --strict)"))
    else:
        rep.add(OKAY, "prose", f"{body} chars of running prose")

    leftovers = scan_unresolved(text)
    if leftovers:
        rep.add(FAIL, "unresolved",
                f"{rel} still contains: " + "; ".join(leftovers)
                + " — fix the source or the exhibit path, then re-assemble")
    else:
        rep.add(OKAY, "unresolved", "no unresolved includes, cross-references or placeholders")

    # State honesty: the file on disk is the authority, the state file is a claim.
    declared = manuscript.get("unresolved_markers")
    if isinstance(declared, list) and not declared and leftovers:
        rep.add(FAIL, "state:markers",
                "manuscript.unresolved_markers is empty but the .docx still carries "
                f"{len(leftovers)} unresolved marker class(es)")
    for key, actual, label in (("exhibits_embedded", tables, "table"),
                               ("figures_embedded", figures, "figure")):
        claimed = manuscript.get(key)
        if isinstance(claimed, int) and claimed > actual:
            rep.add(FAIL, f"state:{key}",
                    f"state claims {claimed} embedded {label}(s) but the .docx has {actual}")

    if status == "verified" and rep.failures:
        rep.add(FAIL, "state:verified",
                "manuscript.docx_status=verified while this gate is red — a verified claim "
                "must be earned by the artifact, not written into state")
    elif status == "verified":
        rep.add(OKAY, "state:verified", "docx_status=verified is consistent with the artifact")
    elif not rep.failures:
        rep.add(INFO, "state:verified",
                f"docx_status={status!r}: every check passed, so this run may now set "
                "manuscript.docx_status=verified")

    pack = state.get("replication_pack") if isinstance(state.get("replication_pack"), dict) else {}
    if str(pack.get("status") or "") == "ready" and status not in {"verified", NOT_REQUIRED}:
        rep.add(FAIL, "ordering:replication_pack",
                f"replication_pack.status=ready while manuscript.docx_status={status!r} — "
                "a submission package cannot be ready while the deliverable it ships is "
                "unbuilt or unchecked")

    return rep


# --------------------------------------------------------------------------- #
# selftest                                                                     #
# --------------------------------------------------------------------------- #
_TEX = r"""
\documentclass{article}
\begin{document}
\section{Results}
The treatment raises log wages by 0.123 (s.e. 0.041), a magnitude that is large
relative to the sample mean and stable across every specification we estimate in
this section, including the ones that absorb region-by-year variation entirely.
\begin{table}[htbp]
\caption{Main results}
\input{results/table2}
\end{table}
\section{Conclusion}
The effect is robust to the full battery of alternative specifications reported
in the appendix, and it survives the placebo tests described immediately above.
Taken together, the evidence points to a single channel rather than the bundle of
mechanisms an earlier literature proposed, and the magnitude is stable whether the
comparison group is built from never-treated units or from not-yet-treated ones.
\end{document}
"""

_TABLE = r"""
\begin{tabular}{lc}
\toprule
Variable & (1) \\
\midrule
treat & 0.123*** \\
N & 12,345 \\
\bottomrule
\end{tabular}
"""


def _workspace(root: Path, *, state: dict) -> Path:
    ws = root / "ws"
    for rel in ("05_draft", "04_results", "09_submission", "00_meta"):
        (ws / rel).mkdir(parents=True, exist_ok=True)
    (ws / "05_draft" / "main.tex").write_text(_TEX, encoding="utf-8")
    (ws / "04_results" / "table2.tex").write_text(_TABLE, encoding="utf-8")
    (ws / "00_meta" / "workflow_state.json").write_text(
        json.dumps(state, ensure_ascii=False), encoding="utf-8")
    return ws


def _base_state(**manuscript: object) -> dict:
    block = {"format": "latex", "deliverable": "docx",
             "deliverable_docx": "09_submission/main.docx", "docx_status": "pending",
             "converter": "", "reference_docx": "", "csl": "",
             "exhibits_embedded": 0, "figures_embedded": 0,
             "unresolved_markers": [], "last_assembly": ""}
    block.update(manuscript)
    return {"schema_version": 14, "manuscript": block}


def _levels(rep: Report, prefix: str) -> list[str]:
    return [lvl for lvl, chk, _ in rep.rows if chk.startswith(prefix)]


def selftest() -> None:
    import assemble_manuscript_docx as assembler

    failures: list[str] = []

    def check(condition: bool, message: str) -> None:
        if not condition:
            failures.append(message)

    # 1. A real assembled deliverable passes.
    with tempfile.TemporaryDirectory(prefix="pw-deliv-ok-") as tmp:
        ws = _workspace(Path(tmp), state=_base_state())
        assembler.run(ws, converter="builtin")
        rep = evaluate(ws, strict=True)
        check(not rep.failures, "a correctly assembled deliverable was rejected: "
                                + "; ".join(f"{c}:{d}" for _, c, d in rep.failures))

    # 2. A missing .docx is a hard failure once the run claims it was assembled.
    with tempfile.TemporaryDirectory(prefix="pw-deliv-missing-") as tmp:
        ws = _workspace(Path(tmp), state=_base_state(docx_status="assembled"))
        rep = evaluate(ws)
        check(FAIL in _levels(rep, "deliverable"),
              "a claimed-but-absent deliverable did not fail")

    # 3. A dropped exhibit is caught by recounting, not by trusting state.
    with tempfile.TemporaryDirectory(prefix="pw-deliv-drop-") as tmp:
        ws = _workspace(Path(tmp), state=_base_state())
        assembler.run(ws, converter="builtin")
        docx = ws / "09_submission" / "main.docx"
        with zipfile.ZipFile(docx) as archive:
            parts = {n: archive.read(n) for n in archive.namelist()}
        stripped = re.sub(rb"<w:tbl>.*?</w:tbl>", b"", parts["word/document.xml"], flags=re.DOTALL)
        parts["word/document.xml"] = stripped
        with zipfile.ZipFile(docx, "w", zipfile.ZIP_DEFLATED) as archive:
            for name, data in parts.items():
                archive.writestr(name, data)
        rep = evaluate(ws)
        check(FAIL in _levels(rep, "exhibits:tables"),
              "a table dropped from the .docx was not detected")

    # 4. A leftover unresolved marker blocks the gate.
    with tempfile.TemporaryDirectory(prefix="pw-deliv-unres-") as tmp:
        ws = _workspace(Path(tmp), state=_base_state())
        assembler.run(ws, converter="builtin")
        docx = ws / "09_submission/main.docx"
        with zipfile.ZipFile(docx) as archive:
            parts = {name: archive.read(name) for name in archive.namelist()}
        parts["word/document.xml"] = parts["word/document.xml"].replace(
            b"</w:body>", b"<w:p><w:r><w:t>[UNRESOLVED TABLE]</w:t></w:r></w:p></w:body>")
        with zipfile.ZipFile(docx, "w", zipfile.ZIP_DEFLATED) as archive:
            for name, data in parts.items():
                archive.writestr(name, data)
        rep = evaluate(ws)
        check(FAIL in _levels(rep, "unresolved"),
              "an [UNRESOLVED …] marker in the .docx did not fail the gate")

    # 5. `verified` cannot be asserted over a red artifact.
    with tempfile.TemporaryDirectory(prefix="pw-deliv-lie-") as tmp:
        ws = _workspace(Path(tmp), state=_base_state(docx_status="verified"))
        rep = evaluate(ws)
        check(FAIL in _levels(rep, "deliverable") or FAIL in _levels(rep, "state:verified"),
              "docx_status=verified passed with no .docx on disk")

    # 6. A declared not-required venue is honoured.
    with tempfile.TemporaryDirectory(prefix="pw-deliv-na-") as tmp:
        ws = _workspace(Path(tmp), state=_base_state(docx_status="not-required", deliverable="tex"))
        rep = evaluate(ws, strict=True)
        check(not rep.failures, "an explicitly not-required deliverable was rejected")

    # 7. …but not when the same run also says it owes a .docx.
    with tempfile.TemporaryDirectory(prefix="pw-deliv-contra-") as tmp:
        ws = _workspace(Path(tmp), state=_base_state(docx_status="not-required", deliverable="docx"))
        rep = evaluate(ws)
        check(FAIL in _levels(rep, "scope:contradiction"),
              "a run that both owes and disclaims the Word file was not flagged")

    # 8. A ready replication pack requires a settled deliverable.
    with tempfile.TemporaryDirectory(prefix="pw-deliv-order-") as tmp:
        state = _base_state(docx_status="assembled")
        state["replication_pack"] = {"status": "ready"}
        ws = _workspace(Path(tmp), state=state)
        assembler.run(ws, converter="builtin")
        # the assembler resets docx_status to "assembled"; that is the point
        rep = evaluate(ws)
        check(FAIL in _levels(rep, "ordering:replication_pack"),
              "replication_pack=ready passed over an unverified deliverable")

    # 9. A deliverable that is exhibits with no paper around them fails --strict.
    with tempfile.TemporaryDirectory(prefix="pw-deliv-prose-") as tmp:
        ws = _workspace(Path(tmp), state=_base_state())
        (ws / "05_draft" / "main.tex").write_text(
            "\\documentclass{article}\n\\begin{document}\n"
            "\\begin{table}\\caption{Main}\\input{results/table2}\\end{table}\n"
            "\\end{document}\n", encoding="utf-8")
        assembler.run(ws, converter="builtin")
        rep = evaluate(ws, strict=True)
        check(FAIL in _levels(rep, "prose"),
              "a .docx with no running prose passed --strict")
        rep_lenient = evaluate(ws, strict=False)
        check(WARN in _levels(rep_lenient, "prose"),
              "the prose floor should stay advisory without --strict")

    # 9b. A Chinese manuscript of equivalent substance must clear the same floor.
    #     The floor is English-calibrated; without CJK weighting this is a false
    #     positive on every 中文期刊 run, which is most of this skill's traffic.
    zh = ("本文利用某项试点政策的分批推行，识别其对企业全要素生产率的因果效应。"
          "基准估计显示，政策使处理组企业的对数产出显著提高，且效应在政策实施后"
          "逐年增强，事件研究图未显示任何预期效应，平行趋势假设在政策实施前各期"
          "均无法被拒绝。\n"
          "机制检验表明，效应主要通过缓解融资约束而非降低要素成本实现。结论对剔除"
          "最大省份、替换聚类层级、改变窗口宽度与一系列安慰剂检验均保持稳健，异质性"
          "分析显示效应集中于民营企业与制造业企业。")
    assert prose_chars(zh) >= MIN_PROSE_CHARS, (
        f"a real Chinese section scored {prose_chars(zh)} < {MIN_PROSE_CHARS}: "
        "the prose floor is rejecting 中文 manuscripts")
    check(prose_chars("| treat | 0.12 |\n| N | 100 |") < MIN_PROSE_CHARS,
          "table cells alone should not count as running prose")

    # 10. State cannot claim more exhibits than the file holds.
    with tempfile.TemporaryDirectory(prefix="pw-deliv-claim-") as tmp:
        ws = _workspace(Path(tmp), state=_base_state())
        assembler.run(ws, converter="builtin")
        path = ws / "00_meta" / "workflow_state.json"
        state = json.loads(path.read_text(encoding="utf-8"))
        state["manuscript"]["exhibits_embedded"] = 99
        path.write_text(json.dumps(state, ensure_ascii=False), encoding="utf-8")
        rep = evaluate(ws)
        check(FAIL in _levels(rep, "state:exhibits_embedded"),
              "an inflated exhibits_embedded claim was not caught")

    if failures:
        print("FAIL: check_deliverable_contract selftest\n  - " + "\n  - ".join(failures),
              file=sys.stderr)
        raise SystemExit(1)
    print("selftest OK: deliverable-contract invariants hold")


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(
        prog="check_deliverable_contract.py",
        description="Verify that the full-text .docx deliverable exists, carries every "
                    "exhibit the manuscript includes, and is not over-claimed in state.",
        epilog="Build it first with scripts/assemble_manuscript_docx.py.",
    )
    parser.add_argument("workspace", nargs="?", help="paper workspace root")
    parser.add_argument("--strict", action="store_true",
                        help="advisory tiers become hard failures (Stage 9)")
    parser.add_argument("--json", action="store_true", help="machine-readable output")
    parser.add_argument("--selftest", action="store_true", help="verify this checker")
    args = parser.parse_args(argv)

    if args.selftest:
        selftest()
        return 0
    if not args.workspace:
        parser.error("workspace is required (or pass --selftest)")

    rep = evaluate(Path(args.workspace).expanduser().resolve(), strict=args.strict)
    print(json.dumps(rep.to_dict(), ensure_ascii=False, indent=2) if args.json else rep.render())
    return 1 if rep.failures else 0


if __name__ == "__main__":
    raise SystemExit(main())