# -*- coding: utf-8 -*- """utils.ci_watch_handoff_schema — task-2642 CI_WATCH_HANDOFF JSON v1 schema. 회장 verbatim (2026-05-23 19:38 KST): ANU 는 CI/Gemini 를 직접 기다리지 않는다. PR open 이후 대기/감시/자동수렴은 반드시 bot 또는 watcher task 에 위임한다. 본 helper 책임 (system_ci_watch_handoff_runner_spec_260523.md §2 + 정책 spec §4/§5): - 12 필수 필드 1:1 박제 (pr_number / head_sha / branch / expected_files / forbidden_paths / watcher_owner / max_watch_minutes / poll_interval_seconds / gemini_nudge_policy / auto_remediation_policy / callback_on_terminal_state / terminal_states) - 5 terminal_states enum (MERGE_READY / CHAIR_REQUIRED / GEMINI_EXTERNAL_TRIGGER_STALE / CI_FAILED_NON_REMEDIABLE / LOOP_BOUNDARY) - validate_handoff(): contract 위반 시 SchemaError raise one-way isolation: utils/ 외부 import 0. live cokacdir / gh CLI 호출 0. frozen anchor: ANCHOR-1: 12 필수 필드 1:1 박제 (회장 verbatim §4) ANCHOR-2: 5 terminal_states enum (회장 verbatim §5) ANCHOR-3: gemini_nudge_policy.max_nudges_per_pr_head <= 1 hard limit (회장 verbatim §9, PR #144 NUDGE_HARD_LIMIT_PER_PR_HEAD 정합) ANCHOR-4: auto_remediation_policy.allow_severities ⊆ {medium, style, quality, non-critical-high} (회장 verbatim §8 watcher 책임) """ from __future__ import annotations from typing import Any, Final SCHEMA: Final[str] = "utils.ci_watch_handoff_schema.v1" # terminal_states enum (회장 verbatim §5 — 정책 spec) TERMINAL_MERGE_READY: Final[str] = "MERGE_READY" TERMINAL_CHAIR_REQUIRED: Final[str] = "CHAIR_REQUIRED" TERMINAL_GEMINI_EXTERNAL_TRIGGER_STALE: Final[str] = "GEMINI_EXTERNAL_TRIGGER_STALE" TERMINAL_CI_FAILED_NON_REMEDIABLE: Final[str] = "CI_FAILED_NON_REMEDIABLE" TERMINAL_LOOP_BOUNDARY: Final[str] = "LOOP_BOUNDARY" ALL_TERMINAL_STATES: Final[frozenset[str]] = frozenset( { TERMINAL_MERGE_READY, TERMINAL_CHAIR_REQUIRED, TERMINAL_GEMINI_EXTERNAL_TRIGGER_STALE, TERMINAL_CI_FAILED_NON_REMEDIABLE, TERMINAL_LOOP_BOUNDARY, } ) # 12 필수 필드 (회장 verbatim §4 — 정책 spec) REQUIRED_FIELDS: Final[tuple[str, ...]] = ( "pr_number", "head_sha", "branch", "expected_files", "forbidden_paths", "watcher_owner", "max_watch_minutes", "poll_interval_seconds", "gemini_nudge_policy", "auto_remediation_policy", "callback_on_terminal_state", "terminal_states", ) # spec default (회장 verbatim §4 / 본 spec §2) DEFAULT_MAX_WATCH_MINUTES: Final[int] = 120 DEFAULT_POLL_INTERVAL_SECONDS: Final[int] = 60 # 회장 verbatim §8 (정책 spec) — watcher 자동수렴 허용 severity ALLOWED_AUTO_REMEDIATION_SEVERITIES: Final[frozenset[str]] = frozenset( {"medium", "style", "quality", "non-critical-high"} ) class SchemaError(ValueError): """CI_WATCH_HANDOFF JSON 의 schema / contract 위반.""" def _is_positive_int(value: Any) -> bool: return isinstance(value, int) and not isinstance(value, bool) and value > 0 def _is_non_negative_int(value: Any) -> bool: return isinstance(value, int) and not isinstance(value, bool) and value >= 0 def _is_non_empty_string_list(value: Any) -> bool: return ( isinstance(value, list) and len(value) > 0 and all(isinstance(x, str) and x for x in value) ) def _is_string_list_allow_empty(value: Any) -> bool: return isinstance(value, list) and all( isinstance(x, str) and x for x in value ) def validate_handoff(handoff: Any) -> dict: """CI_WATCH_HANDOFF JSON 의 contract 검증 + 정규화. Args: handoff: 검증 대상 dict (12 필수 필드 포함). Returns: 정규화된 handoff dict (head_sha lower). Raises: SchemaError: 12 필수 필드 누락 / 타입 위반 / 5 terminal_states enum 위반 / nudge hard limit 위반 / severity 화이트리스트 위반. """ if not isinstance(handoff, dict): raise SchemaError( f"handoff must be dict, got {type(handoff).__name__}" ) missing = [k for k in REQUIRED_FIELDS if k not in handoff] if missing: raise SchemaError(f"missing required fields: {missing}") pr = handoff["pr_number"] if not _is_positive_int(pr): raise SchemaError("pr_number must be positive int") head = handoff["head_sha"] if not isinstance(head, str) or len(head) != 40: raise SchemaError("head_sha must be 40-char hex string") head_norm = head.lower() if any(c not in "0123456789abcdef" for c in head_norm): raise SchemaError("head_sha must be 40-char hex (a-f / 0-9 only)") branch = handoff["branch"] if not isinstance(branch, str) or not branch.strip(): raise SchemaError("branch must be non-empty string") expected_files = handoff["expected_files"] if not _is_non_empty_string_list(expected_files): raise SchemaError("expected_files must be non-empty list[non-empty str]") forbidden_paths = handoff["forbidden_paths"] if not _is_string_list_allow_empty(forbidden_paths): raise SchemaError("forbidden_paths must be list[non-empty str]") watcher_owner = handoff["watcher_owner"] if not isinstance(watcher_owner, str) or not watcher_owner.strip(): raise SchemaError("watcher_owner must be non-empty string") mwm = handoff["max_watch_minutes"] if not _is_positive_int(mwm): raise SchemaError("max_watch_minutes must be positive int") pis = handoff["poll_interval_seconds"] if not _is_positive_int(pis): raise SchemaError("poll_interval_seconds must be positive int") gnp = handoff["gemini_nudge_policy"] if not isinstance(gnp, dict): raise SchemaError("gemini_nudge_policy must be dict") if "enabled" not in gnp or not isinstance(gnp["enabled"], bool): raise SchemaError("gemini_nudge_policy.enabled must be bool") max_nudges = gnp.get("max_nudges_per_pr_head", 1) if not _is_non_negative_int(max_nudges): raise SchemaError( "gemini_nudge_policy.max_nudges_per_pr_head must be non-negative int" ) # ANCHOR-3: 회장 verbatim §9 hard limit 보호 (PR #144 NUDGE_HARD_LIMIT_PER_PR_HEAD 정합) if max_nudges > 1: raise SchemaError( "gemini_nudge_policy.max_nudges_per_pr_head must be <= 1 " "(회장 verbatim §9 nudge 1회 hard limit)" ) on_403 = gnp.get("on_403", "report") if on_403 != "report": raise SchemaError( "gemini_nudge_policy.on_403 must be 'report' (회장 verbatim §8 NUDGE_403 path)" ) arp = handoff["auto_remediation_policy"] if not isinstance(arp, dict): raise SchemaError("auto_remediation_policy must be dict") if "enabled" not in arp or not isinstance(arp["enabled"], bool): raise SchemaError("auto_remediation_policy.enabled must be bool") severities = arp.get("allow_severities", []) if not isinstance(severities, list): raise SchemaError("auto_remediation_policy.allow_severities must be list") # ANCHOR-4: 회장 verbatim §8 watcher 책임 화이트리스트 for sev in severities: if not isinstance(sev, str) or sev not in ALLOWED_AUTO_REMEDIATION_SEVERITIES: raise SchemaError( f"auto_remediation_policy.allow_severities contains invalid " f"severity {sev!r}; allowed=" f"{sorted(ALLOWED_AUTO_REMEDIATION_SEVERITIES)}" ) cb = handoff["callback_on_terminal_state"] if not isinstance(cb, bool): raise SchemaError("callback_on_terminal_state must be bool") terms = handoff["terminal_states"] if not isinstance(terms, list) or not terms: raise SchemaError("terminal_states must be non-empty list") for term in terms: if term not in ALL_TERMINAL_STATES: raise SchemaError( f"terminal_states contains invalid enum {term!r}; " f"allowed={sorted(ALL_TERMINAL_STATES)}" ) normalized = dict(handoff) normalized["head_sha"] = head_norm return normalized __all__ = [ "SCHEMA", "TERMINAL_MERGE_READY", "TERMINAL_CHAIR_REQUIRED", "TERMINAL_GEMINI_EXTERNAL_TRIGGER_STALE", "TERMINAL_CI_FAILED_NON_REMEDIABLE", "TERMINAL_LOOP_BOUNDARY", "ALL_TERMINAL_STATES", "REQUIRED_FIELDS", "DEFAULT_MAX_WATCH_MINUTES", "DEFAULT_POLL_INTERVAL_SECONDS", "ALLOWED_AUTO_REMEDIATION_SEVERITIES", "SchemaError", "validate_handoff", ]