#!/usr/bin/env python3
"""The WAI Quantum documentation, run: every command and every snippet in it,
checked against what the code does.

    python3 wai/tools/quantum-docs/check_docs.py --cli PATH/TO/wai-quantum \\
        [--service URL] [--python PATH/TO/python] [--rust] [--only SUBSTRING] [--keep]

The documents are the ones `docs.json` beside this file names. In each, every
fenced block is one of these, by its info string (the words after the fence):

- ```` ```sh runnable ```` (or bash, shell): run in order, one document's
  blocks in one scratch directory holding `fixtures/`, as a reader would run
  them, with `wai-quantum` the command line given (`in=DIR` runs it in that
  directory of the scratch). Each command must exit 0,
  except a line whose comment begins `# exits N` (or `# exit N`), which must
  exit N; `exit=N` holds the whole block to N. `copy=DIR` copies a directory
  beside the document into the scratch first and `timeout=S` gives it longer.
  `needs=service` needs `--service`; the documented service address is run
  against that one. `cargo=DIR` runs its `cargo run` against the crate in that
  directory beside the document.
- ```` ```console runnable ````: each `$ command` run, and its output held to
  the lines after it.
- ```` ```text output ```` right after a runnable block: that block's output,
  held to it line by line (a run's `time` line, which is wall-clock, aside).
- ```` ```python runnable ````: run in order in one interpreter (`--python`,
  with the package installed), one namespace for the document.
- ```` ```rust runnable ```` or ```` ```rust compile ````: built (and run) as
  an example against the crate (`crate=NAME`, default wai-quantum,
  `features=a,b`), with `--rust`.
- ```` ```qasm ````: read by the engine (`wai-quantum lint`); `refused` holds
  it to a refusal, and `file=NAME` writes it into the scratch for the blocks
  after it.
- ```` ```toml ````: each dependency on a crate of this repository names a
  version its version satisfies.
- ```` ```json ````: parses (`fragment` excepts one).
- ```` ```<lang> norun=REASON ````: a block that cannot run here, for a stated
  reason: `ci:WORKFLOW` (the steps a CI workflow runs: each script it names
  exists, and the workflow names it), `install` (each package it installs is
  this repository's, at its version), `data` (it needs files the repository
  does not hold), `account` (it needs what only the person has: an account,
  a device, a program of their own), `server`
  (it starts a server that does not return), `usage` (a synopsis). Every
  `wai-quantum` command in it must still name a command, and flags, that
  exist.

A shell, Python or Rust block with none of these, or a command in a text
block, is an error: documentation either runs here or says why it cannot.

A translation (`NAME.zh-CN.md`) is marked machine-assisted on its first line
and holds its English page's code (comments aside); one whose code is the
English page's exactly is not run again.

Every JSON file a runnable block writes, and every output that is one, that
carries a receipt is checked with `wai-quantum verify`. In prose, every
`wai-quantum …` in backquotes must name a command that exists, and every
relative link must reach a file.

Exit status: 0 when every check passes, 1 otherwise, 2 on bad arguments.
"""
import argparse
import fnmatch
import json
import os
import re
import shlex
import shutil
import subprocess
import sys
import tempfile
import time
from pathlib import Path

HERE = Path(__file__).resolve().parent
ROOT = HERE.parents[2]
SERVICE_URL = "https://wai.transaction.science/quantum-jobs"
SHELLS = {"sh", "bash", "shell", "zsh"}
REASONS = {"ci", "install", "data", "account", "server", "usage"}
TEXTS = {"", "text", "txt", "plain"}

FENCE = re.compile(r"^(\s*)(```+|~~~+)(.*)$")


class Block:
    def __init__(self, doc, line, info, body):
        self.doc, self.line, self.info, self.body = doc, line, info.strip(), body
        words = self.info.split()
        self.lang = words[0].lower() if words else ""
        self.flags = set()
        self.attrs = {}
        for w in words[1:]:
            if "=" in w:
                k, v = w.split("=", 1)
                self.attrs[k] = v
            else:
                self.flags.add(w)

    @property
    def where(self):
        return f"{self.doc.rel}:{self.line}"

    def text(self):
        return "\n".join(self.body) + ("\n" if self.body else "")


class Doc:
    def __init__(self, path):
        self.path = path
        self.rel = path.relative_to(ROOT).as_posix()
        self.text = path.read_text(encoding="utf-8")
        self.blocks = []
        self.prose = []  # (line, text) outside fences
        lines = self.text.split("\n")
        i = 0
        while i < len(lines):
            m = FENCE.match(lines[i])
            if m:
                indent, fence, info = m.group(1), m.group(2), m.group(3)
                body = []
                j = i + 1
                while j < len(lines) and not re.match(r"^\s*" + re.escape(fence[0]) * len(fence) + r"\s*$", lines[j]):
                    body.append(lines[j][len(indent):] if lines[j].startswith(indent) else lines[j].lstrip())
                    j += 1
                self.blocks.append(Block(self, i + 1, info, body))
                i = j + 1
                continue
            self.prose.append((i + 1, lines[i]))
            i += 1


