"""5-enum status 판정기. 회장 verbatim 필수 구현 9.status_enum_separation_5: - SPAWN_VERIFIED (worktree + executor process + first response NOT-refusal 3 신호) - SPAWN_PENDING (fire 후 짧은 시간 · 부분 신호만) - SPAWN_VISIBILITY_GAP (일부 source 부재 + 다른 source 있음) - CALLBACK_RECOVERED_AFTER_VISIBILITY_GAP (ANU 초기 ABSENT 후 result/report/done 또는 callback envelope 도착 · 2 source 이상 교차) - TRUE_SILENT_DROP (전 source 부재 + 30분 경과 + ANCHOR-3 예외 0) ANCHOR-2: self-attestation 단독 금지 · 최소 2 source 교차. ANCHOR-4: worktree 한쪽 missing 만으로 silent drop 단정 금지. ANCHOR-5: CALLBACK_RECOVERED_AFTER_VISIBILITY_GAP 분리 명시 (task-2657 WARN 1 재발 방지). """ from __future__ import annotations from dataclasses import dataclass, field from enum import Enum from typing import Optional from .source_collector import SourceSnapshot from .timeout_gate import ( HARD_TIMEOUT_SECONDS, TimeoutDecision, evaluate_timeout_gate, ) class SpawnVisibilityStatus(str, Enum): SPAWN_VERIFIED = "SPAWN_VERIFIED" SPAWN_PENDING = "SPAWN_PENDING" SPAWN_VISIBILITY_GAP = "SPAWN_VISIBILITY_GAP" CALLBACK_RECOVERED_AFTER_VISIBILITY_GAP = "CALLBACK_RECOVERED_AFTER_VISIBILITY_GAP" TRUE_SILENT_DROP = "TRUE_SILENT_DROP" @dataclass(frozen=True) class StatusDecision: status: SpawnVisibilityStatus reason: str positive_sources: tuple[str, ...] callback_evidence_sources: tuple[str, ...] crossed_sources_count: int timeout: TimeoutDecision initial_anu_absent_observed: bool = False first_response_not_refusal: Optional[bool] = None notes: tuple[str, ...] = field(default_factory=tuple) def classify_spawn_visibility( sources: SourceSnapshot, *, elapsed_since_fire_seconds: float, initial_anu_absent_observed: bool = False, first_response_not_refusal: Optional[bool] = None, hard_timeout_seconds: int = HARD_TIMEOUT_SECONDS, ) -> StatusDecision: """5-enum 분류 수행. Parameters ---------- sources: SourceSnapshot elapsed_since_fire_seconds: dispatch fire 이후 경과 초. initial_anu_absent_observed: ANU 가 사전에 spawn ABSENT 로 측정한 적이 있는가? True 일 때 callback recovery 라벨로 승격 가능. first_response_not_refusal: SPAWN_VERIFIED 3 신호 중 'first response NOT-refusal' 신호. 모르면 None. hard_timeout_seconds: TRUE_SILENT_DROP hard timeout (default 30분). """ timeout = evaluate_timeout_gate( sources, elapsed_seconds=elapsed_since_fire_seconds, hard_timeout_seconds=hard_timeout_seconds, ) positives = sources.positive_sources() callback_evidence = sources.callback_evidence_sources() notes: list[str] = [] worktree_visible = ( sources.legacy_worktree_present or sources.cokacdir_worktree_present ) # ── Rule 1. TRUE_SILENT_DROP (전 source 부재 + timeout + 예외 0) ── if ( timeout.silent_drop_eligible and not positives and not callback_evidence ): return StatusDecision( status=SpawnVisibilityStatus.TRUE_SILENT_DROP, reason=( "전 source 부재 + " f"elapsed {elapsed_since_fire_seconds:.0f}s ≥ " f"hard_timeout {hard_timeout_seconds}s + ANCHOR-3 예외 0" ), positive_sources=positives, callback_evidence_sources=callback_evidence, crossed_sources_count=0, timeout=timeout, initial_anu_absent_observed=initial_anu_absent_observed, first_response_not_refusal=first_response_not_refusal, notes=tuple(notes), ) # ── Rule 2. CALLBACK_RECOVERED_AFTER_VISIBILITY_GAP ── # (executor process 부재 OR ANU 초기 ABSENT 관측) 이후 # callback evidence ≥ 2 source 교차 → recovery 라벨. # ANCHOR-2: self-attestation 단독 금지 → 최소 2 source 교차 필수. if ( len(callback_evidence) >= 2 and ( initial_anu_absent_observed or not sources.executor_process_present ) ): return StatusDecision( status=SpawnVisibilityStatus.CALLBACK_RECOVERED_AFTER_VISIBILITY_GAP, reason=( "callback evidence ≥ 2 source 교차 (" + ", ".join(callback_evidence) + ")" + ( " · executor process 부재" if not sources.executor_process_present else "" ) + ( " · initial ANU ABSENT 관측" if initial_anu_absent_observed else "" ) ), positive_sources=positives, callback_evidence_sources=callback_evidence, crossed_sources_count=len(callback_evidence), timeout=timeout, initial_anu_absent_observed=initial_anu_absent_observed, first_response_not_refusal=first_response_not_refusal, notes=tuple(notes), ) # ── Rule 3. SPAWN_VERIFIED (3 신호 강결합) ── three_signal = ( worktree_visible and sources.executor_process_present and first_response_not_refusal is True ) if three_signal: return StatusDecision( status=SpawnVisibilityStatus.SPAWN_VERIFIED, reason=( "3 신호 충족: worktree(legacy=" f"{sources.legacy_worktree_present}, " f"cokacdir={sources.cokacdir_worktree_present}) " "+ executor process + first_response NOT-refusal" ), positive_sources=positives, callback_evidence_sources=callback_evidence, crossed_sources_count=len(positives), timeout=timeout, initial_anu_absent_observed=initial_anu_absent_observed, first_response_not_refusal=first_response_not_refusal, notes=tuple(notes), ) # ── Rule 3b. SPAWN_VERIFIED 완화 (regression R1~R3 호환) ── # first_response 가 미지정(None) 이고 worktree + executor process 가 # 모두 양성이면 verified 로 라벨링한다. first_response 가 False 로 # 명시되었으면 verified 아닌 PENDING/GAP 로 떨어진다. if ( worktree_visible and sources.executor_process_present and first_response_not_refusal is None ): notes.append("first_response_not_refusal unspecified · worktree+process 만으로 verified 분류") return StatusDecision( status=SpawnVisibilityStatus.SPAWN_VERIFIED, reason=( "worktree 가시 + executor process 양성 " "(first_response_not_refusal 미관측 — 2 신호 교차 기준 verified)" ), positive_sources=positives, callback_evidence_sources=callback_evidence, crossed_sources_count=len(positives), timeout=timeout, initial_anu_absent_observed=initial_anu_absent_observed, first_response_not_refusal=first_response_not_refusal, notes=tuple(notes), ) # ── Rule 4. SPAWN_PENDING ── # timeout 미경과 + (schedule_history pending OR 양성 source 1+개 이상 # 이지만 visibility 신호 불완전). schedule_pending = ( isinstance(sources.schedule_history_last_status, str) and sources.schedule_history_last_status.strip().lower() in {"pending", "running", "in_progress", "started", "fired"} ) if (not timeout.silent_drop_eligible) and ( schedule_pending or positives ): # SPAWN_VISIBILITY_GAP 와의 구분: # - 일부 source 부재 + 다른 source 있음 → VISIBILITY_GAP # - 시간 짧음 + 부분 신호만 → PENDING # 우선순위: schedule pending 명시이거나 timeout 잔여시간 ≥50% 이면 PENDING. if ( schedule_pending or elapsed_since_fire_seconds < hard_timeout_seconds * 0.5 ): return StatusDecision( status=SpawnVisibilityStatus.SPAWN_PENDING, reason=( "fire 후 짧은 시간 + 부분 신호 — " + ( f"schedule_history_last_status={sources.schedule_history_last_status!r}" if schedule_pending else f"elapsed {elapsed_since_fire_seconds:.0f}s " f"< hard_timeout/2 ({hard_timeout_seconds // 2}s)" ) ), positive_sources=positives, callback_evidence_sources=callback_evidence, crossed_sources_count=len(positives), timeout=timeout, initial_anu_absent_observed=initial_anu_absent_observed, first_response_not_refusal=first_response_not_refusal, notes=tuple(notes), ) # 그 외 (timeout 미경과지만 절반 이상 경과 + 일부 source 부재) return StatusDecision( status=SpawnVisibilityStatus.SPAWN_VISIBILITY_GAP, reason=( "일부 source 부재 + 다른 source 양성 " + f"({len(positives)} positive · timeout 미경과)" ), positive_sources=positives, callback_evidence_sources=callback_evidence, crossed_sources_count=len(positives), timeout=timeout, initial_anu_absent_observed=initial_anu_absent_observed, first_response_not_refusal=first_response_not_refusal, notes=tuple(notes), ) # ── Rule 5. fallback SPAWN_VISIBILITY_GAP ── # 어디에도 해당하지 않으면 (e.g. timeout 경과 + callback evidence 1개 # 단독 — self-attestation 단독은 인정 금지) GAP 으로 라벨링. notes.append("fallback GAP · ANCHOR-2 self-attestation 단독 인정 금지") return StatusDecision( status=SpawnVisibilityStatus.SPAWN_VISIBILITY_GAP, reason=( "교차 검증 미달 (positive=" f"{len(positives)}, " f"callback_evidence={len(callback_evidence)}) — " "GAP 으로 라벨링 후 추가 관측 권고" ), positive_sources=positives, callback_evidence_sources=callback_evidence, crossed_sources_count=len(positives), timeout=timeout, initial_anu_absent_observed=initial_anu_absent_observed, first_response_not_refusal=first_response_not_refusal, notes=tuple(notes), )