#!/usr/bin/env python3 """check_tool_executor_lifecycle_parity.py — LIVE parity for the tool_executor lifecycle family gap ports (dichotomy wave #2479, contract factory-heavy-te-life). Oracle: src/rune/tool_executor.py imported LIVE (in-process). Go side: the tool-executor package via `go test -run TestLifecycleParityDriver` with LIFECYCLE_CASES / LIFECYCLE_OUT (the driver normalizes identically). Design: - cases are authored with {{TMP}}/{{WROOT}} placeholder paths, then materialized into TWO isolated root pairs (oracle side A, Go side B) — neither side inherits the other's side effects (a confirm_write by the oracle must not make the Go run back the file up); - each side's own roots are re-anonymized to {{TMP}}/{{WROOT}} in its outputs before comparison, so identical envelopes compare equal; - normalizations identical on both sides: uuid token → {{TOKEN}}, operation_digest → {{DIGEST}} (structural invariant), diff → summary (head 2 + plus/minus), float-canon via canonical JSON dump; - the broker stub rides the find_cap cases verbatim. Excluded (documented divergences, verification/te-life-summary.md): sibling-gap dispatch arms (notPorted), bool-offset footer rendering, git/lint/format/test EXECUTION rows (environment binaries). Run: python3 scripts/check_tool_executor_lifecycle_parity.py Exit 0 only at 100%. """ import copy import json import os import subprocess import sys import tempfile from pathlib import Path REPO = Path(__file__).resolve().parent.parent sys.path.insert(0, str(REPO / "src")) from rune import tool_executor as te # noqa: E402 W20 = "".join(f"line {i}\n" for i in range(1, 21)) STUB_NONE_QUERY = "run the whole test suite" STUB_TWO_QUERY = "commit my changes" STUB_MATCHES = [ { "name": "git_commit", "description": "Commit staged changes", "risk": "irreversible", "parameters": {"properties": {"message": {"type": "string"}, "files": {"type": "array"}}}, }, {"name": "snapshot_tree", "description": "Snapshot a directory tree", "parameters": {"properties": {}}}, ] FIXTURE_FILES = { "{{TMP}}/w.txt": W20, "{{TMP}}/a.txt": "the spine rebuild parity\n", "{{WROOT}}/exists.txt": "already here\n", } def summarize_diff(diff): lines = diff.split("\n") return { "head": lines[:2], "plus": sum(1 for l in lines if l.startswith("+") and not l.startswith("+++")), "minus": sum(1 for l in lines if l.startswith("-") and not l.startswith("---")), } def norm_envelope(env): out = copy.deepcopy(env) if "diff" in out: out["diff"] = summarize_diff(out["diff"]) return out def norm_prepare(env): out = norm_envelope(env) if isinstance(out.get("pending"), te.PreparedTool): ser = te.serialize_prepared_tool(out["pending"]) ser["token"] = "{{TOKEN}}" ser["operation_digest"] = "{{DIGEST}}" out["pending"] = ser return out def build_cases(): cases = [] def add(op, name, **kw): cases.append({"op": op, "name": name, **kw}) def exec_row(name, tool, args): cases.append({"op": "exec_now", "name": name, "tool": tool, "args": args}) def exec_tool_row(name, tool, args, now=None, ttl=None): kw = {} if now is not None: kw["now"] = now if ttl is not None: kw["ttl"] = ttl cases.append({"op": "execute_tool", "name": name, "tool": tool, "args": args, **kw}) # _execute_tool_now exec_row("read_window", "read_file", {"path": "{{TMP}}/w.txt", "offset": 2, "limit": 3}) exec_row("read_default", "read_file", {"path": "{{TMP}}/a.txt"}) exec_row("read_missing", "read_file", {"path": "{{TMP}}/missing.txt"}) exec_row("read_blocked", "read_file", {"path": "/etc/hostname"}) exec_row("read_offset_none", "read_file", {"path": "{{TMP}}/w.txt", "offset": None}) exec_row("read_limit_str", "read_file", {"path": "{{TMP}}/w.txt", "limit": "x"}) exec_row("read_offset_frac", "read_file", {"path": "{{TMP}}/w.txt", "offset": 1.5}) exec_row("read_offset_zero", "read_file", {"path": "{{TMP}}/w.txt", "offset": 0, "limit": 2}) exec_row("unknown", "no_such_tool", {"x": 1}) exec_row("list_hit", "list_directory", {"path": "{{TMP}}"}) exec_row("list_pattern", "list_directory", {"path": "{{TMP}}", "pattern": "w*"}) exec_row("list_not_dir", "list_directory", {"path": "{{TMP}}/nodir"}) exec_row("grep_no_pattern", "grep_files", {"path": "{{TMP}}"}) exec_row("grep_no_match", "grep_files", {"pattern": "zzz_never", "path": "{{TMP}}"}) exec_row("grep_denied", "grep_files", {"pattern": "x", "path": "/etc/hostname"}) exec_row("write_preview", "write_file", {"path": "{{WROOT}}/exec-new.txt", "content": "alpha\nbeta\n"}) exec_row("write_no_change", "write_file", {"path": "{{WROOT}}/exists.txt", "content": "already here\n"}) exec_row("write_denied_outside", "write_file", {"path": "/usr/share/x.txt", "content": "x"}) exec_row("write_denied_protected", "write_file", {"path": "/etc/passwd", "content": "x"}) exec_row("git_commit_no_message", "git_commit", {}) exec_row("run_tests_no_runner", "run_tests", {"path": "{{TMP}}/a.txt"}) exec_row("lint_missing", "lint_file", {"path": "{{TMP}}/missing.py"}) exec_row("format_denied", "format_code", {"path": "/etc/f.py"}) # serialize / deserialize full = te.serialize_prepared_tool(te.PreparedTool( token="ser-full-token", name="write_file", arguments={"path": "/tmp/x.txt", "content": "abc"}, operation_digest="d" * 64, created_at=10.0, expires_at=310.0, requested_path="/tmp/x.txt", resolved_path="/tmp/x.txt", path_digest="p" * 64, )) minimal = te.serialize_prepared_tool(te.PreparedTool( token="ser-min-token", name="run_command", arguments={"command": "ls"}, operation_digest="e" * 64, created_at=0.0, expires_at=5.0, )) add("serialize", "full", pending=full) add("serialize", "minimal", pending=minimal) def deser_row(name, value): add("deserialize", name, value=value) deser_row("ok_full", full) deser_row("ok_minimal", minimal) deser_row("not_object_str", "nope") deser_row("not_object_null", None) deser_row("extra_field", dict(minimal, extra=1)) deser_row("token_empty", dict(minimal, token="")) deser_row("token_number", dict(minimal, token=5)) deser_row("arguments_list", dict(minimal, arguments=[])) deser_row("created_bool", dict(minimal, created_at=True)) deser_row("expires_null", dict(minimal, expires_at=None)) # confirm_write add("confirm_write", "new_file", path="{{WROOT}}/confirmed-new.txt", content="abc") add("confirm_write", "protected", path="/etc/x.txt", content="x") add("confirm_write", "outside", path="/usr/share/x.txt", content="x") # execute_tool / prepare_tool exec_tool_row("readonly_passthrough", "read_file", {"path": "{{TMP}}/a.txt"}) exec_tool_row("write_prepare", "write_file", {"path": "{{WROOT}}/exec-prep.txt", "content": "one\ntwo\nthree\n"}, now=1000.0) exec_tool_row("format_prepare", "format_code", {"path": "{{WROOT}}/fmt-prep.go"}, now=2000.0) exec_tool_row("write_ttl", "write_file", {"path": "{{WROOT}}/prep-ttl.txt", "content": "x"}, now=5.0, ttl=60.0) exec_tool_row("write_denied", "write_file", {"path": "/usr/share/x.txt", "content": "x"}) exec_tool_row("edit_no_path", "edit_file", {"search": "a", "replace": "b"}) exec_tool_row("args_null", "write_file", None) # find_capability (frozen stub) add("find_cap", "query_missing", args={}, matches=[]) add("find_cap", "query_blank", args={"query": " "}, matches=[]) add("find_cap", "no_matches", args={"query": STUB_NONE_QUERY}, matches=[]) add("find_cap", "two_matches", args={"query": STUB_TWO_QUERY}, matches=copy.deepcopy(STUB_MATCHES)) return cases def subst(obj, tmp_root, w_root): if isinstance(obj, str): return obj.replace("{{TMP}}", tmp_root).replace("{{WROOT}}", w_root) if isinstance(obj, list): return [subst(x, tmp_root, w_root) for x in obj] if isinstance(obj, dict): return {k: subst(v, tmp_root, w_root) for k, v in obj.items()} return obj def run_oracle(cases, tmp_root, w_root): out = {} for c in cases: c = subst(c, tmp_root, w_root) op, name = c["op"], c["name"] if op == "exec_now": r = te._execute_tool_now(c["tool"], copy.deepcopy(c["args"])) out[(op, name)] = norm_envelope(r) elif op == "serialize": p = te.PreparedTool( token=c["pending"]["token"], name=c["pending"]["name"], arguments=c["pending"]["arguments"], operation_digest=c["pending"]["operation_digest"], created_at=c["pending"]["created_at"], expires_at=c["pending"]["expires_at"], requested_path=c["pending"].get("requested_path"), resolved_path=c["pending"].get("resolved_path"), path_digest=c["pending"].get("path_digest"), ) out[(op, name)] = te.serialize_prepared_tool(p) elif op == "deserialize": try: p = te.deserialize_prepared_tool(copy.deepcopy(c["value"])) out[(op, name)] = te.serialize_prepared_tool(p) except ValueError as exc: out[(op, name)] = {"raises": str(exc)} elif op == "confirm_write": out[(op, name)] = norm_envelope(te.confirm_write(c["path"], c["content"])) elif op == "execute_tool": kwargs = {} if "now" in c: kwargs["now"] = c["now"] if "ttl" in c: kwargs["ttl"] = c["ttl"] if kwargs: r = te.prepare_tool(c["tool"], copy.deepcopy(c["args"]), **kwargs) else: r = te.execute_tool(c["tool"], copy.deepcopy(c["args"])) out[(op, name)] = norm_prepare(r) elif op == "find_cap": import rune.capability_broker as broker original = broker.find_capability broker.find_capability = lambda q: copy.deepcopy(c["matches"]) try: r = te._find_capability(copy.deepcopy(c["args"])) finally: broker.find_capability = original out[(op, name)] = norm_envelope(r) return out def main(): cases = build_cases() roots = {} for side in ("py", "go"): tmp_root = tempfile.mkdtemp(prefix=f"rune-life-parity-{side}-", dir="/tmp") w_root = tempfile.mkdtemp(prefix=f"rune-life-parity-{side}w-", dir=os.path.expanduser("~")) for rel, content in FIXTURE_FILES.items(): p = Path(subst(rel, tmp_root, w_root)) p.parent.mkdir(parents=True, exist_ok=True) p.write_text(content, encoding="utf-8") roots[side] = (tmp_root, w_root) py_out = run_oracle(cases, *roots["py"]) workdir = Path(tempfile.mkdtemp(prefix="rune-life-driver-")) cases_file = workdir / "cases.json" out_file = workdir / "go_out.json" cases_file.write_text(json.dumps(subst(cases, *roots["go"]))) env = dict(os.environ, LIFECYCLE_CASES=str(cases_file), LIFECYCLE_OUT=str(out_file)) r = subprocess.run( ["go", "test", "-run", "TestLifecycleParityDriver", "./..."], cwd=REPO / "src" / "go" / "tool-executor", capture_output=True, text=True, env=env, timeout=300, ) if r.returncode != 0: print(r.stdout[-3000:]) print(r.stderr[-3000:]) sys.exit("Go parity driver failed") go_rows = json.loads(out_file.read_text()) go_out = {(row["op"], row["name"]): row["out"] for row in go_rows} def anonymize(obj, tmp_root, w_root): if isinstance(obj, str): return obj.replace(tmp_root, "{{TMP}}").replace(w_root, "{{WROOT}}") if isinstance(obj, list): return [anonymize(x, tmp_root, w_root) for x in obj] if isinstance(obj, dict): return {k: anonymize(v, tmp_root, w_root) for k, v in obj.items()} return obj def canon(v): # int→float pre-pass: Go json renders 10.0 as 10; Python as 10.0 — # numerically identical envelopes must compare equal. def lift(x): if isinstance(x, bool) or isinstance(x, int): if isinstance(x, bool): return x return float(x) if isinstance(x, list): return [lift(i) for i in x] if isinstance(x, dict): return {k: lift(i) for k, i in x.items()} return x return json.dumps(lift(v), sort_keys=True, default=str) fails = 0 checked = 0 for c in cases: key = (c["op"], c["name"]) want = anonymize(py_out[key], *roots["py"]) got = anonymize(go_out[key], *roots["go"]) checked += 1 if canon(want) != canon(got): fails += 1 print(f"FAIL {key}:\n want {canon(want)[:500]}\n got {canon(got)[:500]}") print(f"lifecycle live parity: {checked - fails}/{checked}") sys.exit(1 if fails else 0) if __name__ == "__main__": main()