#!/usr/bin/env python3
"""Reproducible, read-only audit of the 6,297 -> 20,047 sentence goal.

The only writes this program performs are its JSON result and, when explicitly
requested, a byte-for-byte capture of the public documentation endpoints.
"""

from __future__ import annotations

import argparse
import ast
import collections
import datetime as dt
import hashlib
import json
import os
from pathlib import Path
import re
import sqlite3
import subprocess
import sys
import time
from typing import Any, Iterable

import yaml


PROGRAM = Path(__file__).resolve().parents[1]
SYNTH = PROGRAM / "synthetic/claude-synthetic-v1-2026-07-02"
DB = SYNTH / "synthetic.db"
BATCH_SCRIPTS = SYNTH / "batch_scripts"
EXPORT = SYNTH / "export"
REPORTS = PROGRAM / "reports"
PUBLIC_DOCS = PROGRAM / "public-docs"
LIVE_DIR = SYNTH / "live_audits/completion-20000"
DICTIONARY = Path(
    "/mnt/donto-data/workspace/mobtranslate.com/"
    "dictionaries/kuku_yalanji/dictionary.yaml"
)
GRAMMAR = Path(
    "/mnt/donto-data/workspace/mobtranslate.com/"
    "dictionaries/kuku_yalanji/grammar.md"
)
LEXICAL_REPORT = PROGRAM / "completion_audits/lexical-source-coverage-2026-07-10.md"
COMPLETION_REPORT = PROGRAM / "completion_audits/KUKU-YALANJI-20000-COMPLETION-AUDIT-2026-07-10.md"
RUNPOD_HANDOFF = PROGRAM / "docs/kuku-yalanji-v21-parallel-runpod-handoff-2026-07-10.md"

START_BATCH = 125
END_BATCH = 399
BASELINE = 6_297
SCOPE_ROWS = 13_750
FINAL_ROWS = 20_047
FINAL_WORDS = 135_547
LENSES = ("morphology", "lexicon", "syntax", "fidelity", "register")
REMEDIATED_BATCHES = (239, 240, 241, 329)

SOURCE_HASHES = {
    DICTIONARY: "b575ab035ecc60f09f7bfc5537116555f3d290869fe75b2edf0fd8db60fe1b37",
    GRAMMAR: "5f21305f1fca9ee2cb64668e18156d1f322e3bd9e59850fda662bf0673714153",
}
EXPORT_HASHES = {
    "MANIFEST.json": "66f37fc8099406c98ddd7a35a055c49586d60db2cacbafae87ad6d8e47f3f5a4",
    "train.jsonl": "1945595c24b11b4d1ba4da14df38eb09e8f03891307befad8e0723c2efa4fe20",
    "dev.jsonl": "25ace6e437bb945454c35e50768f6a03d5e61a0d11bd136fa84a8b3d06101481",
    "synthtest.jsonl": "af697ff339dd89115cf79f50f99dec868481f7520bd3149355aaa6435e11fe2d",
}
ARTIFACT_HASHES = {
    "batch_scripts/remediate_literal_authorship_239_240_241_329.py":
        "1af79c64cf95081ac39a1df12b43b98b6fd86349329a8cb66e0a9e59bc81afc2",
    "batch_scripts/backfill_process_events_125_140.py":
        "99ffb17708b64d2388be83ab5eb8b995761ea0093e89d94044c8254fc53cb41d",
    "batch_scripts/bank_239.py":
        "eb33c9d200be2a97dd20358360a4e210ab7033ca9eedd731b5f6aaff797055e8",
    "batch_scripts/bank_240.py":
        "88075cc5ff23b5f86df082bb8d93961f91c948f55dd90b0a6a903a964cf9e764",
    "batch_scripts/bank_241.py":
        "5d4babdfc52c9ccaf6c7e454be7e1aee034283481b5b7b6af3d1f04add7f6be4",
    "batch_scripts/bank_329.py":
        "99730e3dd9f371bafaa03744df05fd04b60e4f9c211728bd849dc905e3e349cc",
    "process_logs/literal-authorship-remediation-batches-239-240-241-329.md":
        "93a0e20bd002d9a6f675cf6750af0afd3084ed3b2e200de2588178dee8c2b6ff",
    "process_logs/retrospective-process-backfill-batches-125-140.md":
        "bb7b03a0ffe3fc43be3db545bbe616e8416f6de29010ea92551d6725cd68a343",
}
DOCS = {
    "operator-guide": "KUKU-YALANJI-CORPUS-OPERATOR-GUIDE.md",
    "sample-pages": "SAMPLE-PRODUCTION-PAGES.md",
    "grammar-cheatsheet": "kuku-yalanji-mega-grammar-cheatsheet-2026-07-02.md",
    "dictionary-errata": "kuku-yalanji-dictionary-errata-2026-07-02.md",
}
LIVE_FILES = (
    "index.html",
    "operator-guide.html",
    "operator-guide.md",
    "sample-pages.html",
    "sample-pages.md",
    "grammar-cheatsheet.html",
    "grammar-cheatsheet.md",
    "dictionary-errata.html",
    "dictionary-errata.md",
)


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


