#!/usr/bin/env python3
"""Independent verification for the workqueue-census-v2 artifact.

RE-APPROACH: this checker does NOT import the extractor. It re-derives the
expected rows from the SOURCE BYTES THE MANIFEST PINS with plain string
operations (split on whitespace, startswith); the committed TSV is the
artifact of record. The verdict is the script's exit code: 0 = correct,
1 = a check failed, 2 = could not grade (bad invocation / unusable input).

WHICH BYTES ARE GRADED (Q135, 2026-10-31). The source is the blob named by
`manifest.source_blob_sha`, read with `git cat-file blob`, not the worktree
file. That is what makes the verdict a function of the artifact rather than of
when it is run: a clone made later, with WORK-QUEUE.md hundreds of append-only
records longer, grades the same rows against the same bytes. The worktree is
used only when the manifest declares no pin (a `--allow-dirty` snapshot), and
then only if its sha256 is `manifest.source_sha256`.

A declared pin that git cannot read is a FAULT, never a fallback: an artifact
that cites a blob nobody can open has no provenance, and falling back to
whatever the worktree holds would grade bytes the manifest never named.

Drift is reported, not graded. When the worktree's heading count differs from
the pinned source's, this prints `ADVISORY source-moved worktree=<n>
pinned=<m>` and still exits 0 -- a census is a snapshot, and a snapshot that
goes red because the world moved is a snapshot nobody regenerates.

Run it with the interpreter this file names:

    python3 CHECK                     # portable: every leg the artifact proves
    python3 CHECK --now=2026-10-31T00:00:00Z   # also grade stamp recency
    python3 CHECK --selftest          # grade the tree, then prove every leg fails

Checks:
  1. source pin: the pinned blob resolves and its sha256 is the manifest's;
  2. every '## Q<n>' heading in the pinned source has exactly one committed
     row, in source order, with a verbatim title;
  3. row counts, distinct-q count and reopened-q list in extract_manifest.json
     agree with an independent recomputation;
  4. manifest.source is repo-relative -- a host path is the Q69/Q106/Q109 class;
  5. manifest.row_digest is the sha256 of the committed data rows, so the TSV
     cannot be edited without the manifest disagreeing;
  6. extracted_at_utc parses as an UTC 'Z' stamp and is not the known stale
     constant; its RECENCY is graded only against a caller-supplied --now, so
     the gate does not decay with time;
  7. README names the interpreter and the file carries the python3 shebang and
     the execute bit (the defect that made the documented invocation unrunnable);
  8. no committed row has an empty q or title, and no status/done value
     contains a literal tab (delimiter corruption).
"""

import csv
import hashlib
import json
import os
import shutil
import subprocess
import sys
import tempfile
from datetime import datetime, timedelta, timezone

HERE = os.path.dirname(os.path.abspath(__file__))
REPO_ROOT = os.path.abspath(os.path.join(HERE, "..", "..", ".."))
SRC_NAME = "WORK-QUEUE.md"
TSV_NAME = "workqueue-census.tsv"
MD_NAME = "workqueue-census.md"
MANIFEST_NAME = "extract_manifest.json"
README_NAME = "README.md"
STALE_CONSTANT = "2026-10-28T00:00:00Z"
TS_FMT = "%Y-%m-%dT%H:%M:%SZ"
RECENCY_WINDOW_S = 120
SHEBANG = "#!/usr/bin/env python3"
COLUMNS = ["q", "title", "status", "done_implies"]


def parse_args(argv):
    now_ref, selftest = None, False
    for arg in argv:
        if arg == "--selftest":
            selftest = True
        elif arg.startswith("--now="):
            now_ref = arg.split("=", 1)[1]
        elif arg in ("-h", "--help"):
            print(__doc__)
            sys.exit(0)
        else:
            print("usage: python3 CHECK [--now=YYYY-MM-DDTHH:MM:SSZ] [--selftest]")
            sys.exit(2)
    return now_ref, selftest


def sha256_bytes(data):
    return hashlib.sha256(data).hexdigest()


