"""Callback lifecycle classifier (evidence-only, pure function, decoupled). 회장 승인 2축(+root_cause 태그) schema 의 코드화 — CORE ONLY. evidence snapshot(dict) → (delivery_outcome, normal_callback_miss_cause, root_cause_tags, evidence_completeness) 를 결정적으로 판정한다. doctrines: - evidence-only decoupled: anu_v3 런타임 import 0 - pure function · read-only · 파일/네트워크/subprocess I/O 0 · no daemon - 추정 금지 → 증거 결핍 시 UNKNOWN(INSUFFICIENT_EVIDENCE) - CALLBACK_DELIVERY_GAP 은 residual only (이전 차단 원인 전무일 때만) - 축B precedence = 상태기계상 가장 이른 차단 전이 (스펙 §5) 단일소스 스펙: memory/specs/system_callback_lifecycle_state_schema_260522.md """ from utils.callback_lifecycle_states import ( # 축A NORMAL_CALLBACK_RECEIVED, FALLBACK_COLLECTOR_APPLIED, DUPLICATE_FALLBACK_NO_OP, MANUAL_ANU_REVERIFY, NOT_YET_COLLECTED, UNKNOWN_INSUFFICIENT_EVIDENCE, # 축B NONE, ENVELOPE_PREPARED_NOT_FIRED, FINISH_TASK_GIT_GATE_BLOCKED_BEFORE_CALLBACK, SELF_KEY_FAIL_CLOSED_BEFORE_FIRE, SELF_KEY_FIRED_NON_AUTHORITATIVE, CALLBACK_DELIVERY_GAP, CALLBACK_CONTRACT_VIOLATION, MISS_CAUSE_UNKNOWN, # 축C FOREIGN_DIRTY_BLOCKER, BOT_APP_TOKEN_ABSENT, REFLECTION_NOT_MERGED, GIT_GATE_SHARED_WORKSPACE_MISFIRE, SCOPE_GUARD_MAIN_HEAD_DIVERGENCE, ROOT_CAUSE_TAG_ORDER, # completeness / classification / helpers COMPLETE, PARTIAL, MISSING, INCIDENT, NORMAL, INCIDENT_MISS_CAUSES, DEFAULT_ANU_KEYS, is_anu_owner_key, CANONICAL_EVIDENCE_SOURCES, CORE_EVIDENCE_SOURCES, MAX_ENVELOPE_BYTES, ) 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] if not (CORE_EVIDENCE_SOURCES & set(present)): return MISSING, missing if not missing: return COMPLETE, missing return PARTIAL, missing def classify_callback_lifecycle(evidence, anu_keys=DEFAULT_ANU_KEYS): """evidence snapshot(dict) → 분류 결과(dict). 순수 함수.""" rj = evidence.get("result_json") or {} sh = evidence.get("schedule_history") or {} nc = sh.get("normal_callback") or {} esc = evidence.get("escalate") or {} col = evidence.get("collectors") or {} git = evidence.get("git") or {} applied_count = evidence.get("applied_count") or 0 completeness, missing = _evidence_completeness(evidence) reg_status = rj.get("callback_registration_status") self_collection = col.get("self_collection") or {} fallback_collector = col.get("fallback_collector") or {} manual_reverify = col.get("manual_anu_reverify") or {} normal_collection_ack = bool(col.get("normal_collection_ack")) fallback_present = bool(fallback_collector.get("present")) fallback_authoritative = bool(fallback_collector.get("authoritative")) manual_present = bool(manual_reverify.get("present")) authoritative_cron_collection = normal_collection_ack or (fallback_present and fallback_authoritative) nc_owner = nc.get("owner_key") nc_fired = bool(nc.get("fired")) nc_argv_present = bool(nc.get("argv_present")) normal_fired_with_self_key = ( nc_fired and nc_argv_present and nc_owner is not None and not is_anu_owner_key(nc_owner, anu_keys) ) anu_key_normal_fired = nc_fired and is_anu_owner_key(nc_owner, anu_keys) esc_reason = esc.get("reason") or "" git_gate_blocked = bool(git.get("git_gate_blocked")) or ("git_gate_blocked" in esc_reason) foreign_dirty = git.get("foreign_dirty") or [] git_gate_checked_scope = git.get("git_gate_checked_scope") per_task_scope_clean = bool(git.get("per_task_scope_clean")) scope_guard_blocked = bool(git.get("scope_guard_blocked")) scope_guard_basis = git.get("scope_guard_basis") envelope_utf8_bytes = rj.get("envelope_utf8_bytes") contract_violation = bool(rj.get("contract_violation")) or ( envelope_utf8_bytes is not None and envelope_utf8_bytes > MAX_ENVELOPE_BYTES ) self_key_fail_closed = bool(evidence.get("self_key_fail_closed")) bot_token_absent = (evidence.get("bot_app_token_present") is False) reflection_not_merged = (evidence.get("reflection_merged") is False) # ===== 축A: delivery_outcome ===== if completeness == MISSING: delivery_outcome = UNKNOWN_INSUFFICIENT_EVIDENCE elif fallback_present and normal_collection_ack and applied_count > 1: delivery_outcome = DUPLICATE_FALLBACK_NO_OP elif normal_collection_ack: delivery_outcome = NORMAL_CALLBACK_RECEIVED elif fallback_present and (not normal_collection_ack) and applied_count == 1: delivery_outcome = FALLBACK_COLLECTOR_APPLIED elif manual_present and not authoritative_cron_collection: delivery_outcome = MANUAL_ANU_REVERIFY elif _src_present(evidence, "result_json") or reg_status is not None or nc: delivery_outcome = NOT_YET_COLLECTED else: delivery_outcome = UNKNOWN_INSUFFICIENT_EVIDENCE # ===== 축B: normal_callback_miss_cause (상태기계 precedence 순) ===== if completeness == MISSING: miss_cause = MISS_CAUSE_UNKNOWN elif delivery_outcome == NORMAL_CALLBACK_RECEIVED: miss_cause = NONE elif git_gate_blocked: miss_cause = FINISH_TASK_GIT_GATE_BLOCKED_BEFORE_CALLBACK elif contract_violation: miss_cause = CALLBACK_CONTRACT_VIOLATION elif normal_fired_with_self_key: miss_cause = SELF_KEY_FIRED_NON_AUTHORITATIVE elif reg_status == ENVELOPE_PREPARED_NOT_FIRED: reasons = [] if self_key_fail_closed: reasons.append(SELF_KEY_FAIL_CLOSED_BEFORE_FIRE) if bot_token_absent: reasons.append(BOT_APP_TOKEN_ABSENT) if reflection_not_merged: reasons.append(REFLECTION_NOT_MERGED) if len(reasons) > 1: miss_cause = ENVELOPE_PREPARED_NOT_FIRED # umbrella (§3.1) elif reasons == [SELF_KEY_FAIL_CLOSED_BEFORE_FIRE]: miss_cause = SELF_KEY_FAIL_CLOSED_BEFORE_FIRE # self-key 단독 else: miss_cause = ENVELOPE_PREPARED_NOT_FIRED elif self_key_fail_closed: miss_cause = SELF_KEY_FAIL_CLOSED_BEFORE_FIRE elif anu_key_normal_fired and not authoritative_cron_collection: # residual: 여기 도달 = 이전 모든 차단 전이(git-gate/contract/self-key/envelope-not-fired) 부재 miss_cause = CALLBACK_DELIVERY_GAP elif completeness == COMPLETE: miss_cause = ENVELOPE_PREPARED_NOT_FIRED # 보수적 기본 else: miss_cause = MISS_CAUSE_UNKNOWN # ===== 축C: root_cause_tags (축B 분기로 gating) ===== raw_tags = set() if miss_cause == FINISH_TASK_GIT_GATE_BLOCKED_BEFORE_CALLBACK: if foreign_dirty: raw_tags.add(FOREIGN_DIRTY_BLOCKER) if git_gate_checked_scope == "main_workspace" and per_task_scope_clean: raw_tags.add(GIT_GATE_SHARED_WORKSPACE_MISFIRE) if scope_guard_blocked and scope_guard_basis == "main_head": raw_tags.add(SCOPE_GUARD_MAIN_HEAD_DIVERGENCE) elif miss_cause == SELF_KEY_FIRED_NON_AUTHORITATIVE: if reflection_not_merged: raw_tags.add(REFLECTION_NOT_MERGED) # token 부재는 self-key 가 실제 발사됐으므로 moot elif miss_cause == ENVELOPE_PREPARED_NOT_FIRED: if self_key_fail_closed: raw_tags.add(SELF_KEY_FAIL_CLOSED_BEFORE_FIRE) if bot_token_absent: raw_tags.add(BOT_APP_TOKEN_ABSENT) if reflection_not_merged: raw_tags.add(REFLECTION_NOT_MERGED) # SELF_KEY_FAIL_CLOSED_BEFORE_FIRE(단독), CALLBACK_DELIVERY_GAP 등은 # 추가 root_cause 태그 없음(원인이 축B 값 자체). CALLBACK_ENVELOPE_ONLY 는 # 관측 태그이며 회장 exact 매핑 보존을 위해 자동 방출하지 않는다. root_cause_tags = [t for t in ROOT_CAUSE_TAG_ORDER if t in raw_tags] classification = INCIDENT if miss_cause in INCIDENT_MISS_CAUSES else NORMAL lifecycle_state_evidence = { "callback_registration_status": reg_status, "normal_callback_owner_key_is_anu": (is_anu_owner_key(nc_owner, anu_keys) if nc_owner else None), "normal_callback_fired": nc_fired, "normal_fired_with_self_key": normal_fired_with_self_key, "anu_key_normal_fired": anu_key_normal_fired, "git_gate_blocked": git_gate_blocked, "foreign_dirty_count": len(foreign_dirty), "authoritative_cron_collection": authoritative_cron_collection, "fallback_collector_present": fallback_present, "manual_anu_reverify_present": manual_present, "self_collection_authoritative": bool(self_collection.get("authoritative")), "self_key_fail_closed": self_key_fail_closed, "reflection_merged": evidence.get("reflection_merged"), "bot_app_token_present": evidence.get("bot_app_token_present"), "contract_violation": contract_violation, "applied_count": applied_count, "escalate_reason": esc.get("reason"), } return { "delivery_outcome": delivery_outcome, "normal_callback_miss_cause": miss_cause, "root_cause_tags": root_cause_tags, "evidence_completeness": completeness, "missing_evidence_sources": missing, "classification": classification, "lifecycle_state_evidence": lifecycle_state_evidence, "classified_by": "auto-classifier", }