rune/scripts/check_cron_engine_parity.py

#!/usr/bin/env python3
"""check_cron_engine_parity.py — LIVE parity for the cron_engine GO port
(#2479 dichotomy gap-close): drives the Python oracle (src/rune/
cron_engine.py) and the GO binary through IDENTICAL scenarios and diffs
their JSON outputs — beyond frozen-truth replay, this exercises the two
implementations LIVE (real subprocesses, real files) under a pinned
clock + hermetic HOME.

Python side mirrors the Go binary's `scenario` mode exactly (op set,
output shape). Paths are tokenized to <HOME> before comparison.

Usage: python3 scripts/check_cron_engine_parity.py <go-binary>
  (build first: cd src/go/cron-engine && go build -o /tmp/cron-engine .)
"""
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"))

os.environ.setdefault("RUNE_CRON_ENGINE_CORE", "python")

from rune import cron_engine as ce  # noqa: E402
import rune.subsystem_health as _SH  # noqa: E402

_SH.record_failure = lambda *a, **k: True  # hermetic: no ledger writes

FROZEN_NOW = "2026-10-13T09:20:00"


def frozen_clock(iso):
    from datetime import datetime, timezone

    base = datetime.fromisoformat(iso)

    class FrozenDatetime(datetime):
        @classmethod
        def now(cls, tz=None):
            return base.replace(tzinfo=tz) if tz is not None else base

    return FrozenDatetime


class StopTime:
    def __init__(self, engine):
        self._e = engine

    def sleep(self, _s):
        if self._e is not None:
            self._e.running = False


def run_python(sc):
    ce.datetime = frozen_clock(sc["frozen_now"])
    home = tempfile.mkdtemp(prefix="ce-live-py-")
    os.environ["HOME"] = home
    storage = os.path.join(home, ".rune", "cron.json")
    logp = os.path.join(home, ".rune", "cron.log")

    e = ce.CronEngine.__new__(ce.CronEngine)
    e.storage_path = storage
    e.tasks = {}
    e.running = False
    e.thread = None
    e._last_run_minute = {}
    e._load()

    def rd(p):
        return open(p).read() if os.path.exists(p) else None

    def dump():
        return {
            "running": e.running,
            "storage_path": e.storage_path,
            "log_path": logp,
            "tasks": [{"name": n, "task": t.to_dict()} for n, t in e.tasks.items()],
            "_last_run_minute": dict(e._last_run_minute),  # snapshot (no live aliasing)
        }

    steps = []
    for op in sc["ops"]:
        k = op["op"]
        if k == "add":
            t = e.add_task(op["name"], op["expr"], op["command"], op.get("description", ""))
            steps.append({"op": "add", "task": t.to_dict()})
        elif k == "enable":
            steps.append({"op": "enable", "ok": e.enable_task(op["name"])})
        elif k == "disable":
            steps.append({"op": "disable", "ok": e.disable_task(op["name"])})
        elif k == "remove":
            steps.append({"op": "remove", "ok": e.remove_task(op["name"])})
        elif k == "run":
            steps.append({"op": "run", "result": e.run_task(op["name"])})
        elif k == "fire":
            steps.append({"op": "fire", "fired": e.fire_on_task_complete()})
        elif k == "loop":
            real = ce.time
            ce.time = StopTime(e)
            try:
                e._run_loop()
            finally:
                ce.time = real
            steps.append({"op": "loop"})
        elif k == "start":
            real = ce.time
            ce.time = StopTime(e)
            try:
                e.start()
                e.stop()
            finally:
                ce.time = real
            steps.append({"op": "start+stop"})
        elif k == "stop":
            e.stop()
            steps.append({"op": "stop"})
        elif k == "save":
            e._save()
            steps.append({"op": "save", "ok": True})
        elif k == "load":
            e._load()
            steps.append({"op": "load", "ok": True})
        elif k == "storage":
            steps.append({"op": "storage", "bytes": rd(storage)})
        elif k == "log":
            steps.append({"op": "log", "bytes": rd(logp)})
        elif k == "state":
            steps.append({"op": "state", "state": dump()})
        else:
            raise ValueError(k)
    return {"home": home, "out": {"steps": steps, "final": dump()}}