def is_heading(line):
    """The heading rule, written here and not imported from the extractor."""
    if not line.startswith("## "):
        return False
    parts = line[3:].split(None, 1)
    return bool(parts) and len(parts[0]) > 1 and parts[0][0] == "Q" and parts[0][1:].isdigit()


def rederive(lines):
    """Re-derive (qid, title) in source order with plain string operations."""
    golden = []
    qid, title = None, ""
    for line in lines:
        if not is_heading(line):
            continue
        parts = line[3:].split(None, 1)
        if qid is not None:
            golden.append((qid, title))
        qid = parts[0]
        title = parts[1].strip() if len(parts) > 1 else ""
    if qid is not None:
        golden.append((qid, title))
    return golden


def git_blob(repo_root, sha):
    """The blob's bytes, or None when git or the object is absent."""
    try:
        proc = subprocess.run(["git", "-C", repo_root, "cat-file", "blob", sha],
                              capture_output=True)
    except OSError:
        return None
    if proc.returncode != 0:
        return None
    return proc.stdout


def read_source(repo_root, man):
    """(lines, tier, sha, problem). tier: pinned-blob | worktree-pinned | none."""
    pin = man.get("source_blob_sha")
    want = man.get("source_sha256") or ""
    if pin:
        blob = git_blob(repo_root, pin)
        if blob is None:
            return None, "none", "", ("the manifest pins blob %s and git cannot "
                                      "read it from %s" % (pin[:12], repo_root))
        return (blob.decode("utf-8").splitlines(), "pinned-blob",
                sha256_bytes(blob), "")
    path = os.path.join(repo_root, SRC_NAME)
    try:
        with open(path, "rb") as fh:
            raw = fh.read()
    except OSError:
        return None, "none", "", "no pin, and no %s to fall back to" % SRC_NAME
    got = sha256_bytes(raw)
    if got != want:
        return None, "none", got, ("no pin, and the worktree %s (sha256 %s) is not "
                                   "the bytes the manifest names (%s)"
                                   % (SRC_NAME, got[:12], want[:12]))
    return raw.decode("utf-8").splitlines(), "worktree-pinned", got, ""


def worktree_headings(repo_root):
    """Heading count of the worktree source, or None when unreadable."""
    try:
        with open(os.path.join(repo_root, SRC_NAME), encoding="utf-8") as fh:
            return sum(1 for line in fh.read().splitlines() if is_heading(line))
    except OSError:
        return None