class Report:
    def __init__(self):
        self.fails = []
        self.counts = {}

    def ok(self, kind):
        self.counts[kind] = self.counts.get(kind, 0) + 1

    def fail(self, where, msg):
        self.fails.append((where, msg))
        print(f"FAIL  {where}: {msg}", flush=True)


# ---- the repository's packages ---------------------------------------------

def packages():
    """Every Cargo package under wai/, by name: its version and whether it has a binary."""
    out = {}
    for toml in (ROOT / "wai").rglob("Cargo.toml"):
        if any(p in ("target", "node_modules") for p in toml.parts):
            continue
        t = toml.read_text(encoding="utf-8", errors="replace")
        pkg = re.search(r"^\[package\](.*?)(?=^\[|\Z)", t, re.S | re.M)
        if not pkg:
            continue
        name = re.search(r'^name\s*=\s*"([^"]+)"', pkg.group(1), re.M)
        ver = re.search(r'^version\s*=\s*"([^"]+)"', pkg.group(1), re.M)
        if not name:
            continue
        bins = re.findall(r'^\[\[bin\]\][^\[]*?name\s*=\s*"([^"]+)"', t, re.S | re.M)
        if (toml.parent / "src" / "main.rs").exists():
            bins.append(name.group(1))
        bins += [b.stem for b in (toml.parent / "src" / "bin").glob("*.rs")]
        out[name.group(1)] = {"version": ver.group(1) if ver else None, "bins": bins, "dir": toml.parent}
    return out


def python_packages():
    out = {}
    for py in (ROOT / "wai").rglob("pyproject.toml"):
        if "node_modules" in py.parts:
            continue
        t = py.read_text(encoding="utf-8")
        name = re.search(r'^name\s*=\s*"([^"]+)"', t, re.M)
        if name:
            cargo = py.parent / "Cargo.toml"
            ver = re.search(r'^version\s*=\s*"([^"]+)"', cargo.read_text(encoding="utf-8"), re.M) if cargo.exists() else None
            out[name.group(1).replace("_", "-")] = ver.group(1) if ver else None
    return out


def caret_ok(req, have):
    """Whether the version `have` satisfies the Cargo requirement `req` (caret by default)."""
    req = req.strip()
    if not have:
        return False
    h = [int(x) for x in re.findall(r"\d+", have.split("-")[0])[:3]]
    if req.startswith("="):
        return h == [int(x) for x in re.findall(r"\d+", req)][:3]
    parts = [int(x) for x in re.findall(r"\d+", req.lstrip("^"))[:3]]
    if not parts:
        return False
    if h < parts + [0] * (3 - len(parts)):
        return False
    for i, x in enumerate(parts):
        if x != 0 or i == len(parts) - 1:
            return h[: i + 1] == parts[: i + 1]
    return True


# ---- the command line's shape ----------------------------------------------

