"""utils/critical_escalation_reporter.py — Critical 7종 에스컬레이션 라우터 및 회장 보고 포맷터.""" from __future__ import annotations import argparse import dataclasses import hashlib import json import os import sys import tempfile from collections.abc import Callable from datetime import datetime, timezone from pathlib import Path from typing import Any from utils.automation_contracts import ( # pyright: ignore[reportMissingImports] CriticalEscalationType, EscalationPacket, RiskLevel, ) # --------------------------------------------------------------------------- # Legacy 호환 매핑 테이블 # --------------------------------------------------------------------------- LEGACY_CRITICAL_MAP: dict[str, CriticalEscalationType] = { "FORBIDDEN_PATH_INVASION": CriticalEscalationType.FORBIDDEN_PATH_INTRUSION, "EFFECTIVE_DIFF_CONTAMINATION_REPLACEMENT_FAILED": CriticalEscalationType.REPLACEMENT_PR_AUTO_CREATION_FAILED_FOR_CONTAMINATED_DIFF, "GEMINI_REAL_BUG_SCOPE_EXPANSION": CriticalEscalationType.GEMINI_REAL_BUG_REQUIRES_SCOPE_EXPANSION, "BLOCK_OVERRIDE_REQUIRED_OR_INSUFFICIENT_REASON": CriticalEscalationType.BLOCK_OVERRIDE_REQUIRED_OR_REASON_INSUFFICIENT, "DEPENDENCY_CYCLE_OR_SERIAL_ONLY_CONFLICT": CriticalEscalationType.DEPENDENCY_CYCLE_OR_SERIAL_ONLY_COLLISION, "REPLACEMENT_PR_ALSO_FAILED": CriticalEscalationType.REPLACEMENT_PR_FAILED, "POST_MERGE_SMOKE_FAILURE": CriticalEscalationType.POST_MERGE_SMOKE_FAILED, } # --------------------------------------------------------------------------- # Severity 매핑 (모두 HIGH 이상) # --------------------------------------------------------------------------- SEVERITY_MAP: dict[CriticalEscalationType, RiskLevel] = { CriticalEscalationType.FORBIDDEN_PATH_INTRUSION: RiskLevel.HIGH_CORE, CriticalEscalationType.REPLACEMENT_PR_AUTO_CREATION_FAILED_FOR_CONTAMINATED_DIFF: RiskLevel.HIGH_CORE, CriticalEscalationType.GEMINI_REAL_BUG_REQUIRES_SCOPE_EXPANSION: RiskLevel.HIGH, CriticalEscalationType.BLOCK_OVERRIDE_REQUIRED_OR_REASON_INSUFFICIENT: RiskLevel.HIGH_CORE, CriticalEscalationType.DEPENDENCY_CYCLE_OR_SERIAL_ONLY_COLLISION: RiskLevel.HIGH_CORE, CriticalEscalationType.REPLACEMENT_PR_FAILED: RiskLevel.HIGH, CriticalEscalationType.POST_MERGE_SMOKE_FAILED: RiskLevel.HIGH, } # --------------------------------------------------------------------------- # escalation_type별 기본 텍스트 매핑 # --------------------------------------------------------------------------- _REASON_MAP: dict[CriticalEscalationType, str] = { CriticalEscalationType.FORBIDDEN_PATH_INTRUSION: "PR이 금지 경로(freeze 파일)를 침범했습니다. 자동 병합을 즉시 중단합니다.", CriticalEscalationType.REPLACEMENT_PR_AUTO_CREATION_FAILED_FOR_CONTAMINATED_DIFF: "오염된 diff에 대한 replacement PR 자동 생성이 실패했습니다.", CriticalEscalationType.GEMINI_REAL_BUG_REQUIRES_SCOPE_EXPANSION: "Gemini가 실제 버그를 감지했으며, 수정에 스코프 확장이 필요합니다.", CriticalEscalationType.BLOCK_OVERRIDE_REQUIRED_OR_REASON_INSUFFICIENT: "블록 오버라이드가 필요하거나 사유가 불충분합니다.", CriticalEscalationType.DEPENDENCY_CYCLE_OR_SERIAL_ONLY_COLLISION: "의존성 사이클 또는 직렬 전용 충돌이 감지되었습니다.", CriticalEscalationType.REPLACEMENT_PR_FAILED: "Replacement PR이 실패했습니다.", CriticalEscalationType.POST_MERGE_SMOKE_FAILED: "병합 후 smoke 테스트가 실패했습니다.", } _WHY_AUTO_MAP: dict[CriticalEscalationType, str] = { CriticalEscalationType.FORBIDDEN_PATH_INTRUSION: "freeze 파일 침범은 자동 처리 불가. 회장 수동 확인 및 승인 필요.", CriticalEscalationType.REPLACEMENT_PR_AUTO_CREATION_FAILED_FOR_CONTAMINATED_DIFF: "replacement PR 생성 실패 시 자동 복구 로직이 없음. 수동 개입 필요.", CriticalEscalationType.GEMINI_REAL_BUG_REQUIRES_SCOPE_EXPANSION: "스코프 확장 결정은 자동화 범위 초과. 회장 승인 없이 진행 불가.", CriticalEscalationType.BLOCK_OVERRIDE_REQUIRED_OR_REASON_INSUFFICIENT: "블록 오버라이드는 회장만 승인 가능. 자동 처리 불가.", CriticalEscalationType.DEPENDENCY_CYCLE_OR_SERIAL_ONLY_COLLISION: "의존성 사이클 해소는 수동 재설계 필요. 자동 해결 불가.", CriticalEscalationType.REPLACEMENT_PR_FAILED: "replacement PR도 실패한 경우 자동 재시도 불가. 수동 조사 필요.", CriticalEscalationType.POST_MERGE_SMOKE_FAILED: "smoke 실패 후 롤백/핫픽스 결정은 회장 승인 필요.", } _SAFE_OPTIONS_MAP: dict[CriticalEscalationType, list[str]] = { CriticalEscalationType.FORBIDDEN_PATH_INTRUSION: [ "PR을 즉시 close하고 freeze 파일 제거 후 재제출", "freeze 규칙 예외 승인 후 진행", ], CriticalEscalationType.REPLACEMENT_PR_AUTO_CREATION_FAILED_FOR_CONTAMINATED_DIFF: [ "원본 PR 보존 후 수동으로 replacement PR 생성", "오염된 diff 제거 후 원본 PR 재제출", ], CriticalEscalationType.GEMINI_REAL_BUG_REQUIRES_SCOPE_EXPANSION: [ "스코프 확장 승인 후 별도 태스크로 분리 처리", "현재 PR 범위 내로 버그 수정 범위 제한", ], CriticalEscalationType.BLOCK_OVERRIDE_REQUIRED_OR_REASON_INSUFFICIENT: [ "충분한 사유 보완 후 재제출", "회장 수동 오버라이드 승인", ], CriticalEscalationType.DEPENDENCY_CYCLE_OR_SERIAL_ONLY_COLLISION: [ "의존성 순서 재설계 후 재제출", "충돌하는 PR 중 하나를 먼저 병합 후 재시도", ], CriticalEscalationType.REPLACEMENT_PR_FAILED: [ "replacement PR 수동 생성 및 검토", "원본 PR 롤백 후 재시작", ], CriticalEscalationType.POST_MERGE_SMOKE_FAILED: [ "즉시 롤백(revert commit) 후 원인 분석", "핫픽스 PR 생성 후 긴급 배포", ], } _RECOMMENDED_OPTION_MAP: dict[CriticalEscalationType, str] = { CriticalEscalationType.FORBIDDEN_PATH_INTRUSION: "PR을 즉시 close하고 freeze 파일 제거 후 재제출", CriticalEscalationType.REPLACEMENT_PR_AUTO_CREATION_FAILED_FOR_CONTAMINATED_DIFF: "원본 PR 보존 후 수동으로 replacement PR 생성", CriticalEscalationType.GEMINI_REAL_BUG_REQUIRES_SCOPE_EXPANSION: "스코프 확장 승인 후 별도 태스크로 분리 처리", CriticalEscalationType.BLOCK_OVERRIDE_REQUIRED_OR_REASON_INSUFFICIENT: "충분한 사유 보완 후 재제출", CriticalEscalationType.DEPENDENCY_CYCLE_OR_SERIAL_ONLY_COLLISION: "의존성 순서 재설계 후 재제출", CriticalEscalationType.REPLACEMENT_PR_FAILED: "replacement PR 수동 생성 및 검토", CriticalEscalationType.POST_MERGE_SMOKE_FAILED: "즉시 롤백(revert commit) 후 원인 분석", } # --------------------------------------------------------------------------- # Dedup helpers # --------------------------------------------------------------------------- def _make_evidence_hash( escalation_type: CriticalEscalationType, task_id: str, evidence_keys: list[str], ) -> str: """evidence_hash 생성: sha256(escalation_type.value + '|' + task_id + '|' + sorted_keys_csv).""" sorted_keys_csv = ",".join(sorted(evidence_keys)) raw = escalation_type.value + "|" + task_id + "|" + sorted_keys_csv return hashlib.sha256(raw.encode()).hexdigest() def is_duplicate( escalation_type: CriticalEscalationType, task_id: str, evidence_keys: list[str], audit_log_path: Path, window_sec: int = 3600, *, now: datetime | None = None, ) -> bool: """audit log에서 window_sec 이내 동일 hash 여부 확인.""" if not audit_log_path.exists(): return False target_hash = _make_evidence_hash(escalation_type, task_id, evidence_keys) now_ts = (now or datetime.now(timezone.utc)).timestamp() cutoff_ts = now_ts - window_sec try: lines = audit_log_path.read_text(encoding="utf-8").splitlines() except OSError: return False for line in reversed(lines): line = line.strip() if not line: continue try: record = json.loads(line) except json.JSONDecodeError: continue try: rec_ts = datetime.fromisoformat(record["ts"]).timestamp() except (KeyError, ValueError): continue if rec_ts < cutoff_ts: break if record.get("evidence_hash") == target_hash: return True return False # --------------------------------------------------------------------------- # Audit log helpers # --------------------------------------------------------------------------- def _global_audit_path(workspace_root: Path) -> Path: """글로벌 audit log 경로 반환.""" return workspace_root / "memory" / "orchestration-audit" / "critical-escalations.jsonl" def _per_task_path(workspace_root: Path, task_id: str) -> Path: """per-task audit json 경로 반환.""" return workspace_root / "memory" / "events" / f"{task_id}.escalation.json" def _build_audit_record( *, ts: str, task_id: str, pr_number: int, escalation_type: str, classification: str, severity: str | None, evidence_hash: str, source: str, suppressed_reason: str | None, ) -> dict[str, Any]: """audit record dict 생성.""" return { "ts": ts, "task_id": task_id, "pr_number": pr_number, "escalation_type": escalation_type, "classification": classification, "severity": severity, "evidence_hash": evidence_hash, "source": source, "suppressed_reason": suppressed_reason, } def _append_audit_log( workspace_root: Path, record: dict[str, Any], task_id: str, packet_or_decision: dict[str, Any], ) -> None: """글로벌 JSONL append + per-task JSON atomic write.""" global_path = _global_audit_path(workspace_root) global_path.parent.mkdir(parents=True, exist_ok=True) with global_path.open("a", encoding="utf-8") as f: f.write(json.dumps(record, ensure_ascii=False) + "\n") per_task_path = _per_task_path(workspace_root, task_id) per_task_path.parent.mkdir(parents=True, exist_ok=True) _atomic_write_json(per_task_path, packet_or_decision) def _atomic_write_json(path: Path, data: dict[str, Any]) -> None: """tempfile + os.replace로 atomic write.""" parent = path.parent with tempfile.NamedTemporaryFile( mode="w", dir=parent, delete=False, suffix=".tmp", encoding="utf-8" ) as tmp: json.dump(data, tmp, ensure_ascii=False, indent=2) tmp_path = tmp.name os.replace(tmp_path, str(path)) # --------------------------------------------------------------------------- # EscalationPacket 생성 # --------------------------------------------------------------------------- def _build_packet( *, escalation_type: CriticalEscalationType, task_id: str, pr_number: int, severity: RiskLevel, source: str, merge_commit: str | None, occurred_at: str | None, original_evidence: dict[str, Any], now_iso: str, ) -> EscalationPacket: """Critical escalation 시 EscalationPacket 생성.""" enriched_evidence: dict[str, Any] = { "merge_commit": merge_commit, "severity": severity.value, "created_at": occurred_at or now_iso, "source": source, "risk_level": severity.value, "original_evidence": original_evidence, } return EscalationPacket( task_id=task_id, pr_number=pr_number, escalation_type=escalation_type, reason=_REASON_MAP[escalation_type], why_auto_cannot_continue=_WHY_AUTO_MAP[escalation_type], safe_options=_SAFE_OPTIONS_MAP[escalation_type], recommended_option=_RECOMMENDED_OPTION_MAP[escalation_type], evidence=enriched_evidence, ) # --------------------------------------------------------------------------- # 포맷터 # --------------------------------------------------------------------------- # Critical 카운터 (프로세스 레벨 간단 시퀀스) _critical_counter = 0 def format_packet_for_chair( packet: EscalationPacket, *, max_len: int = 4096, ) -> str: """회장 보고용 텍스트 포맷. 길이 max_len 보장.""" global _critical_counter _critical_counter += 1 merge_commit = packet.evidence.get("merge_commit") or "N/A" short_sha = merge_commit[:8] if merge_commit != "N/A" else "N/A" severity = packet.evidence.get("severity", "UNKNOWN") other_options = [ opt for opt in packet.safe_options if opt != packet.recommended_option ] other_lines = "\n".join(f" - {opt}" for opt in other_options) header = ( f"🚨 [Critical-{_critical_counter}] {packet.escalation_type.value}\n" f"task: {packet.task_id} / PR: #{packet.pr_number} / merge_commit: {short_sha}\n" f"severity: {severity}\n" f"why: {packet.reason}\n" f"auto cannot continue: {packet.why_auto_cannot_continue}\n" f"options:\n" f" → [recommended] {packet.recommended_option}\n" f"{other_lines}\n" ) evidence_str = json.dumps(packet.evidence, ensure_ascii=False) evidence_prefix = "evidence: " max_evidence = max_len - len(header) - len(evidence_prefix) - 3 # "..." if len(evidence_str) > max_evidence: evidence_str = evidence_str[:max_evidence] + "..." result = header + evidence_prefix + evidence_str # 최종 trim 보장 if len(result) > max_len: result = result[:max_len] return result # --------------------------------------------------------------------------- # 이벤트 routing # --------------------------------------------------------------------------- def _route_event_type(event_type: str) -> CriticalEscalationType | None: """event_type → CriticalEscalationType. 실패 시 None(auto-handled).""" try: return CriticalEscalationType(event_type) except ValueError: pass return LEGACY_CRITICAL_MAP.get(event_type) # --------------------------------------------------------------------------- # process_event — main entry # --------------------------------------------------------------------------- def process_event( event: dict[str, Any], *, workspace_root: Path | str = ".", dry_run: bool = True, no_audit: bool = False, dedup_window_sec: int = 3600, now: datetime | None = None, telegram_send: Callable[[str], None] | None = None, ) -> dict[str, Any]: """이벤트를 수신해 Critical 라우팅, dedup, audit, 포맷을 처리하고 결과 dict 반환.""" ws = Path(workspace_root) now_dt = now or datetime.now(timezone.utc) now_iso = now_dt.isoformat() # 1. 입력 검증 missing = [f for f in ("task_id", "pr_number", "event_type") if not event.get(f)] if missing: raise ValueError(f"필수 필드 누락: {missing}") task_id: str = event["task_id"] pr_number: int = int(event["pr_number"]) event_type: str = str(event["event_type"]) source: str = str(event.get("source", "unknown")) merge_commit: str | None = event.get("merge_commit") or None occurred_at: str | None = event.get("occurred_at") or None original_evidence: dict[str, Any] = dict(event.get("evidence") or {}) # 2. routing canonical: CriticalEscalationType | None = _route_event_type(event_type) is_critical = canonical is not None evidence_keys = list(original_evidence.keys()) evidence_hash = _make_evidence_hash( canonical if canonical else CriticalEscalationType.FORBIDDEN_PATH_INTRUSION, task_id, evidence_keys, ) if is_critical else hashlib.sha256( f"auto-handled|{task_id}|{event_type}".encode() ).hexdigest() # auto-handled if not is_critical: audit_record = _build_audit_record( ts=now_iso, task_id=task_id, pr_number=pr_number, escalation_type=event_type, classification="auto-handled", severity=None, evidence_hash=evidence_hash, source=source, suppressed_reason="not in critical 7 enum (canonical or legacy)", ) if not no_audit: _append_audit_log( ws, audit_record, task_id, {"classification": "auto-handled", "event_type": event_type, "ts": now_iso}, ) return { "classification": "auto-handled", "escalation_type": None, "severity": None, "packet": None, "formatted_text": None, "audit_appended": not no_audit, "evidence_hash": evidence_hash, "suppression_reason": "not in critical 7 enum (canonical or legacy)", } # 3. critical path assert canonical is not None severity = SEVERITY_MAP[canonical] audit_log_path = _global_audit_path(ws) # dedup check is_dup = is_duplicate( canonical, task_id, evidence_keys, audit_log_path, window_sec=dedup_window_sec, now=now_dt, ) if is_dup: dup_reason = f"duplicate within {dedup_window_sec}s window" audit_record = _build_audit_record( ts=now_iso, task_id=task_id, pr_number=pr_number, escalation_type=canonical.value, classification="duplicate-suppressed", severity=severity.value, evidence_hash=evidence_hash, source=source, suppressed_reason=dup_reason, ) if not no_audit: _append_audit_log( ws, audit_record, task_id, {"classification": "duplicate-suppressed", "escalation_type": canonical.value, "ts": now_iso}, ) return { "classification": "duplicate-suppressed", "escalation_type": canonical.value, "severity": severity.value, "packet": None, "formatted_text": None, "audit_appended": not no_audit, "evidence_hash": evidence_hash, "suppression_reason": dup_reason, } # packet 생성 packet = _build_packet( escalation_type=canonical, task_id=task_id, pr_number=pr_number, severity=severity, source=source, merge_commit=merge_commit, occurred_at=occurred_at, original_evidence=original_evidence, now_iso=now_iso, ) formatted_text = format_packet_for_chair(packet) # telegram if not dry_run: if telegram_send is not None: telegram_send(formatted_text) # audit audit_record = _build_audit_record( ts=now_iso, task_id=task_id, pr_number=pr_number, escalation_type=canonical.value, classification="critical", severity=severity.value, evidence_hash=evidence_hash, source=source, suppressed_reason=None, ) if not no_audit: _append_audit_log( ws, audit_record, task_id, dataclasses.asdict(packet), ) return { "classification": "critical", "escalation_type": canonical.value, "severity": severity.value, "packet": packet, "formatted_text": formatted_text, "audit_appended": not no_audit, "evidence_hash": evidence_hash, "suppression_reason": None, } # --------------------------------------------------------------------------- # CLI entrypoint # --------------------------------------------------------------------------- def _serialize_result(result: dict[str, Any]) -> dict[str, Any]: """process_event 결과를 JSON 직렬화 가능한 형태로 변환.""" out = dict(result) if out.get("packet") is not None: out["packet"] = dataclasses.asdict(out["packet"]) return out def main() -> None: """CLI entrypoint.""" parser = argparse.ArgumentParser( description="Critical Escalation Reporter — Critical 7종 이벤트 라우터", ) parser.add_argument("--event-file", required=True, help="이벤트 JSON 파일 경로") parser.add_argument("--workspace-root", default=".", help="workspace root 경로 (기본: .)") parser.add_argument("--dedup-window", type=int, default=3600, help="dedup 윈도우(초)") parser.add_argument("--no-audit", action="store_true", help="audit log 기록 skip") mode_group = parser.add_mutually_exclusive_group() mode_group.add_argument("--dry-run", action="store_true", default=True, help="dry-run 모드 (기본)") mode_group.add_argument("--apply", action="store_true", help="실제 telegram 발송") args = parser.parse_args() event_path = Path(args.event_file) if not event_path.exists(): print(f"ERROR: event file not found: {event_path}", file=sys.stderr) sys.exit(1) try: event = json.loads(event_path.read_text(encoding="utf-8")) except json.JSONDecodeError as exc: print(f"ERROR: invalid JSON in event file: {exc}", file=sys.stderr) sys.exit(1) dry_run = not args.apply try: result = process_event( event, workspace_root=args.workspace_root, dry_run=dry_run, no_audit=args.no_audit, dedup_window_sec=args.dedup_window, ) except ValueError as exc: print(f"ERROR: validation failed: {exc}", file=sys.stderr) sys.exit(1) print(json.dumps(_serialize_result(result), ensure_ascii=False, indent=2)) sys.exit(0) if __name__ == "__main__": main()