#!/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] [--observe-spool DIR] Parity target (M0/M3 gate): >= 99%. --observe-spool DIR turns nash observation ON during the replay (spool transport; the env is set identically for bash, where it is inert), proving the gate hooks don't perturb command semantics (docs/NASH.md §10.1). """ 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] = "" 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") ap.add_argument("--observe-spool", help="enable nash observation, spooling to this dir") args = ap.parse_args() observe_env = None if args.observe_spool: os.makedirs(args.observe_spool, exist_ok=True) observe_env = { "NUCLEIC_SHELL_SPOOL": args.observe_spool, "NUCLEIC_SESSION_ID": "corpus-replay", } 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"], extra_env=observe_env) n = run_one(args.nash, entry["cmd"], extra_env=observe_env) 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())