"""tools/poc/termination_classifier.py Phase B termination classifier (DRY-RUN ONLY, 격리 POC). ⚠️ Production 미반영. task-timer.py / dispatch.py / finish-task.sh 변경 없음. ⚠️ classify()는 dataclass를 반환만 한다. 마커 파일 생성 / dispatch 호출 금지. Usage (dry-run): from tools.poc.termination_classifier import classify, TaskEvidence evidence = TaskEvidence(task_id="task-2466", pr_state="MERGED", ...) result = classify(evidence) print(result.classification, result.confidence, result.followup_conditions) 근거 문서: - memory/orchestration/phase_b_integration_items_260507.md (섹션 2.1 — 8종 + 9단계 분류 룰) - memory/feedback/feedback_merge_pending_dependency_classification_260507.md (6항목 자동 감지) - memory/feedback/feedback_merged_close_blocked_external_classification_260507.md (5항목 자동 감지) - .claude/projects/.../memory/feedback_dogfooding_pending_classification_260507.md (4항목 자동 감지) """ from __future__ import annotations import enum from dataclasses import dataclass, field from typing import Optional # --------------------------------------------------------------------------- # Enum # --------------------------------------------------------------------------- class TerminationClassification(enum.Enum): """8종 종료 분류 + UNCLASSIFIED. 우선순위 순 정렬이 아니라 도메인 의미 단위 정렬.""" DONE = "DONE" ESCALATED = "ESCALATED" DOGFOODING_PENDING = "DOGFOODING_PENDING" MERGE_PENDING_DEPENDENCY = "MERGE_PENDING_DEPENDENCY" MERGED_CLOSE_BLOCKED_EXTERNAL = "MERGED_CLOSE_BLOCKED_EXTERNAL" BLOCKED_BY_EXTERNAL_DEPENDENCY = "BLOCKED_BY_EXTERNAL_DEPENDENCY" FAILED_PREEXISTING = "FAILED_PREEXISTING" WAITING_FOR_CHAIR_DECISION = "WAITING_FOR_CHAIR_DECISION" UNCLASSIFIED = "UNCLASSIFIED" # 룰 미해당, 회장 알림 + 수동 분류 필요. 8종 분류 외에 dry-run 안전망 케이스로만 사용. production 반영 시 회장 재결정 영역. # --------------------------------------------------------------------------- # Input dataclass # --------------------------------------------------------------------------- @dataclass class TaskEvidence: """Classifier 입력. 모든 필드 optional — 기본값이 안전하여 UNCLASSIFIED로 흐른다. 필드 설명: pr_state : OPEN | MERGED | CLOSED | None (PR 없음) pr_merged_at : ISO8601 문자열 또는 None ci_rollup : PASS | FAIL | UNKNOWN qc_result : PASS | WARN | FAIL close_lifecycle_state : CLEAN | FAIL | NOT_RUN close_blocker_owner : internal | external | None essence_verdict : PASS | FAIL | UNKNOWN retry_count : 현재 재시도 횟수 retry_max : 재시도 임계값 (초과 시 ESCALATED 후보) dependency_task_id : 의존/후속 task 식별자 dogfooding_external_dependency : BOT 토큰 / ruleset / approval 등 외부 의존으로 dogfooding layer 미완 여부 blocked_external_system : 외부 API/infra로 본질 진행 자체가 차단 failed_preexisting : 기존 결함이 본 task 진행을 차단 (본 task 책임 아님) waiting_chair_decision : 방향성/정책 결정을 회장이 대기 forbidden_violations : 금지 규칙 위반 건수 (0이어야 MERGED_CLOSE_BLOCKED 적용) admin_override_used : admin override 사용 여부 (false이어야 함) branch_protection_bypass : branch protection bypass 여부 (false이어야 함) ci_fail_owner : INTERNAL | EXTERNAL_DEPENDENCY | PREEXISTING | None CI 실패 귀책 주체. None은 fail 없음 또는 미판별. done_marker_exists : .done 마커 파일 존재 여부 (default False) workspace_dirty_owner : internal | external | None 본 task workspace 자체 dirty 여부 귀책 주체. close_blocker_owner와 의미 중복 아님; 본 필드는 본 task workspace 자체 dirty 여부. extra : 확장용 자유 dict """ task_id: str pr_state: Optional[str] = None # OPEN | MERGED | CLOSED | None pr_merged_at: Optional[str] = None # ISO8601 or None ci_rollup: Optional[str] = None # PASS | FAIL | UNKNOWN qc_result: Optional[str] = None # PASS | WARN | FAIL close_lifecycle_state: Optional[str] = None # CLEAN | FAIL | NOT_RUN close_blocker_owner: Optional[str] = None # internal | external | None essence_verdict: Optional[str] = None # PASS | FAIL | UNKNOWN retry_count: int = 0 retry_max: int = 3 dependency_task_id: Optional[str] = None # 후속/의존 task id dogfooding_external_dependency: bool = False # BOT 토큰/ruleset/approval 차단 blocked_external_system: bool = False # API/infra 차단 failed_preexisting: bool = False # 기존 결함이 본 task 진행 차단 waiting_chair_decision: bool = False # 회장 정책 결정 대기 forbidden_violations: int = 0 admin_override_used: bool = False branch_protection_bypass: bool = False ci_fail_owner: Optional[str] = None # INTERNAL | EXTERNAL_DEPENDENCY | PREEXISTING | None done_marker_exists: bool = False # .done 마커 존재 여부 workspace_dirty_owner: Optional[str] = None # internal | external | None (close_blocker_owner와 의미 중복 아님; 본 필드는 본 task workspace 자체 dirty 여부) extra: dict = field(default_factory=dict) # --------------------------------------------------------------------------- # Output dataclass # --------------------------------------------------------------------------- @dataclass class ClassificationResult: """classify() 반환값. 이 오브젝트만 반환; 마커 파일 생성·dispatch 호출 없음. marker_file : (dry-run 참고용) 실제 운영에서 발행될 마커 파일명. 본 POC에서는 생성 안 함. preserve_markers: 동시 보존해야 할 기존 마커 (예: MERGED_CLOSE_BLOCKED_EXTERNAL → .done, .escalate) """ classification: TerminationClassification confidence: float # 0.0 ~ 1.0 matched_rule: str # 룰 식별자 ("R1_DONE", "R2_MERGED_CLOSE_BLOCKED", ...) followup_conditions: list[str] = field(default_factory=list) notes: list[str] = field(default_factory=list) marker_file: Optional[str] = None preserve_markers: list[str] = field(default_factory=list) # --------------------------------------------------------------------------- # Individual rule functions (우선순위: R1 > R2 > ... > R8 > R9_UNCLASSIFIED) # --------------------------------------------------------------------------- def _rule_done(ev: TaskEvidence) -> Optional[ClassificationResult]: """R1: 본질 PASS + merge + lifecycle close 모두 완료 → DONE (성공 종료). 근거: phase_b_integration_items 섹션 2.1 규칙 1. 조건: PR MERGED + pr_merged_at not null + close_lifecycle_state CLEAN + essence_verdict PASS + forbidden_violations == 0. """ if ( ev.pr_state == "MERGED" and ev.pr_merged_at is not None # ★ 추가 and ev.close_lifecycle_state == "CLEAN" and ev.essence_verdict == "PASS" # ★ 추가 and ev.forbidden_violations == 0 # ★ 추가 ): return ClassificationResult( classification=TerminationClassification.DONE, confidence=1.0, matched_rule="R1_DONE", marker_file=".done", ) return None def _rule_merged_close_blocked_external(ev: TaskEvidence) -> Optional[ClassificationResult]: """R2: 머지 완료 + post-merge close lifecycle 외부 차단 → MERGED_CLOSE_BLOCKED_EXTERNAL. 근거: feedback_merged_close_blocked_external 섹션 4 (5항목 자동 감지). 조건: - PR MERGED (mergedAt not null) - admin override 0, branch protection bypass 0, forbidden 0 - close lifecycle FAIL - 차단 사유 외부 workspace dirty (close_blocker_owner == "external") 핵심: 이미 머지됐으므로 FAILED 계열로 분류하면 부정확. """ if ( ev.pr_state == "MERGED" and ev.close_lifecycle_state == "FAIL" and ev.close_blocker_owner == "external" and ev.forbidden_violations == 0 and not ev.admin_override_used and not ev.branch_protection_bypass ): return ClassificationResult( classification=TerminationClassification.MERGED_CLOSE_BLOCKED_EXTERNAL, confidence=1.0, matched_rule="R2_MERGED_CLOSE_BLOCKED_EXTERNAL", marker_file=".close-blocked-external", preserve_markers=[".done", ".escalate"], followup_conditions=[ "외부 workspace dirty 정리 task 발행 (타 task 영역 / 운영 복사본 / 시스템 활동 파일)", "정리 완료 후 finish-task.sh 재실행", "성공 시 .close-blocked-external.resolved 추가 + final_state_resolved=true 박제", ], notes=[ "lifecycle 마커 3개 동시 보존 필요: .done / .escalate / .close-blocked-external", "taskctl 최종 상태 = MERGED_CLOSE_BLOCKED_EXTERNAL (DONE 아님)", ], ) return None def _rule_dogfooding_pending(ev: TaskEvidence) -> Optional[ClassificationResult]: """R3: 본질 PASS + CI PASS + dogfooding layer 외부 의존 미완 → DOGFOODING_PENDING. 근거: feedback_dogfooding_pending 섹션 '발동 조건' (4항목). 조건: - PR MERGED - essence_verdict PASS - ci_rollup PASS - qc_result PASS 또는 WARN (FAIL 아님) - dogfooding_external_dependency True (BOT 토큰 / ruleset / approval) 주의: ESCALATED는 본질 실패 의미; 본질 PASS인데 외부 의존으로 dogfooding 못 끝낸 경우 ESCALATED로 박제하면 retry 룰이 본질 재작업 시도 → 무의미. """ if ( ev.pr_state == "MERGED" and ev.essence_verdict == "PASS" and ev.ci_rollup == "PASS" and ev.qc_result in ("PASS", "WARN") and ev.dogfooding_external_dependency ): return ClassificationResult( classification=TerminationClassification.DOGFOODING_PENDING, confidence=0.95, matched_rule="R3_DOGFOODING_PENDING", marker_file=".dogfooding-pending", followup_conditions=[ "외부 의존 항목 복구 (BOT_GITHUB_TOKEN 갱신 / ruleset 임시 해제 / approval 정책 조정)", "bot-authored PR 재발행 또는 handoff 경로 재시도", "no-admin enqueue-merge 성공", "layer 5 dogfooding evidence 확보 (audit jsonl 박제)", ], notes=[ "수동 사람 approve 우회 금지 (self-approval 정책 위반)", "admin override 사용 금지", "4가지 후속 조건 모두 충족 전까지 DONE 금지", ], ) return None def _rule_merge_pending_dependency(ev: TaskEvidence) -> Optional[ClassificationResult]: """R4: 본질 PASS + PR OPEN + 후속 task 의존으로 merge 자체가 차단 → MERGE_PENDING_DEPENDENCY. 근거: feedback_merge_pending_dependency 섹션 5 (6항목 자동 감지). 조건: - PR OPEN (mergedAt null) - essence_verdict PASS (본질 결함 0) - dependency_task_id 명시 (후속/의존 task 명확히 추적 가능) - done_marker_exists False (.done 미발행 상태) - ci_fail_owner None 또는 EXTERNAL_DEPENDENCY (내부 결함 아님) 주의: 실패가 아니라 조건부 대기 상태; retry 룰 적용 시 본질 재작업 → 무의미. """ if ( ev.pr_state == "OPEN" and ev.pr_merged_at is None # ★ 추가 and ev.essence_verdict == "PASS" and ev.dependency_task_id is not None and not ev.done_marker_exists # ★ 추가 and (ev.ci_fail_owner in (None, "EXTERNAL_DEPENDENCY")) # ★ 추가 ): return ClassificationResult( classification=TerminationClassification.MERGE_PENDING_DEPENDENCY, confidence=0.95, matched_rule="R4_MERGE_PENDING_DEPENDENCY", marker_file=".merge-pending", followup_conditions=[ f"의존 task {ev.dependency_task_id} DONE 대기", f"의존 task {ev.dependency_task_id} DONE 후 본 PR CI rerun", "guard PASS 확인", "PR MERGED", "lifecycle close 진입", ".merge-pending → .done 변환", ], notes=[ "admin override / force merge 금지", "본질 코드 추가 수정 금지 (이미 PASS)", "후속 6조건 미충족 상태에서 .done 발행 금지", ], ) return None def _rule_blocked_by_external_dependency(ev: TaskEvidence) -> Optional[ClassificationResult]: """R5: 외부 시스템(API/infra/ruleset)이 본질 진행 자체를 차단 → BLOCKED_BY_EXTERNAL_DEPENDENCY. 근거: phase_b_integration_items 섹션 2.1 규칙 5. 조건: - blocked_external_system True - essence_verdict None 또는 UNKNOWN (본질 진행 시작조차 못한 상태) - pr_state None 또는 OPEN (이미 MERGED면 해당 없음) """ if ( ev.blocked_external_system and ev.essence_verdict in (None, "UNKNOWN") and ev.pr_state in (None, "OPEN") # ★ 추가 ): return ClassificationResult( classification=TerminationClassification.BLOCKED_BY_EXTERNAL_DEPENDENCY, confidence=0.85, matched_rule="R5_BLOCKED_BY_EXTERNAL_DEPENDENCY", marker_file=".blocked-external", followup_conditions=[ "외부 시스템 복구 확인 (API 다운 / infra 장애 / ruleset 차단 해소)", "재시도 가능 여부 검증", "lifecycle 진입", ], ) return None def _rule_failed_preexisting(ev: TaskEvidence) -> Optional[ClassificationResult]: """R6: 기존 결함이 본 task 진행을 차단 — 본 task 책임 아님 → FAILED_PREEXISTING. 근거: phase_b_integration_items 섹션 2.1 규칙 6. 조건: - failed_preexisting True (예: main lint/test 기존 실패가 본 task CI를 차단) - pr_state None 또는 OPEN (이미 MERGED면 해당 없음) - ci_fail_owner PREEXISTING (CI 실패 귀책이 기존 결함임이 명시) """ if ( ev.failed_preexisting and ev.pr_state in (None, "OPEN") # ★ 추가 and ev.ci_fail_owner == "PREEXISTING" # ★ 추가 ): return ClassificationResult( classification=TerminationClassification.FAILED_PREEXISTING, confidence=0.9, matched_rule="R6_FAILED_PREEXISTING", marker_file=".failed-preexisting", followup_conditions=[ "기존 결함 별도 task 발행 (본 task 범위 외 수정 금지)", "기존 결함 수정 task DONE 후 본 task 영향 확인", "본 task 재시도 가능 여부 재평가", ], ) return None def _rule_waiting_chair(ev: TaskEvidence) -> Optional[ClassificationResult]: """R7: 방향성/우선순위/예외 승인을 회장이 결정 대기 → WAITING_FOR_CHAIR_DECISION. 근거: phase_b_integration_items 섹션 2.1 규칙 7. 조건: waiting_chair_decision True (정책 충돌, 사용자 영향 큰 변경 등). """ if ev.waiting_chair_decision: return ClassificationResult( classification=TerminationClassification.WAITING_FOR_CHAIR_DECISION, confidence=1.0, matched_rule="R7_WAITING_FOR_CHAIR_DECISION", marker_file=".waiting-chair", followup_conditions=["회장 결정 명시 후 분류 재산정"], ) return None def _rule_escalated(ev: TaskEvidence) -> Optional[ClassificationResult]: """R8: retry 초과 + 본질 결함 + 본질 재작업 필요 → ESCALATED (실패 종료). 근거: phase_b_integration_items 섹션 2.1 규칙 8. 조건: - retry_count >= retry_max - essence_verdict FAIL 주의: retry 초과라도 차단 사유가 외부이면 R2~R7에서 먼저 매칭됨. 이 룰은 실제로 본질 결함인 경우만 도달한다. """ if ev.retry_count >= ev.retry_max and ev.essence_verdict == "FAIL": return ClassificationResult( classification=TerminationClassification.ESCALATED, confidence=0.9, matched_rule="R8_ESCALATED", marker_file=".escalate", followup_conditions=[ "본질 재작업 필요 (essence 결함 근본 원인 분석)", "회장/아누 판단 대기", ], ) return None # --------------------------------------------------------------------------- # Rule registry (우선순위 순; 첫 매칭 반환) # --------------------------------------------------------------------------- _RULES = [ _rule_done, # R1 _rule_merged_close_blocked_external, # R2 _rule_dogfooding_pending, # R3 _rule_merge_pending_dependency, # R4 _rule_blocked_by_external_dependency, # R5 _rule_failed_preexisting, # R6 _rule_waiting_chair, # R7 _rule_escalated, # R8 ] # --------------------------------------------------------------------------- # Public API # --------------------------------------------------------------------------- def classify(evidence: TaskEvidence) -> ClassificationResult: """입력 evidence를 8종 룰 트리에 순서대로 매칭하여 최초 매칭 결과 반환. 마커 파일 생성 / dispatch 호출 / 상태 변경 없음 — dry-run only. 어떤 룰도 매칭되지 않으면 UNCLASSIFIED 반환 (회장 알림 + 수동 분류 필요). Args: evidence: TaskEvidence 인스턴스 (필드 기본값 = 안전) Returns: ClassificationResult (dataclass, 읽기 전용 활용 권장) """ for rule in _RULES: result = rule(evidence) if result is not None: return result return ClassificationResult( classification=TerminationClassification.UNCLASSIFIED, confidence=0.0, matched_rule="R9_UNCLASSIFIED", marker_file=None, notes=["룰 미해당. 회장 알림 + 수동 분류 필요."], ) # --------------------------------------------------------------------------- # CLI (dry-run; 마커 파일 생성 없음) # --------------------------------------------------------------------------- def _main() -> int: """fixture JSON → TaskEvidence → classify() → JSON 출력 (dry-run CLI). 실행 예시: python3 tools/poc/termination_classifier.py fixture.json """ import argparse import json import sys parser = argparse.ArgumentParser( prog="termination_classifier", description=( "Phase B termination classifier (DRY-RUN only). " "마커 파일 생성 / dispatch 호출 없음." ), ) parser.add_argument( "fixture", help="fixture JSON 파일 경로 (TaskEvidence 필드 key-value 매핑)", ) args = parser.parse_args() try: with open(args.fixture, encoding="utf-8") as f: data = json.load(f) except FileNotFoundError: print(f"[ERROR] fixture 파일을 찾을 수 없음: {args.fixture}", file=sys.stderr) return 1 except json.JSONDecodeError as exc: print(f"[ERROR] JSON 파싱 실패: {exc}", file=sys.stderr) return 1 known_fields = set(TaskEvidence.__dataclass_fields__.keys()) filtered = {k: v for k, v in data.items() if k in known_fields} unknown = set(data.keys()) - known_fields if unknown: print(f"[WARN] 알 수 없는 필드 무시: {sorted(unknown)}", file=sys.stderr) evidence = TaskEvidence(**filtered) result = classify(evidence) output = { "task_id": evidence.task_id, "classification": result.classification.value, "confidence": result.confidence, "matched_rule": result.matched_rule, "marker_file": result.marker_file, "preserve_markers": result.preserve_markers, "followup_conditions": result.followup_conditions, "notes": result.notes, "dry_run": True, } print(json.dumps(output, ensure_ascii=False, indent=2)) return 0 if __name__ == "__main__": raise SystemExit(_main())