class Cli:
    def __init__(self, path):
        self.path = path
        self.help = {}
        self.lab = None

    def run(self, args, env=None, timeout=60):
        return subprocess.run([self.path, *args], capture_output=True, text=True, timeout=timeout, env=env, stdin=subprocess.DEVNULL)

    def help_of(self, path):
        if list(path) == ["lab"]:
            return "lab"
        key = tuple(path)
        if key not in self.help:
            p = self.run([*path, "--help"], env={**os.environ, "LANG": "C.UTF-8", "WAI_QUANTUM_LANG": "en"})
            self.help[key] = p.stdout if p.returncode == 0 else None
        return self.help[key]

    def lab_tools(self):
        if self.lab is None:
            text = self.run(["lab", "help"]).stdout + self.run(["lab", "help"]).stderr
            self.lab = {}
            for line in text.split("\n"):
                m = re.search(r"wai-quantum lab ([a-z][a-z0-9-]*)(.*)$", line)
                if m:
                    self.lab.setdefault(m.group(1), set()).update(re.findall(r"--[a-z][a-z0-9-]*", m.group(2)))
            # A tool's usage continues on the indented lines after it.
            cur = None
            for line in text.split("\n"):
                m = re.search(r"wai-quantum lab ([a-z][a-z0-9-]*)", line)
                if m:
                    cur = m.group(1)
                elif cur and line.startswith("    "):
                    self.lab[cur].update(re.findall(r"--[a-z][a-z0-9-]*", line))
                elif not line.strip():
                    cur = None
        return self.lab

    def check_command(self, tokens):
        """Why `wai-quantum TOKENS…` names no command or flag the command line has, or None."""
        words = []
        rest = list(tokens)
        while rest and re.match(r"^[a-z][a-z0-9-]*$", rest[0]) and self.help_of(words + [rest[0]]) is not None:
            words.append(rest.pop(0))
            if words == ["lab"]:
                break
        if not words:
            return f"`wai-quantum {' '.join(tokens[:2])}`: no such command" if tokens and not tokens[0].startswith("-") else None
        if words == ["lab"]:
            tools = self.lab_tools()
            if not rest or rest[0] in ("help", "--help", "-h"):
                return None
            tool = rest[0]
            if tool not in tools:
                return f"`wai-quantum lab {tool}`: no such research tool (`wai-quantum lab help` lists them)"
            flags = tools[tool]
            for f in rest[1:]:
                f = f.split("=", 1)[0]
                if f.startswith("--") and f not in flags and f != "--help":
                    return f"`wai-quantum lab {tool}`: no flag {f} in its usage"
            return None
        h = self.help_of(words) or ""
        root = self.help_of([]) or ""
        known = set(re.findall(r"--[a-z][a-z0-9-]*", h)) | set(re.findall(r"--[a-z][a-z0-9-]*", root))
        for f in rest:
            if f == "--":
                break
            f = f.split("=", 1)[0]
            if re.match(r"^--[a-z][a-z0-9-]*$", f) and f not in known:
                return f"`wai-quantum {' '.join(words)}`: no flag {f}"
        # A word after the path where the command takes subcommands must be one of them.
        subs = subcommands(h)
        if subs and rest and re.match(r"^[a-z][a-z0-9-]*$", rest[0]) and rest[0] not in subs:
            return f"`wai-quantum {' '.join(words)} {rest[0]}`: no such command ({', '.join(sorted(subs))})"
        return None


def subcommands(help_text):
    """The subcommands a command's help lists under Commands:, if any."""
    if "\nCommands:" not in "\n" + help_text:
        return set()
    sec = ("\n" + help_text).split("\nCommands:", 1)[1]
    out = set()
    for line in sec.split("\n")[1:]:
        if not line.startswith("  "):
            break
        m = re.match(r"^\s+([a-z][a-z0-9-]*)\b", line)
        if m and m.group(1) != "help":
            out.add(m.group(1))
    return out


def command_tokens(line):
    """The tokens of a shell line's commands, each a list (pipes and && split)."""
    line = line.strip()
    if line.startswith("$ "):
        line = line[2:]
    line = re.sub(r"\s#.*$", "", line) if not line.startswith("#") else ""
    if not line:
        return []
    try:
        toks = shlex.split(line, comments=False)
    except ValueError:
        toks = line.split()
    cmds, cur = [], []
    for t in toks:
        if t in ("|", "&&", "||", ";", "&"):
            cmds.append(cur)
            cur = []
        else:
            cur.append(t)
    cmds.append(cur)
    out = []
    for c in cmds:
        while c and re.match(r"^[A-Z_][A-Z0-9_]*=", c[0]):
            c = c[1:]
        out.append(c)
    return [c for c in out if c]


# ---- the checks --------------------------------------------------------------

def static_line_checks(cli, block, line, report, pkgs, pys):
    for cmd in command_tokens(line):
        head = cmd[0].lstrip("(")
        if head == "wai-quantum":
            why = cli.check_command([t for t in cmd[1:] if not t.startswith(">") and not t.startswith("<")])
            if why:
                report.fail(block.where, why)
        elif head == "cargo" and len(cmd) > 2 and cmd[1] == "install":
            name = next((t for t in cmd[2:] if not t.startswith("-")), None)
            if name and (name not in pkgs or not pkgs[name]["bins"]):
                report.fail(block.where, f"`cargo install {name}`: no package of this repository with a binary is named {name}")
        elif head in ("pip", "pip3", "uv") and "install" in cmd:
            for t in cmd[cmd.index("install") + 1:]:
                if t.startswith("-") or t.startswith("dist/"):
                    continue
                m = re.match(r"^([A-Za-z0-9_.-]+?)-(\d[\w.]*)-.*\.whl", t.split("[")[0])
                if m:
                    name, ver = m.group(1).replace("_", "-"), m.group(2)
                    if name not in pys:
                        report.fail(block.where, f"pip install {t}: no Python package of this repository is named {name}")
                    elif pys[name] != ver:
                        report.fail(block.where, f"pip install {t}: the package is at {pys[name]}, not {ver}")
                else:
                    name = t.split("[")[0].replace("_", "-")
                    if name and name not in pys and not name.endswith(".whl"):
                        report.fail(block.where, f"pip install {t}: no Python package of this repository is named {name}")


