# -*- coding: utf-8 -*-
"""tests/regression/test_ci_watcher_lifecycle_2729p0a.py — task-2729+1 P0-A 회귀.

회귀 보호 시나리오:
  (a) watcher stale + PR pending → CI_WATCHER_SESSION_LIFETIME_GAP 분류
  (b) terminal 도달 → callback 필수, 미발사 시 WATCHER_TERMINAL_CALLBACK_NOT_WIRED
  (c) quiet-window settled 판정 (review-settle 골격)
  (d) fallback = dead-man safety-net only (final-report trigger 아님)
  (e) 기존 6-state/terminal 회귀 무손상 (heartbeat 필드 추가가 6-state dict 오염 없음)

대상 모듈(progress_watcher_gate.py, ci_watch_handoff_runner.py,
pr_watcher_terminal_state_classifier.py, normal_fallback_callback_helper.py)은
수정하지 않고 실제 동작에 맞춰 테스트만 작성한다.
외부 네트워크 / 실제 cokacdir / gh 호출 없음.
"""
from __future__ import annotations

import importlib.util
import pathlib
import sys
import types

# ---------------------------------------------------------------------------
# worktree root sys.path 보강.
#   ★ tests/conftest.py 가 비-worktree 풀체크아웃(/home/jay/workspace)을 sys.path[0]
#     에 주입한다. 그 dispatch/utils 패키지에는 task-2729 신규 모듈
#     (dispatch.progress_watcher_gate)이 없어 ModuleNotFoundError 가 발생한다.
#     → 이 sparse worktree root 를 sys.path 최상단으로 끌어올리고, 이미 캐시된
#       stale dispatch*/utils* 모듈을 제거하여 worktree 사본이 로드되도록 강제한다.
#     (대상 모듈은 수정하지 않고 import 해석만 worktree 로 고정.)
# ---------------------------------------------------------------------------
_ROOT = pathlib.Path(__file__).resolve().parents[2]
_ROOT_STR = str(_ROOT)
# 항상 worktree root 를 최상단에 둔다(중복 제거 후 0번 삽입).
while _ROOT_STR in sys.path:
    sys.path.remove(_ROOT_STR)
sys.path.insert(0, _ROOT_STR)

# worktree 외부에서 먼저 로드된 dispatch/utils 패키지 캐시를 제거(신규 모듈 누락분 회피).
for _name in list(sys.modules):
    if _name == "dispatch" or _name.startswith("dispatch."):
        _mod = sys.modules[_name]
        _f = getattr(_mod, "__file__", None) or ""
        if not _f.startswith(_ROOT_STR):
            del sys.modules[_name]

# ---------------------------------------------------------------------------
# 대상 모듈 import
# ---------------------------------------------------------------------------
from dispatch.progress_watcher_gate import (  # noqa: E402
    DISPATCH_INCOMPLETE,
    WATCHER_TERMINAL_CALLBACK_NOT_WIRED,
    WATCHER_TRACKED_STATES,
    evaluate_progress_watcher_gate,
    watcher_terminal_callback_status,
    all_states_tracked,
)
from utils.pr_watcher_terminal_state_classifier import (  # noqa: E402
    PRSnapshot,
    TERMINAL_STATES,
    CI_WATCHER_SESSION_LIFETIME_GAP,
    LIFETIME_GAP_DIAGNOSTIC_STATES,
    WATCHER_HEARTBEAT_STALE_THRESHOLD_SEC,
    classify_lifetime_gap,
)

# ---------------------------------------------------------------------------
# scripts/ci_watch_handoff_runner.py 는 일반 패키지가 아닐 수 있으므로
# importlib.util.spec_from_file_location 로 로드한다.
# (dataclass 평가가 sys.modules 등록에 의존하므로 exec_module 전에 등록)
# ---------------------------------------------------------------------------
_RUNNER_PATH = _ROOT / "scripts" / "ci_watch_handoff_runner.py"
_spec = importlib.util.spec_from_file_location("ci_watch_handoff_runner", _RUNNER_PATH)
assert _spec is not None and _spec.loader is not None
runner_mod = importlib.util.module_from_spec(_spec)
sys.modules["ci_watch_handoff_runner"] = runner_mod
_spec.loader.exec_module(runner_mod)