def run(here, repo_root, now_ref):
    """Grade one tree. Returns (fails, stats)."""
    fails = []

    def check(name, cond, detail=""):
        if cond:
            print("PASS " + name)
        else:
            print("FAIL " + name + " " + repr(detail))
            fails.append(name)

    tsv = os.path.join(here, TSV_NAME)
    manifest_path = os.path.join(here, MANIFEST_NAME)
    readme_path = os.path.join(here, README_NAME)
    self_path = os.path.join(here, "CHECK")

    with open(manifest_path, encoding="utf-8") as fh:
        man = json.load(fh)

    # --- 1. Source pin --------------------------------------------------------
    lines, tier, got_sha, problem = read_source(repo_root, man)
    if tier == "none":
        check("source pin", False, problem)
        print("")
        print("CANNOT GRADE: %d check(s) failed before any row was read" % len(fails))
        return fails, {"rows": 0, "distinct": 0, "reopened": 0}
    pin = man.get("source_blob_sha") or "(none: --allow-dirty snapshot)"
    check("source pin", got_sha == (man.get("source_sha256") or ""),
          "pinned=%s read-sha256=%s manifest-sha256=%s"
          % (pin[:12], got_sha[:12], (man.get("source_sha256") or "")[:12]))
    print("INFO source %s %s sha256=%s" % (tier, pin[:12], got_sha[:12]))

    # --- 2. Independent re-derivation from the pinned bytes -------------------
    golden = rederive(lines)

    # --- 3. Committed TSV ----------------------------------------------------
    with open(tsv, encoding="utf-8", newline="") as fh:
        rows = list(csv.reader(fh, delimiter="\t"))
    header, committed = rows[0], rows[1:]

    check("tsv header", header == COLUMNS, repr(header))
    check("row count", len(committed) == len(golden),
          "committed=%d golden=%d" % (len(committed), len(golden)))

    mism = [k for k, (c, g) in enumerate(zip(committed, golden))
            if c[0] != g[0] or c[1] != g[1]]
    check("qid+title identical to independent re-derive", not mism,
          "first mismatch row %r" % (mism[0] if mism else None))

    # --- 4. Manifest counts, independent recomputation ------------------------
    counts = {}
    for q, _ in golden:
        counts[q] = counts.get(q, 0) + 1
    distinct = sorted(counts)
    reopened = sorted(q for q, c in counts.items() if c > 1)

    check("manifest.rows", man.get("rows") == len(golden),
          "manifest=%r actual=%d" % (man.get("rows"), len(golden)))
    check("manifest.distinct_q_labels", man.get("distinct_q_labels") == len(distinct),
          "manifest=%r actual=%d" % (man.get("distinct_q_labels"), len(distinct)))
    check("manifest.reopened_q_labels", man.get("reopened_q_labels") == reopened,
          "manifest=%r actual=%r" % (man.get("reopened_q_labels"), reopened))

    # --- 5. Provenance: relative source, row digest ---------------------------
    check("manifest.source is repo-relative", man.get("source") == SRC_NAME,
          "%r -- a host path is the Q69/Q106/Q109 class" % (man.get("source"),))

    recomputed = "".join("\t".join(r) + "\n" for r in committed)
    check("manifest.row_digest is the committed rows",
          man.get("row_digest") == sha256_bytes(recomputed.encode("utf-8")),
          "manifest=%r recomputed=%s"
          % (man.get("row_digest"), sha256_bytes(recomputed.encode("utf-8"))[:12]))

    # --- 6. extracted_at_utc: provenance graded, recency by argument ---------
    auto = man.get("extracted_at_utc", "")
    check("extracted_at_utc key", isinstance(auto, str) and bool(auto), repr(auto))
    dt = None
    if isinstance(auto, str) and auto:
        try:
            dt = datetime.strptime(auto, TS_FMT).replace(tzinfo=timezone.utc)
            check("extracted_at_utc format", True)
        except ValueError:
            check("extracted_at_utc format", False, repr(auto))
    check("extracted_at_utc not the stale constant", auto != STALE_CONSTANT, repr(auto))
    check("manifest provenance names the source",
          man.get("schema") == "workqueue-census-v2"
          and isinstance(man.get("generated_by"), str) and man.get("generated_by"),
          "schema=%r generated_by=%r" % (man.get("schema"), man.get("generated_by")))

    if now_ref is None:
        print("SKIP extracted_at_utc recency "
              "(no reference time; pass --now=YYYY-MM-DDTHH:MM:SSZ to grade it)")
    else:
        try:
            ref = datetime.strptime(now_ref, TS_FMT).replace(tzinfo=timezone.utc)
        except ValueError:
            check("--now format", False, repr(now_ref))
        else:
            if dt is None:
                check("extracted_at_utc recency", False, "unparseable stamp %r" % (auto,))
            else:
                delta = abs((ref - dt).total_seconds())
                check("extracted_at_utc recency", delta < RECENCY_WINDOW_S,
                      "auto=%s ref=%s delta=%ss" % (auto, now_ref, delta))

    # --- 7. Interpreter contract (README vs this file) ------------------------
    with open(self_path, encoding="utf-8") as fh:
        first = fh.readline().rstrip("\n")
    check("CHECK shebang is python3", first == SHEBANG, repr(first))
    check("CHECK is executable", os.access(self_path, os.X_OK),
          oct(os.stat(self_path).st_mode & 0o777))
    with open(readme_path, encoding="utf-8") as fh:
        readme = fh.read()
    # The documented invocation lives in the Verification section; the prose may
    # still quote the old broken command when it explains the repair.
    vsec = readme.split("## Verification", 1)
    vsec = vsec[1].split("\n## ", 1)[0] if len(vsec) > 1 else ""
    check("README documents the python3 invocation",
          "python3 CHECK" in vsec and "bash CHECK" not in vsec,
          "python3 CHECK present=%r bash CHECK present=%r"
          % ("python3 CHECK" in vsec, "bash CHECK" in vsec))

    # --- 8. Corruption checks -------------------------------------------------
    empty = [r for r in committed if not r[0] or not r[1]]
    check("no empty q/title", not empty, repr(empty[:1]))
    bad_tab = [r for r in committed if any("\t" in c for c in r[2:])]
    check("status/done no tabs", not bad_tab, repr(bad_tab[:1]))

    # --- 9. Drift is reported, never graded -----------------------------------
    head_n = worktree_headings(repo_root)
    if head_n is not None and head_n != len(golden):
        print("ADVISORY source-moved worktree=%d pinned=%d "
              "(this census is a snapshot of the pinned bytes; regenerate it to "
              "move the snapshot forward)" % (head_n, len(golden)))

    return fails, {"rows": len(golden), "distinct": len(distinct), "reopened": len(reopened)}