def run_go(binary, sc_path):
    r = subprocess.run([binary, "scenario", str(sc_path)],
                       capture_output=True, text=True, timeout=120)
    if r.returncode != 0:
        raise RuntimeError(f"go scenario failed: {r.stderr}")
    # the binary's temp home differs per run — recover it from final.storage_path
    out = json.loads(r.stdout)
    home = os.path.dirname(os.path.dirname(out["final"]["storage_path"]))
    return {"home": home, "out": out}


def tokenize(obj, home):
    if isinstance(obj, str):
        return obj.replace(home, "<HOME>")
    if isinstance(obj, list):
        return [tokenize(x, home) for x in obj]
    if isinstance(obj, dict):
        return {k: tokenize(v, home) for k, v in obj.items()}
    return obj


SCENARIOS = [
    # 1: lifecycle + run envelopes + storage/log bytes
    {"frozen_now": FROZEN_NOW, "ops": [
        {"op": "add", "name": "z-first", "expr": "on_complete", "command": "echo one"},
        {"op": "add", "name": "a-second", "expr": "*/5 * * * *", "command": "echo '中文 twø'", "description": "d-ü"},
        {"op": "add", "name": "bad", "expr": "* * * * *", "command": "nope-cmd-xyz"},
        {"op": "add", "name": "quoted", "expr": "* * * * *", "command": "echo 'unterminated"},
        {"op": "state"},
        {"op": "run", "name": "z-first"},
        {"op": "run", "name": "bad"},
        {"op": "run", "name": "quoted"},
        {"op": "run", "name": "ghost"},
        {"op": "disable", "name": "a-second"},
        {"op": "fire"},
        {"op": "storage"},
        {"op": "log"},
        {"op": "state"},
    ]},
    # 2: scheduler loop — matching, non-matching, dedup, disabled
    {"frozen_now": FROZEN_NOW, "ops": [
        {"op": "add", "name": "match", "expr": "* * * * *", "command": "echo m"},
        {"op": "add", "name": "nomatch", "expr": "5 4 * * *", "command": "echo n"},
        {"op": "add", "name": "off", "expr": "* * * * *", "command": "echo o"},
        {"op": "disable", "name": "off"},
        {"op": "loop"},
        {"op": "loop"},
        {"op": "storage"},
        {"op": "log"},
        {"op": "state"},
    ]},
    # 3: persistence round-trip + start/stop + remove
    {"frozen_now": FROZEN_NOW, "ops": [
        {"op": "add", "name": "keep", "expr": "0 9 * * 1", "command": "echo k"},
        {"op": "add", "name": "drop", "expr": "0 9 * * 1", "command": "echo d"},
        {"op": "remove", "name": "drop"},
        {"op": "run", "name": "keep"},
        {"op": "storage"},
        {"op": "start"},
        {"op": "state"},
        {"op": "stop"},
    ]},
]


def main():
    binary = sys.argv[1] if len(sys.argv) > 1 else "/tmp/cron-engine"
    ok = 0
    for i, sc in enumerate(SCENARIOS, 1):
        with tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) as f:
            json.dump(sc, f)
            sc_path = f.name
        try:
            py = run_python(sc)
            go = run_go(binary, sc_path)
        finally:
            os.unlink(sc_path)
        pyj = tokenize(py["out"], py["home"])
        goj = tokenize(go["out"], go["home"])
        if pyj == goj:
            ok += 1
            print(f"scenario {i}: LIVE PARITY OK ({len(sc['ops'])} ops)")
        else:
            print(f"scenario {i}: LIVE PARITY FAILED")
            pj, gj = json.dumps(pyj, indent=1, sort_keys=True), json.dumps(goj, indent=1, sort_keys=True)
            for pl, gl in zip(pj.splitlines(), gj.splitlines()):
                if pl != gl:
                    print(f"  py: {pl}")
                    print(f"  go: {gl}")
                    break
            return 1
    print(f"LIVE PARITY: {ok}/{len(SCENARIOS)} scenarios identical (oracle vs GO)")
    return 0


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