"""Merge-ready DRY-RUN executor (Track B · read-only · evidence-only · merge 실행 0). merge_ready_classifier 의 output dict 와 PR identity 를 입력받아 verdict 별 routing artifact 를 결정적으로 생성한다. **본 task = dry-run 까지만** — 실 merge/push/PR/cron 발사 0 · GitHub write 0 · branch protection 우회 0 · admin override 자동화 0. doctrines (spec system_merge_ready_executor_dryrun_spec_260522.md): - evidence-only decoupled: anu_v3/런타임 import 0 · classifier 수정 0(입력 소비만) - pure function · read-only · 파일/네트워크/subprocess/live-git/merge I/O 0 · no daemon - 모든 artifact 에 actually_executed=false 강제 · 모든 executor_action 에 WOULD_* 접두사 강제 - verdict → routing (spec §1): PASS → merge_ready.auto_merge_candidate.v1 (executor_action=WOULD_MERGE) HOLD → merge_ready.remediation_required.v1 (executor_action=WOULD_AUTO_REMEDIATE) CHAIR_REQUIRED → merge_ready.hold_for_chair.v1 (executor_action=WOULD_ESCALATE_CHAIR) UNKNOWN → merge_ready.hold_for_chair.v1 (executor_action=WOULD_REGATHER, reason=INSUFFICIENT_EVIDENCE) - 보수적 추정 금지: classifier 가 UNKNOWN 이면 routing 도 REGATHER (재수집). - 결정적 idempotency: 동일 입력 2회 호출 → byte-identical 출력 (ts_kst 도 caller 가 공급). 본 모듈은 데이터-only 다. merge/push/PR/cron/branch-protection/admin-override 호출 경로 자체가 부재하다 (코드 경로 0 = 안전 불변식의 가장 강한 보증). 단일소스 스펙: memory/specs/system_merge_ready_executor_dryrun_spec_260522.md """ from utils.merge_ready_states import ( PASS, HOLD, CHAIR_REQUIRED, UNKNOWN, 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, ) # ───────────────────────────────────────────────────────────────────────────── # schema id 단일소스 (spec §2.1/2.2/2.3 정본) # ───────────────────────────────────────────────────────────────────────────── SCHEMA_AUTO_MERGE_CANDIDATE = "merge_ready.auto_merge_candidate.v1" SCHEMA_REMEDIATION_REQUIRED = "merge_ready.remediation_required.v1" SCHEMA_HOLD_FOR_CHAIR = "merge_ready.hold_for_chair.v1" # ───────────────────────────────────────────────────────────────────────────── # executor_action enum — 전부 WOULD_* 접두사 (안전 불변식) # ───────────────────────────────────────────────────────────────────────────── ACTION_WOULD_MERGE = "WOULD_MERGE" ACTION_WOULD_AUTO_REMEDIATE = "WOULD_AUTO_REMEDIATE" ACTION_WOULD_ESCALATE_CHAIR = "WOULD_ESCALATE_CHAIR" ACTION_WOULD_REGATHER = "WOULD_REGATHER" # remediation 하위 action (spec §2.2 remediation_plan) ACTION_WOULD_WAIT_RECHECK = "WOULD_WAIT_RECHECK" ACTION_WOULD_OWNER_GEMINI_REVIEW_NUDGE = "WOULD_OWNER_GEMINI_REVIEW_NUDGE" ACTION_WOULD_AUTO_FIX_REGRESS_PUSH_RESOLVE_RECHECK = "WOULD_AUTO_FIX_REGRESS_PUSH_RESOLVE_RECHECK" ACTION_WOULD_RESOLVE_THREAD_RECHECK = "WOULD_RESOLVE_THREAD_RECHECK" # unknown 라우팅 사유 (spec §1: "with reason=INSUFFICIENT_EVIDENCE") REASON_INSUFFICIENT_EVIDENCE = "INSUFFICIENT_EVIDENCE" # loop boundary 가드 메시지 (spec §2.2 정본) LOOP_BOUNDARY_GUARD_TEXT = "같은 함수 HIGH 반복/scope 확장 조짐 → CHAIR_REQUIRED 승격" # next_recheck 액션 (spec §2.2 정본) NEXT_RECHECK_ACTION = "WOULD_RECLASSIFY_AFTER_REMEDIATION" # auto_remediable 항목 → sub-action 결정적 매핑 (spec §2.2 remediation_plan) AUTO_REMEDIABLE_ACTION_MAP = { AR_CI_PENDING: ACTION_WOULD_WAIT_RECHECK, AR_GEMINI_EVIDENCE_STALE: ACTION_WOULD_OWNER_GEMINI_REVIEW_NUDGE, AR_GEMINI_MEDIUM_WITHIN_EXPECTED: ACTION_WOULD_AUTO_FIX_REGRESS_PUSH_RESOLVE_RECHECK, AR_GEMINI_NONCRITICAL_HIGH_WITHIN_EXPECTED: ACTION_WOULD_AUTO_FIX_REGRESS_PUSH_RESOLVE_RECHECK, AR_UNRESOLVED_MEDIUM_THREAD: ACTION_WOULD_RESOLVE_THREAD_RECHECK, AR_MERGE_STATE_TRANSIENT_BLOCKED: ACTION_WOULD_WAIT_RECHECK, } # 전체 WOULD_* enum 집합 — 안전 불변식 정적 검증용 ALL_EXECUTOR_ACTIONS = frozenset({ ACTION_WOULD_MERGE, ACTION_WOULD_AUTO_REMEDIATE, ACTION_WOULD_ESCALATE_CHAIR, ACTION_WOULD_REGATHER, }) ALL_SUB_ACTIONS = frozenset({ ACTION_WOULD_WAIT_RECHECK, ACTION_WOULD_OWNER_GEMINI_REVIEW_NUDGE, ACTION_WOULD_AUTO_FIX_REGRESS_PUSH_RESOLVE_RECHECK, ACTION_WOULD_RESOLVE_THREAD_RECHECK, NEXT_RECHECK_ACTION, }) # ───────────────────────────────────────────────────────────────────────────── # 순수 헬퍼 (I/O 0) # ───────────────────────────────────────────────────────────────────────────── def _as_int(value, default=0): """안전 int 변환 (classifier 입력 견고성 doctrine 동형). pr 번호 등 비숫자 폴백.""" try: return int(value) except (TypeError, ValueError): return default def _identity(pr_identity): """pr_identity dict → 정규화된 dict({pr, head_sha, task_id, branch}). None-guard 포함.""" pi = pr_identity or {} return { "pr": _as_int(pi.get("pr"), 0), "head_sha": str(pi.get("head_sha") or ""), "task_id": str(pi.get("task_id") or ""), "branch": str(pi.get("branch") or ""), } def _ts_kst(pr_identity): """ts_kst 는 caller(=collector/wiring) 가 공급 — 결정적 idempotency 보장. None/누락 → 빈 문자열(스키마 형식 유지).""" pi = pr_identity or {} return str(pi.get("ts_kst") or "") def _smoke_plan(pr_identity): """post_merge_smoke_plan 은 collector 가 사전 수집해 pr_identity 로 전달. 기본값(defined=True/runnable=True/cmd_ref="") — auto_merge_candidate 생성 시점에 smoke 가 정의/실행가능하다는 가정은 회장 executor 활성화 단계에서 재검증된다.""" pi = pr_identity or {} sp = pi.get("post_merge_smoke_plan") or {} return { "defined": bool(sp.get("defined", True)), "runnable": bool(sp.get("runnable", True)), "cmd_ref": str(sp.get("cmd_ref") or ""), } def _callback_plan(pr_identity): """callback_lifecycle_artifact_plan — collector 가 lifecycle artifact 생성 가능성을 사전 통지. 기본 would_generate=True (정상 closeout 경로 가정).""" pi = pr_identity or {} cp = pi.get("callback_lifecycle_artifact_plan") or {} return {"would_generate": bool(cp.get("would_generate", True))} def _assert_safety_invariants(artifact): """artifact 가 안전 불변식을 만족하는지 자체검증. 위반 시 ValueError(설계 버그 신호) — 정상 입력에서는 절대 발생하지 않아야 한다. 불변식: - actually_executed === False - executor_action 은 WOULD_* 접두사 - schema 는 3종 중 하나 """ if artifact.get("actually_executed") is not False: raise ValueError( "safety invariant violated: actually_executed must be False " f"(got {artifact.get('actually_executed')!r})" ) action = artifact.get("executor_action") if not isinstance(action, str) or not action.startswith("WOULD_"): raise ValueError( f"safety invariant violated: executor_action must start with 'WOULD_' (got {action!r})" ) schema = artifact.get("schema") if schema not in (SCHEMA_AUTO_MERGE_CANDIDATE, SCHEMA_REMEDIATION_REQUIRED, SCHEMA_HOLD_FOR_CHAIR): raise ValueError(f"safety invariant violated: unknown schema {schema!r}") return artifact # ───────────────────────────────────────────────────────────────────────────── # verdict 별 artifact 빌더 # ───────────────────────────────────────────────────────────────────────────── def _build_auto_merge_candidate(result, ident, pr_identity, ts_kst): """PASS → merge_ready.auto_merge_candidate.v1 (spec §2.1).""" conditions = dict(result.get("auto_merge_10_conditions") or {}) artifact = { "schema": SCHEMA_AUTO_MERGE_CANDIDATE, "pr": ident["pr"], "head_sha": ident["head_sha"], "task_id": ident["task_id"], "branch": ident["branch"], "verdict": PASS, "auto_merge_10_conditions": conditions, "post_merge_smoke_plan": _smoke_plan(pr_identity), "executor_action": ACTION_WOULD_MERGE, "actually_executed": False, "requires_chair_activation": True, "callback_lifecycle_artifact_plan": _callback_plan(pr_identity), "ts_kst": ts_kst, } return _assert_safety_invariants(artifact) def _build_remediation_required(result, ident, ts_kst): """HOLD → merge_ready.remediation_required.v1 (spec §2.2).""" # classifier 가 결정적 순서(AUTO_REMEDIABLE_ORDER)로 산출한 리스트를 set 으로 정규화 # (Gemini medium: O(1) lookup + leftover 중복 제거 일관). auto_rem_src_set = set(result.get("auto_remediable") or []) # 안전망: classifier 순서가 깨졌더라도 결정적 순서로 정규화한다(idempotency 보장). auto_rem = [a for a in AUTO_REMEDIABLE_ORDER if a in auto_rem_src_set] # 미인지 코드(미래 enum 확장)도 set 기반(중복 제거) + 알파벳 안정정렬. leftover = sorted(a for a in auto_rem_src_set if a not in AUTO_REMEDIABLE_ORDER) auto_rem.extend(leftover) remediation_plan = [ {"item": item, "action": AUTO_REMEDIABLE_ACTION_MAP.get(item, ACTION_WOULD_WAIT_RECHECK)} for item in auto_rem ] artifact = { "schema": SCHEMA_REMEDIATION_REQUIRED, "pr": ident["pr"], "head_sha": ident["head_sha"], "task_id": ident["task_id"], "branch": ident["branch"], "verdict": HOLD, "auto_remediable": auto_rem, "remediation_plan": remediation_plan, "loop_boundary_guard": LOOP_BOUNDARY_GUARD_TEXT, "next_recheck": NEXT_RECHECK_ACTION, "executor_action": ACTION_WOULD_AUTO_REMEDIATE, "actually_executed": False, "ts_kst": ts_kst, } return _assert_safety_invariants(artifact) def _build_hold_for_chair(result, ident, ts_kst, *, verdict, executor_action, reason=None): """CHAIR_REQUIRED/UNKNOWN → merge_ready.hold_for_chair.v1 (spec §2.3). Args: verdict: CHAIR_REQUIRED 또는 UNKNOWN — classifier 값 그대로 보존(추정 0). executor_action: WOULD_ESCALATE_CHAIR (CHAIR_REQUIRED) | WOULD_REGATHER (UNKNOWN). reason: UNKNOWN 시 INSUFFICIENT_EVIDENCE 표식 (spec §1). """ chair_triggers = list(result.get("chair_triggers") or []) blocking_reasons = list(result.get("blocking_reasons") or []) critical7_hits = list(result.get("critical7_hits") or []) if blocking_reasons: one_line = blocking_reasons[0] elif reason == REASON_INSUFFICIENT_EVIDENCE: one_line = "core gate evidence(scope/gates) 결핍 → 재수집 필요 (추정 0)" else: one_line = "" report_envelope = { "one_line": one_line, "evidence_refs": critical7_hits, "recommended_options": [], } artifact = { "schema": SCHEMA_HOLD_FOR_CHAIR, "pr": ident["pr"], "head_sha": ident["head_sha"], "task_id": ident["task_id"], "branch": ident["branch"], "verdict": verdict, "chair_triggers": chair_triggers, "report_envelope": report_envelope, "blocking_reasons": blocking_reasons, "executor_action": executor_action, "actually_executed": False, "ts_kst": ts_kst, } if reason: artifact["reason"] = reason return _assert_safety_invariants(artifact) # ───────────────────────────────────────────────────────────────────────────── # 공개 진입점 (순수함수) # ───────────────────────────────────────────────────────────────────────────── def dryrun_route(merge_ready_result, pr_identity): """classifier 결과 → verdict 별 routing artifact dict (spec §1). Args: merge_ready_result: utils.merge_ready_classifier.classify_merge_ready(...) 반환 dict. verdict/chair_triggers/auto_remediable/critical7_hits/ auto_merge_10_conditions/blocking_reasons/credential_tier 사용. pr_identity: {pr, head_sha, task_id, branch, ts_kst, post_merge_smoke_plan?, callback_lifecycle_artifact_plan?} ts_kst 는 caller 가 결정적으로 공급(idempotency). Returns: dict: 3종 routing artifact 중 verdict 에 맞는 1종. 모든 artifact 는 actually_executed=False · executor_action ∈ {WOULD_*}. Safety invariants (코드 경로 부재로 자동 보장): - merge/push/PR/cron/branch-protection/admin-override 호출 0 - GitHub write 0 · 파일 쓰기 0 · subprocess 0 · network 0 - classifier 수정 0 (입력 dict 만 read-only 소비) """ result = merge_ready_result or {} verdict = result.get("verdict") ident = _identity(pr_identity) ts = _ts_kst(pr_identity) if verdict == PASS: return _build_auto_merge_candidate(result, ident, pr_identity, ts) if verdict == HOLD: return _build_remediation_required(result, ident, ts) if verdict == CHAIR_REQUIRED: return _build_hold_for_chair( result, ident, ts, verdict=CHAIR_REQUIRED, executor_action=ACTION_WOULD_ESCALATE_CHAIR, ) if verdict == UNKNOWN: return _build_hold_for_chair( result, ident, ts, verdict=UNKNOWN, executor_action=ACTION_WOULD_REGATHER, reason=REASON_INSUFFICIENT_EVIDENCE, ) # 방어: 미인지 verdict 는 UNKNOWN 과 동등 처리(추정 0 · 재수집). return _build_hold_for_chair( result, ident, ts, verdict=UNKNOWN, executor_action=ACTION_WOULD_REGATHER, reason=REASON_INSUFFICIENT_EVIDENCE, )