# --- selftest: one cell per leg, each built by mutating a scratch copy ---------

def _git(dst, *args):
    """git in the fixture tree, with an identity, returning the CompletedProcess."""
    return subprocess.run(["git", "-C", dst, "-c", "user.name=selftest",
                           "-c", "user.email=selftest@example.invalid"] + list(args),
                          capture_output=True, text=True)


def build_tree(dst, manifest_mutate=None, tsv_mutate=None, source_append=None):
    """Copy the artifact + its source into dst as a self-contained repo root.

    The copy is a real git repository whose HEAD blob is the pinned source, so
    the pin leg is graded the way it is graded in the real tree; `source_append`
    then moves the worktree out from under that pin.
    """
    here = os.path.join(dst, "wizard-work", "ledger", "workqueue-census-v2")
    os.makedirs(here)
    shutil.copy2(os.path.join(REPO_ROOT, SRC_NAME), os.path.join(dst, SRC_NAME))

    for name in ("CHECK", TSV_NAME, MD_NAME, README_NAME):
        shutil.copy2(os.path.join(HERE, name), os.path.join(here, name))

    _git(dst, "init", "-q")
    _git(dst, "add", SRC_NAME)
    _git(dst, "commit", "-q", "-m", "selftest fixture")

    with open(os.path.join(dst, SRC_NAME), "rb") as fh:
        src_raw = fh.read()
    blob_sha = _git(dst, "rev-parse", "HEAD:" + SRC_NAME).stdout.strip()
    rev = _git(dst, "rev-parse", "HEAD").stdout.strip()

    if source_append:
        with open(os.path.join(dst, SRC_NAME), "a", encoding="utf-8") as fh:
            fh.write(source_append)

    with open(os.path.join(HERE, MANIFEST_NAME), encoding="utf-8") as fh:
        man = json.load(fh)
    man["source_blob_sha"] = blob_sha
    man["source_rev"] = rev
    man["source_sha256"] = sha256_bytes(src_raw)
    man["source_matches_head_blob"] = True
    if manifest_mutate:
        manifest_mutate(man)
    with open(os.path.join(here, MANIFEST_NAME), "w", encoding="utf-8") as fh:
        json.dump(man, fh, indent=2)

    if tsv_mutate:
        path = os.path.join(here, TSV_NAME)
        with open(path, encoding="utf-8", newline="") as fh:
            rows = list(csv.reader(fh, delimiter="\t"))
        tsv_mutate(rows)
        with open(path, "w", encoding="utf-8", newline="") as fh:
            csv.writer(fh, delimiter="\t").writerows(rows)

    return here


def shift_one_day(man):
    dt = datetime.strptime(man["extracted_at_utc"], TS_FMT).replace(tzinfo=timezone.utc)
    man["extracted_at_utc"] = (dt + timedelta(days=1)).strftime(TS_FMT)


