"""anu_v2.gemini_stale_prevention_runner — ANU v2 Gemini stale prevention runner v0 (task-2545). 회장 §명시 (2026-05-10): - Gemini evidence 가 도착한 PR 에 대해, 코드 변경이 필요한 Gemini real bug 를 같은 PR 에 push 하지 않는다. - same-PR push 는 Gemini evidence 를 stale 로 만들고 human trigger 를 강제하는 구조이므로 기본 금지. - expected_files 내부 real bug 수정이 필요한 경우 replacement_pr_runner 를 호출해 clean replacement PR 을 생성한다. - replacement PR 은 pull_request.opened 이벤트로 Gemini 자동 리뷰를 새로 받게 한다. - original PR 은 보존. close / reopen / force / rebase / empty commit 사용 금지. 핵심 5원칙 (회장 §명시): 1) Gemini evidence 도착 PR 에 code-changing fix same-PR push 금지 2) same-PR push 는 stale 강제 → human trigger 유발 구조이므로 기본 차단 3) expected_files 내부 real bug 는 replacement_pr_runner 호출 → clean replacement PR 4) replacement PR opened 이벤트 → Gemini 새 리뷰 자동 트리거 5) original PR 보존 (close / reopen / force / rebase / empty commit 절대 금지) 설계 원칙: - one-way isolation: anu_v2/* 만 import. utils / dispatch / scripts / dashboard 의존성 0. 표준 라이브러리만 보조 사용. - 외부 부수효과 (replacement PR runner / pr open health gate / audit writer / 시간) 는 모두 callable 주입 (DI). 미주입 시 메서드 호출 시점에 명시적 ValueError. - 단, audit_writer 는 jsonl 파일 append fallback 허용. - md / report fallback 분기 금지 (runner decision + check_run + PR diff 기준만). - bot `/gemini review` 코멘트 작성 금지 (코드 문자열 등장 0). - subprocess / git / gh 직접 호출 절대 금지. 모두 callable 주입 경로. - chat_id != 6937032012 audit record 금지 (default chat 이외 record 차단). """ from __future__ import annotations import json from dataclasses import dataclass from datetime import datetime, timezone from pathlib import Path from typing import Any, Callable, Mapping, Sequence # ─── Decision codes (evaluate_push_safety) ─────────────────────────────────── SAME_PR_SAFE = "SAME_PR_SAFE" SAME_PR_BLOCKED_REPLACEMENT_REQUIRED = "SAME_PR_BLOCKED_REPLACEMENT_REQUIRED" SAME_PR_BLOCKED_SCOPE_EXPANSION = "SAME_PR_BLOCKED_SCOPE_EXPANSION" DECISIONS: frozenset[str] = frozenset({ SAME_PR_SAFE, SAME_PR_BLOCKED_REPLACEMENT_REQUIRED, SAME_PR_BLOCKED_SCOPE_EXPANSION, }) # ─── Outcome codes (run entry point) ───────────────────────────────────────── OUTCOME_SAME_PR_RESOLVED = "SAME_PR_RESOLVED" OUTCOME_REPLACEMENT_PR_OPENED = "REPLACEMENT_PR_OPENED" OUTCOME_SCOPE_EXPANSION_REPORTED = "SCOPE_EXPANSION_REPORTED" OUTCOME_EMPTY_COMMIT_BLOCKED = "EMPTY_COMMIT_BLOCKED" # ─── Classification 분류 코드 (auto_gemini_triage 와 1:1) ──────────────────── CLASSIFICATION_FALSE_POSITIVE = "false_positive" CLASSIFICATION_STYLE_ONLY = "style_only" CLASSIFICATION_NO_CODE_CHANGE = "no_code_change" CLASSIFICATION_MINOR_FIX_IN_SCOPE = "minor_fix_in_scope" CLASSIFICATION_REAL_BUG_IN_SCOPE = "real_bug_in_scope" CLASSIFICATION_SCOPE_EXPANSION = "scope_expansion" NON_CODE_CHANGING_CLASSIFICATIONS: frozenset[str] = frozenset({ CLASSIFICATION_FALSE_POSITIVE, CLASSIFICATION_STYLE_ONLY, CLASSIFICATION_NO_CODE_CHANGE, }) CODE_CHANGING_CLASSIFICATIONS: frozenset[str] = frozenset({ CLASSIFICATION_MINOR_FIX_IN_SCOPE, CLASSIFICATION_REAL_BUG_IN_SCOPE, }) # ─── Critical 7종 #3 ───────────────────────────────────────────────────────── CRITICAL_SEVEN_KIND_SCOPE_EXPANSION = 3 CRITICAL_SEVEN_KIND_SCOPE_EXPANSION_NAME = "EXPECTED_FILES_SCOPE_EXPANSION" # ─── Critical 7종 #6 — REPLACEMENT_PR_CONTRACT_FRAMING_INCONSISTENT_WITH_ORIGIN_MAIN_STATE ── # 회장 §명시 (2026-05-11, task-2545+2 corrected replacement 회귀 박제): # "original PR이 OPEN이면 replacement PR expected_files는 original PR이 main에 # 반영되었다고 가정해 축소하면 안 된다. replacement expected_files는 반드시 # origin/main 실제 상태 기준 전체 effective diff로 산정한다." # # 사고 경위 (PR #93 task-2545+1): # - PR #92 (original task-2545) state=OPEN, head=15cf6ad011e1, mergedAt=null # - origin/main 에는 task-2545 의 runner / test / fixtures 0 file 반영 상태 # - 그러나 task-2545+1 replacement contract 가 expected_files=1 (runner.py only) 로 # 축소 산정되어 effective diff (7 files) 와 불일치 → 회귀 미박제 위험 CRITICAL_SEVEN_KIND_REPLACEMENT_CONTRACT_FRAMING = 6 CRITICAL_SEVEN_KIND_REPLACEMENT_CONTRACT_FRAMING_NAME = ( "REPLACEMENT_PR_CONTRACT_FRAMING_INCONSISTENT_WITH_ORIGIN_MAIN_STATE" ) # original PR 이 main 에 반영되지 않았다고 간주되는 상태 집합 (open/escalated/closed-no-merge). ORIGINAL_PR_UNMERGED_STATES: frozenset[str] = frozenset({ "OPEN", "ESCALATED", "CLOSED", # closed-without-merge (mergedAt=null) }) # ─── Empty commit block 라벨 ───────────────────────────────────────────────── EMPTY_COMMIT_BLOCKED_KIND = "EMPTY_COMMIT_TRIGGER_ATTEMPT_BLOCKED" EMPTY_COMMIT_BLOCKED_REASON = ( "PR #76 fixture proof: empty commit does not trigger Gemini App webhook" ) # ─── chat_id 기본값 (회장 §명시) ────────────────────────────────────────────── DEFAULT_CHAT_ID = 6937032012 # ─── Callable 시그니처 (DI 계약) ───────────────────────────────────────────── ReplacementPrRunnerCallable = Callable[..., Mapping[str, Any]] PrOpenHealthGateCallable = Callable[..., Mapping[str, Any]] AuditWriter = Callable[[Mapping[str, Any]], None] NowFactory = Callable[[], str] def _default_now_iso() -> str: """UTC ISO-8601 (seconds 정밀도) — now_factory 미주입 시 fallback.""" return datetime.now(timezone.utc).isoformat(timespec="seconds") @dataclass(frozen=True) class _EvaluateBuckets: """evaluate_push_safety 내부 사용 분류 버킷. code_changing / non_code_changing / scope_expansion 세 버킷으로 thread_id 분리. """ code_changing: tuple[int, ...] non_code_changing: tuple[int, ...] scope_expansion: tuple[int, ...] class GeminiStalePreventionRunner: """ANU v2 Gemini evidence stale 사전 예방 + same-PR push 차단 + replacement PR 자동 전환 v0. 회장 §명시 (2026-05-10): - Gemini evidence 도착한 PR 에 코드 변경 same-PR push 기본 금지 - false_positive / style_only / no_code_change → same PR resolve 허용 - minor_fix_in_scope / real_bug_in_scope → replacement_pr_runner 호출 → clean replacement PR - scope_expansion → Critical 7종 #3 보고 - empty commit / force / rebase / close-reopen 절대 금지 - md/report fallback 금지 (runner decision + marker + check_run + PR diff 기준만) one-way isolation: anu_v2 외부 import 금지. anu_v2 내부에서도 task-2537 / 2538 모듈은 interface 호출만 (직접 코드 수정 X — 본 클래스에서는 callable 주입을 통해서만 접근). """ # task-2544 정책과 동일: PR opened 이후 Gemini evidence 도착 grace window. GRACE_SECONDS_PR_OPEN_HEALTH: int = 180 def __init__( self, *, replacement_pr_runner_callable: ReplacementPrRunnerCallable | None = None, pr_open_health_gate_callable: PrOpenHealthGateCallable | None = None, audit_writer: AuditWriter | None = None, audit_root: Path = Path("memory/orchestration-audit"), chat_id: int = DEFAULT_CHAT_ID, now_factory: NowFactory | None = None, ) -> None: """생성자 — 모든 외부 부수효과는 callable 주입. Args: replacement_pr_runner_callable: task-2537 replacement_pr_runner 진입점 callable. pivot_to_replacement_pr 호출 시 필수. 미주입 시 명시적 ValueError. pr_open_health_gate_callable: task-2544 PROpenGeminiTriggerPrevention 진입점 callable. run_pr_open_health_gate 호출 시 필수. 미주입 시 명시적 ValueError. audit_writer: audit record append callable. 미주입 시 jsonl 파일 append fallback. audit_root: audit jsonl 저장 root (기본 memory/orchestration-audit). chat_id: 회장 §명시 default chat (다른 chat record 노출 차단의 기준값). now_factory: 시간 주입 (테스트 결정성 확보용). 미주입 시 UTC now ISO-8601. 회장 §명시 silent fallback 금지: replacement / pr_open_health_gate callable 미주입은 fallback 처리하지 않고 호출 시점에 ValueError 를 던진다 (audit_writer 만 fallback). """ self._replacement_pr_runner_callable = replacement_pr_runner_callable self._pr_open_health_gate_callable = pr_open_health_gate_callable self._audit_root = Path(audit_root) self._chat_id = int(chat_id) self._now_factory: NowFactory = now_factory or _default_now_iso if audit_writer is not None: self._audit_writer: AuditWriter = audit_writer else: self._audit_writer = self._default_jsonl_audit_writer # 회장 §명시: chat_id != 6937032012 audit record 생성 금지. if self._chat_id != DEFAULT_CHAT_ID: raise ValueError( "chat_id must be 6937032012 (회장 §명시 default chat). " f"got={self._chat_id}" ) # ─── 내부 helper ──────────────────────────────────────────────────────── def _now(self) -> str: """주입된 now_factory 호출 (또는 default UTC ISO).""" return self._now_factory() def _default_jsonl_audit_writer(self, record: Mapping[str, Any]) -> None: """audit_writer 미주입 시 fallback — jsonl 파일에 append.""" kind = record.get("audit_kind") or record.get("kind") or "audit" # kind 에서 안전 파일명 생성 (영문/숫자/_ 만 허용). safe_kind = "".join(c if (c.isalnum() or c == "_") else "_" for c in str(kind)) path = self._audit_root / f"{safe_kind}.jsonl" try: self._audit_root.mkdir(parents=True, exist_ok=True) with path.open("a", encoding="utf-8") as fp: fp.write(json.dumps(dict(record), ensure_ascii=False, sort_keys=True) + "\n") except OSError: # audit 실패는 본 runner 의 decision 자체를 막으면 안 된다 (silent log). # 단, raw token 노출이 발생하지 않도록 record 자체는 호출자에서 sanitize 책임. pass @staticmethod def _thread_id_of(thread: Mapping[str, Any]) -> int: """thread dict 에서 thread_id 추출 (없으면 -1).""" tid = thread.get("thread_id") try: return int(tid) if tid is not None else -1 except (TypeError, ValueError): return -1 @staticmethod def _classification_of(thread: Mapping[str, Any]) -> str: """thread dict 의 classification 문자열 추출 (lowercase normalize).""" cls = thread.get("classification") if not isinstance(cls, str): return "" return cls.strip().lower() @staticmethod def _is_code_change_required( thread: Mapping[str, Any], classification: str | None = None, ) -> bool: """thread dict 에서 code_change_required 추출. 명시되지 않으면 classification 기반. classification 인자가 주어지면 _classification_of 재호출 없이 그 값을 사용한다 (호출부에서 이미 cls 를 계산한 경우 중복 연산 회피). """ explicit = thread.get("code_change_required") if isinstance(explicit, bool): return explicit cls = ( classification if classification is not None else GeminiStalePreventionRunner._classification_of(thread) ) if cls in NON_CODE_CHANGING_CLASSIFICATIONS: return False if cls in CODE_CHANGING_CLASSIFICATIONS: return True if cls == CLASSIFICATION_SCOPE_EXPANSION: # scope_expansion 도 코드 변경을 수반하지만 별도 버킷에서 처리. return True return False def _bucketize( self, triage_classifications: Sequence[Mapping[str, Any]] ) -> _EvaluateBuckets: """triage_classifications 를 3 버킷으로 분류. scope_expansion 판단은 classification 문자열 OR outside_expected_files=True 둘 중 하나라도 만족하면 scope_expansion 버킷으로 분리한다 (Critical 7종 #3 보고 누락 방지). """ code_changing: list[int] = [] non_code_changing: list[int] = [] scope_expansion: list[int] = [] for thread in triage_classifications: tid = self._thread_id_of(thread) cls = self._classification_of(thread) is_outside = bool(thread.get("outside_expected_files", False)) if cls == CLASSIFICATION_SCOPE_EXPANSION or is_outside: scope_expansion.append(tid) continue if cls in NON_CODE_CHANGING_CLASSIFICATIONS: non_code_changing.append(tid) continue if cls in CODE_CHANGING_CLASSIFICATIONS: code_changing.append(tid) continue # 명시 분류 외: code_change_required 플래그 기반 fallback. # P9O: cls 재계산 회피 — 위에서 이미 _classification_of 호출함. if self._is_code_change_required(thread, classification=cls): code_changing.append(tid) else: non_code_changing.append(tid) return _EvaluateBuckets( code_changing=tuple(code_changing), non_code_changing=tuple(non_code_changing), scope_expansion=tuple(scope_expansion), ) # ─── 1. evaluate_push_safety ──────────────────────────────────────────── def evaluate_push_safety( self, pr_number: int, current_head_sha: str, gemini_review_commit_id: str | None, triage_classifications: Sequence[Mapping[str, Any]], ) -> dict[str, Any]: """Gemini review 후 새 commit push 가 필요한지 + same-PR 안전 여부 판단. 분기 로직 (회장 §명시): - 모든 thread 가 비-코드변경 (false_positive / style_only / no_code_change) → SAME_PR_SAFE (same PR resolve 허용) - 코드변경 thread 1개 이상 + scope_expansion 0건 → SAME_PR_BLOCKED_REPLACEMENT_REQUIRED (replacement_pr_runner 호출 필요) - scope_expansion 1건 이상 → SAME_PR_BLOCKED_SCOPE_EXPANSION (Critical 7종 #3 보고 필요) gemini_review_commit_id != current_head_sha 일 때는 reason 에 "evidence_already_stale" 을 포함시키되 분류 자체는 동일 로직 적용. Args: pr_number: PR 번호. current_head_sha: 현재 PR HEAD SHA. gemini_review_commit_id: Gemini review 가 평가한 commit SHA (없으면 None). triage_classifications: auto_gemini_triage.classify_thread() 결과 list. Returns: dict: decision / reason / code_changing_threads / non_code_changing_threads / auto_retry_allowed / next_action / pr_number / current_head_sha / gemini_review_commit_id / evidence_stale. """ buckets = self._bucketize(triage_classifications) # evidence stale 여부 판단 (gemini_review_commit_id 가 None 이거나 head 와 mismatch). evidence_stale = False if gemini_review_commit_id is None: evidence_stale = True elif str(gemini_review_commit_id) != str(current_head_sha): evidence_stale = True # decision 결정 (scope_expansion 우선). if buckets.scope_expansion: decision = SAME_PR_BLOCKED_SCOPE_EXPANSION reason_parts = [ "scope_expansion threads detected: " f"{list(buckets.scope_expansion)} — Critical 7종 #3 보고 필요", ] next_action = ( "classify_scope_expansion_as_critical_three 호출 후 회장 보고 " "(replacement PR 도 회장 승인 후 진행)" ) auto_retry_allowed = False elif buckets.code_changing: decision = SAME_PR_BLOCKED_REPLACEMENT_REQUIRED reason_parts = [ "code-changing threads detected: " f"{list(buckets.code_changing)} — same-PR push 금지, " "replacement_pr_runner 호출 필요", ] next_action = ( "pivot_to_replacement_pr 호출 → clean replacement PR 생성 → " "pull_request.opened 이벤트로 Gemini 새 리뷰 자동 트리거" ) auto_retry_allowed = False else: decision = SAME_PR_SAFE reason_parts = [ "all threads non-code-changing " "(false_positive / style_only / no_code_change) — " "same PR resolve 허용", ] next_action = "comments / threads resolve only — no new commit required" auto_retry_allowed = True if evidence_stale: reason_parts.append("evidence_already_stale") return { "decision": decision, "reason": " | ".join(reason_parts), "code_changing_threads": list(buckets.code_changing), "non_code_changing_threads": list(buckets.non_code_changing), "scope_expansion_threads": list(buckets.scope_expansion), "auto_retry_allowed": auto_retry_allowed, "next_action": next_action, "pr_number": int(pr_number), "current_head_sha": str(current_head_sha), "gemini_review_commit_id": ( None if gemini_review_commit_id is None else str(gemini_review_commit_id) ), "evidence_stale": bool(evidence_stale), } # ─── 2. pivot_to_replacement_pr ───────────────────────────────────────── def pivot_to_replacement_pr( self, original_pr: int, original_branch: str, original_head_sha: str, proposed_fixes: Sequence[Mapping[str, Any]], chat_id: int = DEFAULT_CHAT_ID, ) -> dict[str, Any]: """replacement_pr_runner 호출하여 clean replacement PR 생성. 절대 금지 (회장 §명시): - original PR close / reopen - force / rebase / empty commit on original - bot `/gemini review` 코멘트 작성 original PR 보존 + replacement PR opened 이벤트로 Gemini 새 리뷰 트리거 기대. 본 메서드 자체는 git / gh 직접 호출 0 — 모두 주입된 callable 경유. Args: original_pr: 보존할 original PR 번호. original_branch: original PR branch 이름. original_head_sha: original PR HEAD SHA (audit 박제용). proposed_fixes: expected_files 내부 fix 명세 list. chat_id: 회장 §명시 default chat (6937032012). Returns: dict: replacement_pr_number / replacement_branch / replacement_head_sha / original_pr_preserved=True / pr_open_health_gate / gemini_first_evidence / trigger_missed. Raises: ValueError: replacement_pr_runner_callable 미주입 시 / chat_id mismatch 시. """ if int(chat_id) != DEFAULT_CHAT_ID: raise ValueError( "chat_id must be 6937032012 (회장 §명시 default chat). " f"got={chat_id}" ) if self._replacement_pr_runner_callable is None: raise ValueError( "replacement_pr_runner_callable required for pivot_to_replacement_pr " "(회장 §명시 silent fallback 금지)" ) # task-2537 replacement_pr_runner 호출 (callable 경유, 직접 import 금지). runner_result = self._replacement_pr_runner_callable( original_pr=int(original_pr), original_branch=str(original_branch), original_head_sha=str(original_head_sha), proposed_fixes=[dict(f) for f in proposed_fixes], chat_id=int(chat_id), ) if not isinstance(runner_result, Mapping): raise ValueError( "replacement_pr_runner_callable must return a Mapping. " f"got type={type(runner_result).__name__}" ) # callable 반환 dict 의 키는 자유 → 본 runner 가 normalize. replacement_pr_number = ( runner_result.get("replacement_pr_number") or runner_result.get("replacement_pr") or runner_result.get("pr_number") ) replacement_branch = ( runner_result.get("replacement_branch") or runner_result.get("clean_branch") or runner_result.get("branch") ) replacement_head_sha = ( runner_result.get("replacement_head_sha") or runner_result.get("clean_sha") or runner_result.get("head_sha") ) if replacement_pr_number is None or replacement_branch is None or replacement_head_sha is None: raise ValueError( "replacement_pr_runner_callable result missing required keys " "(replacement_pr_number / replacement_branch / replacement_head_sha). " f"got_keys={sorted(runner_result.keys())}" ) # task-2544 PR_OPEN_HEALTH_GATE 호출 (3분 grace window). health_gate_result = self.run_pr_open_health_gate( replacement_pr=int(replacement_pr_number), replacement_head_sha=str(replacement_head_sha), ) # gemini_first_evidence: health gate 의 normalized 결과 일부 박제. gemini_first_evidence = { "evidence_arrived": bool(health_gate_result.get("evidence_arrived", False)), "elapsed_seconds": int(health_gate_result.get("elapsed_seconds", 0)), "classification": str(health_gate_result.get("classification", "")), } # trigger_missed: PR_OPEN_GEMINI_TRIGGER_MISSED 분류 시 True. classification_str = gemini_first_evidence["classification"].upper() trigger_missed = "TRIGGER_MISSED" in classification_str or ( not gemini_first_evidence["evidence_arrived"] ) ts = self._now() audit_record = { "audit_kind": "replacement_pr_pivot", "ts": ts, "chat_id": self._chat_id, "original_pr": int(original_pr), "original_branch": str(original_branch), "original_head_sha": str(original_head_sha), "original_pr_preserved": True, "replacement_pr_number": int(replacement_pr_number), "replacement_branch": str(replacement_branch), "replacement_head_sha": str(replacement_head_sha), "trigger_missed": bool(trigger_missed), "evidence_arrived": gemini_first_evidence["evidence_arrived"], "classification": gemini_first_evidence["classification"], } self._audit_writer(audit_record) return { "replacement_pr_number": int(replacement_pr_number), "replacement_branch": str(replacement_branch), "replacement_head_sha": str(replacement_head_sha), "original_pr_preserved": True, "original_pr": int(original_pr), "pr_open_health_gate": dict(health_gate_result), "gemini_first_evidence": dict(gemini_first_evidence), "trigger_missed": bool(trigger_missed), "ts": ts, } # ─── 3. run_pr_open_health_gate ───────────────────────────────────────── def run_pr_open_health_gate( self, replacement_pr: int, replacement_head_sha: str, grace_seconds: int = GRACE_SECONDS_PR_OPEN_HEALTH, ) -> dict[str, Any]: """task-2544 PROpenGeminiTriggerPrevention 호출. Gemini opened evidence 3분 watch. evidence 미도착 시 PR_OPEN_GEMINI_TRIGGER_MISSED 로 즉시 분류. long polling / self-register cron 절대 금지. Args: replacement_pr: replacement PR 번호. replacement_head_sha: replacement PR HEAD SHA. grace_seconds: grace window (기본 GRACE_SECONDS_PR_OPEN_HEALTH=180). Returns: dict: evidence_arrived / elapsed_seconds / classification / long_polling_invoked=False (강제 어설션). Raises: ValueError: pr_open_health_gate_callable 미주입 시 / callable 결과에 long_polling_invoked=True 포함 시. """ if self._pr_open_health_gate_callable is None: raise ValueError( "pr_open_health_gate_callable required for run_pr_open_health_gate " "(회장 §명시 silent fallback 금지)" ) gate_result = self._pr_open_health_gate_callable( replacement_pr=int(replacement_pr), replacement_head_sha=str(replacement_head_sha), grace_seconds=int(grace_seconds), ) if not isinstance(gate_result, Mapping): raise ValueError( "pr_open_health_gate_callable must return a Mapping. " f"got type={type(gate_result).__name__}" ) # 강제 어설션: long polling 사용 시 즉시 fail. long_polling = bool(gate_result.get("long_polling_invoked", False)) if long_polling: raise ValueError( "pr_open_health_gate_callable returned long_polling_invoked=True — " "회장 §명시: long polling / self-register cron 절대 금지 (3분 grace 만 허용)" ) evidence_arrived = bool(gate_result.get("evidence_arrived", False)) elapsed_seconds_raw = gate_result.get("elapsed_seconds", 0) try: elapsed_seconds = int(elapsed_seconds_raw) except (TypeError, ValueError): elapsed_seconds = 0 classification = str(gate_result.get("classification", "")) return { "evidence_arrived": evidence_arrived, "elapsed_seconds": elapsed_seconds, "classification": classification, "long_polling_invoked": False, "replacement_pr": int(replacement_pr), "replacement_head_sha": str(replacement_head_sha), "grace_seconds": int(grace_seconds), } # ─── 4. block_empty_commit_attempt ────────────────────────────────────── def block_empty_commit_attempt(self, pr_number: int, ts: str) -> dict[str, Any]: """empty commit 으로 Gemini webhook re-trigger 시도 차단. PR #76 사고 fixture 재현: empty commit 으로도 Gemini 도착 X 입증됨. 본 메서드 호출은 audit jsonl 박제만 수행 — 실제 empty commit 은 절대 수행 X. 본 클래스 어디에서도 git / gh / subprocess 직접 호출 없음. Args: pr_number: PR 번호. ts: 시도 시각 (ISO-8601 권장). Returns: dict: blocked=True / kind / reason / audit_jsonl_path / ts / pr_number. """ audit_jsonl_path = self._audit_root / "empty_commit_block.jsonl" record = { "audit_kind": "empty_commit_block", "ts": str(ts), "chat_id": self._chat_id, "pr_number": int(pr_number), "blocked": True, "kind": EMPTY_COMMIT_BLOCKED_KIND, "reason": EMPTY_COMMIT_BLOCKED_REASON, } self._audit_writer(record) return { "blocked": True, "kind": EMPTY_COMMIT_BLOCKED_KIND, "reason": EMPTY_COMMIT_BLOCKED_REASON, "audit_jsonl_path": str(audit_jsonl_path), "ts": str(ts), "pr_number": int(pr_number), } # ─── 5. classify_scope_expansion_as_critical_three ────────────────────── def classify_scope_expansion_as_critical_three( self, pr_number: int, proposed_fixes: Sequence[Mapping[str, Any]], expected_files_original: Sequence[str], ) -> dict[str, Any]: """expected_files 외 수정 필요 시 Critical 7종 #3 (scope expansion) 분류. proposed_fixes 의 각 항목에서 다음 중 하나라도 해당하면 outside 로 분류: - outside_expected_files=True 명시 - target_file 이 expected_files_original 에 없음 Args: pr_number: PR 번호. proposed_fixes: expected_files 외부 후보 포함 fix 명세 list. expected_files_original: 원래 task 의 expected_files list. Returns: dict: critical_seven_kind=3 / kind_name / pr_number / expected_files_original (정렬) / proposed_outside_files (정렬) / report_to_chairman_required=True. """ expected_set = {str(p) for p in expected_files_original} outside_files: set[str] = set() for fix in proposed_fixes: outside_flag = bool(fix.get("outside_expected_files", False)) target_file = fix.get("target_file") or fix.get("file") or fix.get("path") if outside_flag and target_file: outside_files.add(str(target_file)) continue if target_file and str(target_file) not in expected_set: outside_files.add(str(target_file)) sorted_outside = sorted(outside_files) sorted_expected = sorted({str(p) for p in expected_files_original}) ts = self._now() audit_record = { "audit_kind": "critical_seven_scope_expansion", "ts": ts, "chat_id": self._chat_id, "pr_number": int(pr_number), "critical_seven_kind": CRITICAL_SEVEN_KIND_SCOPE_EXPANSION, "kind_name": CRITICAL_SEVEN_KIND_SCOPE_EXPANSION_NAME, "expected_files_original": sorted_expected, "proposed_outside_files": sorted_outside, "report_to_chairman_required": True, } self._audit_writer(audit_record) return { "critical_seven_kind": CRITICAL_SEVEN_KIND_SCOPE_EXPANSION, "kind_name": CRITICAL_SEVEN_KIND_SCOPE_EXPANSION_NAME, "pr_number": int(pr_number), "expected_files_original": sorted_expected, "proposed_outside_files": sorted_outside, "report_to_chairman_required": True, "ts": ts, } # ─── 5b. classify_replacement_contract_framing (회귀 박제) ────────────── def classify_replacement_contract_framing( self, replacement_task_id: str, original_pr_number: int, original_pr_state: str, original_pr_merged_at: str | None, replacement_expected_files: Sequence[str], origin_main_effective_diff_files: Sequence[str], ) -> dict[str, Any]: """REPLACEMENT_PR_CONTRACT_FRAMING_INCONSISTENT_WITH_ORIGIN_MAIN_STATE 회귀 박제. 조건 (회장 §명시 2026-05-11, task-2545+2 corrected replacement): 1) original PR 이 OPEN / ESCALATED / CLOSED-without-merge 상태 (즉 origin/main 에 미반영) 2) replacement task expected_files 가 origin/main 실 상태 기준 effective diff 와 불일치 (특히 축소된 경우) 원칙: replacement expected_files 는 반드시 origin/main 실제 상태 기준 전체 effective diff 로 산정한다. original PR 머지 여부와 무관. Args: replacement_task_id: replacement task id (예: "task-2545+1"). original_pr_number: original PR 번호. original_pr_state: original PR 상태 ("OPEN" / "MERGED" / "CLOSED" / "ESCALATED"). original_pr_merged_at: original PR mergedAt (None 이면 미머지). replacement_expected_files: replacement task 가 선언한 expected_files. origin_main_effective_diff_files: origin/main 기준 실제 effective diff 파일 list. Returns: dict: contract_framing_inconsistent (bool) / critical_seven_kind=6 / kind_name / replacement_task_id / original_pr / original_pr_state / original_pr_unmerged (bool) / replacement_expected_files (정렬) / origin_main_effective_diff_files (정렬) / missing_from_replacement (정렬) — effective diff 에는 있는데 replacement expected_files 에는 없는 파일 list / report_to_chairman_required / ts. """ original_state_norm = str(original_pr_state).upper() original_pr_unmerged = ( original_state_norm in ORIGINAL_PR_UNMERGED_STATES and original_pr_merged_at in (None, "", "null") ) replacement_set = {str(p) for p in replacement_expected_files} effective_set = {str(p) for p in origin_main_effective_diff_files} missing_from_replacement = sorted(effective_set - replacement_set) # framing inconsistent 조건 (회장 §명시): # original_pr_unmerged AND replacement 가 effective diff 의 어느 파일도 누락 contract_framing_inconsistent = bool( original_pr_unmerged and missing_from_replacement ) ts = self._now() audit_record = { "audit_kind": "critical_seven_replacement_contract_framing", "ts": ts, "chat_id": self._chat_id, "replacement_task_id": str(replacement_task_id), "original_pr": int(original_pr_number), "original_pr_state": original_state_norm, "original_pr_merged_at": original_pr_merged_at, "original_pr_unmerged": original_pr_unmerged, "critical_seven_kind": CRITICAL_SEVEN_KIND_REPLACEMENT_CONTRACT_FRAMING, "kind_name": CRITICAL_SEVEN_KIND_REPLACEMENT_CONTRACT_FRAMING_NAME, "replacement_expected_files": sorted(replacement_set), "origin_main_effective_diff_files": sorted(effective_set), "missing_from_replacement": missing_from_replacement, "contract_framing_inconsistent": contract_framing_inconsistent, "report_to_chairman_required": contract_framing_inconsistent, } self._audit_writer(audit_record) return { "contract_framing_inconsistent": contract_framing_inconsistent, "critical_seven_kind": CRITICAL_SEVEN_KIND_REPLACEMENT_CONTRACT_FRAMING, "kind_name": CRITICAL_SEVEN_KIND_REPLACEMENT_CONTRACT_FRAMING_NAME, "replacement_task_id": str(replacement_task_id), "original_pr": int(original_pr_number), "original_pr_state": original_state_norm, "original_pr_unmerged": original_pr_unmerged, "replacement_expected_files": sorted(replacement_set), "origin_main_effective_diff_files": sorted(effective_set), "missing_from_replacement": missing_from_replacement, "report_to_chairman_required": contract_framing_inconsistent, "ts": ts, } # ─── 6. run (entry point) ─────────────────────────────────────────────── def run( self, pr_number: int, current_head_sha: str, gemini_review_commit_id: str | None, triage_classifications: Sequence[Mapping[str, Any]], original_branch: str, chat_id: int = DEFAULT_CHAT_ID, ) -> dict[str, Any]: """evaluate → pivot (필요 시) → pr_open_health_gate → 결정 통합 entry point. 흐름 (회장 §명시): 1) evaluate_push_safety 호출 2) decision == SAME_PR_SAFE → outcome=SAME_PR_RESOLVED (pivot/health 호출 X) 3) decision == SAME_PR_BLOCKED_SCOPE_EXPANSION → classify_scope_expansion_as_critical_three 호출 → outcome=SCOPE_EXPANSION_REPORTED 4) decision == SAME_PR_BLOCKED_REPLACEMENT_REQUIRED → triage_classifications 중 code_change_required=True 항목을 proposed_fixes 로 변환 → pivot_to_replacement_pr 호출 → outcome=REPLACEMENT_PR_OPENED (replacement_pr_runner_callable 미주입 시 ValueError) Args: pr_number: PR 번호. current_head_sha: 현재 PR HEAD SHA. gemini_review_commit_id: Gemini review 가 평가한 commit SHA (없으면 None). triage_classifications: auto_gemini_triage.classify_thread() 결과 list. original_branch: original PR branch (replacement 시 audit 박제용). chat_id: 회장 §명시 default chat (6937032012). Returns: dict: outcome / evaluate_result / pivot_result | None / health_gate_result | None / critical_seven_classification | None / pr_number / ts. """ if int(chat_id) != DEFAULT_CHAT_ID: raise ValueError( "chat_id must be 6937032012 (회장 §명시 default chat). " f"got={chat_id}" ) ts = self._now() evaluate_result = self.evaluate_push_safety( pr_number=int(pr_number), current_head_sha=str(current_head_sha), gemini_review_commit_id=gemini_review_commit_id, triage_classifications=triage_classifications, ) decision = evaluate_result["decision"] pivot_result: dict[str, Any] | None = None health_gate_result: dict[str, Any] | None = None critical_seven_classification: int | None = None if decision == SAME_PR_SAFE: outcome = OUTCOME_SAME_PR_RESOLVED elif decision == SAME_PR_BLOCKED_SCOPE_EXPANSION: # scope_expansion 분류 → Critical 7종 #3 보고. proposed_fixes 는 # triage_classifications 에서 outside_expected_files=True 또는 # scope_expansion 분류로 표시된 항목으로 변환. proposed_fixes_for_critical: list[dict[str, Any]] = [] expected_files_original: list[str] = [] # P9P: list O(N²) 멤버십 체크 회피 — set 으로 dedupe 후 list 보존. # 순서 보존이 필요하므로 set + list 조합 (단순 set 변환 X). expected_files_seen: set[str] = set() # expected_files_original 은 scope_expansion 분류 외 thread 에도 # 박제될 수 있으므로 분류 여부와 무관하게 전체 thread 에서 1차 수집. for thread in triage_classifications: ef = thread.get("expected_files_original") or thread.get("expected_files") if isinstance(ef, (list, tuple)): for f in ef: s = str(f) if s not in expected_files_seen: expected_files_seen.add(s) expected_files_original.append(s) for thread in triage_classifications: cls = self._classification_of(thread) is_outside = bool(thread.get("outside_expected_files", False)) # evaluate_push_safety 의 scope_expansion 버킷 정의와 일관: # cls == SCOPE_EXPANSION OR outside_expected_files=True 인 thread 만 포함. if cls != CLASSIFICATION_SCOPE_EXPANSION and not is_outside: continue proposed_fixes_for_critical.append({ "thread_id": self._thread_id_of(thread), # P9Q: path 키 일관성 — classify_scope_expansion_as_critical_three 의 # fix.get("target_file") or fix.get("file") or fix.get("path") lookup 순서와 # 일치하도록 thread 에서도 path 키 fallback 추가. "target_file": ( thread.get("target_file") or thread.get("file") or thread.get("path") ), "outside_expected_files": True, "classification": CLASSIFICATION_SCOPE_EXPANSION, }) critical_result = self.classify_scope_expansion_as_critical_three( pr_number=int(pr_number), proposed_fixes=proposed_fixes_for_critical, expected_files_original=expected_files_original, ) critical_seven_classification = critical_result["critical_seven_kind"] outcome = OUTCOME_SCOPE_EXPANSION_REPORTED elif decision == SAME_PR_BLOCKED_REPLACEMENT_REQUIRED: if self._replacement_pr_runner_callable is None: raise ValueError( "replacement_pr_runner_callable required for " "SAME_PR_BLOCKED_REPLACEMENT_REQUIRED outcome " "(회장 §명시 silent fallback 금지)" ) # 명시 루프: classification / is_code_change_required 1회만 호출하여 # 효율성·가독성 확보 (list comprehension 내부 중복 호출 제거). proposed_fixes: list[dict[str, Any]] = [] for thread in triage_classifications: cls = self._classification_of(thread) if cls == CLASSIFICATION_SCOPE_EXPANSION: continue # P9U: cls 재계산 회피 — 위에서 이미 _classification_of 호출함. if not self._is_code_change_required(thread, classification=cls): continue proposed_fixes.append({ "thread_id": self._thread_id_of(thread), "classification": cls, # P9Y: path 키 일관성 — P9Q 와 동일 lookup 순서 적용. "target_file": ( thread.get("target_file") or thread.get("file") or thread.get("path") ), "summary": thread.get("summary") or thread.get("body"), }) pivot_result = self.pivot_to_replacement_pr( original_pr=int(pr_number), original_branch=str(original_branch), original_head_sha=str(current_head_sha), proposed_fixes=proposed_fixes, chat_id=int(chat_id), ) health_gate_result = pivot_result.get("pr_open_health_gate") outcome = OUTCOME_REPLACEMENT_PR_OPENED else: raise ValueError( f"unknown decision from evaluate_push_safety: {decision!r}" ) return { "outcome": outcome, "evaluate_result": evaluate_result, "pivot_result": pivot_result, "health_gate_result": health_gate_result, "critical_seven_classification": critical_seven_classification, "pr_number": int(pr_number), "ts": ts, } __all__ = [ "GeminiStalePreventionRunner", # Decision codes "SAME_PR_SAFE", "SAME_PR_BLOCKED_REPLACEMENT_REQUIRED", "SAME_PR_BLOCKED_SCOPE_EXPANSION", "DECISIONS", # Outcome codes "OUTCOME_SAME_PR_RESOLVED", "OUTCOME_REPLACEMENT_PR_OPENED", "OUTCOME_SCOPE_EXPANSION_REPORTED", "OUTCOME_EMPTY_COMMIT_BLOCKED", # Classifications "CLASSIFICATION_FALSE_POSITIVE", "CLASSIFICATION_STYLE_ONLY", "CLASSIFICATION_NO_CODE_CHANGE", "CLASSIFICATION_MINOR_FIX_IN_SCOPE", "CLASSIFICATION_REAL_BUG_IN_SCOPE", "CLASSIFICATION_SCOPE_EXPANSION", "NON_CODE_CHANGING_CLASSIFICATIONS", "CODE_CHANGING_CLASSIFICATIONS", # Critical 7종 #3 "CRITICAL_SEVEN_KIND_SCOPE_EXPANSION", "CRITICAL_SEVEN_KIND_SCOPE_EXPANSION_NAME", # Critical 7종 #6 (REPLACEMENT_PR_CONTRACT_FRAMING_INCONSISTENT_WITH_ORIGIN_MAIN_STATE) "CRITICAL_SEVEN_KIND_REPLACEMENT_CONTRACT_FRAMING", "CRITICAL_SEVEN_KIND_REPLACEMENT_CONTRACT_FRAMING_NAME", "ORIGINAL_PR_UNMERGED_STATES", # Empty commit block "EMPTY_COMMIT_BLOCKED_KIND", "EMPTY_COMMIT_BLOCKED_REASON", # chat_id "DEFAULT_CHAT_ID", ]