#!/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())