def set_stale(man):
    man["extracted_at_utc"] = STALE_CONSTANT


def set_unreadable_pin(man):
    man["source_blob_sha"] = "0" * 40


def set_host_path(man):
    man["source"] = os.path.join(REPO_ROOT, SRC_NAME)


def set_bad_row_digest(man):
    man["row_digest"] = "0" * 64


def corrupt_title(rows):
    rows[1][1] = rows[1][1] + " (corrupted)"


def selftest():
    with open(os.path.join(HERE, MANIFEST_NAME), encoding="utf-8") as fh:
        true_stamp = json.load(fh)["extracted_at_utc"]

    cells = []

    def cell(name, mutate_kwargs, argv, want_rc, want_substr):
        with tempfile.TemporaryDirectory(prefix="census2-selftest-") as tmp:
            here = build_tree(tmp, **mutate_kwargs)
            proc = subprocess.run([sys.executable, os.path.join(here, "CHECK")] + argv,
                                  capture_output=True, text=True)
        out = proc.stdout + proc.stderr
        ok = proc.returncode == want_rc and want_substr in out
        cells.append((name, ok))
        print("%s %-30s rc=%d (want %d) %r%s"
              % ("PASS" if ok else "FAIL", name, proc.returncode, want_rc,
                 want_substr, "" if ok else "\n" + out))

    # controls
    cell("control-no-reference", {}, [], 0, "ALL CHECKS PASSED")
    cell("control-with-reference", {}, ["--now=" + true_stamp], 0,
         "PASS extracted_at_utc recency")
    # the pin is graded, not assumed
    cell("unreadable-pin", {"manifest_mutate": set_unreadable_pin}, [], 1,
         "FAIL source pin")
    # the host-pin cell (Q69/Q106/Q109 class)
    cell("host-path-source", {"manifest_mutate": set_host_path}, [], 1,
         "FAIL manifest.source is repo-relative")
    # the artifact's own digest
    cell("row-digest-corrupted", {"manifest_mutate": set_bad_row_digest}, [], 1,
         "FAIL manifest.row_digest")
    # the rows still bite
    cell("tsv-title-corrupted", {"tsv_mutate": corrupt_title}, [], 1,
         "FAIL qid+title identical to independent re-derive")
    # the time value is a measurement, not a constant
    cell("stale-constant", {"manifest_mutate": set_stale}, [], 1,
         "FAIL extracted_at_utc not the stale constant")
    cell("timestamp-plus-one-day", {"manifest_mutate": shift_one_day},
         ["--now=" + true_stamp], 1, "FAIL extracted_at_utc recency")
    # and drift is an advisory, not a fault
    cell("drift-is-advisory", {"source_append": "\n## Q999 - selftest drift probe\n"},
         [], 0, "ADVISORY source-moved")

    bad = [n for n, ok in cells if not ok]
    print("")
    if bad:
        print("SELFTEST FAILED (%d of %d cells): %s" % (len(bad), len(cells), ", ".join(bad)))
        return 1
    print("SELFTEST OK (%d cells: 3 green controls, 6 red mutations)" % len(cells))
    return 0


def main():
    now_ref, do_selftest = parse_args(sys.argv[1:])
    if do_selftest:
        fails, _ = run(HERE, REPO_ROOT, None)
        if fails:
            print("")
            print("SELFTEST ABORTED: the in-tree artifact is already red (%d check(s)); "
                  "fix the tree before grading the gate" % len(fails))
            return 1
        print("")
        print("in-tree gate green; grading the legs")
        print("")
        return selftest()

    fails, stats = run(HERE, REPO_ROOT, now_ref)
    if fails:
        print("")
        print("%d check(s) FAILED" % len(fails))
        return 1

    print("")
    print("ALL CHECKS PASSED (%d rows, %d distinct labels, %d reopened)"
          % (stats["rows"], stats["distinct"], stats["reopened"]))
    return 0


if __name__ == "__main__":
    sys.exit(main())