# ---------------------------------------------------------------------------
# fake runner (subprocess.run 호환 시그니처)
# register_terminal_callback 은 runner(cmd, capture_output=True, text=True,
# timeout=..., check=False) 형태로 호출하고 returncode==0 일 때 fired=True.
# ---------------------------------------------------------------------------
def _make_fake_runner(returncode: int):
    def _runner(cmd, capture_output=True, text=True, timeout=None, check=False, **kwargs):
        return types.SimpleNamespace(
            returncode=returncode,
            stdout="ok" if returncode == 0 else "",
            stderr="" if returncode == 0 else "boom",
        )

    return _runner


# ===========================================================================
# (a) watcher stale + PR pending → CI_WATCHER_SESSION_LIFETIME_GAP
# ===========================================================================

def test_a_classify_lifetime_gap_stale_pending():
    """stale=True + pr 비-terminal → CI_WATCHER_SESSION_LIFETIME_GAP."""
    state_str, reason = classify_lifetime_gap(watcher_stale=True, pr_terminal_state="")
    assert state_str == CI_WATCHER_SESSION_LIFETIME_GAP
    assert reason  # 비어있지 않음


def test_a_is_watcher_stale_none_last_poll():
    """last_poll_ts=None → stale=True (heartbeat 한 번도 없음)."""
    assert runner_mod.is_watcher_stale(last_poll_ts=None, now_ts=100.0) is True


def test_a_is_watcher_stale_at_threshold():
    """now_ts - last_poll_ts >= threshold → stale=True."""
    assert runner_mod.is_watcher_stale(
        last_poll_ts=0.0, now_ts=3600.0, stale_threshold_sec=3600
    ) is True


def test_a_is_watcher_stale_below_threshold():
    """now_ts - last_poll_ts < threshold → stale=False."""
    assert runner_mod.is_watcher_stale(
        last_poll_ts=0.0, now_ts=3599.0, stale_threshold_sec=3600
    ) is False


def test_a_classify_watcher_lifetime_gap_no_heartbeat():
    """WatcherState last_poll_ts=None → classify_watcher_lifetime_gap returns gap."""
    s = runner_mod.WatcherState()
    assert s.last_poll_ts is None
    out = runner_mod.classify_watcher_lifetime_gap(
        state=s, now_ts=100.0, pr_terminal_state=""
    )
    assert out["lifetime_gap_state"] == CI_WATCHER_SESSION_LIFETIME_GAP
    assert out["lifetime_gap_detected"] is True
    assert out["watcher_stale"] is True
    assert out["active"] is False


def test_a_classify_watcher_lifetime_gap_after_heartbeat():
    """record_poll_heartbeat 직후 → not stale → no lifetime gap."""
    s = runner_mod.WatcherState()
    runner_mod.record_poll_heartbeat(s, 100.0)
    out = runner_mod.classify_watcher_lifetime_gap(
        state=s, now_ts=120.0, pr_terminal_state="", stale_threshold_sec=3600
    )
    assert out["lifetime_gap_state"] == ""
    assert out["lifetime_gap_detected"] is False


def test_a_record_poll_heartbeat_sets_fields():
    """record_poll_heartbeat → state.last_poll_ts 와 state.heartbeat_ts 가 설정된다."""
    s = runner_mod.WatcherState()
    ret = runner_mod.record_poll_heartbeat(s, 999.0)
    assert ret is s
    assert s.last_poll_ts == 999.0
    assert s.heartbeat_ts == 999.0


def test_a_ci_watcher_session_lifetime_gap_not_in_terminal_states():
    """★ CI_WATCHER_SESSION_LIFETIME_GAP 는 TERMINAL_STATES 에 포함되지 않는다
    (diagnostic, non-terminal-merge 보장)."""
    assert CI_WATCHER_SESSION_LIFETIME_GAP not in TERMINAL_STATES


def test_a_lifetime_gap_diagnostic_states_contains_sentinel():
    """LIFETIME_GAP_DIAGNOSTIC_STATES 는 CI_WATCHER_SESSION_LIFETIME_GAP 를 포함한다."""
    assert CI_WATCHER_SESSION_LIFETIME_GAP in LIFETIME_GAP_DIAGNOSTIC_STATES


