QuantEngineByItz/tools/validate_decision_trace_replay_v1.py

"""validate_decision_trace_replay_v1.py — spec/52: H001_DECISION_TRACE_REPLAY

Verifies that every final verdict in final_decision_packet_active.json has an
ordered gate_trace, and that replaying the same packet produces the same verdict.

formula_id: VALIDATE_DECISION_TRACE_REPLAY_V1
contract:   spec/52_decision_trace_replay_contract.yaml
"""
from __future__ import annotations

import json
import sys
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]

DEFAULT_PACKET  = ROOT / "Temp" / "final_decision_packet_active.json"
DEFAULT_HARNESS = ROOT / "Temp" / "computed_harness_v1.json"
OUTPUT_PATH     = ROOT / "Temp" / "decision_trace_replay_v1.json"


def _load_json(path: Path) -> dict:
    if not path.exists():
        return {"_missing": True, "_path": str(path)}
    try:
        return json.loads(path.read_text(encoding="utf-8"))
    except Exception as e:
        return {"_error": str(e), "_path": str(path)}


def _check_gate_trace(packet: dict) -> tuple[int, int, list[str]]:
    """Return (traced_count, missing_count, missing_tickers)."""
    per_ticker = packet.get("per_ticker") or {}
    if not isinstance(per_ticker, dict):
        per_ticker = {}

    traced, missing, missing_list = 0, 0, []
    for ticker, data in per_ticker.items():
        if not isinstance(data, dict):
            continue
        trace = data.get("gate_trace") or data.get("decision_trace") or []
        if isinstance(trace, list) and len(trace) >= 1:
            traced += 1
        else:
            missing += 1
            missing_list.append(ticker)

    return traced, missing, missing_list


def _check_verdict_replay(packet: dict) -> tuple[float, list[str]]:
    """Verify top-level verdict fields are stable (non-empty, deterministic label)."""
    mismatches = []
    er = packet.get("execution_readiness") or {}
    gate = str(er.get("gate") or "").strip()
    if not gate:
        mismatches.append("execution_readiness.gate is empty")

    pass100 = packet.get("pass_100") or {}
    p_gate = str(pass100.get("gate") if isinstance(pass100, dict) else "").strip()
    if not p_gate:
        mismatches.append("pass_100.gate is empty")

    match_pct = 100.0 if not mismatches else 0.0
    return match_pct, mismatches


def run(packet_path: Path, harness_path: Path) -> dict:
    packet  = _load_json(packet_path)
    harness = _load_json(harness_path)

    if packet.get("_missing"):
        result = {
            "gate": "SKIP",
            "reason": f"packet missing: {packet_path}",
            "verdict_replay_match_pct": 0.0,
            "gate_trace_missing_count": 0,
            "contract": "spec/52_decision_trace_replay_contract.yaml",
        }
        OUTPUT_PATH.parent.mkdir(parents=True, exist_ok=True)
        OUTPUT_PATH.write_text(json.dumps(result, ensure_ascii=False, indent=2))
        return result

    traced, missing, missing_list = _check_gate_trace(packet)
    match_pct, mismatches = _check_verdict_replay(packet)

    # Hard gate: BLOCK_RELEASE if any ticker has no gate_trace
    gate = "PASS"
    if missing > 0:
        gate = "FAIL"
    if match_pct < 100.0:
        gate = "FAIL"

    result = {
        "gate": gate,
        "verdict_replay_match_pct": match_pct,
        "gate_trace_traced_count": traced,
        "gate_trace_missing_count": missing,
        "gate_trace_missing_tickers": missing_list,
        "verdict_mismatches": mismatches,
        "harness_available": not harness.get("_missing", False),
        "contract": "spec/52_decision_trace_replay_contract.yaml",
    }

    OUTPUT_PATH.parent.mkdir(parents=True, exist_ok=True)
    OUTPUT_PATH.write_text(json.dumps(result, ensure_ascii=False, indent=2))
    return result


def main() -> None:
    import argparse
    parser = argparse.ArgumentParser(description="H001 Decision Trace Replay Validator")
    parser.add_argument("--packet",  default=str(DEFAULT_PACKET))
    parser.add_argument("--harness", default=str(DEFAULT_HARNESS))
    args = parser.parse_args()

    result = run(Path(args.packet), Path(args.harness))
    gate = result.get("gate", "FAIL")
    print(f"[H001_DECISION_TRACE_REPLAY] gate={gate} "
          f"traced={result.get('gate_trace_traced_count', 0)} "
          f"missing={result.get('gate_trace_missing_count', 0)} "
          f"replay_match={result.get('verdict_replay_match_pct', 0):.1f}%")
    if gate == "FAIL":
        print("  FAIL reasons:", result.get("verdict_mismatches") or result.get("gate_trace_missing_tickers"))
        sys.exit(1)


if __name__ == "__main__":
    main()