def check_norun(cli, block, report, pkgs, pys):
    reason = block.attrs.get("norun", "")
    kind, _, detail = reason.partition(":")
    if kind not in REASONS:
        report.fail(block.where, f"norun={reason}: the reasons are {', '.join(sorted(REASONS))}")
        return
    for line in block.body:
        static_line_checks(cli, block, line, report, pkgs, pys)
    if kind == "usage":
        bins = {b for p in pkgs.values() for b in p["bins"]}
        for line in block.body:
            words = line.split()
            if words and not words[0].startswith("#") and words[0] not in bins:
                report.fail(block.where, f"{words[0]}: no package of this repository builds a binary of that name")
    if kind == "ci":
        wf = ROOT / ".github" / "workflows" / detail
        if not detail or not wf.exists():
            report.fail(block.where, f"norun=ci:{detail}: no such workflow")
            return
        flow = " ".join(wf.read_text(encoding="utf-8").split())
        for line in block.body:
            line = re.sub(r"\s+#.*$", "", line).strip()
            if not line or line.startswith("#"):
                continue
            for seg in re.split(r"&&|;", line):
                seg = seg.strip().strip("()").strip().rstrip("&").strip()
                seg = re.sub(r"^([A-Z_][A-Z0-9_]*=\S*\s+)+", "", seg)
                if not seg or seg.startswith("cd "):
                    continue
                # A placeholder (PATH/TO/PKG, FILE, $VAR, <x>) stands for any one word.
                pat = r"\s+".join(r"\S+" if re.fullmatch(r"[A-Z][A-Z0-9_]*(/[A-Z][A-Z0-9_]*)*|\$\S+|<[^>]+>", t) else re.escape(t) for t in seg.split())
                if not re.search(pat, flow):
                    report.fail(block.where, f"`{seg}`: the workflow {detail} does not run this")
    report.ok("norun")


def check_toml(block, report, pkgs):
    for line in block.body:
        m = re.match(r'^\s*([A-Za-z0-9_-]+)\s*=\s*(?:"([^"]+)"|\{[^}]*version\s*=\s*"([^"]+)")', line)
        if not m:
            continue
        name, req = m.group(1), m.group(2) or m.group(3)
        if name in pkgs:
            have = pkgs[name]["version"]
            if not caret_ok(req, have):
                report.fail(block.where, f'{name} = "{req}": the crate is at {have}, which that requirement does not take')
            else:
                report.ok("toml")


def check_prose(cli, doc, report):
    for n, line in doc.prose:
        for span in re.findall(r"`(wai-quantum [^`]+)`", line):
            toks = command_tokens(span)
            if not toks:
                continue
            words = [t for t in toks[0][1:] if not re.search(r"[A-Z]{2,}|…|\.\.\.|\[|<", t)]
            why = cli.check_command(words)
            if why:
                report.fail(f"{doc.rel}:{n}", why)
            else:
                report.ok("prose command")
        for target in re.findall(r"\]\(([^)\s]+)\)", re.sub(r"`[^`]*`", "", line)):
            if re.match(r"^[a-z]+:", target) or target.startswith("#"):
                if re.match(r"^https://(crates\.io/crates|docs\.rs)/([A-Za-z0-9_-]+)", target):
                    name = re.match(r"^https://(?:crates\.io/crates|docs\.rs)/([A-Za-z0-9_-]+)", target).group(1)
                    if name not in PKGS:
                        report.fail(f"{doc.rel}:{n}", f"{target}: no crate of this repository is named {name}")
                continue
            path = target.split("#")[0]
            if path and not (doc.path.parent / path).exists():
                report.fail(f"{doc.rel}:{n}", f"the link {target} reaches no file")
            else:
                report.ok("link")


def normalize(text, scratch):
    out = []
    for line in text.replace(str(scratch), ".").split("\n"):
        line = line.rstrip()
        # A key's did:key is the fixture's here and the reader's there.
        line = re.sub(r"did:key:z6Mk[1-9A-HJ-NP-Za-km-z]+", "did:key:z6Mk…", line)
        if re.match(r"^time\s+\S", line):
            line = re.sub(r"^time\s+.*$", "time <wall clock>", line)
        out.append(line)
    while out and not out[-1]:
        out.pop()
    while out and not out[0]:
        out.pop(0)
    return "\n".join(out)


