NODEDC_MISSION_CORE/scripts/check_stationary_bootstrap.py

"""Frozen stationary-to-fresh functional experiment on independent archived walks."""

import argparse
import json
import platform
import shutil
import time
from pathlib import Path

import numpy as np

from k1link.artifacts import utc_now_iso
from k1link.device_plugins.xgrids_k1.planning_replay import iter_planning_events
from k1link.missions.causal_replay import digest
from k1link.missions.stationary_bootstrap import BOOTSTRAP_POLICY
from k1link.missions.stationary_replay import replay_stationary


def write(path, value):
    path.write_text(json.dumps(value, indent=2, allow_nan=False))


def code_hashes():
    root = Path(__file__).resolve().parents[1]
    paths = [
        Path(__file__).resolve(),
        *sorted((root / "src/k1link/missions").glob("*.py")),
        root / "src/k1link/device_plugins/xgrids_k1/planning_replay.py",
        root / "src/k1link/device_plugins/xgrids_k1/planning_live.py",
        root / "src/k1link/device_plugins/xgrids_k1/localization_source.py",
        root / "tests/test_stationary_bootstrap.py",
    ]
    return {str(p.relative_to(root)): digest(p) for p in paths}


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--stage", choices=["prepare", "correct-entry", "wrong-region"], required=True
    )
    parser.add_argument("--predecessor", type=Path)
    parser.add_argument("--output", type=Path, required=True)
    args = parser.parse_args()
    root = args.output
    if args.stage == "prepare":
        previous = args.predecessor
        old = json.loads((previous / "manifest.json").read_text())
        inputs = {**old["input_digests"], str(previous / "reference.npz"): old["reference_sha256"]}
        inputs[str(previous / "manifest.json")] = digest(previous / "manifest.json")
        if any(digest(Path(p)) != sha for p, sha in inputs.items()):
            raise ValueError("Predecessor input digest mismatch.")
        root.mkdir(parents=True, exist_ok=False)
        code = code_hashes()
        for name in code:
            target = root / "executed-source" / name
            target.parent.mkdir(parents=True, exist_ok=True)
            shutil.copyfile(name, target)
        a_path = next(
            p
            for p in old["input_digests"]
            if p.endswith(".json")
            and json.loads(Path(p).read_text()).get("session_id") == old["reference_session"]
        )
        poses = json.loads(Path(a_path).read_text())["poses"]
        wrong_path = np.array([p["position"] for p in poses if 130 <= p["distance_m"] <= 155])
        with np.load(old["negative_file"], allow_pickle=False) as data:
            np.savez_compressed(
                root / "wrong-reference.npz", reference=data["reference"], reference_path=wrong_path
            )
        # The old archive supplies fixed A geometry only; no previous B fitted
        # transform, mask or future B pose is used as an initialization seed.
        inputs[str(root / "wrong-reference.npz")] = digest(root / "wrong-reference.npz")
        write(
            root / "manifest.json",
            dict(
                schema_version="missioncore.stationary-bootstrap-probe/v1",
                created_at_utc=utc_now_iso(),
                created_monotonic_ns=time.monotonic_ns(),
                query_raw=old["query_raw"],
                query_session=old["query_session"],
                reference_session=old["reference_session"],
                references={
                    "correct-entry": str(previous / "reference.npz"),
                    "wrong-region": str(root / "wrong-reference.npz"),
                },
                input_digests=inputs,
                implementation_sha256=code,
                protocol=BOOTSTRAP_POLICY,
                maximum_seconds=65.0,
                maximum_distance_m=40.0,
                expectations=dict(
                    correct_entry="complete search -> provisional -> three fresh consistent fits",
                    wrong_region="no current candidate or tracking",
                    freshness="all validation observations after ready; windows have disjoint IDs",
                    pace="original receipt clocks, 1x; no offline B heading or transform",
                ),
                notes="Engineering qualification only. Frame continuity is unverified; "
                "one pre-ready receipt gap permits a hypothesis, not a tracking result. "
                "No new capture, device commands, UI changes or live activation.",
                vehicle_control=False,
                localization_confirmed=False,
            ),
        )
        print(json.dumps(dict(stage="prepared", output=str(root))), flush=True)
        return
    manifest = json.loads((root / "manifest.json").read_text())
    inputs = manifest["input_digests"]
    code = code_hashes()
    if code != manifest["implementation_sha256"]:
        raise ValueError("Implementation changed since protocol freeze.")
    if any(digest(Path(p)) != sha for p, sha in inputs.items()):
        raise ValueError("Source changed before replay.")
    with np.load(manifest["references"][args.stage], allow_pickle=False) as data:
        reference, path = data["reference"], data["reference_path"]
    events = iter_planning_events(Path(manifest["query_raw"]), manifest["query_session"])
    report = replay_stationary(
        events,
        reference,
        path,
        root / args.stage,
        max_seconds=manifest["maximum_seconds"],
        max_distance=manifest["maximum_distance_m"],
    )
    report.update(
        input_digests=inputs,
        source_integrity_verified=all(digest(Path(p)) == h for p, h in inputs.items()),
        implementation_sha256=code,
        code_integrity_verified=code_hashes() == code,
        runtime=dict(
            system=platform.system(), machine=platform.machine(), python=platform.python_version()
        ),
    )
    write(root / args.stage / "report.json", report)
    print(
        json.dumps(
            {
                k: report.get(k)
                for k in (
                    "state",
                    "first_prior_s",
                    "first_candidate_s",
                    "first_tracking_s",
                    "transitions",
                    "source_integrity_verified",
                    "code_integrity_verified",
                )
            }
        ),
        flush=True,
    )


if __name__ == "__main__":
    main()