QuantEngineByItz/tools/validate_truthful_decision_ledger_v2.py

from __future__ import annotations

import argparse
import json
import hashlib
from pathlib import Path
from typing import Any


ROOT = Path(__file__).resolve().parents[1]
DEFAULT_INPUT = ROOT / "Temp" / "truthful_decision_ledger_v2.json"
DEFAULT_REPORT = ROOT / "Temp" / "operational_report.json"


def _load_json(path: Path) -> dict[str, Any]:
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
    except Exception:
        return {}
    return data if isinstance(data, dict) else {}


def _is_number(v: Any) -> bool:
    return isinstance(v, (int, float)) and not isinstance(v, bool)


def _canonical(obj: Any) -> str:
    return json.dumps(obj, ensure_ascii=False, sort_keys=True, separators=(",", ":"))


def _sha256_text(text: str) -> str:
    return hashlib.sha256(text.encode("utf-8")).hexdigest()


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--input", default=str(DEFAULT_INPUT))
    ap.add_argument("--report", default=str(DEFAULT_REPORT))
    args = ap.parse_args()

    input_path = Path(args.input)
    if not input_path.is_absolute():
        input_path = ROOT / input_path
    report_path = Path(args.report)
    if not report_path.is_absolute():
        report_path = ROOT / report_path

    payload = _load_json(input_path)
    if not payload:
        print("TRUTHFUL_DECISION_LEDGER_V2_FAIL")
        print("- invalid input json")
        return 1

    errors: list[str] = []

    if str(payload.get("formula_id") or "") != "TRUTHFUL_DECISION_LEDGER_V2":
        errors.append("formula_id mismatch")
    if payload.get("llm_numeric_generated_flag") is not False:
        errors.append("llm_numeric_generated_flag must be false")
    if not str(payload.get("input_hash") or "").strip():
        errors.append("input_hash missing")
    if not str(payload.get("route_id") or "").strip():
        errors.append("route_id missing")
    if not str(payload.get("report_hash") or "").strip():
        errors.append("report_hash missing")
    elif report_path.exists():
        try:
            report_payload = json.loads(report_path.read_text(encoding="utf-8"))
        except Exception:
            report_payload = None
        if report_payload is None:
            errors.append("report_json invalid")
        else:
            expected_report_hash = _sha256_text(_canonical(report_payload))
            if payload.get("report_hash") != expected_report_hash:
                errors.append("report_hash mismatch")

    ledger_rows = payload.get("ledger_rows")
    if not isinstance(ledger_rows, list):
        errors.append("ledger_rows must be a list")
        ledger_rows = []
    elif not ledger_rows:
        print("WARNING: ledger_rows is empty (no active proposals)")
        # Do not append to errors to allow gate to pass

    for i, row in enumerate(ledger_rows):
        if not isinstance(row, dict):
            errors.append(f"ledger_rows[{i}] must be object")
            continue
        required = [
            "decision_id",
            "proposal_id",
            "ticker",
            "action",
            "state",
            "formula_id",
            "input_hash",
            "output_hash",
            "source_fields",
            "source_snapshot_hash",
            "price_basis",
            "qty_basis",
            "reason_codes",
            "gate_stack",
            "export_gate",
            "renderer_section",
            "llm_numeric_generated_flag",
            "outcome_binding_id",
            "record_type",
        ]
        for key in required:
            if key not in row:
                errors.append(f"ledger_rows[{i}].{key} missing")
        if row.get("llm_numeric_generated_flag") is not False:
            errors.append(f"ledger_rows[{i}].llm_numeric_generated_flag must be false")
        if not isinstance(row.get("source_fields"), list) or not row.get("source_fields"):
            errors.append(f"ledger_rows[{i}].source_fields must be non-empty list")
        if not isinstance(row.get("reason_codes"), list) or not row.get("reason_codes"):
            errors.append(f"ledger_rows[{i}].reason_codes must be non-empty list")
        if not isinstance(row.get("gate_stack"), list) or not row.get("gate_stack"):
            errors.append(f"ledger_rows[{i}].gate_stack must be non-empty list")
        if row.get("record_type") not in {"order_blueprint", "shadow_ledger"}:
            errors.append(f"ledger_rows[{i}].record_type invalid")
        if not str(row.get("output_hash") or "").strip():
            errors.append(f"ledger_rows[{i}].output_hash missing")
        else:
            recompute = dict(row)
            recompute.pop("output_hash", None)
            expected = _sha256_text(_canonical(recompute))
            if row.get("output_hash") != expected:
                errors.append(f"ledger_rows[{i}].output_hash mismatch")

    if errors:
        print("TRUTHFUL_DECISION_LEDGER_V2_FAIL")
        for e in errors:
            print(f"- {e}")
        return 1

    print("TRUTHFUL_DECISION_LEDGER_V2_OK")
    print(f"rows={len(ledger_rows)}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())