#!/usr/bin/env python3 """ build_p1_01_execution_verdict_unify.py ──────────────────────────────────────────────────────────────────────── P1_01: 실행 권위 단일화 — '발행금지(FAIL_BLOCK_PUBLISH)인데 주문 7건' 충돌 제거 문제: - operational_report.json: published_verdict=FAIL_BLOCK_PUBLISH (발행 금지) - final_execution_decision_v4.json: hts_order_count=7, sell_allowed=true - v8 README: hts_order_count=0, sell_allowed=false 해결: 1. global_execution_gate를 final_decision_packet_v2 단 한 곳에서만 산출 2. FAIL_BLOCK_PUBLISH → hts_order_count=0 강제 3. 모든 섹션이 이 단일 게이트를 '복사'만 함 출력: - Temp/p1_01_execution_authority_report.json """ from __future__ import annotations import json import sys from pathlib import Path from datetime import datetime ROOT = Path(__file__).resolve().parent.parent # 입력 파일 OP_REPORT = ROOT / "Temp" / "operational_report.json" FINAL_EXEC = ROOT / "Temp" / "final_execution_decision_v4.json" # 출력 파일 REPORT_P101 = ROOT / "Temp" / "p1_01_execution_authority_report.json" if sys.stdout.encoding and sys.stdout.encoding.lower() not in ("utf-8", "utf8"): sys.stdout = open(sys.stdout.fileno(), mode="w", encoding="utf-8", buffering=1) def load_json(p: Path) -> dict: if not p.exists(): return {} try: return json.loads(p.read_text(encoding="utf-8")) except Exception as e: print(f"[WARN] Failed to load {p.name}: {e}") return {} def audit_execution_verdict_collision(op_report: dict, final_exec: dict) -> dict: """실행 권위 충돌 감사.""" audit = { "generated_at": datetime.now().isoformat(), "authority_sources": {}, "collisions_found": [] } # 1. operational_report 판정 op_verdict = op_report.get("sections", []) op_final_judgment = {} for section in op_verdict: if section.get("name") == "final_judgment_table": op_final_judgment = section break op_verdict_summary = "UNKNOWN" if op_final_judgment: # operational_report의 최종 판정 찾기 markdown = op_final_judgment.get("markdown", "") if "EXPORT_READY" in markdown or "JSON_VALID" in markdown: op_verdict_summary = "EXPORT_READY" elif "FAIL" in markdown or "BLOCK" in markdown: op_verdict_summary = "FAIL_BLOCK_PUBLISH" audit["authority_sources"]["operational_report"] = { "source_file": "Temp/operational_report.json", "verdict": op_verdict_summary, "hts_order_count_implied": 0 if "FAIL" in op_verdict_summary else "CHECK_FINAL_EXEC" } # 2. final_execution_decision_v4 판정 final_exec_verdict = final_exec.get("global_execution_gate", "UNKNOWN") hts_order_count = final_exec.get("hts_order_count", 0) sell_allowed = final_exec.get("sell_allowed", False) audit["authority_sources"]["final_execution_decision_v4"] = { "source_file": "Temp/final_execution_decision_v4.json", "global_execution_gate": final_exec_verdict, "hts_order_count": hts_order_count, "sell_allowed": sell_allowed } # 3. 충돌 검사 if op_verdict_summary == "FAIL_BLOCK_PUBLISH" and hts_order_count > 0: audit["collisions_found"].append({ "collision_type": "VERDICT_ORDER_COUNT_MISMATCH", "severity": "CRITICAL_FINANCIAL_LOSS_RISK", "description": f"발행금지 판정인데 {hts_order_count}건 주문 생성", "source_1": "operational_report.json → FAIL_BLOCK_PUBLISH", "source_2": f"final_execution_decision_v4.json → hts_order_count={hts_order_count}", "consequence": "금지된 주문이 실제 HTS에 입력될 수 있음", "remediation": "hts_order_count=0 강제, 또는 published_verdict=EXPORT_READY 정정" }) if sell_allowed and op_verdict_summary == "FAIL_BLOCK_PUBLISH": audit["collisions_found"].append({ "collision_type": "SELL_PERMISSION_MISMATCH", "severity": "CRITICAL", "description": "발행금지인데 sell_allowed=true", "source_1": "operational_report.json → FAIL_BLOCK_PUBLISH", "source_2": "final_execution_decision_v4.json → sell_allowed=true", "remediation": "operational_report의 최종 판정 = final_execution_decision의 권위로 통일" }) return audit def build_p101_report(audit: dict) -> dict: """P1_01 보고서.""" report = { "phase": "P1_01_ONE_GATE_TO_RULE_THEM", "generated_at": datetime.now().isoformat(), "audit_findings": audit, "authority_ladder": [ { "rank": 1, "authority": "Temp/final_decision_packet_v2.json", "field": "global_execution_gate", "meaning": "실행 여부의 단일 진실 (ONLY SOURCE)", "current_status": "NOT VERIFIED" }, { "rank": 2, "authority": "spec/33_execution_precedence_lock.yaml", "field": "execution_rules", "meaning": "FAIL_BLOCK_PUBLISH이면 hts_order_count=0 강제" }, { "rank": 3, "authority": "모든 보고서 섹션", "field": "global_execution_gate (복사만)", "meaning": "rank 1의 값을 '복사'만. 재해석/조건부 판정 금지" } ], "required_corrections": [ { "correction_id": "P1_01_A", "title": "final_decision_packet_v2 = single truth", "steps": [ "final_decision_packet_v2.json에 global_execution_gate 필드 필수", "값: EXPORT_READY | FAIL_BLOCK_PUBLISH | ... (명시적 열거)", "모든 다른 파일은 이 값을 참조만 함" ] }, { "correction_id": "P1_01_B", "title": "FAIL_BLOCK_PUBLISH → hts_order_count=0 강제", "condition": "if global_execution_gate == FAIL_BLOCK_PUBLISH", "action": "hts_order_count = 0 (override)", "location": "spec/33_execution_precedence_lock.yaml" }, { "correction_id": "P1_01_C", "title": "모든 섹션 = 복사 렌더링만", "files": [ "operational_report.json", "final_execution_decision_v4.json", "prompts/analysis_prompt.md" ], "rule": "global_execution_gate = final_decision_packet_v2.global_execution_gate (조건 없음)" } ], "validation_gates": [ { "gate": "COLLISION_COUNT", "expected": 0, "actual": len(audit.get("collisions_found", [])), "status": "PASS" if len(audit.get("collisions_found", [])) == 0 else "FAIL" }, { "gate": "AUTHORITY_SOURCES", "expected": 1, "actual": "multiple", "status": "FAIL" if len(audit.get("authority_sources", {})) > 1 else "PASS" }, { "gate": "EXECUTION_VERDICT_SOURCE_COUNT", "expected": 1, "rule": "final_decision_packet_v2에서만 산출" } ] } return report def main() -> int: print("=" * 80) print(" P1_01: 실행 권위 단일화 — '발행금지인데 주문 7건' 충돌 제거") print("=" * 80) # 입력 로드 op_report = load_json(OP_REPORT) final_exec = load_json(FINAL_EXEC) # 감사 audit = audit_execution_verdict_collision(op_report, final_exec) print(f"\n[1] 권위 출처 분석") for source_name, details in audit["authority_sources"].items(): print(f" {source_name}") for key, val in details.items(): if key != "source_file": print(f" {key}: {val}") print(f"\n[2] 충돌 발견") if not audit["collisions_found"]: print(f" 충돌 0개 — 데이터 미검증") else: for i, collision in enumerate(audit["collisions_found"], 1): print(f" [{i}] {collision['collision_type']} (심각도: {collision['severity']})") print(f" → {collision['description']}") # 보고서 생성 p101_report = build_p101_report(audit) print(f"\n[3] 필수 수정사항") for corr in p101_report["required_corrections"]: print(f" {corr['correction_id']}: {corr['title']}") if "condition" in corr: print(f" 조건: {corr['condition']}") if "action" in corr: print(f" 조치: {corr['action']}") # 보고서 저장 REPORT_P101.write_text( json.dumps(p101_report, ensure_ascii=False, indent=2), encoding="utf-8" ) print(f"\n✓ P1_01 보고서 저장: {REPORT_P101.name}") return 0 if len(audit["collisions_found"]) == 0 else 1 if __name__ == "__main__": sys.exit(main())