from __future__ import annotations import json from pathlib import Path import yaml ROOT = Path(__file__).resolve().parents[1] def main() -> int: spec_path = ROOT / "spec" / "36_goal_risk_budget_harness.yaml" if not spec_path.exists(): print(json.dumps({"formula_id": "GOAL_RISK_BUDGET_HARNESS_V1", "gate": "FAIL", "error": f"spec not found: {spec_path}"})) return 1 spec = yaml.safe_load(spec_path.read_text(encoding="utf-8")) required_fields = spec.get("required_fields", []) # D+2 cash counts as immediate defense — confirmed by cash_defense_line_d2_used in spec d2_cash_in_spec = "cash_defense_line_d2_used" in required_fields cash_floor_rule = "D+2 cash counts as immediate defense" if d2_cash_in_spec else "MISSING" # Check for risk budget violations from cash ledger artifact ledger_path = ROOT / "Temp" / "cash_ledger_v2.json" risk_budget_violation_count = 0 ledger_status = "DATA_MISSING" if ledger_path.exists(): ledger = json.loads(ledger_path.read_text(encoding="utf-8")) ledger_status = ledger.get("gate", "UNKNOWN") if ledger_status != "PASS": risk_budget_violation_count = 1 cross_leak = ledger.get("cross_account_cash_leak_count", 0) if cross_leak > 0: risk_budget_violation_count += cross_leak gate = "PASS" if d2_cash_in_spec and risk_budget_violation_count == 0 else "FAIL" result = { "formula_id": "GOAL_RISK_BUDGET_HARNESS_V1", "spec_required_fields": required_fields, "cash_floor_rule": cash_floor_rule, "d2_cash_defense_in_spec": d2_cash_in_spec, "ledger_status": ledger_status, "risk_budget_violation_count": risk_budget_violation_count, "gate": gate, } out = ROOT / "Temp" / "goal_risk_budget_harness_v1.json" out.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8") print(json.dumps(result, ensure_ascii=True, indent=2)) return 0 if gate == "PASS" else 1 if __name__ == "__main__": raise SystemExit(main())