def transform(lines):
    """A block's lines as a script: each `# exits N` line held to its status."""
    out = ["set -eo pipefail"]
    heredoc = None
    cont = False
    for line in lines:
        if heredoc:
            out.append(line)
            if line.strip() == heredoc:
                heredoc = None
            continue
        m = re.search(r"<<-?\s*'?\"?([A-Za-z_]+)'?\"?", line)
        if m and not line.lstrip().startswith("#"):
            heredoc = m.group(1)
        e = re.match(r"^(?P<cmd>[^#]*?\S)\s+#\s*exits?\s+(?P<n>\d+)\b", line)
        if e and not cont and not heredoc:
            cmd, n = e.group("cmd"), e.group("n")
            q = shlex.quote(cmd)
            out.append(f'__s=0; {cmd} || __s=$?; [ "$__s" -eq {n} ] || {{ echo "docs: expected exit {n}, got $__s: "{q} >&2; exit 97; }}')
        else:
            out.append(line)
        cont = line.rstrip().endswith("\\")
    return "\n".join(out) + "\n"


class Runner:
    def __init__(self, args, report):
        self.args = args
        self.report = report
        self.cli = Cli(args.cli)
        self.tmp = Path(tempfile.mkdtemp(prefix="wai-quantum-docs-"))
        self.bin = self.tmp / "bin"
        self.bin.mkdir()
        (self.bin / "wai-quantum").symlink_to(Path(args.cli).resolve())

    def env(self, scratch):
        e = dict(os.environ)
        e.update({
            "PATH": f"{self.bin}{os.pathsep}{e.get('PATH', '')}",
            "HOME": str(scratch / ".home"),
            "XDG_CONFIG_HOME": str(scratch / ".config"),
            "WAI_QUANTUM_KEY": str(scratch / "quantum-key"),
            # Credentials in a file under the scratch's configuration, never the keychain.
            "WAI_QUANTUM_CREDENTIALS": "file",
            "LANG": "C.UTF-8",
            "LC_ALL": "C.UTF-8",
            "WAI_QUANTUM_LANG": "",
        })
        e.pop("WAI_QUANTUM_LEARN", None)
        # cargo outside the checkout uses the toolchain the checkout names.
        e.setdefault("RUSTUP_TOOLCHAIN", TOOLCHAIN)
        if self.args.service:
            e["WAI_QUANTUM_SERVICE"] = self.args.service
        else:
            e.pop("WAI_QUANTUM_SERVICE", None)
        return e

    def scratch_for(self, doc):
        d = self.tmp / re.sub(r"[^A-Za-z0-9]+", "-", doc.rel)
        if d.exists():
            shutil.rmtree(d)
        shutil.copytree(HERE / "fixtures", d)
        (d / ".home").mkdir(exist_ok=True)
        (d / ".config").mkdir(exist_ok=True)
        return d

    def verify_receipts(self, block, scratch, since, stdout, env):
        """Every JSON the block wrote, or printed, that carries a receipt, checked."""
        files = [p for p in scratch.rglob("*.json") if p.stat().st_mtime >= since and ".config" not in p.parts and ".home" not in p.parts]
        n = 0
        cands = [(str(p.relative_to(scratch)), p.read_text(encoding="utf-8", errors="replace")) for p in files]
        if stdout.strip().startswith("{"):
            cands.append(("its output", stdout))
        for name, text in cands:
            try:
                v = json.loads(text)
            except ValueError:
                continue
            if not isinstance(v, dict) or not isinstance(v.get("result"), dict) or "request" not in v:
                continue
            if "receipt" not in v["result"]:
                continue
            p = subprocess.run([self.args.cli, "verify", "-"], input=text, capture_output=True, text=True, env=env, cwd=scratch, timeout=600)
            if p.returncode != 0:
                self.report.fail(block.where, f"the receipt in {name} does not verify: {(p.stdout + p.stderr).strip()[-300:]}")
            else:
                n += 1
        if n:
            self.report.ok("receipt")
        return n

    def run_shell(self, doc, block, scratch, expected):
        if "service" in block.attrs.get("needs", "") and not self.args.service:
            self.report.fail(block.where, "needs the service: give --service URL")
            return
        for line in block.body:
            static_line_checks(self.cli, block, line, self.report, PKGS, PYS)
        if "copy" in block.attrs:
            src = (doc.path.parent / block.attrs["copy"]).resolve()
            dst = scratch / src.name
            if src.is_dir():
                shutil.copytree(src, dst, dirs_exist_ok=True)
            else:
                shutil.copy(src, dst)
        lines = block.body
        if "cargo" in block.attrs:
            # `cargo run` as the document runs it beside the crate, here in the scratch.
            manifest = (doc.path.parent / block.attrs["cargo"] / "Cargo.toml").resolve()
            lines = [re.sub(r"\bcargo (run|build|test)\b", lambda m: f"cargo {m.group(1)} --manifest-path {shlex.quote(str(manifest))}", l) for l in lines]
        if self.args.service:
            lines = [l.replace(SERVICE_URL, self.args.service.rstrip("/")) for l in lines]
        script = scratch / f".block-{block.line}.sh"
        script.write_text(transform(lines), encoding="utf-8")
        cwd = scratch / block.attrs["in"] if "in" in block.attrs else scratch
        env = self.env(scratch)
        since = time.time() - 1
        try:
            p = subprocess.run(["bash", str(script)], cwd=cwd, env=env, capture_output=True, text=True, timeout=int(block.attrs.get("timeout", "300")), stdin=subprocess.DEVNULL)
        except subprocess.TimeoutExpired:
            self.report.fail(block.where, f"did not finish within {block.attrs.get('timeout', '300')} s")
            return
        want = int(block.attrs.get("exit", "0"))
        if p.returncode != want:
            self.report.fail(block.where, f"exit {p.returncode}, not {want}:\n{(p.stderr or p.stdout).strip()[-1500:]}")
            return
        self.verify_receipts(block, scratch, since, p.stdout, env)
        if expected is not None:
            got, want_text = normalize(p.stdout, scratch), normalize(expected.text(), scratch)
            if got != want_text:
                self.report.fail(expected.where, "the output shown is not the output:\n" + diff(want_text, got))
                return
            self.report.ok("output")
        self.report.ok("shell")

    def run_console(self, doc, block, scratch):
        items, cur = [], None
        for line in block.body:
            if line.startswith("$ "):
                cur = [line[2:], []]
                items.append(cur)
            elif cur is not None:
                cur[1].append(line)
        env = self.env(scratch)
        for cmd, out in items:
            static_line_checks(self.cli, block, cmd, self.report, PKGS, PYS)
            p = subprocess.run(["bash", "-o", "pipefail", "-c", cmd], cwd=scratch, env=env, capture_output=True, text=True, timeout=300, stdin=subprocess.DEVNULL)
            if p.returncode != 0:
                self.report.fail(block.where, f"$ {cmd}: exit {p.returncode}:\n{p.stderr.strip()[-1000:]}")
                continue
            got, want = normalize(p.stdout, scratch), normalize("\n".join(out), scratch)
            if got != want:
                self.report.fail(block.where, f"$ {cmd}: the output shown is not the output:\n" + diff(want, got))
            else:
                self.report.ok("console")

    def run_qasm(self, block, scratch):
        name = block.attrs.get("file")
        if name:
            (scratch / name).write_text(block.text(), encoding="utf-8")
        p = subprocess.run([self.args.cli, "lint", "-"], input=block.text(), capture_output=True, text=True, env=self.env(scratch), timeout=60)
        refused = "refused" in block.flags
        if (p.returncode == 0) == refused:
            self.report.fail(block.where, f"the program {'reads' if not refused else 'is refused'}, but the block says otherwise: {(p.stdout + p.stderr).strip()}")
        else:
            self.report.ok("qasm")

    def run_python(self, doc, blocks, scratch):
        if not self.args.python:
            for b in blocks:
                self.report.fail(b.where, "a Python block: give --python, an interpreter with the package installed")
            return
        payload = scratch / ".python-blocks.json"
        payload.write_text(json.dumps([[b.where, b.text()] for b in blocks]), encoding="utf-8")
        driver = scratch / ".python-driver.py"
        driver.write_text(PY_DRIVER, encoding="utf-8")
        env = self.env(scratch)
        since = time.time() - 1
        p = subprocess.run([self.args.python, "-I", str(driver), str(payload)], cwd=scratch, env=env, capture_output=True, text=True, timeout=900)
        done = set(re.findall(r"^__BLOCK_OK__ (\d+)$", p.stdout, re.M))
        for i, b in enumerate(blocks):
            if str(i) in done:
                self.report.ok("python")
            else:
                self.report.fail(b.where, f"the Python block did not run:\n{p.stderr.strip()[-1500:]}")
                break
        self.verify_receipts(blocks[0], scratch, since, "", env)

    def run_rust(self, doc, blocks):
        if not self.args.rust:
            for b in blocks:
                self.report.fail(b.where, "a Rust block: give --rust (cargo builds it)")
            return
        for b in blocks:
            crate = b.attrs.get("crate", "wai-quantum")
            if crate not in PKGS:
                self.report.fail(b.where, f"crate={crate}: no such crate here")
                continue
            proj = self.tmp / f"rust-{re.sub(r'[^A-Za-z0-9]+', '-', doc.rel)}-{b.line}"
            (proj / "examples").mkdir(parents=True, exist_ok=True)
            feats = [f for f in b.attrs.get("features", "").split(",") if f]
            dep = f'{crate} = {{ path = "{PKGS[crate]["dir"].as_posix()}", features = {json.dumps(feats)} }}'
            (proj / "Cargo.toml").write_text(f'[package]\nname = "docs-snippet"\nversion = "0.0.0"\nedition = "2024"\npublish = false\n\n[dependencies]\n{dep}\n\n[workspace]\n', encoding="utf-8")
            lock = PKGS[crate]["dir"] / "Cargo.lock"
            if lock.exists():
                shutil.copy(lock, proj / "Cargo.lock")
            code = b.text()
            if "fn main" not in code:
                uses = "\n".join(l for l in code.split("\n") if l.startswith("use "))
                body = "\n".join(l for l in code.split("\n") if not l.startswith("use "))
                code = f"{uses}\n\n#[allow(unused_variables)]\nfn main() -> Result<(), Box<dyn std::error::Error>> {{\n{body}\nOk(())\n}}\n"
            (proj / "examples" / "snippet.rs").write_text(code, encoding="utf-8")
            (proj / "src").mkdir(exist_ok=True)
            (proj / "src" / "lib.rs").write_text("", encoding="utf-8")
            cmd = ["cargo", "run" if "runnable" in b.flags else "build", "--release", "--quiet", "--example", "snippet"]
            env = dict(os.environ)
            env.setdefault("RUSTUP_TOOLCHAIN", TOOLCHAIN)
            p = subprocess.run(cmd, cwd=proj, env=env, capture_output=True, text=True, timeout=3600)
            if p.returncode != 0:
                self.report.fail(b.where, f"the Rust block does not {'run' if 'runnable' in b.flags else 'build'}:\n{p.stderr.strip()[-2000:]}")
            else:
                self.report.ok("rust")

    def check_translation(self, doc):
        """A translation is marked machine-assisted and holds its English page's code."""
        if not doc.path.name.endswith(".zh-CN.md"):
            return False
        stem = doc.path.name[: -len(".zh-CN.md")]
        english = next((p for p in (doc.path.with_name(stem + ".md"), doc.path.with_name(stem + ".en.md")) if p.exists()), None)
        first = next((l for l in doc.text.split("\n") if l.strip()), "")
        if "机器辅助翻译" not in first or "machine-assisted" not in first:
            self.report.fail(doc.rel, "a translation's first line marks it machine-assisted, to be reviewed by a fluent reader")
        if english is None:
            self.report.fail(doc.rel, "no English page beside the translation")
            return False
        def code(d, lang=True):
            out = []
            for b in d.blocks:
                if b.lang in TEXTS and "output" not in b.flags:
                    continue
                body = [re.sub(r"\s+#.*$|^\s*#.*$", "", l) for l in b.body]
                if lang:
                    body = [l.replace(" --lang zh-CN", "") for l in body]
                out.append((b.info, [l.rstrip() for l in body if l.strip()]))
            return out
        mine, theirs = code(doc), code(Doc(english))
        if mine != theirs:
            at = next((i for i, (a, b) in enumerate(zip(mine, theirs)) if a != b), min(len(mine), len(theirs)))
            self.report.fail(doc.rel, f"its code differs from {english.name}'s, from code block {at + 1} on: a translation holds the same commands and snippets")
            return False
        self.report.ok("translation")
        # The same commands as its English page, which runs them: nothing more to run.
        return code(doc, lang=False) == code(Doc(english), lang=False)

    def run_doc(self, doc):
        scratch = self.scratch_for(doc)
        same = self.check_translation(doc)
        check_prose(self.cli, doc, self.report)
        if same:
            if not self.args.keep:
                shutil.rmtree(scratch, ignore_errors=True)
            return
        pending_python = []
        rust = []
        blocks = doc.blocks
        for i, b in enumerate(blocks):
            nxt = blocks[i + 1] if i + 1 < len(blocks) else None
            if b.lang in TEXTS and "output" in b.flags:
                continue  # held to the block before it
            if "norun" in b.attrs:
                check_norun(self.cli, b, self.report, PKGS, PYS)
                continue
            if b.lang in SHELLS:
                if "runnable" not in b.flags:
                    self.report.fail(b.where, f"a {b.lang} block that neither runs (runnable) nor says why it cannot (norun=REASON)")
                    continue
                expected = nxt if nxt and nxt.lang in TEXTS and "output" in nxt.flags else None
                self.run_shell(doc, b, scratch, expected)
            elif b.lang == "console":
                if "runnable" not in b.flags:
                    self.report.fail(b.where, "a console block that neither runs (runnable) nor says why it cannot (norun=REASON)")
                    continue
                self.run_console(doc, b, scratch)
            elif b.lang in ("python", "py"):
                if "runnable" not in b.flags:
                    self.report.fail(b.where, "a Python block that neither runs (runnable) nor says why it cannot (norun=REASON)")
                    continue
                pending_python.append(b)
            elif b.lang == "rust":
                if not ({"runnable", "compile"} & b.flags):
                    self.report.fail(b.where, "a Rust block that is neither runnable nor compile, nor says why it cannot run (norun=REASON)")
                    continue
                rust.append(b)
            elif b.lang == "qasm":
                self.run_qasm(b, scratch)
            elif b.lang == "toml":
                check_toml(b, self.report, PKGS)
            elif b.lang == "json":
                if "fragment" in b.flags:
                    continue
                try:
                    json.loads(b.text())
                    self.report.ok("json")
                except ValueError as e:
                    self.report.fail(b.where, f"the JSON does not parse: {e}")
            elif b.lang in TEXTS:
                for line in b.body:
                    l = line.strip()
                    if re.match(r"^(\$ )?(wai-quantum|cargo|curl|node|npm|pip|python3?) ", l):
                        self.report.fail(b.where, f"a command in a text block: `{l[:60]}`; make it a shell block that runs, or say why it cannot (norun=REASON)")
                        break
        if pending_python:
            self.run_python(doc, pending_python, scratch)
        if rust:
            self.run_rust(doc, rust)
        if not self.args.keep:
            shutil.rmtree(scratch, ignore_errors=True)


