nash/corpus/replay.py

#!/usr/bin/env python3
"""Transcript-replay compat harness for nash (docs/NASH.md §10.3, M0 gate).

Runs every corpus command under bash and nash in identical fresh workspaces and
compares exit code, stdout, and resulting filesystem state. stderr is compared
separately and does NOT count against parity (error-message wording legitimately
differs between shells); a stderr-only difference is reported for review.

Usage: replay.py [--nash PATH] [--bash PATH] [--corpus PATH] [--json PATH]
Parity target (M0/M3 gate): >= 99%.
"""

import argparse
import hashlib
import json
import os
import shutil
import subprocess
import sys
import tempfile

FIXTURES = {
    "a.txt": "first line\nsecond line\nthird line\n",
    "b.txt": "banana\napple\nbanana\ncherry\napple\n",
    "data.csv": "name,team,score\nalice,red,10\nbob,blue,7\ncarol,red,9\ndave,blue,7\n",
    "logs/app.log": (
        "2026-01-01 INFO boot ok\n"
        "2026-01-01 ERROR disk full\n"
        "2026-01-01 INFO retry\n"
        "2026-01-01 WARN slow\n"
        "2026-01-01 ERROR net down\n"
        "2026-01-01 INFO done\n"
    ),
    "src/main.py": "print('main')\n",
    "src/util.py": "def add(a, b):\n    return a + b\n",
    "README.md": "# Sample\n",
}

TIMEOUT_S = 20


def seed(workdir):
    for rel, content in FIXTURES.items():
        path = os.path.join(workdir, rel)
        os.makedirs(os.path.dirname(path), exist_ok=True)
        with open(path, "w") as f:
            f.write(content)


def snapshot(workdir):
    """Relative path -> sha256 of contents, for every file in the tree."""
    out = {}
    for root, _dirs, files in os.walk(workdir):
        for name in files:
            path = os.path.join(root, name)
            rel = os.path.relpath(path, workdir)
            try:
                with open(path, "rb") as f:
                    out[rel] = hashlib.sha256(f.read()).hexdigest()
            except OSError:
                out[rel] = "<unreadable>"
    return out


def run_one(shell, cmd, mode="-c", extra_env=None):
    parent = tempfile.mkdtemp(prefix="nash-corpus-")
    workdir = os.path.join(parent, "workspace")
    os.makedirs(workdir)
    seed(workdir)
    env = {
        "PATH": "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin",
        "HOME": workdir,
        "LC_ALL": "C",
        "LANG": "C",
        "TERM": "dumb",
        "SHELL": shell,
    }
    if extra_env:
        env.update(extra_env)
    argv = [shell, mode, cmd] if mode else [shell, cmd]
    try:
        proc = subprocess.run(
            argv, cwd=workdir, env=env, capture_output=True, timeout=TIMEOUT_S
        )
        code, out, err = proc.returncode, proc.stdout, proc.stderr
    except subprocess.TimeoutExpired:
        code, out, err = "TIMEOUT", b"", b""
    fs = snapshot(workdir)
    shutil.rmtree(parent, ignore_errors=True)

    def norm(b):
        text = b.decode("utf-8", "replace")
        return text.replace(workdir, "__WORK__").replace(parent, "__TMP__")

    return {"code": code, "stdout": norm(out), "stderr": norm(err), "fs": fs}


def classify(b, n):
    core_equal = (
        b["code"] == n["code"] and b["stdout"] == n["stdout"] and b["fs"] == n["fs"]
    )
    if core_equal and b["stderr"] == n["stderr"]:
        return "PASS"
    if core_equal:
        return "STDERR_ONLY"
    return "DIVERGE"


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--nash", default=os.environ.get("NASH_BIN", "nash"))
    ap.add_argument("--bash", default="/bin/bash")
    ap.add_argument(
        "--corpus",
        default=os.path.join(os.path.dirname(os.path.abspath(__file__)), "corpus.jsonl"),
    )
    ap.add_argument("--json", help="also write the full report to this path")
    args = ap.parse_args()

    entries = []
    with open(args.corpus) as f:
        for line in f:
            line = line.strip()
            if line:
                entries.append(json.loads(line))

    results = []
    for entry in entries:
        b = run_one(args.bash, entry["cmd"])
        n = run_one(args.nash, entry["cmd"])
        verdict = classify(b, n)
        results.append({"id": entry["id"], "cmd": entry["cmd"], "verdict": verdict,
                        "bash": b, "nash": n})
        marker = {"PASS": ".", "STDERR_ONLY": "s", "DIVERGE": "X"}[verdict]
        print(marker, end="", flush=True)
    print()

    # Invocation-mode smoke tests (Nucleic exec paths use -lc, scripts, stdin).
    modes = []
    for mode_id, mode, cmd in [
        ("mode--lc", "-lc", "echo login-mode"),
        ("mode--c-args", "-c", "echo argv0-test"),
    ]:
        b = run_one(args.bash, cmd, mode=mode)
        n = run_one(args.nash, cmd, mode=mode)
        modes.append({"id": mode_id, "verdict": classify(b, n), "bash": b, "nash": n})

    passes = sum(1 for r in results if r["verdict"] in ("PASS", "STDERR_ONLY"))
    stderr_only = [r for r in results if r["verdict"] == "STDERR_ONLY"]
    diverges = [r for r in results if r["verdict"] == "DIVERGE"]
    parity = 100.0 * passes / len(results) if results else 0.0

    print(f"\ncorpus: {len(results)}  parity: {passes}/{len(results)} = {parity:.1f}%")
    print(f"  clean pass:  {len(results) - len(stderr_only) - len(diverges)}")
    print(f"  stderr-only: {len(stderr_only)}  ({', '.join(r['id'] for r in stderr_only) or '-'})")
    print(f"  diverge:     {len(diverges)}")
    for r in diverges:
        print(f"\nDIVERGE {r['id']}: {r['cmd']!r}")
        print(f"  bash: code={r['bash']['code']} stdout={r['bash']['stdout']!r}")
        print(f"  nash: code={r['nash']['code']} stdout={r['nash']['stdout']!r}")
        if r["bash"]["fs"] != r["nash"]["fs"]:
            keys = set(r["bash"]["fs"]) ^ set(r["nash"]["fs"])
            same = {
                k for k in set(r["bash"]["fs"]) & set(r["nash"]["fs"])
                if r["bash"]["fs"][k] != r["nash"]["fs"][k]
            }
            print(f"  fs delta: only-one-side={sorted(keys)} content-differs={sorted(same)}")
    for m in modes:
        status = "ok" if m["verdict"] in ("PASS", "STDERR_ONLY") else "DIVERGE"
        print(f"mode {m['id']}: {status}")

    if args.json:
        with open(args.json, "w") as f:
            json.dump({"parity_percent": parity, "results": results, "modes": modes}, f, indent=1)

    gate = parity >= 99.0 and all(m["verdict"] != "DIVERGE" for m in modes)
    print(f"\nM0 gate (>=99% parity, modes ok): {'PASS' if gate else 'FAIL'}")
    return 0 if gate else 1


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