rune/scripts/gen_structural_migration_truth.py

#!/usr/bin/env python3
"""gen_structural_migration_truth.py — freeze/replay parity truth for
structural_migration (literal-100% residue lane, dichotomy wave #2479).

Freeze mode (default): exercises EVERY public function of the LIVE Python
oracle (src/rune/structural_migration.py) across every branch and records
{fn, kwargs, now, expected} rows into
src/rust/structural_migration/src/frozen_smig_truth.json.

No-clock law: apply_migration reads `time.strftime(..., time.gmtime())`
directly — a module-global `import time` — so the rig patches the oracle's
`time` symbol with a frozen shim whose gmtime() returns a settable
instant. Every row carries its epoch (`now`); the battery passes it as
the clock argument. Two instants prove the civil-date law, incl. a leap
day and a pre-epoch negative.

Replay mode (--replay): re-runs the same scenarios against the Rust
kernel (via the frozen battery's own comparison law) is the cargo test
job; this script's replay just re-freezes and byte-compares (regen x2
law). Poisoned-truth law: no loader pin exists for this module (pure
Python facade) — the oracle IS the module.
"""
import json
import os
import sys
import time as _time
from pathlib import Path

REPO = Path(__file__).resolve().parent.parent
TRUTH_PATH = REPO / "src" / "rust" / "structural_migration" / "src" / "frozen_smig_truth.json"

sys.path.insert(0, str(REPO / "src"))
sys.path.insert(0, str(REPO / "src" / "rune"))

import structural_migration as oracle_mod  # noqa: E402

# ── the frozen-clock rig (settable instant) ─────────────────────────────────
FROZEN = {"epoch": None}


def _frozen_struct():
    return _time.gmtime(FROZEN["epoch"])


class _FrozenTime:
    """Module-global `time` shim: gmtime() honors the settable frozen
    instant; strftime stays the real implementation (struct is real)."""

    @staticmethod
    def gmtime(secs=None):
        return _frozen_struct() if secs is None else _time.gmtime(secs)

    @staticmethod
    def strftime(fmt, t=None):
        return _time.strftime(fmt, t if t is not None else _frozen_struct())


oracle_mod.time = _FrozenTime


def case(fn, kwargs, now=None):
    try:
        result = getattr(oracle_mod, fn)(**kwargs)
        expected = {"ok": True, "result": result}
        if isinstance(result, dict):
            expected["key_order"] = list(result.keys())
    except Exception as ex:  # the message is the contract (quirks law)
        expected = {"ok": False, "error": str(ex)}
    return {"fn": fn, "kwargs": kwargs, "now": now, "expected": expected}


def main():
    cases = []

    # classify_migration — every marker + case-sensitivity + empty
    for changes, tag in [
        (["add optional kind"], "minor_additive"),
        (["Add Optional Tags list"], "minor_case_fold"),
        (["rename field x to y"], "material_rename"),
        (["drop legacy field"], "material_drop"),
        (["remove the old index"], "material_remove"),
        (["delete stale docs"], "material_delete"),
        (["id format now uuid"], "material_id_format"),
        (["semantic shift of x"], "material_semantic"),
        (["restructure the tree"], "material_restructure"),
        (["change format of ids"], "material_change_format"),
        (["merge the two kinds"], "material_merge"),
        (["split the id field"], "material_split"),
        ([], "minor_empty"),
        (["add optional kind", "drop legacy field"], "material_wins_mixed"),
        (["dropship unrelated word"], "material_substring_drop"),
    ]:
        cases.append(case("classify_migration",
                          {"from_v": 3, "to_v": 4, "changes": changes}))

    # affected_projects — gating, default schema_version, sorted output
    stores = {
        "delta": {"schema_version": 3},
        "alpha": {"schema_version": 5},
        "charlie": {"schema_version": 5},
        "bravo": {},
    }
    for proposed_v, tag in [(5, "lt_strict"), (3, "default_zero"), (6, "all")]:
        cases.append(case("affected_projects",
                          {"stores": stores, "proposed_v": proposed_v}))

    # apply_migration — every branch of the consent gate.
    # Two clock instants: a leap day and a pre-epoch negative (civil law).
    LEAP = 1_709_164_800        # 2024-02-29T00:00:00Z
    NEG = -1                    # 1969-12-31T23:59:59Z
    material = ["rename field x to y", "drop legacy field",
                "semantic shift of x", "merge the two kinds"]
    minor = ["add optional kind"]

    FROZEN["epoch"] = LEAP
    cases.append(case("apply_migration", {
        "model_store": {"schema_version": 3}, "from_v": 3, "to_v": 4,
        "changes": material, "user_approved": False,
    }, now=LEAP))
    cases.append(case("apply_migration", {
        "model_store": None, "from_v": 3, "to_v": 4,
        "changes": material, "user_approved": True, "approved_by": "Nick",
    }, now=LEAP))
    cases.append(case("apply_migration", {
        "model_store": None, "from_v": 3, "to_v": 4,
        "changes": minor, "user_approved": False,
    }, now=LEAP))  # MINOR: auto — approved_by dropped to None
    cases.append(case("apply_migration", {
        "model_store": None, "from_v": 3, "to_v": 4,
        "changes": minor, "user_approved": True, "approved_by": "",
    }, now=LEAP))  # MINOR + approved: empty string is carried verbatim
    cases.append(case("apply_migration", {
        "model_store": None, "from_v": 0, "to_v": 1,
        "changes": ["split the id field"], "user_approved": False,
    }, now=NEG))  # error branch on the pre-epoch clock (message has no stamp)

    # consent_brief — pure presentation dict, insertion order frozen
    cases.append(case("consent_brief", {
        "from_v": 3, "to_v": 4,
        "changes": ["rename field x to y", "id format now uuid"],
        "benefits": ["faster projections", "one canonical id"],
    }))
    cases.append(case("consent_brief", {
        "from_v": 0, "to_v": 1, "changes": [], "benefits": [],
    }))

    TRUTH_PATH.write_text(json.dumps(cases, indent=1) + "\n")
    print(f"wrote {TRUTH_PATH} ({len(cases)} cases)")


if __name__ == "__main__":
    main()