PY_DRIVER = r'''
import json, sys, traceback
blocks = json.load(open(sys.argv[1], encoding="utf-8"))
g = {"__name__": "__main__"}
for i, (where, code) in enumerate(blocks):
    try:
        exec(compile(code, where, "exec"), g)
    except BaseException:
        traceback.print_exc()
        print(f"__BLOCK_FAIL__ {i}", flush=True)
        sys.exit(1)
    print(f"__BLOCK_OK__ {i}", flush=True)
'''


def diff(want, got):
    import difflib
    return "\n".join(list(difflib.unified_diff(want.split("\n"), got.split("\n"), "shown", "run", lineterm=""))[:60])


PKGS = {}
PYS = {}
_tc = re.search(r'channel\s*=\s*"([^"]+)"', (ROOT / "rust-toolchain.toml").read_text(encoding="utf-8")) if (ROOT / "rust-toolchain.toml").exists() else None
TOOLCHAIN = _tc.group(1) if _tc else "stable"


def main():
    global PKGS, PYS
    ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
    ap.add_argument("--cli", required=True, help="the wai-quantum binary")
    ap.add_argument("--service", help="a job service to run the service's documented commands against")
    ap.add_argument("--python", help="a Python with the wai_quantum package installed")
    ap.add_argument("--rust", action="store_true", help="build and run the Rust blocks with cargo")
    ap.add_argument("--only", action="append", default=[], help="only the documents whose path holds this")
    ap.add_argument("--skip", action="append", default=[], help="leave out the documents whose path holds this")
    ap.add_argument("--keep", action="store_true", help="keep the scratch directories")
    args = ap.parse_args()
    if not os.access(args.cli, os.X_OK):
        print(f"--cli {args.cli}: not an executable", file=sys.stderr)
        return 2
    args.cli = str(Path(args.cli).resolve())
    PKGS, PYS = packages(), python_packages()
    manifest = json.loads((HERE / "docs.json").read_text(encoding="utf-8"))
    report = Report()
    paths = []
    for pat in manifest["docs"]:
        if args.only and not any(o in pat or fnmatch.fnmatch(o, pat) for o in args.only):
            continue
        hits = sorted(p for p in ROOT.glob(pat) if p.is_file())
        if not hits:
            report.fail("docs.json", f"{pat} names no file")
        paths.extend(h for h in hits if h not in paths)
    if args.only:
        paths = [p for p in paths if any(o in p.relative_to(ROOT).as_posix() for o in args.only)]
    paths = [p for p in paths if not any(k in p.relative_to(ROOT).as_posix() for k in args.skip)]
    runner = Runner(args, report)
    try:
        for p in paths:
            doc = Doc(p)
            before = len(report.fails)
            t0 = time.time()
            runner.run_doc(doc)
            n = len(report.fails) - before
            print(f"{'ok  ' if n == 0 else 'FAIL'}  {doc.rel} ({len(doc.blocks)} blocks, {time.time() - t0:.1f} s){'' if n == 0 else f': {n} failed'}", flush=True)
    finally:
        if not args.keep:
            shutil.rmtree(runner.tmp, ignore_errors=True)
        else:
            print(f"scratch kept in {runner.tmp}")
    counts = ", ".join(f"{v} {k}" for k, v in sorted(report.counts.items()))
    print(f"\n{len(paths)} documents; checked: {counts}")
    if report.fails:
        print(f"{len(report.fails)} check(s) failed")
        return 1
    print("every check passed")
    return 0


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