"""anu_v2.ci_gemini_watcher_runner — CI + Gemini watcher 단일 멱등 cycle runner (task-2718). 회장 인가: 기존 anu_v2 빌딩블록을 import/call 로 결선하는 얇은 runner. 신규 로직 최소화. 실제 GitHub write 0 (decision/dry-run only). task-2718. 책임: - 기존 빌딩블록(freshness_checker / validate_decision / AutoGeminiTriage / ExecutorScheduler) 을 결선해 단일 1회 실행 → terminal enum 1 개 반환. - 세션-bound sleep/polling 절대 금지 (1 회 실행 → 즉시 반환). - 실제 GitHub write 0 (github_writes 카운터 항상 0 유지). - 모든 부수효과는 주입형 callable 로 추상화. one-way isolation: anu_v2/* 만 import. 외부 (utils/dispatch/scripts/dashboard) 의존성 0. """ from __future__ import annotations import dataclasses from typing import Any, Callable, Final, Mapping, Sequence # ── 기존 빌딩블록 genuine wiring (4 개 import — 실제 결선 입증) ───────────── from anu_v2.gemini_evidence_freshness_checker import ( check_gemini_evidence_fresh, RESULT_FRESH, ) from anu_v2.owner_trigger_decision import ( validate_decision, SCHEMA_NAME, ALLOWED_ACTION, ALLOWED_COMMENT_BODY, DecisionInvalidError, ) from anu_v2.auto_gemini_triage import AutoGeminiTriage from anu_v2.executor_scheduler import ExecutorScheduler # ────────────────────────────────────────────────────────────────────────────── # terminal 상수 (모듈 레벨 Final) # ────────────────────────────────────────────────────────────────────────────── TERMINAL_MERGE_READY_CANDIDATE: Final[str] = "MERGE_READY_CANDIDATE" TERMINAL_LOOP_BOUNDARY: Final[str] = "LOOP_BOUNDARY" TERMINAL_BLOCKED_BY_CAPABILITY: Final[str] = "BLOCKED_BY_CAPABILITY" TERMINAL_GEMINI_EXTERNAL_TRIGGER_REQUIRED: Final[str] = "GEMINI_EXTERNAL_TRIGGER_REQUIRED" TERMINAL_CI_FAILED_NON_REMEDIABLE: Final[str] = "CI_FAILED_NON_REMEDIABLE" TERMINAL_HOLD_STALE_HEAD: Final[str] = "HOLD_STALE_HEAD" TERMINAL_HOLD_SCOPE_UNCLEAN: Final[str] = "HOLD_SCOPE_UNCLEAN" ALL_TERMINALS: Final[frozenset[str]] = frozenset({ TERMINAL_MERGE_READY_CANDIDATE, TERMINAL_LOOP_BOUNDARY, TERMINAL_BLOCKED_BY_CAPABILITY, TERMINAL_GEMINI_EXTERNAL_TRIGGER_REQUIRED, TERMINAL_CI_FAILED_NON_REMEDIABLE, TERMINAL_HOLD_STALE_HEAD, TERMINAL_HOLD_SCOPE_UNCLEAN, }) # ── trigger decision 상수 ───────────────────────────────────────────────────── TRIGGER_ALLOW_OWNER: Final[str] = "ALLOW_OWNER_TRIGGER" TRIGGER_OWNER_DEDUPED: Final[str] = "OWNER_TRIGGER_DEDUPED" TRIGGER_EXTERNAL_REQUIRED: Final[str] = "GEMINI_EXTERNAL_TRIGGER_REQUIRED" TRIGGER_SELF_BLOCKED: Final[str] = "SELF_GEMINI_TRIGGER_BLOCKED" TRIGGER_NONE: Final[str] = "NONE" # ────────────────────────────────────────────────────────────────────────────── # WatchResult (frozen dataclass) # ────────────────────────────────────────────────────────────────────────────── @dataclasses.dataclass(frozen=True) class WatchResult: """run_watch_cycle 결과 객체 (frozen — 불변 값 객체). fields: terminal : ALL_TERMINALS 중 하나. trigger_decision: TRIGGER_* 상수 중 하나. pr_number : 평가된 PR 번호. expected_head : queue/expected head SHA (입력). actual_head : gh_runner 로 실측한 PR HEAD SHA. ci_state : CI rollup state ("SUCCESS" | "FAILURE" | "PENDING" | "UNKNOWN"). gemini_freshness: FRESH | STALE | NO_REVIEW | "UNKNOWN". triage_summary : triage_batch 결과 count 요약 (FRESH 시 채움). decision_json : ALLOW_OWNER_TRIGGER 시 validate 통과 decision dict. github_writes : 항상 0 (실 write 0 입증). dry_run : 기본 True. reason : 짧은 사유 문자열. """ terminal: str trigger_decision: str pr_number: int expected_head: str actual_head: str ci_state: str gemini_freshness: str triage_summary: dict | None decision_json: dict | None github_writes: int dry_run: bool reason: str def to_json(self) -> dict: """JSON 직렬화용 dict 반환 (dataclasses.asdict 기반).""" return dataclasses.asdict(self) # ────────────────────────────────────────────────────────────────────────────── # scheduler_backed_dedupe 헬퍼 (ExecutorScheduler 결선용) # ────────────────────────────────────────────────────────────────────────────── def scheduler_backed_dedupe( scheduler: ExecutorScheduler, *, pr_number: int, head_sha: str, ) -> bool: """ExecutorScheduler._has_active_trigger_for_head 를 thin-wrap 하는 dedupe 헬퍼. 프로덕션: run_watch_cycle(dedupe_checker=lambda pr, sha: scheduler_backed_dedupe(sched, pr_number=pr, head_sha=sha)) 테스트 : dedupe_checker=mock 주입. ExecutorScheduler 구성은 runner 외부에서 담당 (thin runner 책임 최소화). """ return scheduler._has_active_trigger_for_head(pr_number=pr_number, head_sha=head_sha) # ────────────────────────────────────────────────────────────────────────────── # 내부 유틸 # ────────────────────────────────────────────────────────────────────────────── def _validate_hex_sha(value: str, name: str) -> str: """40-char hex SHA 검증 후 lower 정규화. 실패 시 ValueError.""" if not isinstance(value, str) or len(value) != 40: raise ValueError(f"{name} 는 40-char hex SHA 문자열이어야 합니다, got {value!r}") lowered = value.lower() if any(c not in "0123456789abcdef" for c in lowered): raise ValueError(f"{name} 는 40-char hex SHA (hex chars only), got {value!r}") return lowered def _triage_count_summary(out: dict) -> dict: """triage_batch 결과에서 count 요약 dict 생성.""" return { "applied_count": len(out.get("applied") or []), "dismissed_count": len(out.get("dismissed") or []), "escalated_count": len(out.get("escalated") or []), } # ────────────────────────────────────────────────────────────────────────────── # 진입점: run_watch_cycle # ────────────────────────────────────────────────────────────────────────────── def run_watch_cycle( *, pr_number: int, expected_head: str, expected_files: Sequence[str], gh_runner: Callable[..., Any], owner: str = "", repo: str = "", task_id: str = "unknown", owner_proof: Mapping[str, Any] | None = None, dedupe_checker: Callable[[int, str], bool] | None = None, record_trigger: Callable[[int, str], None] | None = None, triage: AutoGeminiTriage | None = None, audit_writer: Callable[[Mapping[str, Any]], None] | None = None, dry_run: bool = True, ) -> WatchResult: """CI + Gemini watcher 단일 멱등 cycle runner (1회 실행 → terminal enum 1개 반환). 세션-bound sleep/polling 절대 금지. github_writes 는 항상 0. 결정 로직 순서는 명세 §결정로직 1~7 1:1. Args: pr_number : 평가 대상 PR 번호 (positive int). expected_head : queue/expected head SHA (40-char hex). expected_files : 기대 diff 파일 경로 목록 (scope 검사 기준). gh_runner : 주입형 부수효과 callable (op 분기 — 명세 §gh_runner 참조). owner : GitHub owner (freshness checker 에 전달). repo : GitHub repo (freshness checker 에 전달). task_id : 감사 기록용 task ID. owner_proof : OWNER 권한 증빙 dict (is_owner/admin/push/self_key). None 가능. dedupe_checker : (pr_number, head_sha) → bool. 이미 trigger 된 head 중복 방지. record_trigger : (pr_number, head_sha) → None. dry-run trigger 기록 (부수효과 없음). triage : AutoGeminiTriage 인스턴스. None 이면 내부 생성. audit_writer : triage 내부 audit 기록용 callable. dry_run : 기본 True (실 write 0 보장). Returns: WatchResult (frozen dataclass). Raises: ValueError: pr_number <= 0 또는 expected_head 형식 오류. """ # ─── 공통 WatchResult 팩토리 (반복 코드 최소화) ───────────────────────── def _make( terminal: str, *, trigger_decision: str = TRIGGER_NONE, actual_head: str = "", ci_state: str = "UNKNOWN", gemini_freshness: str = "UNKNOWN", triage_summary: dict | None = None, decision_json: dict | None = None, reason: str = "", ) -> WatchResult: """내부 WatchResult 생성 헬퍼. github_writes 는 항상 0.""" assert terminal in ALL_TERMINALS, f"terminal {terminal!r} not in ALL_TERMINALS" return WatchResult( terminal=terminal, trigger_decision=trigger_decision, pr_number=pr_number, expected_head=expected_head, actual_head=actual_head, ci_state=ci_state, gemini_freshness=gemini_freshness, triage_summary=triage_summary, decision_json=decision_json, github_writes=0, # 실 write 0 — 절대 증가 금지 dry_run=dry_run, reason=reason, ) # ─── 1. 입력 검증 ───────────────────────────────────────────────────── if not isinstance(pr_number, int) or isinstance(pr_number, bool) or pr_number <= 0: raise ValueError(f"pr_number 은 양의 정수여야 합니다, got {pr_number!r}") expected_head_norm = _validate_hex_sha(expected_head, "expected_head") # ─── 2. capability gate: gh_runner 없으면 즉시 BLOCKED ──────────────── if gh_runner is None: return _make( TERMINAL_BLOCKED_BY_CAPABILITY, trigger_decision=TRIGGER_NONE, reason="gh_runner 이 None — capability gate 차단", ) # ─── 3. actual_head 실측 (lower 정규화) ────────────────────────────── actual_head_raw = gh_runner("actual_head", pr_number=pr_number) # invalid head 방어: None/빈값을 'None' 같은 암묵 문자열화로 정상 head 처럼 # 흐르지 않게 빈 문자열로 정규화 → head 비교에서 stale 로 fail-closed. actual_head = str(actual_head_raw).lower().strip() if actual_head_raw else "" # ─── 4. head 비교 — 불일치 시 HOLD_STALE_HEAD ──────────────────────── if actual_head != expected_head_norm: return _make( TERMINAL_HOLD_STALE_HEAD, actual_head=actual_head, reason=( f"actual HEAD ({actual_head[:8]}...) != " f"expected HEAD ({expected_head_norm[:8]}...) — stale head" ), ) # ─── 5. scope 검사 — unexpected diff paths 있으면 HOLD_SCOPE_UNCLEAN ── diff_paths: list[str] = gh_runner("diff_paths", pr_number=pr_number) or [] expected_files_set = set(expected_files) out_of_scope = [p for p in diff_paths if p not in expected_files_set] if out_of_scope: return _make( TERMINAL_HOLD_SCOPE_UNCLEAN, actual_head=actual_head, reason=( f"diff 에 scope 외 경로 {len(out_of_scope)}건 포함: " f"{out_of_scope[:3]!r}{'...' if len(out_of_scope) > 3 else ''}" ), ) # ─── 6. CI rollup — FAILURE + not remediable → CI_FAILED_NON_REMEDIABLE ─ ci: dict = gh_runner("ci_rollup", pr_number=pr_number) or {} ci_state: str = ci.get("state", "UNKNOWN") if ci_state == "FAILURE" and not ci.get("remediable", False): return _make( TERMINAL_CI_FAILED_NON_REMEDIABLE, actual_head=actual_head, ci_state=ci_state, reason=f"CI FAILURE (non-remediable) — 자동 복구 불가", ) # ─── 7. Gemini freshness 판정 ────────────────────────────────────────── # freshness checker 에 넘길 github_api 어댑터 (reviews op 결선) _github_api: Callable[[str, str], Any] = ( lambda method, path: gh_runner("reviews", method=method, path=path) ) fr = check_gemini_evidence_fresh( pr_number=pr_number, current_head_sha=actual_head, github_api=_github_api, owner=owner, repo=repo, ) gemini_freshness: str = fr.status # ─── 7-A. FRESH: triage 실행 ────────────────────────────────────────── if fr.status == RESULT_FRESH: findings: list[dict] = gh_runner("findings", pr_number=pr_number) or [] # triage 인스턴스 없으면 내부 생성 (audit_writer 기본 noop) _triage = triage if _triage is None: _audit_writer: Callable[[Mapping[str, Any]], None] = ( audit_writer if audit_writer is not None else (lambda rec: None) ) _triage = AutoGeminiTriage( audit_writer=_audit_writer, task_id=task_id, ) # triage_batch 실제 호출 (입증 대상) triage_out: dict = _triage.triage_batch(findings, expected_files) summary = _triage_count_summary(triage_out) unresolved: list = triage_out.get("escalated") or [] if unresolved: # escalated 있음 → loop 경계 (보수적) return _make( TERMINAL_LOOP_BOUNDARY, actual_head=actual_head, ci_state=ci_state, gemini_freshness=gemini_freshness, triage_summary=summary, reason=f"triage escalated {len(unresolved)}건 — LOOP_BOUNDARY", ) if ci_state == "SUCCESS": # escalated 없음 + CI SUCCESS → merge ready return _make( TERMINAL_MERGE_READY_CANDIDATE, actual_head=actual_head, ci_state=ci_state, gemini_freshness=gemini_freshness, triage_summary=summary, reason="triage clean + CI SUCCESS → MERGE_READY_CANDIDATE", ) # CI PENDING 등 — 보수적으로 LOOP_BOUNDARY return _make( TERMINAL_LOOP_BOUNDARY, actual_head=actual_head, ci_state=ci_state, gemini_freshness=gemini_freshness, triage_summary=summary, reason=f"triage clean 이지만 CI={ci_state} — 보수적 LOOP_BOUNDARY", ) # ─── 7-B. STALE 또는 NO_REVIEW: Gemini 재트리거 필요 ───────────────── # ① self_key 검사 — SELF_GEMINI_TRIGGER_BLOCKED if owner_proof is not None and owner_proof.get("self_key"): return _make( TERMINAL_BLOCKED_BY_CAPABILITY, trigger_decision=TRIGGER_SELF_BLOCKED, actual_head=actual_head, ci_state=ci_state, gemini_freshness=gemini_freshness, reason="owner_proof.self_key=True — self Gemini trigger 차단", ) # ② is_owner 판정 (admin 또는 push 권한 보유) is_owner: bool = ( bool(owner_proof) and bool(owner_proof.get("is_owner")) and (bool(owner_proof.get("admin")) or bool(owner_proof.get("push"))) ) if not is_owner: # owner 가 아님 → 외부 trigger 필요 return _make( TERMINAL_GEMINI_EXTERNAL_TRIGGER_REQUIRED, trigger_decision=TRIGGER_EXTERNAL_REQUIRED, actual_head=actual_head, ci_state=ci_state, gemini_freshness=gemini_freshness, reason="owner proof 없음/불충분 — GEMINI_EXTERNAL_TRIGGER_REQUIRED", ) # ③ dedupe 검사 — 이미 trigger 됐으면 OWNER_TRIGGER_DEDUPED already_triggered: bool = ( dedupe_checker(pr_number, actual_head) if dedupe_checker is not None else False ) if already_triggered: return _make( TERMINAL_GEMINI_EXTERNAL_TRIGGER_REQUIRED, trigger_decision=TRIGGER_OWNER_DEDUPED, actual_head=actual_head, ci_state=ci_state, gemini_freshness=gemini_freshness, reason=f"(pr={pr_number}, head={actual_head[:8]}...) 이미 trigger 됨 — DEDUPED", ) # ④ decision dict 구성 후 validate_decision 실제 호출 (입증 대상) decision: dict = { "schema": SCHEMA_NAME, "task_id": task_id, "pr": pr_number, "current_head": actual_head, "queue_head": True, "current_head_confirmed": True, "gemini_evidence_fresh": False, "nudge_count_for_pr_head": 0, "allowed_action": ALLOWED_ACTION, "comment_body": ALLOWED_COMMENT_BODY, "allowed": True, } try: validate_decision(decision, current_head_actual=actual_head) except DecisionInvalidError as exc: # fail-closed: 검증 실패 → BLOCKED_BY_CAPABILITY return _make( TERMINAL_BLOCKED_BY_CAPABILITY, trigger_decision=TRIGGER_NONE, actual_head=actual_head, ci_state=ci_state, gemini_freshness=gemini_freshness, reason=f"validate_decision 실패 code={exc.code}: {exc.message}", ) # ⑤ dry-run trigger 기록 (부수효과 없음 — record_trigger 주입형) if record_trigger is not None: record_trigger(pr_number, actual_head) # github_writes 는 그대로 0 유지 (실 write 0) return _make( TERMINAL_GEMINI_EXTERNAL_TRIGGER_REQUIRED, trigger_decision=TRIGGER_ALLOW_OWNER, actual_head=actual_head, ci_state=ci_state, gemini_freshness=gemini_freshness, decision_json=decision, reason=( f"owner trigger 허가 — pr={pr_number}, head={actual_head[:8]}..., " f"freshness={fr.status}, dry_run={dry_run}" ), ) # ────────────────────────────────────────────────────────────────────────────── # __all__ # ────────────────────────────────────────────────────────────────────────────── __all__ = [ # terminal 상수 "TERMINAL_MERGE_READY_CANDIDATE", "TERMINAL_LOOP_BOUNDARY", "TERMINAL_BLOCKED_BY_CAPABILITY", "TERMINAL_GEMINI_EXTERNAL_TRIGGER_REQUIRED", "TERMINAL_CI_FAILED_NON_REMEDIABLE", "TERMINAL_HOLD_STALE_HEAD", "TERMINAL_HOLD_SCOPE_UNCLEAN", "ALL_TERMINALS", # trigger decision 상수 "TRIGGER_ALLOW_OWNER", "TRIGGER_OWNER_DEDUPED", "TRIGGER_EXTERNAL_REQUIRED", "TRIGGER_SELF_BLOCKED", "TRIGGER_NONE", # 공개 API "WatchResult", "run_watch_cycle", "scheduler_backed_dedupe", ]