def canonical_hash(rows: Iterable[Iterable[Any]]) -> str:
    digest = hashlib.sha256()
    for row in rows:
        payload = json.dumps(list(row), ensure_ascii=False, separators=(",", ":"))
        digest.update(payload.encode("utf-8"))
        digest.update(b"\n")
    return digest.hexdigest()


class Audit:
    def __init__(self) -> None:
        self.checks: list[dict[str, Any]] = []
        self.metrics: dict[str, Any] = {}
        self.hashes: dict[str, str] = {}

    def check(self, name: str, passed: bool, detail: Any) -> None:
        self.checks.append({"name": name, "status": "PASS" if passed else "FAIL", "detail": detail})

    @property
    def failures(self) -> list[dict[str, Any]]:
        return [item for item in self.checks if item["status"] == "FAIL"]


def batch_number(label: str) -> int:
    return int(label.removeprefix("batch-"))


def assignment_value(tree: ast.Module, name: str) -> ast.AST | None:
    found: ast.AST | None = None
    for node in tree.body:
        if not isinstance(node, (ast.Assign, ast.AnnAssign)):
            continue
        targets = node.targets if isinstance(node, ast.Assign) else [node.target]
        if any(isinstance(target, ast.Name) and target.id == name for target in targets):
            found = node.value
    return found


def literal_rows(path: Path, name: str = "ROWS") -> tuple[list[tuple[int, str, str, str]], list[str]]:
    tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
    value = assignment_value(tree, name)
    issues: list[str] = []
    if not isinstance(value, (ast.List, ast.Tuple)):
        return [], [f"{name} is not a literal list/tuple"]
    records: list[tuple[int, str, str, str]] = []
    for position, element in enumerate(value.elts, 1):
        fields: list[ast.AST]
        if isinstance(element, (ast.Tuple, ast.List)):
            fields = list(element.elts)
        elif (
            isinstance(element, ast.Call)
            and isinstance(element.func, ast.Name)
            and element.func.id == "row"
            and not element.keywords
        ):
            fields = list(element.args)
        else:
            issues.append(f"row {position}: runtime/nonliteral row shape")
            continue
        if len(fields) < 4:
            issues.append(f"row {position}: fewer than four fields")
            continue
        try:
            seq, english, kuku, analysis = (ast.literal_eval(field) for field in fields[:4])
        except Exception as exc:
            issues.append(f"row {position}: nonliteral authorship field ({exc})")
            continue
        if seq != position or not all(isinstance(x, str) and x for x in (english, kuku, analysis)):
            issues.append(f"row {position}: malformed sequence or text field")
            continue
        records.append((seq, english, kuku, analysis))
    if len(value.elts) != 50:
        issues.append(f"inventory has {len(value.elts)} rows, expected 50")
    return records, issues


def dictionary_heads(document: Any) -> set[str]:
    heads: set[str] = set()
    if isinstance(document, dict):
        word = document.get("word")
        if isinstance(word, str) and word.strip():
            heads.add(word.strip().lower())
        for value in document.values():
            heads.update(dictionary_heads(value))
    elif isinstance(document, list):
        for value in document:
            heads.update(dictionary_heads(value))
    return heads


def resolve_artifact(raw: str | None) -> bool:
    if not raw:
        return False
    path = Path(raw)
    candidates = [path] if path.is_absolute() else [SYNTH / path, PROGRAM / path]
    return any(candidate.exists() for candidate in candidates)