def test_a_classify_lifetime_gap_pr_terminal_no_gap():
    """★ PR 이 terminal(MERGE_READY) 이면 stale 여도 gap 이 아님."""
    state_str, reason = classify_lifetime_gap(
        watcher_stale=True, pr_terminal_state="MERGE_READY"
    )
    assert state_str == ""
    assert reason == "no_lifetime_gap"


def test_a_classify_lifetime_gap_not_stale_no_gap():
    """watcher_stale=False → 무조건 gap 없음."""
    state_str, reason = classify_lifetime_gap(watcher_stale=False, pr_terminal_state="")
    assert state_str == ""
    assert reason == "no_lifetime_gap"


def test_a_watcher_heartbeat_stale_threshold_value():
    """WATCHER_HEARTBEAT_STALE_THRESHOLD_SEC == 3600 (60분 convention)."""
    assert WATCHER_HEARTBEAT_STALE_THRESHOLD_SEC == 60 * 60


# ===========================================================================
# (b) terminal 도달 → callback 필수, 미발사 시 WATCHER_TERMINAL_CALLBACK_NOT_WIRED
# ===========================================================================

def test_b_fire_terminal_callback_success_runner():
    """returncode=0 → fired=True, callback_mandatory=True, not_wired=False."""
    state = runner_mod.WatcherState()
    assert state.normal_callback is False
    out = runner_mod.fire_terminal_callback(
        task_id="2729",
        pr_number=42,
        terminal_state=TERMINAL_STATES[0],  # MERGE_READY
        reason="test_b_wired",
        state=state,
        runner=_make_fake_runner(0),
    )
    assert out["fired"] is True
    assert out["callback_status"] == "WATCHER_TERMINAL_CALLBACK_WIRED"
    assert out["callback_mandatory"] is True
    assert out["not_wired"] is False
    assert state.normal_callback is True


def test_b_fire_terminal_callback_failure_runner():
    """returncode=1 → fired=False, callback_mandatory=True, not_wired=True, NOT_WIRED."""
    state = runner_mod.WatcherState()
    out = runner_mod.fire_terminal_callback(
        task_id="2729",
        pr_number=42,
        terminal_state="HOLD_FOR_CHAIR",
        reason="test_b_not_wired",
        state=state,
        runner=_make_fake_runner(1),
    )
    assert out["fired"] is False
    assert out["callback_status"] == WATCHER_TERMINAL_CALLBACK_NOT_WIRED
    assert out["callback_mandatory"] is True
    assert out["not_wired"] is True
    assert state.normal_callback is False


def test_b_watcher_terminal_callback_status_not_wired():
    """watcher_terminal_callback_status(terminal_reached=True, callback_fired=False)."""
    assert (
        watcher_terminal_callback_status(terminal_reached=True, callback_fired=False)
        == WATCHER_TERMINAL_CALLBACK_NOT_WIRED
    )


def test_b_fire_terminal_callback_returns_terminal_state():
    """fire_terminal_callback 반환 dict 에 terminal_state 키 포함."""
    state = runner_mod.WatcherState()
    out = runner_mod.fire_terminal_callback(
        task_id="2729",
        pr_number=42,
        terminal_state="LOOP_BOUNDARY",
        reason="test_b_ts",
        state=state,
        runner=_make_fake_runner(0),
    )
    assert out["terminal_state"] == "LOOP_BOUNDARY"


# ===========================================================================
# (c) quiet-window settled 판정
# ===========================================================================

def test_c_quiet_window_settled():
    """now_sec - last_event_sec >= quiet_window_sec → True."""
    assert (
        runner_mod.is_quiet_window_settled(
            last_event_sec=0, now_sec=120, quiet_window_sec=120
        )
        is True
    )


def test_c_quiet_window_not_settled():
    """now_sec - last_event_sec < quiet_window_sec → False."""
    assert (
        runner_mod.is_quiet_window_settled(
            last_event_sec=0, now_sec=119, quiet_window_sec=120
        )
        is False
    )


def test_c_review_settle_quiet_window_constant():
    """REVIEW_SETTLE_QUIET_WINDOW_SEC == 120."""
    assert runner_mod.REVIEW_SETTLE_QUIET_WINDOW_SEC == 120


# ===========================================================================
# (d) fallback = dead-man safety-net only (final-report trigger 아님)
# ===========================================================================

