"""anu_v2.second_review_recovery — GEMINI_SECOND_REVIEW_BOTTLENECK 자동 복구 모듈 (task-2565). follow-up commit 이후 Gemini evidence 가 old head 에만 존재하고 current head 에는 없어 CI gate 가 SHA mismatch 로 실패하는 병목을 코드/파일 자동화로 고정한다. 9 상태 enum (SecondReviewState): FOLLOW_UP_COMMIT_CREATED → GEMINI_EVIDENCE_STALE_ON_HEAD → SECOND_REVIEW_PENDING → SECOND_REVIEW_TRIGGER_REQUIRED → SECOND_REVIEW_OWNER_TRIGGER_POSTED → SECOND_REVIEW_FRESH_DETECTED → CI_RERUN_REQUIRED_AFTER_FRESH → MERGE_READY_AFTER_SECOND_REVIEW (timeout 분기: SECOND_REVIEW_TIMEOUT) schema 버전: second_review_decision.v1 (상수: SECOND_REVIEW_DECISION_SCHEMA) pre_merge_commit_decision.v1 (상수: PRE_MERGE_COMMIT_DECISION_SCHEMA) 불변 원칙 (task-2565 §8): - manual_owner_input_requested 항상 False — 회장 수동 요청 금지 doctrine - long_polling_used 항상 False — long polling 금지 (task-2556 §9) one-way isolation: anu_v2/ 외부 import 금지. stdlib + enum + json + pathlib 만 사용. """ from __future__ import annotations import json import os import tempfile from dataclasses import dataclass, field from enum import Enum from pathlib import Path from typing import Any, Callable, Final # ─── Phase 2: grace period 상수 import ─────────────────────────────────────── from anu_v2.polling_policy import SECOND_REVIEW_GRACE_SECONDS # ─── Schema 상수 ────────────────────────────────────────────────────────────── SECOND_REVIEW_DECISION_SCHEMA: Final[str] = "anu_v2.second_review_decision.v1" PRE_MERGE_COMMIT_DECISION_SCHEMA: Final[str] = "anu_v2.pre_merge_commit_decision.v1" # ─── Exception ──────────────────────────────────────────────────────────────── class SchemaViolation(ValueError): """schema 검증 실패 — 필수 필드 누락, 잘못된 enum 값, 또는 불변 원칙 위반.""" # ─── 9 상태 enum ────────────────────────────────────────────────────────────── class SecondReviewState(Enum): """GEMINI_SECOND_REVIEW_BOTTLENECK 자동화 상태 머신 9 상태 (task-2565 §2). 전이 순서: FOLLOW_UP_COMMIT_CREATED → GEMINI_EVIDENCE_STALE_ON_HEAD → SECOND_REVIEW_PENDING → SECOND_REVIEW_TRIGGER_REQUIRED → SECOND_REVIEW_OWNER_TRIGGER_POSTED → SECOND_REVIEW_FRESH_DETECTED → CI_RERUN_REQUIRED_AFTER_FRESH → MERGE_READY_AFTER_SECOND_REVIEW (timeout 분기: → SECOND_REVIEW_TIMEOUT) """ FOLLOW_UP_COMMIT_CREATED = "FOLLOW_UP_COMMIT_CREATED" GEMINI_EVIDENCE_STALE_ON_HEAD = "GEMINI_EVIDENCE_STALE_ON_HEAD" SECOND_REVIEW_PENDING = "SECOND_REVIEW_PENDING" SECOND_REVIEW_TRIGGER_REQUIRED = "SECOND_REVIEW_TRIGGER_REQUIRED" SECOND_REVIEW_OWNER_TRIGGER_POSTED = "SECOND_REVIEW_OWNER_TRIGGER_POSTED" SECOND_REVIEW_FRESH_DETECTED = "SECOND_REVIEW_FRESH_DETECTED" SECOND_REVIEW_TIMEOUT = "SECOND_REVIEW_TIMEOUT" CI_RERUN_REQUIRED_AFTER_FRESH = "CI_RERUN_REQUIRED_AFTER_FRESH" MERGE_READY_AFTER_SECOND_REVIEW = "MERGE_READY_AFTER_SECOND_REVIEW" # ─── stale_reason 허용 값 ──────────────────────────────────────────────────── _VALID_STALE_REASON: Final[frozenset[str]] = frozenset({ "SHA_MISMATCH_AFTER_FOLLOW_UP_COMMIT", "GEMINI_EVIDENCE_NOT_ON_CURRENT_HEAD", "OUTDATED_THREAD_ON_OLD_HEAD", "NONE", }) _VALID_CI_GATE_REASON: Final[frozenset[str]] = frozenset({ "GEMINI_EVIDENCE_STALE", "SHA_MISMATCH", "NONE", }) _VALID_FINAL_DECISION: Final[frozenset[str]] = frozenset({ "WAITING", "FRESH_DETECTED", "MERGE_READY", "ESCALATED", "TIMEOUT", }) _VALID_CHANGE_TYPE: Final[frozenset[str]] = frozenset({ "report_only", "code_fix", "test_fix", "evidence_required", }) # ─── second_review_decision.v1 빌더 ───────────────────────────────────────── def build_second_review_decision( *, task_id: str, pr_number: int, old_head_sha: str, current_head_sha: str, latest_gemini_commit_id: str, gemini_fresh_on_current_head: bool, stale_reason: str, ci_gate_reason: str, owner_trigger_required: bool, owner_trigger_result: str, fresh_detected_at: str | None, ci_rerun_required: bool, final_decision: str, ) -> dict[str, Any]: """``second_review_decision.v1`` schema dict 를 생성한다 (task-2565 §4.1). 박제 필드 (절대 변경 금지): - ``manual_owner_input_requested``: 항상 False (회장 수동 요청 금지 doctrine) - ``long_polling_used``: 항상 False (long polling 금지) - ``schema``: SECOND_REVIEW_DECISION_SCHEMA Args: task_id: 관련 task ID (예: "task-2565"). pr_number: PR 번호. old_head_sha: follow-up commit 이전 head SHA. current_head_sha: 현재 head SHA. latest_gemini_commit_id: 최신 Gemini review commit_id. gemini_fresh_on_current_head: current head 기준 fresh evidence 존재 여부. stale_reason: stale 판정 이유 (허용 값: _VALID_STALE_REASON). ci_gate_reason: CI gate 실패 이유 (허용 값: _VALID_CI_GATE_REASON). owner_trigger_required: owner_trigger_only 자동 호출 필요 여부. owner_trigger_result: "POSTED" | "DEDUPED" | "FAILED" | "NONE". fresh_detected_at: fresh evidence 감지 ISO timestamp 또는 None. ci_rerun_required: CI rerun 필요 여부. final_decision: 최종 결정 (허용 값: _VALID_FINAL_DECISION). Returns: 검증되지 않은 decision dict (validate_second_review_decision 으로 후속 검증 권장). """ return { "schema": SECOND_REVIEW_DECISION_SCHEMA, "task_id": task_id, "pr_number": pr_number, "old_head_sha": old_head_sha, "current_head_sha": current_head_sha, "latest_gemini_commit_id": latest_gemini_commit_id, "gemini_fresh_on_current_head": gemini_fresh_on_current_head, "stale_reason": stale_reason, "ci_gate_reason": ci_gate_reason, "owner_trigger_required": owner_trigger_required, "owner_trigger_result": owner_trigger_result, "manual_owner_input_requested": False, # 박제: 절대 True 금지 "long_polling_used": False, # 박제: 절대 True 금지 "fresh_detected_at": fresh_detected_at, "ci_rerun_required": ci_rerun_required, "final_decision": final_decision, } def validate_second_review_decision(decision: dict[str, Any]) -> None: """``second_review_decision.v1`` schema 검증 (task-2565 §4.1). 검증 항목: 1. schema 일치 2. manual_owner_input_requested 항상 False 3. long_polling_used 항상 False 4. 필수 필드 전수 존재 5. stale_reason, ci_gate_reason, final_decision enum 허용 값 Args: decision: build_second_review_decision 으로 생성된 dict. Raises: SchemaViolation: 위반 항목 발견 시. """ _required_fields = { "schema", "task_id", "pr_number", "old_head_sha", "current_head_sha", "latest_gemini_commit_id", "gemini_fresh_on_current_head", "stale_reason", "ci_gate_reason", "owner_trigger_required", "owner_trigger_result", "manual_owner_input_requested", "long_polling_used", "fresh_detected_at", "ci_rerun_required", "final_decision", } # 필수 필드 존재 확인 missing = _required_fields - set(decision.keys()) if missing: raise SchemaViolation( f"second_review_decision 누락 필드: {sorted(missing)}" ) # schema 일치 if decision["schema"] != SECOND_REVIEW_DECISION_SCHEMA: raise SchemaViolation( f"schema 불일치: expected={SECOND_REVIEW_DECISION_SCHEMA!r}, " f"got={decision['schema']!r}" ) # 불변 원칙: manual_owner_input_requested 항상 False if decision["manual_owner_input_requested"] is not False: raise SchemaViolation( "manual_owner_input_requested must always be False " "— 회장 수동 요청 금지 doctrine (task-2565 §8 #1)" ) # 불변 원칙: long_polling_used 항상 False if decision["long_polling_used"] is not False: raise SchemaViolation( "long_polling_used must always be False " "— long polling 금지 (task-2556 §9, task-2565 §8 #4)" ) # stale_reason enum 검증 if decision["stale_reason"] not in _VALID_STALE_REASON: raise SchemaViolation( f"stale_reason 허용 값 위반: {decision['stale_reason']!r} " f"not in {sorted(_VALID_STALE_REASON)}" ) # ci_gate_reason enum 검증 if decision["ci_gate_reason"] not in _VALID_CI_GATE_REASON: raise SchemaViolation( f"ci_gate_reason 허용 값 위반: {decision['ci_gate_reason']!r} " f"not in {sorted(_VALID_CI_GATE_REASON)}" ) # final_decision enum 검증 if decision["final_decision"] not in _VALID_FINAL_DECISION: raise SchemaViolation( f"final_decision 허용 값 위반: {decision['final_decision']!r} " f"not in {sorted(_VALID_FINAL_DECISION)}" ) # ─── pre_merge_commit_decision.v1 빌더 ────────────────────────────────────── def build_pre_merge_commit_decision( *, merge_ready: bool, changed_files: list[str], change_type: str, pre_merge_commit_allowed: bool, redirect_to_post_merge_evidence: bool, ) -> dict[str, Any]: """``pre_merge_commit_decision.v1`` schema dict 를 생성한다 (task-2565 §4.2). Args: merge_ready: MERGE_READY 5+1 조건 충족 여부. changed_files: 변경 파일 목록. change_type: 변경 유형 (허용: "report_only","code_fix","test_fix","evidence_required"). pre_merge_commit_allowed: merge 전 commit 허용 여부. redirect_to_post_merge_evidence: post-merge evidence redirect 활성화 여부. Returns: 검증되지 않은 decision dict (validate_pre_merge_commit_decision 으로 후속 검증 권장). """ return { "schema": PRE_MERGE_COMMIT_DECISION_SCHEMA, "merge_ready": merge_ready, "changed_files": changed_files, "change_type": change_type, "pre_merge_commit_allowed": pre_merge_commit_allowed, "redirect_to_post_merge_evidence": redirect_to_post_merge_evidence, } def validate_pre_merge_commit_decision(decision: dict[str, Any]) -> None: """``pre_merge_commit_decision.v1`` schema 검증 (task-2565 §4.2). 검증 항목: 1. schema 일치 2. 필수 필드 전수 존재 3. change_type 허용 값 Args: decision: build_pre_merge_commit_decision 으로 생성된 dict. Raises: SchemaViolation: 위반 항목 발견 시. """ _required_fields = { "schema", "merge_ready", "changed_files", "change_type", "pre_merge_commit_allowed", "redirect_to_post_merge_evidence", } # 필수 필드 존재 확인 missing = _required_fields - set(decision.keys()) if missing: raise SchemaViolation( f"pre_merge_commit_decision 누락 필드: {sorted(missing)}" ) # schema 일치 if decision["schema"] != PRE_MERGE_COMMIT_DECISION_SCHEMA: raise SchemaViolation( f"schema 불일치: expected={PRE_MERGE_COMMIT_DECISION_SCHEMA!r}, " f"got={decision['schema']!r}" ) # change_type enum 검증 if decision["change_type"] not in _VALID_CHANGE_TYPE: raise SchemaViolation( f"change_type 허용 값 위반: {decision['change_type']!r} " f"not in {sorted(_VALID_CHANGE_TYPE)}" ) # ─── Phase 2: SecondReviewInput + 판정 함수 ────────────────────────────────── @dataclass(frozen=True) class SecondReviewInput: """second-review 판정 입력 (caller가 채워서 주입). Attributes: task_id: 관련 task ID (예: "task-2565"). pr_number: PR 번호. old_head_sha: follow-up commit 이전 head SHA. current_head_sha: 현재 head SHA. latest_gemini_commit_id: 최신 Gemini review commit_id (없으면 None). ci_gate_failure_reason: CI gate 실패 이유 — "GEMINI_EVIDENCE_STALE" | "SHA_MISMATCH" | None. unresolved_thread_outdated: unresolved thread가 old head 기준 OUTDATED인지. follow_up_commit_detected: follow-up commit 감지 여부. elapsed_since_follow_up_seconds: follow-up 발생 후 경과 (caller 계산). owner_trigger_audit_entries: 기존 audit entries (dedupe 검사용). """ task_id: str pr_number: int old_head_sha: str current_head_sha: str latest_gemini_commit_id: str | None ci_gate_failure_reason: str | None # "GEMINI_EVIDENCE_STALE" | "SHA_MISMATCH" | None unresolved_thread_outdated: bool # unresolved thread가 old head 기준 OUTDATED follow_up_commit_detected: bool # follow-up commit 감지 여부 elapsed_since_follow_up_seconds: int # follow-up 발생 후 경과 (caller 계산) owner_trigger_audit_entries: tuple[dict, ...] = () # 기존 audit (dedupe 검사용) def is_gemini_stale_on_head_after_followup(inp: SecondReviewInput) -> bool: """4조건 모두 충족 시 True — Gemini evidence가 follow-up commit 이후 stale 상태임을 판정. 4조건: 1. follow-up commit 감지 + head SHA 변경 2. latest_gemini_commit_id 존재 + current head에 없음 3. CI gate 실패 이유가 GEMINI_EVIDENCE_STALE 또는 SHA_MISMATCH 4. unresolved thread가 old head 기준 OUTDATED Args: inp: SecondReviewInput 판정 입력. Returns: 4조건 모두 True면 True, 하나라도 False면 False. """ cond1 = inp.follow_up_commit_detected and inp.current_head_sha != inp.old_head_sha cond2 = bool(inp.latest_gemini_commit_id) and inp.latest_gemini_commit_id != inp.current_head_sha cond3 = inp.ci_gate_failure_reason in ("GEMINI_EVIDENCE_STALE", "SHA_MISMATCH") cond4 = inp.unresolved_thread_outdated return bool(cond1 and cond2 and cond3 and cond4) def grace_period_elapsed(inp: SecondReviewInput) -> bool: """follow-up commit 이후 grace period가 경과했는지 판정. Args: inp: SecondReviewInput 판정 입력. Returns: elapsed_since_follow_up_seconds >= SECOND_REVIEW_GRACE_SECONDS 이면 True. """ return inp.elapsed_since_follow_up_seconds >= SECOND_REVIEW_GRACE_SECONDS def determine_state(inp: SecondReviewInput) -> SecondReviewState: """상태 전이 결정 함수 — SecondReviewInput을 받아 현재 상태를 결정. 전이 우선순위: 1. follow_up_commit만 감지 → FOLLOW_UP_COMMIT_CREATED 2. 4조건 True + grace 경과 → SECOND_REVIEW_TRIGGER_REQUIRED 3. 4조건 True + grace 미경과 → SECOND_REVIEW_PENDING 4. 4조건 True → GEMINI_EVIDENCE_STALE_ON_HEAD (grace 로직은 2/3에서 처리) Args: inp: SecondReviewInput 판정 입력. Returns: SecondReviewState enum 값. """ # follow-up commit만 감지 (아직 stale 판정 조건 미충족) if inp.follow_up_commit_detected and not is_gemini_stale_on_head_after_followup(inp): return SecondReviewState.FOLLOW_UP_COMMIT_CREATED # 4조건 모두 충족 — stale 판정 if is_gemini_stale_on_head_after_followup(inp): if grace_period_elapsed(inp): return SecondReviewState.SECOND_REVIEW_TRIGGER_REQUIRED else: return SecondReviewState.SECOND_REVIEW_PENDING # stale 조건 미충족 — 기본 stale 상태 반환 return SecondReviewState.GEMINI_EVIDENCE_STALE_ON_HEAD def is_owner_trigger_deduped(inp: SecondReviewInput) -> bool: """same-head dedupe 판정 — 동일 (pr_number, current_head_sha) POSTED 이미 존재하는지 확인. dedupe_key 포맷: "{pr_number}+{current_head_sha}" Args: inp: SecondReviewInput 판정 입력. Returns: 동일 dedupe key로 result=POSTED 항목이 audit_entries에 있으면 True. """ dedupe_key = f"{inp.pr_number}+{inp.current_head_sha}" for entry in inp.owner_trigger_audit_entries: entry_key = f"{entry.get('pr_number') or entry.get('pr') or ''}+{entry.get('head_sha') or entry.get('head') or ''}" if entry_key == dedupe_key and entry.get("result") == "POSTED": return True return False def auto_trigger_owner_review( inp: SecondReviewInput, *, trigger_callable: Callable[[int, str], dict] | None = None, ) -> dict: """owner trigger 자동 호출 진입점 (4조건 + grace + dedupe 확인 후 호출). 흐름: 1. 4조건 미충족 → CONDITIONS_NOT_MET 2. grace 미경과 → GRACE_PENDING 3. dedupe True → DEDUPED 4. trigger_callable None → NOT_TRIGGERED (테스트용) 5. trigger_callable 호출 → POSTED (성공) / FAILED (예외) Args: inp: SecondReviewInput 판정 입력. trigger_callable: (pr_number, head_sha) → dict 형태의 호출 가능 객체. None이면 NOT_TRIGGERED 반환 (테스트/dry-run용). Returns: dict with keys: - triggered: bool - result: "POSTED"|"DEDUPED"|"FAILED"|"NOT_TRIGGERED"|"GRACE_PENDING"|"CONDITIONS_NOT_MET" - dedupe_key: str - reason: str """ dedupe_key = f"{inp.pr_number}+{inp.current_head_sha}" # 1. 4조건 확인 if not is_gemini_stale_on_head_after_followup(inp): return { "triggered": False, "result": "CONDITIONS_NOT_MET", "dedupe_key": dedupe_key, "reason": "4조건 미충족 — stale 상태 아님", } # 2. grace period 확인 if not grace_period_elapsed(inp): return { "triggered": False, "result": "GRACE_PENDING", "dedupe_key": dedupe_key, "reason": ( f"grace period 미경과 — " f"{inp.elapsed_since_follow_up_seconds}s < {SECOND_REVIEW_GRACE_SECONDS}s" ), } # 3. dedupe 확인 if is_owner_trigger_deduped(inp): return { "triggered": False, "result": "DEDUPED", "dedupe_key": dedupe_key, "reason": f"동일 (pr={inp.pr_number}, head={inp.current_head_sha}) POSTED 이미 존재", } # 4. trigger_callable 없으면 NOT_TRIGGERED (테스트/dry-run) if trigger_callable is None: return { "triggered": False, "result": "NOT_TRIGGERED", "dedupe_key": dedupe_key, "reason": "trigger_callable=None — dry-run 모드", } # 5. 실제 호출 try: trigger_callable(inp.pr_number, inp.current_head_sha) return { "triggered": True, "result": "POSTED", "dedupe_key": dedupe_key, "reason": f"owner trigger 호출 성공 (pr={inp.pr_number}, head={inp.current_head_sha})", } except Exception as exc: # noqa: BLE001 — 회귀 방지용 광의 catch return { "triggered": False, "result": "FAILED", "dedupe_key": dedupe_key, "reason": f"trigger_callable 예외: {type(exc).__name__}: {exc}", } # ─── Phase 2: Marker 헬퍼 ──────────────────────────────────────────────────── # 상태 → marker 이름 매핑 (task-2565 §5 Phase 2) _STATE_MARKER_MAP: Final[dict[str, str]] = { SecondReviewState.GEMINI_EVIDENCE_STALE_ON_HEAD.value: "task-2565.gemini-stale-on-head-after-followup", SecondReviewState.SECOND_REVIEW_TRIGGER_REQUIRED.value: "task-2565.second-review-owner-trigger-requested", SecondReviewState.SECOND_REVIEW_OWNER_TRIGGER_POSTED.value: "task-2565.second-review-owner-trigger-posted", SecondReviewState.SECOND_REVIEW_FRESH_DETECTED.value: "task-2565.second-review-fresh-detected", SecondReviewState.CI_RERUN_REQUIRED_AFTER_FRESH.value: "task-2565.ci-rerun-after-fresh", SecondReviewState.MERGE_READY_AFTER_SECOND_REVIEW.value: "task-2565.merge-ready-after-second-review", } SECOND_REVIEW_DECISION_JSON: Final[str] = "task-2565.second-review-decision.json" """항상 갱신되는 second-review decision JSON marker 파일명.""" def emit_phase2_markers( task_id: str, state: SecondReviewState, decision: dict, *, marker_dir: Path | None = None, ) -> list[Path]: """상태별 marker 파일 생성 + second-review-decision.json 갱신. 상태 → marker 파일 매핑: GEMINI_EVIDENCE_STALE_ON_HEAD → task-2565.gemini-stale-on-head-after-followup SECOND_REVIEW_TRIGGER_REQUIRED → task-2565.second-review-owner-trigger-requested SECOND_REVIEW_OWNER_TRIGGER_POSTED → task-2565.second-review-owner-trigger-posted SECOND_REVIEW_FRESH_DETECTED → task-2565.second-review-fresh-detected CI_RERUN_REQUIRED_AFTER_FRESH → task-2565.ci-rerun-after-fresh MERGE_READY_AFTER_SECOND_REVIEW → task-2565.merge-ready-after-second-review 또한 항상 task-2565.second-review-decision.json 갱신. Args: task_id: 관련 task ID. state: 현재 SecondReviewState. decision: auto_trigger_owner_review 반환 dict 또는 임의 decision dict. marker_dir: marker 저장 디렉토리. None이면 MARKER_DIR 사용. Returns: 생성/갱신된 marker 경로 list. """ created: list[Path] = [] # Finding 1 (HIGH): decision이 second_review_decision.v1 필수 필드를 완전히 포함하는지 검증. # 호출자가 부분 dict를 넘기면 schema-aware consumer가 validation 실패를 일으킬 수 있으므로 # 함수 내부에서 선제적으로 누락 필드를 감지하여 ValueError를 raise한다. _SCHEMA_V1_REQUIRED: frozenset[str] = frozenset({ "schema", "pr_number", "old_head_sha", "current_head_sha", "latest_gemini_commit_id", "gemini_fresh_on_current_head", "stale_reason", "ci_gate_reason", "owner_trigger_required", "owner_trigger_result", "manual_owner_input_requested", "long_polling_used", "fresh_detected_at", "ci_rerun_required", "final_decision", }) missing_fields = _SCHEMA_V1_REQUIRED - set(decision.keys()) if missing_fields: raise ValueError( f"emit_phase2_markers: decision 누락 필드 {sorted(missing_fields)} — " "build_second_review_decision()의 완전한 결과를 전달해야 합니다." ) # 상태별 marker 생성 (매핑에 있는 경우만) marker_name = _STATE_MARKER_MAP.get(state.value) if marker_name: p = write_marker(marker_name, marker_dir=marker_dir) created.append(p) # 항상 second-review-decision.json 갱신 decision_payload: dict[str, Any] = { "task_id": task_id, "state": state.value, **decision, } p = write_marker(SECOND_REVIEW_DECISION_JSON, payload=decision_payload, marker_dir=marker_dir) created.append(p) return created # ─── Phase 2: owner_trigger_audit append 헬퍼 ──────────────────────────────── _DEFAULT_AUDIT_REL_PATH: Final[str] = "memory/audit/owner_trigger_audit.jsonl" """기본 owner_trigger audit JSONL 경로 (workspace-relative).""" def append_owner_trigger_audit( entry: dict, *, audit_path: Path | None = None, ) -> Path: """owner_trigger audit JSONL에 한 줄 atomic append. 기본 경로: memory/audit/owner_trigger_audit.jsonl entry는 dict (timestamp, pr_number, head_sha, result, source 등 포함 권장). Args: entry: audit 기록 dict. timestamp/pr_number/head_sha/result/source 포함 권장. audit_path: JSONL 파일 경로. None이면 기본 경로 사용. Returns: audit 파일 절대 경로. Raises: OSError: 디렉토리 생성 또는 파일 기록 실패 시. """ # Finding 3 (MEDIUM): fcntl은 POSIX 전용 모듈이므로 ImportError 시 명확한 # NotImplementedError를 raise한다. ANU CI는 Linux self-hosted 전용이므로 # 실제 운영 환경에서 이 분기가 실행될 일은 없지만, 비-Linux 환경 진입 시 즉시 감지 가능. try: import fcntl except ImportError as e: raise NotImplementedError( "append_owner_trigger_audit requires POSIX fcntl (Linux only). " "ANU CI is Linux self-hosted; non-Linux environments not supported." ) from e import os target = (audit_path if audit_path is not None else Path(_DEFAULT_AUDIT_REL_PATH)).resolve() target.parent.mkdir(parents=True, exist_ok=True) with open(target, "a", encoding="utf-8") as fh: fcntl.flock(fh.fileno(), fcntl.LOCK_EX) try: fh.write(json.dumps(entry, ensure_ascii=False, sort_keys=True) + "\n") fh.flush() os.fsync(fh.fileno()) finally: fcntl.flock(fh.fileno(), fcntl.LOCK_UN) return target.resolve() # ─── Marker 헬퍼 (Phase 2/3 에서 확장될 골격) ──────────────────────────────── MARKER_DIR: Final[Path] = Path("memory/events") """기본 marker 디렉토리 (workspace-relative). 함수 인자 marker_dir 으로 override 가능.""" def write_marker( marker_name: str, payload: dict[str, Any] | None = None, *, marker_dir: Path | None = None, ) -> Path: """marker 파일을 생성하고 절대 경로를 반환한다. Args: marker_name: marker 파일 이름. ``.json`` 으로 끝나면 payload 를 JSON 으로 기록. 그 외이면 빈 파일 touch (payload 무시). payload: JSON marker 일 때 기록할 dict. None 이면 {} 로 기록. marker_dir: marker 저장 디렉토리. None 이면 ``MARKER_DIR`` 사용. Returns: 생성된 marker 파일의 절대 경로. Raises: OSError: 디렉토리 생성 또는 파일 기록 실패 시. """ target_dir = (marker_dir if marker_dir is not None else MARKER_DIR).resolve() target_dir.mkdir(parents=True, exist_ok=True) marker_path = target_dir / marker_name if marker_name.endswith(".json"): data = payload if payload is not None else {} serialized = json.dumps(data, ensure_ascii=False, indent=2) tf = tempfile.NamedTemporaryFile( "w", dir=target_dir, prefix=f".{marker_name}.", suffix=".tmp", delete=False, encoding="utf-8", ) tmp_path = Path(tf.name) try: with tf: tf.write(serialized) tf.flush() os.fsync(tf.fileno()) os.replace(tmp_path, marker_path) except Exception: tmp_path.unlink(missing_ok=True) raise else: marker_path.touch(exist_ok=True) return marker_path.resolve() # ─── Phase 3: CI rerun 상수 ────────────────────────────────────────────────── MAX_CI_RERUN_PER_HEAD: Final[int] = 3 """같은 (head_sha, job_id) 조합으로 CI rerun 허용 최대 횟수 (task-2565 §5 Phase 3).""" # ─── Phase 3: CI rerun input dataclass ─────────────────────────────────────── @dataclass(frozen=True) class CIRerunInput: """CI rerun 자동 판정 입력 (caller가 채워서 주입). Attributes: pr_number: PR 번호. current_head_sha: 현재 head SHA. gemini_fresh_on_current_head: 최신 review commit_id == current head 여부. ci_gate_failed_with_stale_reason: failure_reason이 EVIDENCE_STALE 또는 SHA_MISMATCH인지. current_head_unchanged_since_failure: CI 실패 이후 head SHA 변경 없음 여부. unresolved_thread_count: 미해결 thread 수 (0 = 허용, >0 = triage 별도 판정). failed_job_identifiers: 실패한 CI job 식별자 목록. previous_rerun_attempts: 같은 (head_sha, job_id) 누적 rerun 횟수. """ pr_number: int current_head_sha: str gemini_fresh_on_current_head: bool # latest review commit_id == current head ci_gate_failed_with_stale_reason: bool # failure_reason in {EVIDENCE_STALE, SHA_MISMATCH} current_head_unchanged_since_failure: bool unresolved_thread_count: int # 0 = 허용, >0 = triage 진행 가능 여부 별도 판정 failed_job_identifiers: tuple[str, ...] = () previous_rerun_attempts: int = 0 # 같은 (head_sha, job_id) 누적 횟수 def should_rerun_failed_ci_jobs(inp: CIRerunInput) -> bool: """CI rerun 4조건 + cap 미초과 여부 판정 (task-2565 §5 Phase 3.1). 4조건 (모두 충족 + cap 미초과 시 True): 1. gemini_fresh_on_current_head: Gemini fresh evidence가 current head에 존재 2. ci_gate_failed_with_stale_reason: stale reason으로 실패한 CI gate 존재 3. current_head_unchanged_since_failure: CI 실패 이후 head 변경 없음 4. unresolved_thread_count == 0: 미해결 thread 없음 추가 cap: previous_rerun_attempts >= MAX_CI_RERUN_PER_HEAD(=3) 시 False. Args: inp: CIRerunInput 판정 입력. Returns: 4조건 모두 충족 + cap 미초과 시 True, 하나라도 불충족 시 False. """ # cap 초과 먼저 확인 (cap_exceeded → False) if inp.previous_rerun_attempts >= MAX_CI_RERUN_PER_HEAD: return False cond1 = inp.gemini_fresh_on_current_head cond2 = inp.ci_gate_failed_with_stale_reason cond3 = inp.current_head_unchanged_since_failure cond4 = inp.unresolved_thread_count == 0 return bool(cond1 and cond2 and cond3 and cond4) def build_ci_rerun_decision( *, pr_number: int, current_head_sha: str, failed_jobs: tuple[str, ...], rerun_allowed: bool, reason: str, attempt_number: int, ) -> dict[str, Any]: """``anu_v2.ci_rerun_decision.v1`` schema dict 생성 (task-2565 §5 Phase 3). 박제 필드: - ``schema``: "anu_v2.ci_rerun_decision.v1" - ``long_polling_used``: 항상 False (long polling 금지 박제) Args: pr_number: PR 번호. current_head_sha: 현재 head SHA. failed_jobs: rerun 대상 CI job 식별자 목록. rerun_allowed: rerun 허용 여부. reason: 허용/차단 이유 문자열. attempt_number: 이번 rerun 시도 번호. Returns: ci_rerun_decision.v1 schema dict. """ return { "schema": "anu_v2.ci_rerun_decision.v1", "pr_number": pr_number, "current_head_sha": current_head_sha, "failed_jobs": list(failed_jobs), "rerun_allowed": rerun_allowed, "reason": reason, "attempt_number": attempt_number, "long_polling_used": False, # 박제: 절대 True 금지 } def auto_rerun_failed_ci_jobs( inp: CIRerunInput, *, rerun_callable: Callable[[int, tuple[str, ...]], dict] | None = None, ) -> dict[str, Any]: """CI rerun 자동 호출 진입점 (adapter pattern, task-2565 §5 Phase 3). 흐름: 1. should_rerun_failed_ci_jobs(inp) 호출 2. cap 초과 → result="CAP_EXCEEDED" 3. 조건 미충족 (cap 미초과) → result="SKIPPED" 4. 조건 충족 + rerun_callable=None → result="DECISION_ONLY" 5. 조건 충족 + rerun_callable 있음 → 호출 후 result="POSTED" 또는 "FAILED" Args: inp: CIRerunInput 판정 입력. rerun_callable: (pr_number, failed_jobs) → dict 형태 callable. None이면 DECISION_ONLY 반환 (dry-run 모드). Returns: dict with keys: result (str), reason (str), decision (dict) """ attempt_number = inp.previous_rerun_attempts + 1 # cap 초과 여부 먼저 확인 if inp.previous_rerun_attempts >= MAX_CI_RERUN_PER_HEAD: decision = build_ci_rerun_decision( pr_number=inp.pr_number, current_head_sha=inp.current_head_sha, failed_jobs=inp.failed_job_identifiers, rerun_allowed=False, reason=f"CI rerun cap 초과 — previous_rerun_attempts={inp.previous_rerun_attempts} >= {MAX_CI_RERUN_PER_HEAD}", attempt_number=attempt_number, ) return { "result": "CAP_EXCEEDED", "reason": decision["reason"], "decision": decision, } # 4조건 판정 if not should_rerun_failed_ci_jobs(inp): # 조건 미충족 이유 결정 if not inp.gemini_fresh_on_current_head: reason = "Gemini fresh evidence 미존재 — rerun 건너뜀" elif not inp.ci_gate_failed_with_stale_reason: reason = "CI gate 실패가 stale reason 아님 — rerun 건너뜀" elif not inp.current_head_unchanged_since_failure: reason = "CI 실패 이후 head 변경됨 — rerun 건너뜀" else: reason = f"unresolved thread 존재 ({inp.unresolved_thread_count}개) — rerun 건너뜀" decision = build_ci_rerun_decision( pr_number=inp.pr_number, current_head_sha=inp.current_head_sha, failed_jobs=inp.failed_job_identifiers, rerun_allowed=False, reason=reason, attempt_number=attempt_number, ) return { "result": "SKIPPED", "reason": reason, "decision": decision, } # 조건 충족 — rerun_callable 없으면 DECISION_ONLY if rerun_callable is None: decision = build_ci_rerun_decision( pr_number=inp.pr_number, current_head_sha=inp.current_head_sha, failed_jobs=inp.failed_job_identifiers, rerun_allowed=True, reason="CI rerun 조건 충족 — dry-run 모드 (rerun_callable=None)", attempt_number=attempt_number, ) return { "result": "DECISION_ONLY", "reason": decision["reason"], "decision": decision, } # 실제 rerun 호출 try: rerun_callable(inp.pr_number, inp.failed_job_identifiers) decision = build_ci_rerun_decision( pr_number=inp.pr_number, current_head_sha=inp.current_head_sha, failed_jobs=inp.failed_job_identifiers, rerun_allowed=True, reason=f"CI rerun 호출 성공 (pr={inp.pr_number}, attempt={attempt_number})", attempt_number=attempt_number, ) return { "result": "POSTED", "reason": decision["reason"], "decision": decision, } except Exception as exc: # noqa: BLE001 reason = f"rerun_callable 예외: {type(exc).__name__}: {exc}" decision = build_ci_rerun_decision( pr_number=inp.pr_number, current_head_sha=inp.current_head_sha, failed_jobs=inp.failed_job_identifiers, rerun_allowed=False, reason=reason, attempt_number=attempt_number, ) return { "result": "FAILED", "reason": reason, "decision": decision, } # ─── Phase 3: MERGE_READY 6조건 판정 ───────────────────────────────────────── @dataclass(frozen=True) class MergeReadyInput: """MERGE_READY 6조건 판정 입력 (task-2565 §5 Phase 3.2, context.md §7.5). Attributes: ci_all_green: CI 11/11 SUCCESS 여부. gemini_fresh_on_current_head: latest review commit_id == current head 여부. unresolved_thread_count: 미해결 thread 수. merge_state_status: mergeStateStatus 값 ("CLEAN" | "BEHIND" | "BLOCKED" | ...). effective_diff_matches_expected_files: effective diff == expected_files 여부. forbidden_path_count: forbidden path 위반 파일 수 (0 = 허용). """ ci_all_green: bool # 11/11 SUCCESS gemini_fresh_on_current_head: bool unresolved_thread_count: int merge_state_status: str # "CLEAN" | "BEHIND" | "BLOCKED" | ... effective_diff_matches_expected_files: bool forbidden_path_count: int def is_merge_ready(inp: MergeReadyInput) -> bool: """MERGE_READY 6조건 전수 충족 여부 판정 (task-2565 §3.2 + context.md §7.5). 6조건 (모두 충족 시 True): 1. ci_all_green: CI 11/11 SUCCESS 2. gemini_fresh_on_current_head: Gemini fresh evidence on current head 3. unresolved_thread_count == 0: 미해결 thread 없음 4. merge_state_status == "CLEAN": mergeStateStatus CLEAN 5. effective_diff_matches_expected_files: effective diff == expected_files 6. forbidden_path_count == 0: forbidden path 위반 없음 Args: inp: MergeReadyInput 판정 입력. Returns: 6조건 모두 True면 True, 하나라도 False면 False. """ return bool( inp.ci_all_green and inp.gemini_fresh_on_current_head and inp.unresolved_thread_count == 0 and inp.merge_state_status == "CLEAN" and inp.effective_diff_matches_expected_files and inp.forbidden_path_count == 0 ) # ─── Phase 3: pre-merge commit 차단 판정 ───────────────────────────────────── @dataclass(frozen=True) class PreMergeCommitInput: """pre-merge commit 차단 판정 입력 (task-2565 §5 Phase 3.2). Attributes: merge_ready: MERGE_READY 6조건 충족 여부. changed_files: 변경 파일 경로 목록. commit_message_hint: commit message prefix 힌트 ("report:", "qc:", "lifecycle:", "evidence:", ...). only_touches_report_files: 변경 파일이 비기능 영역(memory/reports/, memory/events/)만인지. """ merge_ready: bool changed_files: tuple[str, ...] commit_message_hint: str # "report:" "qc:" "lifecycle:" 등 commit message prefix only_touches_report_files: bool # 변경 파일이 memory/reports/, memory/events/ 등 비기능 영역만 def _detect_change_type(inp: PreMergeCommitInput) -> str: """commit 변경 유형 5종 중 매칭 결정 (내부 헬퍼). 5종 wording 패턴: 1. "report wording only" — only_touches_report_files & hint starts with "report" 2. "QC Verdict wording" — hint starts with "qc" 3. "lifecycle marker wording" — only_touches lifecycle marker dir 4. "non-functional summary" — only_touches_report_files & generic 5. "evidence text 보강" — hint starts with "evidence" 내부적으로 matched change_type을 결정: - report_only: 1, 4 - report_only (qc): 2 - report_only (lifecycle): 3 - evidence_required: 5 - code_fix: 코드/테스트 변경 포함 - test_fix: 테스트만 변경 """ hint = inp.commit_message_hint.lower().strip() # 패턴 1: report wording only if inp.only_touches_report_files and (hint.startswith("report") or hint.startswith("report:")): return "report_only" # 패턴 2: QC Verdict wording if hint.startswith("qc"): return "report_only" # 패턴 3: lifecycle marker wording # Finding 2 (MEDIUM): f.startswith("memory/lifecycle/")는 # (f.startswith("memory/") and "/lifecycle/" in f) 조건에 포함되므로 중복 제거. lifecycle_files = [ f for f in inp.changed_files if f.startswith("memory/events/") or (f.startswith("memory/") and "/lifecycle/" in f) ] if lifecycle_files and len(lifecycle_files) == len(inp.changed_files): return "report_only" # 패턴 5: evidence text 보강 if hint.startswith("evidence"): return "evidence_required" # 패턴 4: non-functional summary if inp.only_touches_report_files: return "report_only" # 코드/테스트 변경 확인 test_files = [f for f in inp.changed_files if "test_" in f or "_test." in f or "/tests/" in f] code_files = [ f for f in inp.changed_files if f.endswith(".py") and f not in test_files ] if test_files and not code_files: return "test_fix" if code_files: return "code_fix" # 기본: evidence_required return "evidence_required" def classify_pre_merge_commit(inp: PreMergeCommitInput) -> tuple[str, bool]: """pre-merge commit 유형 분류 + 허용 여부 결정 (task-2565 §5 Phase 3.2). 반환 규칙: - merge_ready=True + 비기능 wording 패턴 → ("report_only", False) — 차단 - merge_ready=True + 코드/테스트 변경 포함 → ("code_fix" | "test_fix", True) — 허용 - merge_ready=False → (change_type, True) — 허용 (merge_ready 아직 아니면 차단 불필요) Args: inp: PreMergeCommitInput 판정 입력. Returns: (change_type, pre_merge_commit_allowed) 튜플. """ change_type = _detect_change_type(inp) if not inp.merge_ready: # merge_ready 아님 — 차단하지 않음 (pre-merge 작업 가능) return (change_type, True) # merge_ready=True인 경우 if change_type == "report_only": # 비기능 wording-only commit → 차단 return ("report_only", False) if change_type in ("code_fix", "test_fix"): # 코드/테스트 변경 → 허용 (실제 수정 필요) return (change_type, True) # evidence_required 등 → 허용 return (change_type, True) def build_pre_merge_block_decision(inp: PreMergeCommitInput) -> dict[str, Any]: """pre-merge commit 차단 decision dict 생성 (task-2565 §5 Phase 3.2). schema: PRE_MERGE_COMMIT_DECISION_SCHEMA merge_ready=True + report_only 시 redirect_to_post_merge_evidence=True. Args: inp: PreMergeCommitInput 판정 입력. Returns: pre_merge_commit_decision.v1 schema dict. """ change_type, allowed = classify_pre_merge_commit(inp) redirect = bool(inp.merge_ready and change_type == "report_only" and not allowed) return { "schema": PRE_MERGE_COMMIT_DECISION_SCHEMA, "merge_ready": inp.merge_ready, "changed_files": list(inp.changed_files), "change_type": change_type, "pre_merge_commit_allowed": allowed, "redirect_to_post_merge_evidence": redirect, } # ─── Phase 3: post-merge evidence redirect 헬퍼 ────────────────────────────── def post_merge_evidence_redirect_payload(blocked_decision: dict[str, Any]) -> dict[str, Any]: """차단된 pre-merge commit decision → post-merge marker용 payload 생성 (task-2565 §5 Phase 3). 차단된 report_only commit을 post-merge 단계로 redirect하기 위한 payload. redirect_to 결정: - change_type="report_only" + original hint 기반: - "lifecycle marker"가 changed_files에 있으면 "lifecycle marker" - 그 외 "post-merge smoke" 또는 "reconcile evidence" Args: blocked_decision: build_pre_merge_block_decision 반환 dict. Returns: dict with keys: - original_change_type: str - original_changed_files: list[str] - redirect_to: "post-merge smoke" | "reconcile evidence" | "lifecycle marker" """ change_type = blocked_decision.get("change_type", "") changed_files = blocked_decision.get("changed_files", []) # redirect_to 결정 lifecycle_files = [f for f in changed_files if "lifecycle" in f or "events" in f] if lifecycle_files: redirect_to = "lifecycle marker" elif change_type == "report_only": # report_only 는 post-merge evidence 또는 reconcile 중 택일 report_files = [f for f in changed_files if "reports" in f] if report_files: redirect_to = "reconcile evidence" else: redirect_to = "post-merge smoke" else: redirect_to = "post-merge smoke" return { "original_change_type": change_type, "original_changed_files": list(changed_files), "redirect_to": redirect_to, } # ─── __all__ ────────────────────────────────────────────────────────────────── __all__ = [ "SECOND_REVIEW_DECISION_SCHEMA", "PRE_MERGE_COMMIT_DECISION_SCHEMA", "MARKER_DIR", "SchemaViolation", "SecondReviewState", "build_second_review_decision", "validate_second_review_decision", "build_pre_merge_commit_decision", "validate_pre_merge_commit_decision", "write_marker", # Phase 2 "SecondReviewInput", "is_gemini_stale_on_head_after_followup", "grace_period_elapsed", "determine_state", "is_owner_trigger_deduped", "auto_trigger_owner_review", "emit_phase2_markers", "append_owner_trigger_audit", "SECOND_REVIEW_DECISION_JSON", # Phase 3 "MAX_CI_RERUN_PER_HEAD", "CIRerunInput", "should_rerun_failed_ci_jobs", "build_ci_rerun_decision", "auto_rerun_failed_ci_jobs", "MergeReadyInput", "is_merge_ready", "PreMergeCommitInput", "classify_pre_merge_commit", "build_pre_merge_block_decision", "post_merge_evidence_redirect_payload", ]