clay-borg/tools/replay-test.py

#!/usr/bin/env python3
"""K10 acceptance: a replay bundle must re-execute, and must be able to fail.

CB-WP-0006 T06, implementing ADR-0005 §6's four controls verbatim. The
reviewer supplied two of them, and both are the kind a round-trip test
usually omits.

Without controls, `make replay-test` reports `ok` under at least four
silent failures:

  * no scenario fails, so zero bundles are round-tripped;
  * the writer emits nothing and identical error strings satisfy
    "same failure both times";
  * the replayed hash is compared to one recomputed in the same process —
    `assert_eq!(h, h)`, which is the AM-7 defect exactly;
  * the truncation path is never exercised, so K11's operative clause is
    dead code while the gate is green.

So each control below **corrupts a real bundle and requires the replay to
reject it**. A round-trip that cannot fail proves nothing.

Usage:
  python3 tools/replay-test.py
  python3 tools/replay-test.py --self-test
"""
import os
import shutil
import subprocess
import sys

from repo import ROOT, cargo_bin, cargo_env, enter_root

FIXTURE = "scenarios/fixtures/k10-deliberate-failure.yaml"
REPLAY_DIR = "replays"
BUNDLE = f"{REPLAY_DIR}/ground-k10-deliberate-failure.cbreplay"
# cb_events::store — magic (5) + version (1).
HEADER_LEN = 6


def sim(*args):
    """(exit code, combined output) for one cb-sim invocation."""
    cargo = cargo_bin()
    r = subprocess.run(
        [cargo, "run", "-q", "-p", "cb-sim", "--", *args],
        cwd=ROOT, env=cargo_env(), capture_output=True, text=True)
    return r.returncode, (r.stdout + r.stderr)


def produce_bundle():
    """Run the deliberately-failing fixture; return (bundles, output)."""
    shutil.rmtree(os.path.join(ROOT, REPLAY_DIR), ignore_errors=True)
    code, out = sim(FIXTURE)
    base = os.path.join(ROOT, REPLAY_DIR)
    bundles = sorted(d for d in os.listdir(base)) if os.path.isdir(base) else []
    return code, out, bundles


def read(name):
    with open(os.path.join(ROOT, BUNDLE, name), "rb") as fh:
        return fh.read()


def write(name, data):
    with open(os.path.join(ROOT, BUNDLE, name), "wb") as fh:
        fh.write(data)


def run():
    results = []

    def check(name, ok, detail=""):
        results.append((name, ok, detail))
        return ok

    print("K10 replay acceptance — a bundle must re-execute, and must be "
          "able to fail\n")

    # --- Control 1: something must actually fail, or zero bundles exist ---
    code, out, bundles = produce_bundle()
    check("the fixture fails (all 21 real scenarios pass)", code != 0,
          f"cb-sim exit {code}")
    check("exactly one bundle was produced", len(bundles) == 1,
          f"{len(bundles)} bundle(s): {bundles}")
    for f in ("manifest.yaml", "commands.log", "initial.snapshot",
              "expected.yaml"):
        check(f"bundle contains {f}",
              os.path.isfile(os.path.join(ROOT, BUNDLE, f)))
    # A writer that emitted empty files would satisfy "contains" above.
    check("bundle files are non-empty",
          all(len(read(f)) > 0 for f in
              ("manifest.yaml", "commands.log", "initial.snapshot",
               "expected.yaml")))

    good_log = read("commands.log")
    good_manifest = read("manifest.yaml")

    # --- The happy path, once, so the negatives mean something ---
    code, out = sim("--replay", BUNDLE)
    check("a well-formed bundle replays and reproduces the recorded hash",
          code == 0 and "reproduced" in out, out.strip().splitlines()[0][:70])

    # --- Control 2: the hash is READ FROM the bundle, not recomputed ---
    # Corrupt only the recorded hash. If the replay recomputed its own
    # comparison value, this would still pass — `assert_eq!(h, h)`.
    m = good_manifest.decode()
    tampered = m.replace("end_state_hash: ", "end_state_hash: ff", 1)
    write("manifest.yaml", tampered.encode())
    code, out = sim("--replay", BUNDLE)
    check("a tampered recorded hash fails the replay",
          code != 0 and "did not reproduce" in out,
          "proves the comparison value comes from the bundle")
    write("manifest.yaml", good_manifest)

    # --- Control 3a: truncate the command log by one byte ---
    write("commands.log", good_log[:-1])
    code, out = sim("--replay", BUNDLE)
    check("a command log short by one byte is rejected",
          code != 0 and "truncated tail" in out, out.strip().splitlines()[0][:70])

    # --- Control 3b: corrupt the length prefix ---
    bad = bytearray(good_log)
    bad[HEADER_LEN] = 0xFF
    bad[HEADER_LEN + 1] = 0xFF
    write("commands.log", bytes(bad))
    code, out = sim("--replay", BUNDLE)
    check("a corrupted length prefix is rejected",
          code != 0 and ("truncated tail" in out or "exceeds" in out),
          out.strip().splitlines()[0][:70])
    write("commands.log", good_log)

    # --- Control 4: a mutated seed must fail to reproduce ---
    m = good_manifest.decode()
    seed_line = next(ln for ln in m.splitlines() if ln.startswith("seed:"))
    mutated = m.replace(seed_line, "seed: 999", 1)
    check("the seed mutation actually changed the manifest", mutated != m)
    write("manifest.yaml", mutated.encode())
    code, out = sim("--replay", BUNDLE)
    check("a mutated seed fails the replay",
          code != 0 and "inconsistent" in out,
          "without this the seed would be inert — replay restores from the "
          "snapshot")
    write("manifest.yaml", good_manifest)

    # --- Restored bundle must still replay, or the controls corrupted it ---
    code, out = sim("--replay", BUNDLE)
    check("the bundle still replays after every control restored it",
          code == 0, "controls must not leave the tree broken")

    ok = True
    for name, passed, detail in results:
        print(f"  [{'ok  ' if passed else 'FAIL'}] {name}"
              + (f"\n         {detail}" if detail else ""))
        ok &= passed
    print(f"\n  {sum(1 for _, p, _ in results if p)}/{len(results)} controls "
          f"passed")
    return 0 if ok else 1


def self_test():
    """Positive controls on the harness itself."""
    results = []

    def check(name, ok, detail=""):
        results.append((name, ok, detail))

    check("the deliberately-failing fixture is committed",
          os.path.isfile(os.path.join(ROOT, FIXTURE)), FIXTURE)
    # It must not be in the corpus `make sim` and `make coverage` sweep, or
    # it would fail the build and pollute AM-1.
    check("the fixture is outside scenarios/ground/",
          "scenarios/ground/" not in FIXTURE,
          "a deliberately-failing scenario in the corpus would fail `make sim`")
    text = open(os.path.join(ROOT, FIXTURE)).read()
    check("the fixture claims no rules", "covers: []" in text,
          "a fixture that claimed rules would inflate AM-1")
    check("cb-sim is locatable", bool(cargo_bin()))

    print("replay-test self-test (positive control)")
    ok = True
    for name, passed, detail in results:
        print(f"  [{'ok ' if passed else 'FAIL'}] {name}"
              + (f"  — {detail}" if detail else ""))
        ok &= passed
    return 0 if ok else 1


def main():
    enter_root()
    if "--self-test" in sys.argv:
        return self_test()
    return run()


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