"""Merge-ready classifier (evidence-only, pure function, decoupled). PR/Gemini/CI/phase3/scope/credential/lifecycle evidence snapshot(dict) 를 종합해 merge 가능 여부(verdict)와 회장 보고 필요 여부(chair_triggers)를 결정적으로 판정한다. callback lifecycle classifier 와 동형 — evidence-only · I/O 0 · merge 실행 0. doctrines (spec §1~§6): - evidence-only decoupled: anu_v3/런타임 import 0 - pure function · read-only · 파일/네트워크/subprocess/live-git/merge I/O 0 · no daemon - 추정 금지 → 핵심 gate evidence 결핍 시 UNKNOWN(재수집) - verdict precedence (상태기계): UNKNOWN > CHAIR_REQUIRED > HOLD > PASS - credential 3계층: BLOCKING_SECRET/NET_NEW → CHAIR · EXISTING/NONE → fail 아님 - Critical7 1:1 → 전부 CHAIR_REQUIRED - Gemini auto-remediation: medium/style/non-critical HIGH(+expected_files 내부) → HOLD(auto_remediable) - callback lifecycle classification==INCIDENT → CHAIR_REQUIRED 승격 evidence collector(gh/git 조회)는 별도 read-only collector 가 담당한다. classifier 는 dict 만 입력받는 순수함수다 (git grep/gh/merge 직접 호출 X). 단일소스 스펙: memory/specs/system_merge_ready_executor_spec_260522.md """ from utils.merge_ready_states import ( # verdict PASS, HOLD, CHAIR_REQUIRED, UNKNOWN, # credential CRED_NONE, CRED_EXISTING_SYSTEM_IDENTIFIER, CRED_NET_NEW, CRED_BLOCKING_SECRET, CREDENTIAL_CHAIR_TIERS, # completeness COMPLETE, PARTIAL, MISSING, # Critical7 C7_FORBIDDEN_PATH, C7_OUT_OF_SCOPE_REPLACEMENT_FAIL, C7_SCOPE_EXPANSION, C7_OVERRIDE, C7_DEPENDENCY_CYCLE_OR_SERIAL, C7_REPLACEMENT_FAIL, C7_POST_MERGE_SMOKE_FAIL, CRITICAL7, CRITICAL7_ORDER, # chair triggers CHAIR_CRITICAL7, CHAIR_CREDENTIAL_PERMISSION_EXPANSION, CHAIR_OUT_OF_EXPECTED_FILES, CHAIR_ADMIN_OVERRIDE, CHAIR_REPLACEMENT_PR_FAILURE, CHAIR_POST_MERGE_SMOKE_FAILURE, CHAIR_DEPENDENCY_OR_SERIAL, CHAIR_MERGE_POLICY_CHANGE, CHAIR_LIFECYCLE_INCIDENT, CHAIR_AUTO_REMEDIATION_LOOP_BOUNDARY, CHAIR_TRIGGER_ORDER, # auto_remediable AR_CI_PENDING, AR_GEMINI_EVIDENCE_STALE, AR_GEMINI_MEDIUM_WITHIN_EXPECTED, AR_GEMINI_NONCRITICAL_HIGH_WITHIN_EXPECTED, AR_UNRESOLVED_MEDIUM_THREAD, AR_MERGE_STATE_TRANSIENT_BLOCKED, AUTO_REMEDIABLE_ORDER, GEMINI_AUTO_REMEDIABLE_SEVERITIES, # 10조건 AUTO_MERGE_10_CONDITION_KEYS, # lifecycle LIFECYCLE_NORMAL, LIFECYCLE_INCIDENT, LIFECYCLE_INCIDENT_MISS_CAUSES, # ci / merge-state 정규화 CI_PENDING_STATES, CI_PASS_STATES, CI_FAIL_STATES, MERGE_STATE_CLEAN, MERGE_STATE_BLOCKED, MERGEABLE_OK, # evidence sources / helpers CANONICAL_EVIDENCE_SOURCES, CORE_EVIDENCE_SOURCES, DEFAULT_ANU_KEYS, is_anu_owner_key, CLASSIFIED_BY, NEXT_ACTION, ) # ───────────────────────────────────────────────────────────────────────────── # 순수 헬퍼 (I/O 0) # ───────────────────────────────────────────────────────────────────────────── def _truthy_signal(value): """리스트/집합은 비어있지 않으면 True, 수치는 >0, bool 은 그대로.""" if value is None: return False if isinstance(value, bool): return value if isinstance(value, (list, tuple, set, frozenset, dict)): return len(value) > 0 if isinstance(value, (int, float)): return value > 0 return bool(value) def _src_present(evidence, key): return key in evidence and evidence.get(key) is not None def _evidence_completeness(evidence): present = [s for s in CANONICAL_EVIDENCE_SOURCES if _src_present(evidence, s)] missing = [s for s in CANONICAL_EVIDENCE_SOURCES if s not in present] # Gemini HIGH·추정 금지: 핵심 evidence(scope/gates) 중 하나라도 없으면 CI/scope 판정 # 불가 → MISSING(→UNKNOWN). (기존: 둘 다 없을 때만 MISSING = 과소판정) if any(s not in set(present) for s in CORE_EVIDENCE_SOURCES): return MISSING, missing if not missing: return COMPLETE, missing return PARTIAL, missing def _lower(value): return str(value).strip().lower() if value is not None else "" def _as_int(value, default=0): """안전 int 변환 (Gemini medium): evidence 값이 비숫자('N/A' 등)여도 ValueError 없이 default 로 폴백. 입력 정합성이 완벽히 보장되지 않는 frozen evidence 방어.""" try: return int(value) except (TypeError, ValueError): return default # ───────────────────────────────────────────────────────────────────────────── # 핵심 분류기 (순수함수) # ───────────────────────────────────────────────────────────────────────────── def classify_merge_ready(evidence, *, anu_keys=DEFAULT_ANU_KEYS, expected_files=None): """evidence snapshot(dict) → merge-ready 분류 결과(dict). 순수함수. Args: evidence: scope/credential/gates/gemini/critical7/merge_mechanics/lifecycle 섹션을 가진 frozen evidence dict (collector 가 사전 수집). anu_keys: ANU authoritative owner key 집합 (라우팅 식별자, 비밀 아님). expected_files: 선언된 expected_files (없으면 evidence['scope']['expected_files'] 사용). Returns: dict: verdict · blocking_reasons · chair_triggers · auto_remediable · auto_merge_10_conditions · critical7_hits · credential_tier · evidence_completeness · next_action · classified_by (+부가 진단 필드) """ evidence = evidence or {} # None-guard (Gemini medium): evidence=None 시 AttributeError 방지 scope = evidence.get("scope") or {} cred = evidence.get("credential") or {} gates = evidence.get("gates") or {} gem = evidence.get("gemini") or {} c7 = evidence.get("critical7") or {} mech = evidence.get("merge_mechanics") or {} life = evidence.get("lifecycle") or {} completeness, missing = _evidence_completeness(evidence) # ── 1) credential 3계층 (위험도 높은 순) ── blocking_present = _truthy_signal(cred.get("blocking_secret_hits")) net_new = bool(cred.get("net_new_identifier_exposure")) existing = bool(cred.get("existing_system_identifier_reuse")) if blocking_present: credential_tier = CRED_BLOCKING_SECRET elif net_new: credential_tier = CRED_NET_NEW elif existing: credential_tier = CRED_EXISTING_SYSTEM_IDENTIFIER else: credential_tier = CRED_NONE # ── 핵심 gate evidence 결핍 → UNKNOWN (추정 금지 · escalate/merge 보류) ── if completeness == MISSING: conditions = {k: False for k in AUTO_MERGE_10_CONDITION_KEYS} return _result( verdict=UNKNOWN, blocking_reasons=["core gate evidence(scope/gates) 결핍 → 재수집 필요 (추정 0)"], chair_triggers=[], auto_remediable=[], conditions=conditions, critical7_hits=[], credential_tier=credential_tier, completeness=completeness, missing=missing, diag={"reason": "core_gate_evidence_missing"}, ) # ── 2) scope / forbidden / out-of-expected ── declared_expected = list(expected_files) if expected_files is not None else (scope.get("expected_files") or []) expected_set = set(declared_expected) effective = scope.get("effective_diff_files") forbidden_present = _truthy_signal(scope.get("forbidden_path_hits")) oof_provided = scope.get("out_of_expected_files_modification") # Gemini HIGH: `and expected_set` 가드 제거 — 빈 expected_files([]) 엣지케이스 정합. # expected_set 비어있어도 effective 와 비교가 항상 수행되어야 한다(빈 expected에 파일 # 변경 → out_of_expected True / 변경 0 → exact_match True). # Gemini medium: effective 가 있으면 실측 우선(exact_match 로직과 대칭). 단 caller 가 # oof_provided=True 로 명시 override 한 경우(실측에 안 보여도 caller 가 알고 있는 out-of-scope # 신호 — replacement-fail 조합 등) 그것도 OR 로 반영. 양쪽 어느 하나라도 True 면 True. out_of_expected_observed = ( any(f not in expected_set for f in effective) if effective is not None else False ) out_of_expected = bool(out_of_expected_observed) or bool(oof_provided) em_provided = scope.get("exact_match") if effective is not None: # Gemini medium: effective 가 있으면 em_provided 유무와 무관하게 실측 set 동일성으로 # 판정(subset 도 exact_match 되던 비일관 제거). "exact" = effective == expected 정확 일치. exact_match = (set(effective) == expected_set) and not forbidden_present else: exact_match = bool(em_provided) and not forbidden_present and not out_of_expected # ── 3) Critical7 신호 (spec §4) ── replacement_pr_failure = bool(c7.get("replacement_pr_failure")) admin_override_required = bool(c7.get("admin_override_required")) bp_bypass_required = bool(c7.get("branch_protection_bypass_required")) dependency_cycle = bool(c7.get("dependency_cycle")) serial_only_collision = bool(c7.get("serial_only_collision")) merge_policy_change = bool(c7.get("merge_policy_change_required")) post_merge_smoke_failure = bool(c7.get("post_merge_smoke_failure")) # Gemini 분석 (spec §5) findings = gem.get("gemini_findings") or [] gemini_gate = _lower(gates.get("gemini_review_gate")) freshness = _lower(gem.get("evidence_freshness")) gemini_stale = gemini_gate == "stale" or freshness == "stale" gemini_pending = gemini_gate == "pending" medium_status = _lower(gem.get("medium_auto_remediation_status")) repeated_high = bool(gem.get("repeated_high_same_function")) scope_expansion_required = bool(gem.get("scope_expansion_required")) or bool(c7.get("scope_expansion_required")) def _unresolved(f): return not bool(f.get("resolved")) def _sev(f): return _lower(f.get("severity")) # Gemini real bug 가 expected_files 밖/scope 확장 요구 → C7#3 gemini_requires_oof = any( _unresolved(f) and (f.get("within_expected_files") is False) for f in findings ) high_or_critical_unresolved = gem.get("high_or_critical_unresolved") unresolved_high_or_critical_finding = any( _unresolved(f) and _sev(f) in ("high", "critical") for f in findings ) medium_within_unresolved = any( _unresolved(f) and _sev(f) in GEMINI_AUTO_REMEDIABLE_SEVERITIES and f.get("within_expected_files", True) for f in findings ) or medium_status == "pending" high_within_unresolved = any( _unresolved(f) and _sev(f) == "high" and f.get("within_expected_files", True) for f in findings ) critical7 = set() if forbidden_present: critical7.add(C7_FORBIDDEN_PATH) if out_of_expected and replacement_pr_failure: critical7.add(C7_OUT_OF_SCOPE_REPLACEMENT_FAIL) if gemini_requires_oof or scope_expansion_required: critical7.add(C7_SCOPE_EXPANSION) if admin_override_required or bp_bypass_required: critical7.add(C7_OVERRIDE) if dependency_cycle or serial_only_collision: critical7.add(C7_DEPENDENCY_CYCLE_OR_SERIAL) if replacement_pr_failure: critical7.add(C7_REPLACEMENT_FAIL) if post_merge_smoke_failure: critical7.add(C7_POST_MERGE_SMOKE_FAIL) # 명시적 critical7_hits passthrough (collector 가 직접 식별한 경우) for h in (c7.get("critical7_hits") or []): if h in CRITICAL7: critical7.add(h) critical7_hits = [c for c in CRITICAL7_ORDER if c in critical7] # ── 4) callback lifecycle classification (fields 10~14) ── life_classification = _lower(life.get("classification")) or LIFECYCLE_NORMAL life_miss_cause = life.get("normal_callback_miss_cause") # owner key 교차확인: closeout callback 이 ANU authoritative key 가 아닌 # executor self-key 로 발사됐다면(non-authoritative) production 신뢰성 훼손 → incident. life_owner_key = life.get("owner_key") or life.get("normal_callback_owner_key") life_fired = bool(life.get("fired")) self_key_fired_non_authoritative = ( life_fired and life_owner_key is not None and not is_anu_owner_key(life_owner_key, anu_keys) ) lifecycle_incident = ( life_classification == LIFECYCLE_INCIDENT or (life_miss_cause in LIFECYCLE_INCIDENT_MISS_CAUSES) or self_key_fired_non_authoritative ) # ── 5) gates 평가 (CI / merge-state / threads) ── ci_checks = gates.get("ci_checks") or {} ci_states = [_lower(s) for s in ci_checks.values()] if isinstance(ci_checks, dict) else [] ci_pending = bool(gates.get("ci_pending")) or any(s in CI_PENDING_STATES for s in ci_states) ci_failed = any(s in CI_FAIL_STATES for s in ci_states) ci_all_pass_provided = gates.get("ci_all_pass") if ci_all_pass_provided is None: ci_all_pass = bool(ci_states) and all(s in CI_PASS_STATES for s in ci_states) else: ci_all_pass = bool(ci_all_pass_provided) and not ci_pending and not ci_failed merge_state = str(gates.get("mergeStateStatus") or "").upper() mergeable_state = str(gates.get("mergeable") or "").upper() threads = gates.get("unresolved_threads") if isinstance(threads, dict): ut_high = _as_int(threads.get("high")) ut_medium = _as_int(threads.get("medium")) # Gemini medium: ut_total 을 항상 actionable(high+medium) 로 일관 산출 — low 제외. # (threads_resolved 정책[high/medium 0]과 일치 · total_raw 의 low 포함 비결정성 제거) ut_total = ut_high + ut_medium elif isinstance(threads, (int, float)): ut_total = _as_int(threads) ut_high = ut_total ut_medium = 0 else: ut_high = ut_medium = ut_total = 0 phase3 = gates.get("phase3_merge_gate") phase3_pass = (phase3 is True) or (_lower(phase3) == "pass") # ── 6) auto-merge 10조건 ── gemini_review_pass = ( gemini_gate == "pass" # 추정 금지(Gemini medium): 빈 gate 를 PASS 로 간주하지 않음 and not gemini_stale and not gemini_pending and (high_or_critical_unresolved in (None, 0)) and not unresolved_high_or_critical_finding and medium_status != "pending" # Gemini medium: low severity 는 threads_resolved 와 일관되게 제외 — actionable(high/medium) # 미해소만 PASS 차단(low 단독으로 영구 HOLD 만들지 않음). and not any(_unresolved(f) and _sev(f) in ("high", "medium") for f in findings) ) conditions = { # exact_match 정의(§scope)가 이미 not forbidden_present·(not out_of_expected)를 포함하므로 # 중복 조건 제거(Gemini medium). 정의 드리프트 방어는 chair_forbidden_path/chair_out_of_scope fixture. "exact_scope_match": exact_match, "ci_all_green": ci_all_pass and not ci_pending and not ci_failed, "gemini_review_pass": bool(gemini_review_pass), "phase3_merge_gate_pass": bool(phase3_pass), "merge_state_clean": merge_state == MERGE_STATE_CLEAN, "mergeable": mergeable_state == MERGEABLE_OK, # Gemini medium: high/medium thread 0 이 정책 기준(states 정의). low thread 는 # auto_remediable 아니어서 ut_total 포함 시 영구 HOLD 위험 + 입력 필드 구성 비결정성. # low thread 미해소는 mergeStateStatus(BLOCKED)가 merge_state_clean 으로 별도 차단. "threads_resolved": (ut_high == 0 and ut_medium == 0), "credential_clean": credential_tier in (CRED_NONE, CRED_EXISTING_SYSTEM_IDENTIFIER), "no_critical7": len(critical7_hits) == 0, "lifecycle_normal": not lifecycle_incident, } # ── 7) chair_triggers (8 + lifecycle incident + loop boundary) ── chair = set() blocking_reasons = [] if critical7_hits: chair.add(CHAIR_CRITICAL7) blocking_reasons.append("Critical7 적중: " + ", ".join(critical7_hits)) if credential_tier in CREDENTIAL_CHAIR_TIERS: chair.add(CHAIR_CREDENTIAL_PERMISSION_EXPANSION) blocking_reasons.append(f"credential·permission expansion (tier={credential_tier})") if out_of_expected: chair.add(CHAIR_OUT_OF_EXPECTED_FILES) blocking_reasons.append("expected_files 밖 수정 필요") if admin_override_required or bp_bypass_required: chair.add(CHAIR_ADMIN_OVERRIDE) blocking_reasons.append("admin override / branch protection 우회 필요") if replacement_pr_failure: chair.add(CHAIR_REPLACEMENT_PR_FAILURE) blocking_reasons.append("replacement PR failure") if post_merge_smoke_failure: chair.add(CHAIR_POST_MERGE_SMOKE_FAILURE) blocking_reasons.append("post-merge smoke failure") if dependency_cycle or serial_only_collision: chair.add(CHAIR_DEPENDENCY_OR_SERIAL) blocking_reasons.append("dependency cycle / serial_only collision") if merge_policy_change: chair.add(CHAIR_MERGE_POLICY_CHANGE) blocking_reasons.append("merge policy 변경 필요") if lifecycle_incident: chair.add(CHAIR_LIFECYCLE_INCIDENT) blocking_reasons.append( "callback lifecycle classification=INCIDENT (production 신뢰성 훼손)" ) # 같은 함수/file-boundary HIGH 반복 → auto-remediation loop 경계 회장 요약보고 if high_within_unresolved and repeated_high: chair.add(CHAIR_AUTO_REMEDIATION_LOOP_BOUNDARY) blocking_reasons.append("auto-remediation loop boundary: 같은 함수 HIGH 반복") chair_triggers = [t for t in CHAIR_TRIGGER_ORDER if t in chair] # ── 8) auto_remediable (HOLD 자동수렴 후보) ── auto = set() if ci_pending: auto.add(AR_CI_PENDING) if gemini_stale: auto.add(AR_GEMINI_EVIDENCE_STALE) cred_clean_for_remediation = credential_tier in (CRED_NONE, CRED_EXISTING_SYSTEM_IDENTIFIER) if medium_within_unresolved and cred_clean_for_remediation and not critical7_hits: auto.add(AR_GEMINI_MEDIUM_WITHIN_EXPECTED) if ( high_within_unresolved and not repeated_high and not scope_expansion_required and cred_clean_for_remediation and not critical7_hits ): auto.add(AR_GEMINI_NONCRITICAL_HIGH_WITHIN_EXPECTED) if ut_medium > 0 and ut_high == 0: auto.add(AR_UNRESOLVED_MEDIUM_THREAD) if merge_state == MERGE_STATE_BLOCKED and not chair_triggers: auto.add(AR_MERGE_STATE_TRANSIENT_BLOCKED) auto_remediable = [a for a in AUTO_REMEDIABLE_ORDER if a in auto] # ── 9) verdict precedence: UNKNOWN > CHAIR_REQUIRED > HOLD > PASS ── if chair_triggers: verdict = CHAIR_REQUIRED elif all(conditions.values()) and not auto_remediable: verdict = PASS else: verdict = HOLD if not blocking_reasons: failed = [k for k, v in conditions.items() if not v] if failed: blocking_reasons.append("auto-merge 10조건 미충족: " + ", ".join(failed)) if auto_remediable: blocking_reasons.append("자동수렴 후보: " + ", ".join(auto_remediable)) diag = { "exact_match": exact_match, "out_of_expected_files_modification": out_of_expected, "forbidden_path_present": forbidden_present, "ci_pending": ci_pending, "ci_failed": ci_failed, "ci_all_pass": ci_all_pass, "gemini_stale": gemini_stale, "gemini_pending": gemini_pending, "lifecycle_classification": LIFECYCLE_INCIDENT if lifecycle_incident else LIFECYCLE_NORMAL, "unresolved_threads": {"high": ut_high, "medium": ut_medium, "total": ut_total}, "merge_state": merge_state, "mergeable": mergeable_state, "queue_head": bool(mech.get("queue_head")), "serial_conflict": bool(mech.get("serial_conflict")), "declared_expected_files": declared_expected, } return _result( verdict=verdict, blocking_reasons=blocking_reasons, chair_triggers=chair_triggers, auto_remediable=auto_remediable, conditions=conditions, critical7_hits=critical7_hits, credential_tier=credential_tier, completeness=completeness, missing=missing, diag=diag, ) def _result(*, verdict, blocking_reasons, chair_triggers, auto_remediable, conditions, critical7_hits, credential_tier, completeness, missing, diag): """반환 dict 조립 (필드 순서 고정).""" return { "verdict": verdict, "blocking_reasons": blocking_reasons, "chair_triggers": chair_triggers, "auto_remediable": auto_remediable, "auto_merge_10_conditions": conditions, "critical7_hits": critical7_hits, "credential_tier": credential_tier, "evidence_completeness": completeness, "missing_evidence_sources": missing, "next_action": NEXT_ACTION[verdict], "merge_ready_evidence": diag, "classified_by": CLASSIFIED_BY, }