def test_d_fallback_prune_does_not_set_normal_callback():
    """mark_fallback_prune → state.fallback_prune==True 이지만 state.normal_callback==False."""
    state = runner_mod.WatcherState()
    runner_mod.mark_fallback_prune(state)
    assert state.fallback_prune is True
    assert state.normal_callback is False  # fallback 은 normal callback 트리거 아님


def test_d_fallback_only_gate_incomplete():
    """fallback만 등록 시 evaluate_progress_watcher_gate → fallback_only=True, DISPATCH_INCOMPLETE."""
    gr = evaluate_progress_watcher_gate(watcher_registered=False, fallback_registered=True)
    assert gr.incomplete is True
    assert gr.fallback_only is True
    assert gr.dispatch_status == DISPATCH_INCOMPLETE


def test_d_fallback_role_single_purpose_doctrine():
    """FALLBACK_ROLE_SINGLE_PURPOSE == 'RECOVERY_ONLY_NO_FINAL_REPORT_TRIGGER' doctrine 상수 검증."""
    from dispatch.normal_fallback_callback_helper import FALLBACK_ROLE_SINGLE_PURPOSE
    assert FALLBACK_ROLE_SINGLE_PURPOSE == "RECOVERY_ONLY_NO_FINAL_REPORT_TRIGGER"


# ===========================================================================
# (e) 기존 6-state/terminal 회귀 무손상
# ===========================================================================

def test_e_all_tracked_initially_false():
    """새 WatcherState → all_tracked() == False."""
    state = runner_mod.WatcherState()
    assert state.all_tracked() is False


def test_e_all_tracked_after_six_marks():
    """6개 mark_* 모두 호출 후 all_tracked() == True."""
    state = runner_mod.WatcherState()
    runner_mod.mark_head_change(state)
    runner_mod.mark_non_force_push(state)
    runner_mod.mark_ci(state)
    runner_mod.mark_finish_done(state)
    runner_mod.mark_normal_callback(state)
    runner_mod.mark_fallback_prune(state)
    assert state.all_tracked() is True
    assert all_states_tracked(state.as_tracked_dict()) is True


def test_e_as_tracked_dict_keys_match_constant():
    """★ tuple(state.as_tracked_dict().keys()) == WATCHER_TRACKED_STATES
    (heartbeat 필드 추가가 6-state dict 를 오염시키지 않음 보장)."""
    state = runner_mod.WatcherState()
    assert tuple(state.as_tracked_dict().keys()) == WATCHER_TRACKED_STATES


def test_e_terminal_states_length_five():
    """TERMINAL_STATES 길이 == 5."""
    assert len(TERMINAL_STATES) == 5


def test_e_terminal_states_excludes_lifetime_gap():
    """CI_WATCHER_SESSION_LIFETIME_GAP 는 TERMINAL_STATES 에 미포함."""
    assert CI_WATCHER_SESSION_LIFETIME_GAP not in TERMINAL_STATES


def test_e_run_once_head_drift_returns_hold_for_chair():
    """run_once head drift → HOLD_FOR_CHAIR (terminal), state.polls==1."""
    snap = PRSnapshot(head_ref_oid="deadbeefdrifted", merge_state_status="BLOCKED")
    state = runner_mod.WatcherState()
    terminal_state, reason = runner_mod.run_once(
        snap,
        expected_head="expectedhead000",
        elapsed_watcher_sec=10,
        state=state,
    )
    assert terminal_state in TERMINAL_STATES
    assert terminal_state == "HOLD_FOR_CHAIR"
    assert "head_drift" in reason
    assert state.polls == 1
    assert state.elapsed_sec == 10


def test_e_watcher_state_new_has_no_heartbeat():
    """새 WatcherState → last_poll_ts=None, heartbeat_ts=None."""
    state = runner_mod.WatcherState()
    assert state.last_poll_ts is None
    assert state.heartbeat_ts is None


def test_e_heartbeat_fields_not_in_tracked_dict():
    """last_poll_ts / heartbeat_ts 는 as_tracked_dict() 에 포함되지 않는다."""
    state = runner_mod.WatcherState()
    d = state.as_tracked_dict()
    assert "last_poll_ts" not in d
    assert "heartbeat_ts" not in d