"""utils/owner_gemini_trigger.py — task-2729 Phase 2 결함6 전용 entrypoint. OWNER /gemini review 자동 발사 전용 모듈. 정책 (회장 2026-06-03 verbatim): - request-only /gemini review 자동 발사만 허용. - thread resolve / 판단대행 / merge approval / merge 실행은 절대 하지 않음. - head-lock 검증 + dedupe(중복 발사 차단) 필수. - raw token 노출 금지 — sha256 앞 12자(token_hash_prefix)만 기록. - ACTIVE=false 골격: production 전환 없이 구조·결선만 제공. - 실제 gh 네트워크 호출 0: router 주입 시만 router.route_for_pr() 통해 발사, router=None이면 REQUEST_PREPARED 반환(network 0). dedupe 전략: - already_fired_heads(set, 주입 가능)에 current_head_sha가 있으면 DEDUPED 반환. - NUDGE_HARD_LIMIT_PER_PR_HEAD 기반 nudge limit 정책 적용(router 위임). - dedupe key = (pr_number, current_head_sha) 조합으로 중복 발사 차단. """ from __future__ import annotations import hashlib import json import logging from pathlib import Path from typing import Any, Callable, Optional, Set logger = logging.getLogger(__name__) # ── 상수 ──────────────────────────────────────────────────────────────────── OWNER_GEMINI_REVIEW_BODY: str = "/gemini review" # anu_v2.owner_gemini_trigger_router import (없으면 fallback 상수) try: from anu_v2.owner_gemini_trigger_router import ( OwnerGeminiTriggerRouter, NUDGE_HARD_LIMIT_PER_PR_HEAD, ) except ImportError: # pragma: no cover — anu_v2 미설치 환경 fallback OwnerGeminiTriggerRouter = None # type: ignore[assignment,misc] NUDGE_HARD_LIMIT_PER_PR_HEAD: int = 1 # type: ignore[no-redef] # ── head-lock 검증 ──────────────────────────────────────────────────────────── def _validate_head_lock(head_sha: str) -> None: """head_sha가 40자 16진수 문자열인지 검증. 형식 불일치 시 ValueError 발생 — caller가 잘못된 SHA로 dedupe/발사하는 것을 원천 차단. """ if ( not isinstance(head_sha, str) or len(head_sha) != 40 or not all(c in "0123456789abcdefABCDEF" for c in head_sha) ): raise ValueError( f"head_sha must be a 40-char hex string, got: {head_sha!r}" ) # ── audit jsonl append ──────────────────────────────────────────────────────── def _audit_append(audit_path: Optional[Any], record: dict) -> None: """audit_path에 record를 UTF-8 JSONL 한 줄로 append. audit_path가 None이면 skip(audit 비활성). """ if audit_path is None: return try: p = Path(audit_path) p.parent.mkdir(parents=True, exist_ok=True) with p.open("a", encoding="utf-8") as fh: fh.write(json.dumps(record, ensure_ascii=False) + "\n") except Exception as exc: # noqa: BLE001 — audit 실패가 발사 자체를 막으면 안 됨 logger.warning("owner_gemini_trigger audit append failed: %s", exc) # ── 메인 entrypoint ─────────────────────────────────────────────────────────── def fire_owner_gemini_review( *, pr_number: int, current_head_sha: str, owner: str, repo: str, task_id: str = "", observed_comment: Optional[str] = None, audit_path: Optional[Any] = None, router: Optional[Any] = None, request_only: bool = True, token_provider: Optional[Callable[[], str]] = None, already_fired_heads: Optional[Set[str]] = None, ) -> dict: """OWNER /gemini review 자동 발사 전용 entrypoint. 정책: - request_only=True 고정 (request-only 자동발사 정책). - thread resolve / 판단대행 / merge approval / merge 실행 절대 하지 않음. - head-lock 검증 → dedupe 체크 → router 발사(또는 골격 반환). - ACTIVE=false: router=None이면 실 gh 네트워크 호출 0. dedupe: - already_fired_heads(set) 안에 current_head_sha가 있으면 DEDUPED 반환. - 중복 발사 차단은 dedupe set 주입으로 외부에서도 제어 가능. Args: pr_number: PR 번호. current_head_sha: 현재 PR head SHA (40자 hex, head-lock 검증). owner: 저장소 owner. repo: 저장소 이름. task_id: 태스크 ID (audit 기록용). observed_comment: 관측된 최근 코멘트 본문 (router에 전달). audit_path: audit JSONL 경로 (None이면 skip). router: OwnerGeminiTriggerRouter 인스턴스 (None이면 ACTIVE=false 골격). request_only: True 고정 (변경 금지). token_provider: OWNER PAT 토큰 provider callable (None이면 token_hash_prefix=None). already_fired_heads: 이미 발사한 head SHA set (dedupe용, 주입 가능). Returns: dict: fired, status, body, pr_number, head_sha, dedupe, request_only, token_hash_prefix, active 포함. """ # (1) head-lock 검증 _validate_head_lock(current_head_sha) # (2) token hash prefix (raw 노출 금지 — sha256 앞 12자만) token_hash_prefix: Optional[str] = None if token_provider is not None: try: token_raw = token_provider() if token_raw: token_hash_prefix = hashlib.sha256( token_raw.encode("utf-8") ).hexdigest()[:12] except Exception as exc: # noqa: BLE001 logger.warning("owner_gemini_trigger token_provider failed: %s", exc) # (3) dedupe 체크 — head-lock + nudge limit 기반 중복 발사 차단 if already_fired_heads is not None and current_head_sha in already_fired_heads: result: dict = { "fired": False, "status": "DEDUPED", "body": OWNER_GEMINI_REVIEW_BODY, "pr_number": pr_number, "head_sha": current_head_sha, "dedupe": True, "request_only": True, "token_hash_prefix": token_hash_prefix, "active": False, } logger.info( "owner_gemini_trigger DEDUPED: pr=%s head=%s (already_fired_heads dedupe)", pr_number, current_head_sha, ) _audit_append(audit_path, {"event": "DEDUPED", **result, "task_id": task_id}) return result # (4) router 발사 또는 ACTIVE=false 골격 if router is not None: # router 주입 시 route_for_pr 호출 (실 gh 미호출 여부는 router 구현 책임) try: router_result = router.route_for_pr( pr_number=pr_number, current_head_sha=current_head_sha, owner=owner, repo=repo, task_id=task_id, observed_comment=observed_comment, nudge_hard_limit=NUDGE_HARD_LIMIT_PER_PR_HEAD, ) fired = getattr(router_result, "fired", False) status = str(getattr(router_result, "final_state", "ROUTER_RESULT")) except Exception as exc: # noqa: BLE001 logger.warning("owner_gemini_trigger router.route_for_pr failed: %s", exc) fired = False status = f"ROUTER_ERROR: {type(exc).__name__}" else: # ACTIVE=false 골격: 실 gh 네트워크 호출 0 fired = False status = "REQUEST_PREPARED" result = { "fired": fired, "status": status, "body": OWNER_GEMINI_REVIEW_BODY, "pr_number": pr_number, "head_sha": current_head_sha, "dedupe": False, "request_only": True, "token_hash_prefix": token_hash_prefix, "active": False, } # (5) audit 기록 (발사 시) _audit_append( audit_path, {"event": "FIRE_ATTEMPT", **result, "task_id": task_id}, ) logger.info( "owner_gemini_trigger: pr=%s head=%s fired=%s status=%s active=False", pr_number, current_head_sha, fired, status, ) return result