"""anu_v2.owner_gemini_trigger_router — task-2641 신규 상위 layer router. 회장 verbatim 12 (2026-05-23) 1:1 박제 — OWNER_GEMINI_TRIGGER_UI_FALLBACK_MISROUTE 재발 방지 router 신설. spec §2 (회장 verbatim 12 필수 구현) 매핑: §1, §2 PR Review comment ≠ Gemini trigger → ``is_gemini_trigger_comment`` §3 PR-backed issue comment body="/gemini review" 만 trigger 인정 §4, §11 ANU OWNER issue comment 발사 = **1차 nudge** (회장 UI 1차 안내 금지) §5 OWNER_GEMINI_TRIGGER_TOKEN / owner_trigger_only capability 재사용 §6 raw token 출력 금지 (token_hash_prefix 12 hex 만) §7 gh api / 안전한 경로만 (subprocess 직접 호출 0) §8 403 → X-Accepted-GitHub-Permissions header 기록 §9 nudge 1회 hard limit per PR/head §10 fresh review 미도착 → GEMINI_EXTERNAL_TRIGGER_STALE §12 OWNER_GEMINI_TRIGGER_UI_FALLBACK_MISROUTE 재발 방지 fixture 검증 state machine (spec §3.2): 1. ``is_gemini_trigger_comment(observed_comment)`` — PR Review 오인 차단 2. ``check_gemini_evidence_fresh`` — FRESH 면 통과 (action 0) 3. STALE → nudge_count_for_pr_head == 0 이면 nudge 발사 4. 결과 분류: - POSTED → polling timeout 후 재freshness → FRESH 면 통과 / STALE 면 GEMINI_EXTERNAL_TRIGGER_STALE - DEDUPED → nudge_count > 0 (재시도 차단) - FAILED 403 → CHAIR_UI_FALLBACK_REQUIRED + permission_header_diagnostics 기록 - FAILED other → NUDGE_FAILED 본 router 는 ``owner_trigger_only.invoke_from_scheduler`` 호출만 사용 (spec §3.2, token 직접 노출 0). 기존 owner_trigger_only / decision / audit / pat 무수정. one-way isolation: anu_v2/ 외부 import 금지. live cokacdir / gh / merge / push 0. """ from __future__ import annotations from dataclasses import dataclass from pathlib import Path from typing import Any, Callable, Final from anu_v2.gemini_evidence_freshness_checker import ( RESULT_FRESH, FreshnessCheckerError, FreshnessResult, check_gemini_evidence_fresh, is_gemini_trigger_comment, ) from anu_v2.owner_gemini_trigger_router_audit import ( AUDIT_SCHEMA, OwnerGeminiTriggerRouterAudit, RouterAuditRedactionError, STATE_CHAIR_UI_FALLBACK_REQUIRED, STATE_FRESH, STATE_GEMINI_EXTERNAL_TRIGGER_STALE, STATE_NOT_GEMINI_TRIGGER, STATE_NUDGE_DEDUPED, STATE_NUDGE_FAILED, STATE_NUDGE_PERMISSION_DENIED, STATE_NUDGE_POSTED, extract_403_headers, redact_diagnostics, token_hash_prefix as router_token_hash_prefix, ) from anu_v2.owner_trigger_audit import ( RESULT_DEDUPED, RESULT_FAILED, RESULT_POSTED, ) # RouterResult enum aliases (spec §3.2). 외부 caller 가 직접 비교할 수 있도록 export. ROUTER_RESULT_FRESH: Final[str] = STATE_FRESH ROUTER_RESULT_NUDGE_POSTED: Final[str] = STATE_NUDGE_POSTED ROUTER_RESULT_NUDGE_DEDUPED: Final[str] = STATE_NUDGE_DEDUPED ROUTER_RESULT_GEMINI_EXTERNAL_TRIGGER_STALE: Final[str] = ( STATE_GEMINI_EXTERNAL_TRIGGER_STALE ) ROUTER_RESULT_CHAIR_UI_FALLBACK_REQUIRED: Final[str] = ( STATE_CHAIR_UI_FALLBACK_REQUIRED ) ROUTER_RESULT_NUDGE_PERMISSION_DENIED: Final[str] = STATE_NUDGE_PERMISSION_DENIED ROUTER_RESULT_NUDGE_FAILED: Final[str] = STATE_NUDGE_FAILED ROUTER_RESULT_NOT_GEMINI_TRIGGER: Final[str] = STATE_NOT_GEMINI_TRIGGER # 회장 verbatim §9: nudge 1회 hard limit per PR/head NUDGE_HARD_LIMIT_PER_PR_HEAD: Final[int] = 1 # 회장 verbatim §10: fresh review 미도착 timeout (spec §2.9 기본 600s). # polling 간격은 caller / scheduler 가 조정. 본 router 는 단일 호출에서 1회만 # re-check (no in-process sleep — cron 또는 caller polling 책임). DEFAULT_FRESH_REVIEW_TIMEOUT_S: Final[int] = 600 class RouterContractError(RuntimeError): """router 입력 schema 위반 (PR / head SHA / decision).""" @dataclass(frozen=True) class RouterResult: """OwnerGeminiTriggerRouter.route_for_pr 반환 객체. final_state 는 STATE_* enum 중 하나. caller 는 final_state 로 분기: - FRESH / NUDGE_POSTED / NUDGE_DEDUPED → pass 또는 polling - GEMINI_EXTERNAL_TRIGGER_STALE → 별도 보고 / 회장 escalation (spec §2.9) - CHAIR_UI_FALLBACK_REQUIRED → ANU 가 회장 UI 입력 요청 (최후 수단) - NUDGE_PERMISSION_DENIED → 403 header diagnostics 기록됨 - NUDGE_FAILED → transient, caller retry 정책에 위임 - NOT_GEMINI_TRIGGER → PR Review 등 trigger 오인 차단 (task-2640 사고 박제) """ final_state: str pr_number: int current_head_sha: str freshness_state: str gemini_commit_id_observed: str | None nudge_attempted: bool nudge_result: str | None permission_header_diagnostics: dict | None token_present: bool token_hash_prefix: str reason: str class OwnerGeminiTriggerRouter: """OWNER_GEMINI_TRIGGER_ROUTER 단일 진입점 (spec §3.2). side-effect 추상화 (test injection 가능): - ``freshness_checker``: ``Callable[**kwargs, FreshnessResult]`` - ``invoke_scheduler``: ``Callable[**kwargs, str]`` — owner_trigger_only ``invoke_from_scheduler`` 시그니처 ("POSTED"/"DEDUPED"/"FAILED"/"PENDING"). - ``permission_diagnostics_provider``: ``Callable[[], dict | None]`` — last invoke 의 403 response headers (없으면 None). spec §2.7 / §3.2-4. - ``token_provider``: ``Callable[[], str]`` — token_hash_prefix 계산용. token 자체는 본 router 가 보관/노출 0 (회장 verbatim §6). spec §3.2 state machine: 1. observed_comment 가 제공된 경우 ``is_gemini_trigger_comment`` 검사 (PR Review 오인 차단 — task-2640 사고 박제) 2. ``check_gemini_evidence_fresh`` → FRESH 면 통과 3. STALE / NO_REVIEW → audit 에서 nudge_count_for_pr_head 조회 · == 0 이면 nudge 발사 · >= 1 이면 NUDGE_DEDUPED (회장 verbatim §9 hard limit) 4. invoke_scheduler 결과 분류: · POSTED → fresh_review_arrived_post_nudge 인자 True 시 FRESH/STALE 재평가 False 시 GEMINI_EXTERNAL_TRIGGER_STALE (timeout 경로) · DEDUPED → NUDGE_DEDUPED · FAILED + permission_diagnostics_provider 가 403 header 회수 → NUDGE_PERMISSION_DENIED + CHAIR_UI_FALLBACK_REQUIRED · FAILED 그 외 → NUDGE_FAILED → CHAIR_UI_FALLBACK_REQUIRED (spec §2.10) """ def __init__( self, *, workspace_root: str | Path, freshness_checker: Callable[..., FreshnessResult] = check_gemini_evidence_fresh, invoke_scheduler: Callable[..., str] | None = None, permission_diagnostics_provider: Callable[[], dict | None] | None = None, token_provider: Callable[[], str] | None = None, audit: OwnerGeminiTriggerRouterAudit | None = None, github_api: Callable[[str, str], Any] | None = None, ) -> None: if invoke_scheduler is None: raise NotImplementedError( "invoke_scheduler callable must be injected — " "owner_trigger_only.invoke_from_scheduler 권장" ) if github_api is None: raise NotImplementedError( "github_api callable must be injected (regression mock 가능)" ) self._workspace_root = Path(workspace_root).resolve() self._freshness_checker = freshness_checker self._invoke_scheduler = invoke_scheduler self._permission_diagnostics_provider = ( permission_diagnostics_provider or (lambda: None) ) self._token_provider = token_provider self._github_api = github_api self._audit = ( audit if audit is not None else OwnerGeminiTriggerRouterAudit(self._workspace_root) ) # ─── public API ──────────────────────────────────────────────────────── def route_for_pr( self, *, pr_number: int, current_head_sha: str, owner: str, repo: str, task_id: str = "", decision_path: str | Path | None = None, observed_comment: dict | None = None, fresh_review_arrived_post_nudge: bool = False, nudge_hard_limit: int = NUDGE_HARD_LIMIT_PER_PR_HEAD, ) -> RouterResult: """router state machine 1회 호출. spec §3.2. Args: pr_number: 평가 대상 PR 번호. current_head_sha: 실측 PR HEAD SHA (40-char hex). owner / repo: GitHub repository 좌표. task_id: audit 추적용 (optional). decision_path: invoke_scheduler 가 owner_trigger_only.invoke_from_scheduler 를 호출할 때 필요한 decision JSON 경로. STALE 시 필수. observed_comment: 평가 시점에 관측된 comment dict (kind / body / user). 제공되면 ``is_gemini_trigger_comment`` 검사 — PR Review 오인 차단 (task-2640 사고 박제). 미제공 시 freshness 만 검사. fresh_review_arrived_post_nudge: nudge POSTED 후 caller 가 polling 끝에 fresh review 도착을 관측했는지. True → FRESH 재평가, False → timeout classification (GEMINI_EXTERNAL_TRIGGER_STALE). nudge_hard_limit: PR/head 당 nudge 한도 (회장 verbatim §9, default 1). Returns: RouterResult — final_state enum + diagnostics. Raises: RouterContractError: 입력 schema 위반. """ self._validate_input(pr_number, current_head_sha, owner, repo) head_norm = current_head_sha.lower() token_present, hash_prefix = self._compute_token_hash_prefix() # 1) observed_comment 가 제공된 경우 PR Review 오인 차단 (회장 §1/§2/§3) if observed_comment is not None and not is_gemini_trigger_comment( observed_comment ): reason = ( "observed_comment 가 PR-backed issue comment body='/gemini review' " "이 아님 — PR Review / empty body 오인 차단 (task-2640 사고 박제)" ) result = RouterResult( final_state=STATE_NOT_GEMINI_TRIGGER, pr_number=pr_number, current_head_sha=head_norm, freshness_state="N/A", gemini_commit_id_observed=None, nudge_attempted=False, nudge_result=None, permission_header_diagnostics=None, token_present=token_present, token_hash_prefix=hash_prefix, reason=reason, ) self._audit_append(result, task_id=task_id) return result # 2) freshness 평가 try: freshness = self._freshness_checker( pr_number=pr_number, current_head_sha=head_norm, github_api=self._github_api, owner=owner, repo=repo, ) except FreshnessCheckerError as exc: # input schema 위반 → router contract error 로 변환 (fail-closed) raise RouterContractError( f"freshness check failed: {exc!r}" ) from exc if freshness.status == RESULT_FRESH: obs_prefix = ( freshness.gemini_commit_id_observed[:8] if isinstance(freshness.gemini_commit_id_observed, str) else "????????" ) reason = ( f"Gemini fresh evidence 일치 ({obs_prefix}" f"... == {head_norm[:8]}...) — nudge 미발사" ) result = RouterResult( final_state=STATE_FRESH, pr_number=pr_number, current_head_sha=head_norm, freshness_state=freshness.status, gemini_commit_id_observed=freshness.gemini_commit_id_observed, nudge_attempted=False, nudge_result=None, permission_header_diagnostics=None, token_present=token_present, token_hash_prefix=hash_prefix, reason=reason, ) self._audit_append(result, task_id=task_id) return result # freshness.status in {STALE, NO_REVIEW} → 1차 OWNER nudge 검토 # 3) nudge_count_for_pr_head 확인 (회장 verbatim §9 hard limit) prior_nudge_count = self._audit.nudge_count_for_pr_head( pr_number=pr_number, head=head_norm ) if prior_nudge_count >= nudge_hard_limit: reason = ( f"nudge_count_for_pr_head={prior_nudge_count} >= hard limit " f"({nudge_hard_limit}) — DEDUPED (회장 verbatim §9)" ) result = RouterResult( final_state=STATE_NUDGE_DEDUPED, pr_number=pr_number, current_head_sha=head_norm, freshness_state=freshness.status, gemini_commit_id_observed=freshness.gemini_commit_id_observed, nudge_attempted=False, nudge_result=RESULT_DEDUPED, permission_header_diagnostics=None, token_present=token_present, token_hash_prefix=hash_prefix, reason=reason, ) self._audit_append(result, task_id=task_id) return result # 4) decision_path 필수 (owner_trigger_only.invoke_from_scheduler 호출용) if decision_path is None: raise RouterContractError( "decision_path required when freshness is STALE/NO_REVIEW — " "owner_trigger_only.invoke_from_scheduler 호출 시 필수" ) # 5) nudge 발사 (owner_trigger_only.invoke_from_scheduler — token 직접 미노출) try: invoke_status = self._invoke_scheduler( decision_path=decision_path, owner=owner, repo=repo, current_head_actual=head_norm, ) except Exception as exc: # noqa: BLE001 — caller injection; fail-closed 분류 # invoke 호출 자체가 예외 → 403 diagnostics 회수 시도 permission_headers = self._safe_permission_diagnostics() if permission_headers: reason = ( f"invoke_scheduler raised + 403 diagnostics 회수 → " f"NUDGE_PERMISSION_DENIED · exc={type(exc).__name__}" ) result = RouterResult( final_state=STATE_NUDGE_PERMISSION_DENIED, pr_number=pr_number, current_head_sha=head_norm, freshness_state=freshness.status, gemini_commit_id_observed=freshness.gemini_commit_id_observed, nudge_attempted=True, nudge_result=RESULT_FAILED, permission_header_diagnostics=permission_headers, token_present=token_present, token_hash_prefix=hash_prefix, reason=reason, ) else: reason = ( f"invoke_scheduler raised — CHAIR_UI_FALLBACK_REQUIRED " f"(spec §2.10 최후 수단) · exc={type(exc).__name__}" ) result = RouterResult( final_state=STATE_CHAIR_UI_FALLBACK_REQUIRED, pr_number=pr_number, current_head_sha=head_norm, freshness_state=freshness.status, gemini_commit_id_observed=freshness.gemini_commit_id_observed, nudge_attempted=True, nudge_result=RESULT_FAILED, permission_header_diagnostics=None, token_present=token_present, token_hash_prefix=hash_prefix, reason=reason, ) self._audit_append(result, task_id=task_id) return result # 6) invoke_status 분류 if invoke_status == RESULT_POSTED: return self._handle_posted( pr_number=pr_number, head_norm=head_norm, freshness=freshness, token_present=token_present, hash_prefix=hash_prefix, fresh_review_arrived_post_nudge=fresh_review_arrived_post_nudge, task_id=task_id, owner=owner, repo=repo, ) if invoke_status == RESULT_DEDUPED: reason = ( "invoke_scheduler returned DEDUPED — 이전 trigger 가 active " "(POSTED|PENDING) — 회장 verbatim §9 hard limit 보호" ) result = RouterResult( final_state=STATE_NUDGE_DEDUPED, pr_number=pr_number, current_head_sha=head_norm, freshness_state=freshness.status, gemini_commit_id_observed=freshness.gemini_commit_id_observed, nudge_attempted=True, nudge_result=RESULT_DEDUPED, permission_header_diagnostics=None, token_present=token_present, token_hash_prefix=hash_prefix, reason=reason, ) self._audit_append(result, task_id=task_id) return result # FAILED / PENDING / 기타 → 403 diagnostics 회수 permission_headers = self._safe_permission_diagnostics() if permission_headers: reason = ( f"invoke_scheduler returned {invoke_status} + 403 diagnostics " "회수 → NUDGE_PERMISSION_DENIED (회장 verbatim §8 header 기록)" ) final_state = STATE_NUDGE_PERMISSION_DENIED else: reason = ( f"invoke_scheduler returned {invoke_status} (transient/unknown) — " "CHAIR_UI_FALLBACK_REQUIRED (spec §2.10 최후 수단)" ) final_state = STATE_CHAIR_UI_FALLBACK_REQUIRED result = RouterResult( final_state=final_state, pr_number=pr_number, current_head_sha=head_norm, freshness_state=freshness.status, gemini_commit_id_observed=freshness.gemini_commit_id_observed, nudge_attempted=True, nudge_result=invoke_status, permission_header_diagnostics=permission_headers, token_present=token_present, token_hash_prefix=hash_prefix, reason=reason, ) self._audit_append(result, task_id=task_id) return result # ─── internal helpers ───────────────────────────────────────────────── @staticmethod def _validate_input( pr_number: Any, current_head_sha: Any, owner: Any, repo: Any ) -> None: if ( not isinstance(pr_number, int) or isinstance(pr_number, bool) or pr_number <= 0 ): raise RouterContractError("pr_number must be a positive int") if ( not isinstance(current_head_sha, str) or len(current_head_sha) != 40 or any(c not in "0123456789abcdefABCDEF" for c in current_head_sha) ): raise RouterContractError( "current_head_sha must be 40-char hex SHA" ) if not isinstance(owner, str) or not owner or "/" in owner: raise RouterContractError( "owner must be a non-empty string without '/'" ) if not isinstance(repo, str) or not repo or "/" in repo: raise RouterContractError( "repo must be a non-empty string without '/'" ) def _compute_token_hash_prefix(self) -> tuple[bool, str]: """token_provider 가 주입된 경우 raw token → hash prefix 12 hex 계산. spec §3.3 / 회장 verbatim §6: raw token 노출 0. 본 router 는 hash prefix 만 보관. token_provider 호출 자체가 실패하면 token_present=False. """ if self._token_provider is None: return False, "" try: token = self._token_provider() except Exception: # noqa: BLE001 — token 부재도 fail-closed False return False, "" if not isinstance(token, str) or not token: return False, "" try: return True, router_token_hash_prefix(token, length=12) except Exception: # noqa: BLE001 return True, "" def _safe_permission_diagnostics(self) -> dict | None: """permission_diagnostics_provider 호출 + redaction.""" try: headers = self._permission_diagnostics_provider() except Exception: # noqa: BLE001 return None if not headers: return None extracted = extract_403_headers(headers) if not extracted: return None return extracted def _handle_posted( self, *, pr_number: int, head_norm: str, freshness: FreshnessResult, token_present: bool, hash_prefix: str, fresh_review_arrived_post_nudge: bool, task_id: str, owner: str, repo: str, ) -> RouterResult: """POSTED 분류 후 fresh review 도착 여부 확인 (spec §3.2 POSTED 경로).""" if fresh_review_arrived_post_nudge: # caller / scheduler 가 polling 끝에 fresh review 도착 확인. # router 는 1회 재freshness check 로 confirm. try: refresh = self._freshness_checker( pr_number=pr_number, current_head_sha=head_norm, github_api=self._github_api, owner=owner, repo=repo, ) except FreshnessCheckerError as exc: raise RouterContractError( f"post-nudge freshness check failed: {exc!r}" ) from exc if refresh.status == RESULT_FRESH: reason = ( "POSTED + fresh review 도착 + re-check FRESH 일치 — " "router 통과 (spec §3.2 POSTED-FRESH)" ) result = RouterResult( final_state=STATE_FRESH, pr_number=pr_number, current_head_sha=head_norm, freshness_state=refresh.status, gemini_commit_id_observed=refresh.gemini_commit_id_observed, nudge_attempted=True, nudge_result=RESULT_POSTED, permission_header_diagnostics=None, token_present=token_present, token_hash_prefix=hash_prefix, reason=reason, ) self._audit_append(result, task_id=task_id) return result # fresh review 도착 신호가 거짓 (caller bug or race) → STALE 분류 reason = ( f"POSTED + fresh_review_arrived 신호 True 였으나 re-check " f"{refresh.status} — GEMINI_EXTERNAL_TRIGGER_STALE" ) result = RouterResult( final_state=STATE_GEMINI_EXTERNAL_TRIGGER_STALE, pr_number=pr_number, current_head_sha=head_norm, freshness_state=refresh.status, gemini_commit_id_observed=refresh.gemini_commit_id_observed, nudge_attempted=True, nudge_result=RESULT_POSTED, permission_header_diagnostics=None, token_present=token_present, token_hash_prefix=hash_prefix, reason=reason, ) self._audit_append(result, task_id=task_id) return result # POSTED 만 했고 fresh review 도착 신호 없음 → caller 가 timeout 처리. # 일단 NUDGE_POSTED 로 표시 (caller 가 후속 polling 책임). reason = ( "OWNER nudge POSTED — fresh review 도착 polling 후 추가 호출 필요 " "(spec §3.2 POSTED 분기, default timeout=" f"{DEFAULT_FRESH_REVIEW_TIMEOUT_S}s)" ) result = RouterResult( final_state=STATE_NUDGE_POSTED, pr_number=pr_number, current_head_sha=head_norm, freshness_state=freshness.status, gemini_commit_id_observed=freshness.gemini_commit_id_observed, nudge_attempted=True, nudge_result=RESULT_POSTED, permission_header_diagnostics=None, token_present=token_present, token_hash_prefix=hash_prefix, reason=reason, ) self._audit_append(result, task_id=task_id) return result def _audit_append(self, result: RouterResult, *, task_id: str) -> None: """RouterResult → audit JSONL record append. redaction guard 적용.""" diag = result.permission_header_diagnostics # diag 가 dict 이면 한 번 더 redact (defense in depth) if isinstance(diag, dict): redacted = redact_diagnostics(diag) diag = redacted if isinstance(redacted, dict) else None record = { "schema": AUDIT_SCHEMA, "task_id": task_id or "", "pr_number": result.pr_number, "current_head_sha": result.current_head_sha, "freshness_state": result.freshness_state, "gemini_commit_id_observed": result.gemini_commit_id_observed, "nudge_attempted": result.nudge_attempted, "nudge_result": result.nudge_result, "permission_header_diagnostics": diag, "token_present": result.token_present, "token_hash_prefix": result.token_hash_prefix, "token_value_logged": False, "final_state": result.final_state, "reason": result.reason, } try: self._audit.append(record) except RouterAuditRedactionError: # audit redaction 실패 = fail-closed. router 호출자에게 propagate. raise __all__ = [ "ROUTER_RESULT_FRESH", "ROUTER_RESULT_NUDGE_POSTED", "ROUTER_RESULT_NUDGE_DEDUPED", "ROUTER_RESULT_GEMINI_EXTERNAL_TRIGGER_STALE", "ROUTER_RESULT_CHAIR_UI_FALLBACK_REQUIRED", "ROUTER_RESULT_NUDGE_PERMISSION_DENIED", "ROUTER_RESULT_NUDGE_FAILED", "ROUTER_RESULT_NOT_GEMINI_TRIGGER", "NUDGE_HARD_LIMIT_PER_PR_HEAD", "DEFAULT_FRESH_REVIEW_TIMEOUT_S", "RouterContractError", "RouterResult", "OwnerGeminiTriggerRouter", ]