def fetch_live_docs(audit: Audit) -> dict[str, Any]:
    LIVE_DIR.mkdir(parents=True, exist_ok=True)
    nonce = f"{int(time.time())}"
    user_agent = "Mozilla/5.0 (compatible; MobTranslateMilestoneAudit/1.0)"
    route_url = f"https://mobtranslate.com/docs?milestone_audit={nonce}"
    route = subprocess.run(
        [
            "curl", "-sS", "--max-time", "90", "--retry", "4", "--retry-all-errors",
            "-A", user_agent, "-o", "/dev/null", "-w", "%{http_code}\n%{redirect_url}", route_url,
        ],
        text=True,
        capture_output=True,
        timeout=180,
        check=False,
    )
    lines = route.stdout.splitlines()
    no_slash = {
        "url": route_url,
        "curl_exit": route.returncode,
        "status": int(lines[0]) if lines and lines[0].isdigit() else None,
        "redirect_url": lines[1] if len(lines) > 1 else "",
        "stderr": route.stderr.strip(),
    }
    resources: dict[str, Any] = {}
    for filename in LIVE_FILES:
        suffix = "" if filename == "index.html" else filename
        url = f"https://mobtranslate.com/docs/{suffix}?milestone_audit={nonce}"
        temporary = LIVE_DIR / f".{filename}.tmp"
        result = subprocess.run(
            [
                "curl", "-sS", "-fL", "--max-time", "120", "--retry", "5",
                "--retry-delay", "2", "--retry-all-errors", "-A", user_agent,
                "-o", str(temporary), "-w", "%{http_code}", url,
            ],
            text=True,
            capture_output=True,
            timeout=240,
            check=False,
        )
        status = int(result.stdout.strip()) if result.stdout.strip().isdigit() else None
        if result.returncode == 0 and status == 200 and temporary.exists():
            os.replace(temporary, LIVE_DIR / filename)
        else:
            temporary.unlink(missing_ok=True)
        local = PUBLIC_DOCS / filename
        live = LIVE_DIR / filename
        resources[filename] = {
            "url": url,
            "curl_exit": result.returncode,
            "status": status,
            "stderr": result.stderr.strip(),
            "local_sha256": sha256(local) if local.exists() else None,
            "live_sha256": sha256(live) if live.exists() else None,
            "byte_equal": local.exists() and live.exists() and local.read_bytes() == live.read_bytes(),
        }
    status_record = {
        "captured_at": dt.datetime.now(dt.timezone.utc).isoformat().replace("+00:00", "Z"),
        "canonical_domain": "mobtranslate.com",
        "no_slash": no_slash,
        "resources": resources,
    }
    status_path = LIVE_DIR / "route-status.json"
    status_path.write_text(json.dumps(status_record, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
    audit.hashes[str(status_path.relative_to(PROGRAM))] = sha256(status_path)
    return status_record


def read_live_status() -> dict[str, Any] | None:
    path = LIVE_DIR / "route-status.json"
    if not path.exists():
        return None
    try:
        return json.loads(path.read_text(encoding="utf-8"))
    except (json.JSONDecodeError, OSError):
        return None


def audit_database(audit: Audit, db: sqlite3.Connection) -> dict[int, list[tuple[int, str, str, str]]]:
    db.row_factory = sqlite3.Row
    status_rows = db.execute("SELECT status,COUNT(*) n FROM sentences GROUP BY status").fetchall()
    statuses = {row["status"]: row["n"] for row in status_rows}
    verified = statuses.get("accepted", 0) + statuses.get("revised", 0)
    word_count = db.execute(
        """SELECT SUM(LENGTH(kuku)-LENGTH(REPLACE(kuku,' ',''))+1)
             FROM sentences WHERE status IN ('accepted','revised')"""
    ).fetchone()[0]
    audit.metrics["corpus"] = {
        "rows": sum(statuses.values()),
        "verified": verified,
        "accepted": statuses.get("accepted", 0),
        "revised": statuses.get("revised", 0),
        "open_or_rejected": sum(v for k, v in statuses.items() if k not in {"accepted", "revised"}),
        "kuku_words": word_count,
    }
    audit.check(
        "final corpus count and state",
        (sum(statuses.values()), verified, statuses.get("accepted"), statuses.get("revised"), word_count)
        == (FINAL_ROWS, FINAL_ROWS, 19_136, 911, FINAL_WORDS),
        audit.metrics["corpus"],
    )

    baseline = db.execute(
        """SELECT COUNT(*) FROM sentences s JOIN batches b ON b.id=s.batch_id
            WHERE CAST(substr(b.label,7) AS INTEGER)<=124
              AND s.status IN ('accepted','revised')"""
    ).fetchone()[0]
    batch_inventory = db.execute(
        """SELECT b.label,COUNT(s.id) rows,
                  SUM(s.status IN ('accepted','revised')) verified,
                  SUM(s.status NOT IN ('accepted','revised')) open
             FROM batches b LEFT JOIN sentences s ON s.batch_id=b.id
            WHERE CAST(substr(b.label,7) AS INTEGER) BETWEEN ? AND ?
            GROUP BY b.id ORDER BY CAST(substr(b.label,7) AS INTEGER)""",
        (START_BATCH, END_BATCH),
    ).fetchall()
    scope_sum = sum(row["verified"] for row in batch_inventory)
    bad_batches = [dict(row) for row in batch_inventory if (row["rows"], row["verified"], row["open"]) != (50, 50, 0)]
    audit.metrics["goal_scope"] = {
        "baseline_through_batch_124": baseline,
        "batch_count": len(batch_inventory),
        "scope_rows": scope_sum,
        "first_batch": batch_inventory[0]["label"] if batch_inventory else None,
        "last_batch": batch_inventory[-1]["label"] if batch_inventory else None,
        "bad_batches": bad_batches,
    }
    audit.check(
        "baseline plus 275 complete 50-row batches",
        baseline == BASELINE
        and len(batch_inventory) == 275
        and scope_sum == SCOPE_ROWS
        and not bad_batches
        and baseline + scope_sum == FINAL_ROWS,
        audit.metrics["goal_scope"],
    )

    scope_rows = db.execute(
        """SELECT CAST(substr(b.label,7) AS INTEGER) batch_no,b.label,s.seq,s.english,s.kuku,
                  s.analysis,s.frame,s.tier,s.words_used,s.loans_used,s.evidence,s.status,
                  s.confidence,s.rights_status,s.author
             FROM sentences s JOIN batches b ON b.id=s.batch_id
            WHERE CAST(substr(b.label,7) AS INTEGER) BETWEEN ? AND ?
            ORDER BY batch_no,s.seq""",
        (START_BATCH, END_BATCH),
    ).fetchall()
    malformed: list[str] = []
    all_words: set[str] = set()
    for row in scope_rows:
        tag = f"{row['label']}.{row['seq']:02d}"
        if not row["english"].strip() or not row["kuku"].strip() or " | " not in row["analysis"]:
            malformed.append(f"{tag}: text/analysis")
        if not row["frame"].strip() or not row["evidence"] or not row["evidence"].strip():
            malformed.append(f"{tag}: frame/evidence")
        try:
            words = json.loads(row["words_used"])
            loans = json.loads(row["loans_used"])
            if not isinstance(words, list) or not words or not isinstance(loans, list):
                raise ValueError("words/loans shape")
            all_words.update(str(word).strip().lower() for word in words if str(word).strip())
        except (json.JSONDecodeError, TypeError, ValueError):
            malformed.append(f"{tag}: words_used/loans_used")
    authors = collections.Counter(row["author"] for row in scope_rows)
    rights = collections.Counter(row["rights_status"] for row in scope_rows)
    audit.metrics["row_evidence"] = {
        "rows": len(scope_rows),
        "distinct_words_used_labels": len(all_words),
        "malformed": malformed,
        "authors": dict(sorted(authors.items())),
        "rights": dict(sorted(rights.items())),
    }
    audit.check(
        "row evidence, analysis, structured word inventories, authors, and rights",
        len(scope_rows) == SCOPE_ROWS
        and not malformed
        and authors == {"claude": 6_150, "claude-main": 5_400, "codex-main": 2_200}
        and rights == {"project_approved_synthetic_pending_elder_verification": SCOPE_ROWS},
        audit.metrics["row_evidence"],
    )

    lens_rows = db.execute(
        """SELECT CAST(substr(b.label,7) AS INTEGER) batch_no,r.lens,COUNT(*) n
             FROM reviews r JOIN sentences s ON s.id=r.sentence_id JOIN batches b ON b.id=s.batch_id
            WHERE CAST(substr(b.label,7) AS INTEGER) BETWEEN ? AND ?
              AND r.round=1 AND r.verdict='pass'
            GROUP BY batch_no,r.lens""",
        (START_BATCH, END_BATCH),
    ).fetchall()
    lens_counts = {(row["batch_no"], row["lens"]): row["n"] for row in lens_rows}
    missing_lenses = [
        f"batch-{number}:{lens}={lens_counts.get((number, lens), 0)}"
        for number in range(START_BATCH, END_BATCH + 1)
        for lens in LENSES
        if lens_counts.get((number, lens)) != 50
    ]
    round_counts = dict(
        db.execute(
            """SELECT r.round,COUNT(*) FROM reviews r JOIN sentences s ON s.id=r.sentence_id
                 JOIN batches b ON b.id=s.batch_id
                WHERE CAST(substr(b.label,7) AS INTEGER) BETWEEN ? AND ? GROUP BY r.round""",
            (START_BATCH, END_BATCH),
        ).fetchall()
    )
    audit.metrics["reviews"] = {
        "total": sum(round_counts.values()),
        "rounds": round_counts,
        "round1_lens_cells": len(lens_counts),
        "missing_or_wrong_lenses": missing_lenses,
    }
    audit.check(
        "original five-lens review for every sentence",
        round_counts == {1: 68_750, 2: 1_000} and len(lens_counts) == 1_375 and not missing_lenses,
        audit.metrics["reviews"],
    )

    lesson_counts = db.execute(
        """SELECT CAST(substr(b.label,7) AS INTEGER) batch_no,COUNT(l.id) n
             FROM batches b LEFT JOIN lessons l ON l.batch_id=b.id
            WHERE CAST(substr(b.label,7) AS INTEGER) BETWEEN ? AND ? GROUP BY b.id""",
        (START_BATCH, END_BATCH),
    ).fetchall()
    log_rows = db.execute("SELECT lower(COALESCE(action,'')),lower(COALESCE(detail,'')) FROM lexeme_log").fetchall()
    lexeme_log_counts = {
        number: sum(
            f"batch-{number}" in action or f"batch-{number}" in detail
            for action, detail in log_rows
        )
        for number in range(START_BATCH, END_BATCH + 1)
    }
    audit.metrics["learning_records"] = {
        "lessons": sum(row["n"] for row in lesson_counts),
        "lesson_min": min(row["n"] for row in lesson_counts),
        "lesson_max": max(row["n"] for row in lesson_counts),
        "batch_linked_lexeme_logs": sum(lexeme_log_counts.values()),
        "lexeme_log_min": min(lexeme_log_counts.values()),
        "lexeme_log_max": max(lexeme_log_counts.values()),
        "missing_lexeme_logs": [n for n, count in lexeme_log_counts.items() if not count],
    }
    audit.check(
        "lessons and batch-linked lexeme/source logs",
        len(lesson_counts) == 275
        and min(row["n"] for row in lesson_counts) >= 6
        and not audit.metrics["learning_records"]["missing_lexeme_logs"],
        audit.metrics["learning_records"],
    )

    event_rows = db.execute(
        """SELECT batch_label,artifact_path FROM process_events
            WHERE CAST(substr(batch_label,7) AS INTEGER) BETWEEN ? AND ?""",
        (START_BATCH, END_BATCH),
    ).fetchall()
    event_counts = collections.Counter(row["batch_label"] for row in event_rows)
    artifact_batches = {
        row["batch_label"] for row in event_rows if resolve_artifact(row["artifact_path"])
    }
    missing_events = [f"batch-{n}" for n in range(START_BATCH, END_BATCH + 1) if event_counts[f"batch-{n}"] < 2]
    missing_artifacts = [f"batch-{n}" for n in range(START_BATCH, END_BATCH + 1) if f"batch-{n}" not in artifact_batches]
    audit.metrics["process_events"] = {
        "events": len(event_rows),
        "minimum_per_batch": min(event_counts.values()),
        "maximum_per_batch": max(event_counts.values()),
        "missing_events": missing_events,
        "missing_existing_artifact": missing_artifacts,
    }
    audit.check(
        "structured process history and resolvable artifacts",
        not missing_events and not missing_artifacts and len(event_counts) == 275,
        audit.metrics["process_events"],
    )

    revisions = db.execute(
        """SELECT r.*,CAST(substr(b.label,7) AS INTEGER) batch_no,s.english current_english,
                  s.kuku current_kuku,s.analysis current_analysis
             FROM revisions r JOIN sentences s ON s.id=r.sentence_id JOIN batches b ON b.id=s.batch_id
            WHERE CAST(substr(b.label,7) AS INTEGER) BETWEEN ? AND ? ORDER BY r.id""",
        (START_BATCH, END_BATCH),
    ).fetchall()
    latest: dict[int, sqlite3.Row] = {row["sentence_id"]: row for row in revisions}
    revision_mismatch = []
    for row in latest.values():
        for field in ("english", "kuku", "analysis"):
            new = row[f"new_{field}"]
            if new is not None and new != row[f"current_{field}"]:
                revision_mismatch.append(f"revision-{row['id']}:{field}")
    revision_batches = collections.Counter(row["batch_no"] for row in revisions)
    audit.metrics["revisions"] = {
        "rows": len(revisions),
        "sentences": len(latest),
        "by_batch": dict(sorted(revision_batches.items())),
        "current_field_mismatches": revision_mismatch,
    }
    audit.check(
        "I3 revision ledger resolves to current sentence fields",
        len(revisions) == 16
        and len(latest) == 16
        and revision_batches == {159: 1, 392: 15}
        and not revision_mismatch,
        audit.metrics["revisions"],
    )

    fk_rows = db.execute("PRAGMA foreign_key_check").fetchall()
    integrity = [row[0] for row in db.execute("PRAGMA integrity_check").fetchall()]
    audit.metrics["sqlite"] = {"foreign_key_violations": len(fk_rows), "integrity": integrity}
    audit.check("SQLite foreign keys and integrity", not fk_rows and integrity == ["ok"], audit.metrics["sqlite"])

    all_verified = db.execute(
        """SELECT b.label,s.seq,s.english,s.kuku,s.analysis,s.frame,s.tier,s.words_used,
                  s.evidence,s.status,s.confidence,s.rights_status,s.author
             FROM sentences s JOIN batches b ON b.id=s.batch_id
            WHERE s.status IN ('accepted','revised') ORDER BY b.id,s.seq"""
    ).fetchall()
    normalized_kuku = collections.Counter(row["kuku"].strip().lower() for row in all_verified)
    normalized_pairs = collections.Counter(
        (row["english"].strip().lower(), row["kuku"].strip().lower()) for row in all_verified
    )
    duplicate_kuku = sum(count > 1 for count in normalized_kuku.values())
    duplicate_pairs = sum(count > 1 for count in normalized_pairs.values())
    audit.metrics["duplicates"] = {
        "normalized_kuku_groups": duplicate_kuku,
        "normalized_pair_groups": duplicate_pairs,
    }
    audit.check("zero exact normalized duplicates", duplicate_kuku == 0 and duplicate_pairs == 0, audit.metrics["duplicates"])

    corpus_fields = (
        "label", "seq", "english", "kuku", "analysis", "frame", "tier", "words_used",
        "evidence", "status", "confidence", "rights_status", "author",
    )
    audit.hashes["corpus-content-20047"] = canonical_hash(
        tuple(row[field] for field in corpus_fields) for row in all_verified
    )
    scope_by_batch: dict[int, list[tuple[int, str, str, str]]] = collections.defaultdict(list)
    for row in scope_rows:
        scope_by_batch[row["batch_no"]].append(
            (row["seq"], row["english"], row["kuku"], row["analysis"])
        )
    audit.hashes["goal-scope-content-13750"] = canonical_hash(
        (number, *record) for number in sorted(scope_by_batch) for record in scope_by_batch[number]
    )
    return scope_by_batch


def audit_literal_authorship(
    audit: Audit,
    db: sqlite3.Connection,
    scope_by_batch: dict[int, list[tuple[int, str, str, str]]],
) -> None:
    direct_count = 0
    failures: list[str] = []
    source_records: list[tuple[Any, ...]] = []
    for number in range(START_BATCH, END_BATCH + 1):
        bank = BATCH_SCRIPTS / f"bank_{number}.py"
        close = BATCH_SCRIPTS / f"close_{number}.py"
        if not bank.exists() or not close.exists():
            failures.append(f"batch-{number}: missing bank/close script")
            continue
        try:
            ast.parse(close.read_text(encoding="utf-8"), filename=str(close))
        except SyntaxError as exc:
            failures.append(f"batch-{number}: close syntax {exc}")
        records, issues = literal_rows(bank)
        if number in REMEDIATED_BATCHES:
            if not issues:
                failures.append(f"batch-{number}: historical runtime assembly unexpectedly changed")
            continue
        direct_count += 1
        if issues:
            failures.extend(f"batch-{number}: {issue}" for issue in issues)
        if records != scope_by_batch[number]:
            failures.append(f"batch-{number}: literal source != current database")
        source_records.extend((number, *record) for record in records)

    remediation = BATCH_SCRIPTS / "remediate_literal_authorship_239_240_241_329.py"
    for number in REMEDIATED_BATCHES:
        records, issues = literal_rows(remediation, f"ROWS_{number}")
        if issues:
            failures.extend(f"remediation-{number}: {issue}" for issue in issues)
        if records != scope_by_batch[number]:
            failures.append(f"remediation-{number}: literal source != current database")
        source_records.extend((number, *record) for record in records)

    remediation_rows = db.execute(
        """SELECT ar.*,CAST(substr(b.label,7) AS INTEGER) batch_no
             FROM authorship_remediation ar JOIN sentences s ON s.id=ar.sentence_id
             JOIN batches b ON b.id=s.batch_id ORDER BY batch_no,s.seq"""
    ).fetchall()
    remediation_counts = collections.Counter(row["batch_no"] for row in remediation_rows)
    remediation_hashes = {row["artifact_sha256"] for row in remediation_rows}
    round2 = db.execute(
        """SELECT CAST(substr(b.label,7) AS INTEGER) batch_no,r.lens,COUNT(*) n
             FROM reviews r JOIN sentences s ON s.id=r.sentence_id JOIN batches b ON b.id=s.batch_id
            WHERE r.round=2 AND r.verdict='pass' GROUP BY batch_no,r.lens"""
    ).fetchall()
    round2_counts = {(row["batch_no"], row["lens"]): row["n"] for row in round2}
    expected_round2 = {(number, lens): 50 for number in REMEDIATED_BATCHES for lens in LENSES}
    audit.metrics["literal_authorship"] = {
        "direct_literal_batches": direct_count,
        "remediated_literal_batches": len(REMEDIATED_BATCHES),
        "literal_rows": len(source_records),
        "authorship_remediation_rows": len(remediation_rows),
        "remediation_by_batch": dict(sorted(remediation_counts.items())),
        "round2_lens_cells": len(round2_counts),
        "failures": failures,
    }
    audit.check(
        "literal authorship and exact source-to-database parity",
        direct_count == 271
        and len(source_records) == SCOPE_ROWS
        and not failures
        and remediation_counts == {number: 50 for number in REMEDIATED_BATCHES}
        and remediation_hashes == {ARTIFACT_HASHES["batch_scripts/remediate_literal_authorship_239_240_241_329.py"]}
        and round2_counts == expected_round2,
        audit.metrics["literal_authorship"],
    )
    audit.hashes["literal-authorship-source-13750"] = canonical_hash(sorted(source_records))


def audit_lexical_sources(audit: Audit, db: sqlite3.Connection) -> None:
    for path, expected in SOURCE_HASHES.items():
        actual = sha256(path)
        audit.hashes[str(path)] = actual
        audit.check(f"canonical source hash: {path.name}", actual == expected, {"expected": expected, "actual": actual})

    labels: set[str] = set()
    for (raw,) in db.execute(
        """SELECT s.words_used FROM sentences s JOIN batches b ON b.id=s.batch_id
            WHERE CAST(substr(b.label,7) AS INTEGER) BETWEEN ? AND ?""",
        (START_BATCH, END_BATCH),
    ):
        labels.update(str(item).strip().lower() for item in json.loads(raw) if str(item).strip())
    verified_heads = {
        row[0] for row in db.execute("SELECT lower(headword) FROM lexemes WHERE status='verified'")
    }
    dictionary_document = yaml.safe_load(DICTIONARY.read_text(encoding="utf-8"))
    dict_heads = dictionary_heads(dictionary_document)
    grammar_text = GRAMMAR.read_text(encoding="utf-8").lower()
    grammar_tokens = set(re.findall(r"[a-z-]+", grammar_text))
    lex_hits = labels & verified_heads
    dict_hits = labels & dict_heads
    grammar_hits = {
        label
        for label in labels
        if (
            (re.fullmatch(r"[a-z-]+", label) is not None and label in grammar_tokens)
            or (
                re.fullmatch(r"[a-z-]+", label) is None
                and re.search(rf"(?<![a-z-]){re.escape(label)}(?![a-z-])", grammar_text)
            )
        )
    }
    direct = lex_hits | dict_hits | grammar_hits
    residual = labels - direct
    report_text = LEXICAL_REPORT.read_text(encoding="utf-8") if LEXICAL_REPORT.exists() else ""
    report_labels = set(re.findall(r"^\| `([^`]+)` \|", report_text, flags=re.MULTILINE))
    report_hash = sha256(LEXICAL_REPORT) if LEXICAL_REPORT.exists() else None
    if report_hash:
        audit.hashes[str(LEXICAL_REPORT.relative_to(PROGRAM))] = report_hash
    audit.metrics["lexical_sources"] = {
        "labels": len(labels),
        "verified_lexeme_hits": len(lex_hits),
        "dictionary_head_hits": len(dict_hits),
        "patz_exact_token_hits": len(grammar_hits),
        "direct_union": len(direct),
        "reviewed_residual": len(residual),
        "report_labels": len(report_labels),
        "missing_from_report": sorted(residual - report_labels),
        "extra_in_report": sorted(report_labels - residual),
        "report_sha256": report_hash,
    }
    audit.check(
        "1,412/1,412 lexical labels source-accounted",
        (len(labels), len(lex_hits), len(dict_hits), len(grammar_hits), len(direct), len(residual))
        == (1_412, 1_303, 1_183, 447, 1_354, 58)
        and report_labels == residual
        and "All 58 residuals are source-anchored" in report_text,
        audit.metrics["lexical_sources"],
    )


def audit_artifact_hashes(audit: Audit) -> None:
    mismatches: list[dict[str, str | None]] = []
    for relative, expected in ARTIFACT_HASHES.items():
        path = SYNTH / relative
        actual = sha256(path) if path.exists() else None
        if actual:
            audit.hashes[str(path.relative_to(PROGRAM))] = actual
        if actual != expected:
            mismatches.append({"path": relative, "expected": expected, "actual": actual})
    audit.check("preserved remediation/backfill artifact hashes", not mismatches, {"files": len(ARTIFACT_HASHES), "mismatches": mismatches})


def audit_linter(audit: Audit) -> None:
    result = subprocess.run(
        [sys.executable, str(SYNTH / "lint_corpus.py")],
        cwd=SYNTH,
        text=True,
        capture_output=True,
        timeout=300,
        check=False,
    )
    output = (result.stdout + result.stderr).strip()
    tail = output.splitlines()[-4:]
    passed = (
        result.returncode == 0
        and f"lint: {FINAL_ROWS} rows scanned, 0 issues flagged" in output
        and f"GOAL METRIC: {FINAL_ROWS} / 20000 verified sentences" in output
    )
    audit.metrics["linter"] = {"exit": result.returncode, "tail": tail}
    audit.check("whole-corpus linter", passed, audit.metrics["linter"])


def audit_export(audit: Audit, db: sqlite3.Connection) -> None:
    hashes: dict[str, str | None] = {}
    for filename, expected in EXPORT_HASHES.items():
        path = EXPORT / filename
        actual = sha256(path) if path.exists() else None
        hashes[filename] = actual
        if actual:
            audit.hashes[str(path.relative_to(PROGRAM))] = actual
    manifest = json.loads((EXPORT / "MANIFEST.json").read_text(encoding="utf-8"))
    exported: list[dict[str, Any]] = []
    line_counts: dict[str, int] = {}
    malformed: list[str] = []
    for split in ("train", "dev", "synthtest"):
        path = EXPORT / f"{split}.jsonl"
        records = []
        for line_no, line in enumerate(path.read_text(encoding="utf-8").splitlines(), 1):
            try:
                record = json.loads(line)
                records.append(record)
                if record.get("src_tagged") != f"<synth> {record.get('en', '')}":
                    malformed.append(f"{split}:{line_no}:src_tagged")
                if not isinstance(record.get("meta"), dict):
                    malformed.append(f"{split}:{line_no}:meta")
            except json.JSONDecodeError:
                malformed.append(f"{split}:{line_no}:json")
        line_counts[split] = len(records)
        exported.extend(records)
    export_counter = collections.Counter(
        (
            row.get("en"), row.get("ku"), row.get("analysis"),
            row.get("meta", {}).get("frame"), row.get("meta", {}).get("status"),
        )
        for row in exported
    )
    db_counter = collections.Counter(
        tuple(row) for row in db.execute(
            """SELECT english,kuku,analysis,frame,status FROM sentences
                WHERE status IN ('accepted','revised')"""
        ).fetchall()
    )
    expected_splits = {"train": 16_820, "dev": 1_609, "synthtest": 1_618}
    audit.metrics["export"] = {
        "manifest_total": manifest.get("total_rows"),
        "manifest_splits": manifest.get("splits"),
        "line_counts": line_counts,
        "hashes": hashes,
        "malformed": malformed,
        "five_field_multiset_equal": export_counter == db_counter,
    }
    audit.check(
        "training export manifest, hashes, lines, and five-field database parity",
        manifest.get("total_rows") == FINAL_ROWS
        and manifest.get("splits") == expected_splits
        and line_counts == expected_splits
        and hashes == EXPORT_HASHES
        and not malformed
        and export_counter == db_counter,
        audit.metrics["export"],
    )


def audit_docs(audit: Audit, preflight: bool, fetch_live: bool) -> None:
    source_texts: dict[str, str] = {}
    raw_mismatch: list[str] = []
    rendered_issues: list[str] = []
    source_hashes: dict[str, str] = {}
    public_hashes: dict[str, str] = {}
    for slug, source_name in DOCS.items():
        source = REPORTS / source_name
        raw = PUBLIC_DOCS / f"{slug}.md"
        html = PUBLIC_DOCS / f"{slug}.html"
        source_texts[slug] = source.read_text(encoding="utf-8")
        source_hashes[slug] = sha256(source)
        if not raw.exists() or raw.read_bytes() != source.read_bytes():
            raw_mismatch.append(slug)
        if html.exists():
            rendered = html.read_text(encoding="utf-8")
            public_hashes[f"{slug}.html"] = sha256(html)
            if "<style" not in rendered or "prefers-color-scheme" not in rendered or 'id="TOC"' not in rendered:
                rendered_issues.append(slug)
        else:
            rendered_issues.append(slug)
        if raw.exists():
            public_hashes[f"{slug}.md"] = sha256(raw)
    index = PUBLIC_DOCS / "index.html"
    public_hashes["index.html"] = sha256(index) if index.exists() else ""
    index_text = index.read_text(encoding="utf-8") if index.exists() else ""
    missing_index_links = [
        filename for filename in LIVE_FILES[1:] if f"/docs/{filename}" not in index_text
    ]
    combined = "\n".join(source_texts.values()).lower()
    missing_batch_markers = [
        f"batch-{number}" for number in range(START_BATCH, END_BATCH + 1)
        if f"batch-{number}" not in combined
    ]
    milestone_missing = [
        slug for slug, text in source_texts.items()
        if "milestone completion audit" not in text.lower() or "20,047" not in text
    ]
    audit.metrics["documentation"] = {
        "source_hashes": source_hashes,
        "public_hashes": public_hashes,
        "raw_mismatch": raw_mismatch,
        "rendered_issues": rendered_issues,
        "missing_index_links": missing_index_links,
        "batch_markers": 275 - len(missing_batch_markers),
        "missing_batch_markers": missing_batch_markers,
        "milestone_marker_missing": milestone_missing,
    }
    local_pass = not raw_mismatch and not rendered_issues and not missing_index_links and not missing_batch_markers
    if not preflight:
        local_pass = local_pass and not milestone_missing
    audit.check("four living books rendered with complete batch history", local_pass, audit.metrics["documentation"])
    for slug, digest in source_hashes.items():
        audit.hashes[f"reports/{DOCS[slug]}"] = digest
    for filename, digest in public_hashes.items():
        audit.hashes[f"public-docs/{filename}"] = digest

    if preflight:
        audit.check("final completion report and parallel RunPod handoff", True, "skipped in --preflight mode")
        audit.check("live mobtranslate.com documentation", True, "skipped in --preflight mode")
        return
    audit.check(
        "final completion report and parallel RunPod handoff",
        COMPLETION_REPORT.exists() and RUNPOD_HANDOFF.exists(),
        {
            "completion_report": str(COMPLETION_REPORT),
            "completion_report_exists": COMPLETION_REPORT.exists(),
            "runpod_handoff": str(RUNPOD_HANDOFF),
            "runpod_handoff_exists": RUNPOD_HANDOFF.exists(),
        },
    )
    status = fetch_live_docs(audit) if fetch_live else read_live_status()
    live_pass = False
    if status:
        no_slash = status.get("no_slash", {})
        resources = status.get("resources", {})
        live_pass = (
            no_slash.get("curl_exit") == 0
            and no_slash.get("status") == 301
            and "/docs/" in no_slash.get("redirect_url", "")
            and set(resources) == set(LIVE_FILES)
            and all(
                item.get("curl_exit") == 0 and item.get("status") == 200 and item.get("byte_equal")
                for item in resources.values()
            )
        )
    audit.metrics["live_docs"] = status
    audit.check("live mobtranslate.com documentation", live_pass, status or "missing route-status.json")


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--preflight", action="store_true", help="skip final report/handoff/live requirements")
    parser.add_argument("--fetch-live", action="store_true", help="refresh the final live documentation capture")
    parser.add_argument(
        "--output",
        type=Path,
        default=PROGRAM / "completion_audits/audit-goal-20000-result.json",
    )
    args = parser.parse_args()
    if args.preflight and args.fetch_live:
        parser.error("--fetch-live is only meaningful for the final audit")

    started = time.monotonic()
    audit = Audit()
    with sqlite3.connect(f"file:{DB}?mode=ro", uri=True) as db:
        db.execute("PRAGMA foreign_keys=ON")
        scope_by_batch = audit_database(audit, db)
        audit_literal_authorship(audit, db, scope_by_batch)
        audit_lexical_sources(audit, db)
        audit_export(audit, db)
    audit_artifact_hashes(audit)
    audit_linter(audit)
    audit_docs(audit, args.preflight, args.fetch_live)
    audit.hashes[str(Path(__file__).relative_to(PROGRAM))] = sha256(Path(__file__))

    result = {
        "audit": "Kuku Yalanji 6,297-to-20,047 sentence goal",
        "status": "PASS" if not audit.failures else "FAIL",
        "mode": "preflight" if args.preflight else "final",
        "completed_at": dt.datetime.now(dt.timezone.utc).isoformat().replace("+00:00", "Z"),
        "elapsed_seconds": round(time.monotonic() - started, 3),
        "program_root": str(PROGRAM),
        "checks_passed": len(audit.checks) - len(audit.failures),
        "checks_failed": len(audit.failures),
        "checks": audit.checks,
        "metrics": audit.metrics,
        "hashes": dict(sorted(audit.hashes.items())),
    }
    args.output.parent.mkdir(parents=True, exist_ok=True)
    temporary = args.output.with_suffix(args.output.suffix + ".tmp")
    temporary.write_text(json.dumps(result, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
    os.replace(temporary, args.output)
    print(json.dumps({
        "status": result["status"],
        "mode": result["mode"],
        "checks_passed": result["checks_passed"],
        "checks_failed": result["checks_failed"],
        "output": str(args.output),
    }, indent=2))
    if audit.failures:
        for failure in audit.failures:
            print(f"FAIL: {failure['name']}: {failure['detail']}", file=sys.stderr)
        return 1
    return 0


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