"""tests/regression/test_schedule_id_freshness_2535.py — 회귀 6건.

회장 명시 task-2535 P0 (신호등 sync fix C / 2026-05-10):
  cron schedule_id 신선도 검증 부재 → 오래된 schedule_id 로 잘못된 활성 판정 발생.
  Fix: utils/schedule_id_freshness.py (validator) + lifecycle_reconciliation_manager
  STALE_SCHEDULE_ID stuck case.

회귀 6건 (회장 §명시):
  1. fresh (5분 전 응답)        → classify_freshness = FRESH
  2. stale (90분 전 응답)       → classify_freshness = STALE
  3. missing (history 부재)     → classify_freshness = MISSING
  4. lifecycle stuck STALE_SCHEDULE_ID 분류
  5. chat=6937032012 격리 (다른 chat record 는 freshness 판정에 포함 X)
  6. token raw 0 (validator 출력 / stuck detail 에 ghs_/ghp_/github_pat_ prefix 부재)
"""
from __future__ import annotations

import json
import sys
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any

import pytest

_WORKTREE_ROOT = Path(__file__).resolve().parent.parent.parent
if str(_WORKTREE_ROOT) in sys.path:
    sys.path.remove(str(_WORKTREE_ROOT))
sys.path.insert(0, str(_WORKTREE_ROOT))

from utils.schedule_id_freshness import (  # noqa: E402  # pyright: ignore[reportMissingImports]
    SCHEDULE_FRESHNESS_THRESHOLD_MIN,
    CHAIRMAN_CHAT_ID,
    classify_freshness,
    is_schedule_id_fresh,
    schedule_id_age_seconds,
)
from utils.lifecycle_reconciliation_manager import (  # noqa: E402  # pyright: ignore[reportMissingImports]
    LifecycleEvidence,
    StuckReason,
    detect_stuck_cases,
)


# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------

NOW = datetime(2026, 5, 10, 14, 0, 0, tzinfo=timezone.utc)
TOKEN_PREFIXES = ("ghs_", "ghp_", "github_pat_")


def _write_record(path: Path, *, ts: datetime, chat_id: int, schedule_id: str,
                  status: str = "ok", response: str = "ok", prompt: str = "p") -> None:
    record: dict[str, Any] = {
        "ts": ts.isoformat(),
        "chat_id": chat_id,
        "schedule_id": schedule_id,
        "status": status,
        "response": response,
        "prompt": prompt,
        "duration_ms": 1234,
        "workspace_path": "/tmp/workspace",
    }
    with path.open("a", encoding="utf-8") as f:
        f.write(json.dumps(record, ensure_ascii=False) + "\n")


def _make_evidence(**overrides: Any) -> LifecycleEvidence:
    base: dict[str, Any] = dict(
        task_id="task-9999",
        pr_number=None,
        pr_state=None,
        merge_commit=None,
        merged_into_main=False,
        ci_status=None,
        smoke_status=None,
        timer_status=None,
        timer_end_time=None,
        has_done=False,
        has_done_acked=False,
        has_merge_done=False,
        has_qc_result=False,
        has_followup=False,
        has_escalate_marker=False,
        escalate_marker_age_minutes=None,
        telegram_reply_truncated=False,
        bot_session_status=None,
        worktree_exists=False,
        branch_pushed_to_remote=False,
    )
    base.update(overrides)
    return LifecycleEvidence(**base)


# ===========================================================================
# 1. FRESH (5분 전 응답)
# ===========================================================================

class TestFreshRecord:
    """회귀 #1: 마지막 chairman record 가 5분 전이면 FRESH."""

    def test_classify_fresh(self, tmp_path: Path):
        sid = "ABCDEF12"
        log = tmp_path / f"{sid}.log"
        _write_record(log, ts=NOW - timedelta(minutes=5), chat_id=CHAIRMAN_CHAT_ID,
                      schedule_id=sid, response="ok")

        state = classify_freshness(sid, now=NOW, history_dir=tmp_path)
        assert state == "FRESH"

        is_fresh, token = is_schedule_id_fresh(sid, NOW, history_dir=tmp_path)
        assert is_fresh is True
        assert token == "FRESH"

    def test_age_seconds_for_fresh(self, tmp_path: Path):
        sid = "F12RESH00"
        log = tmp_path / f"{sid}.log"
        _write_record(log, ts=NOW - timedelta(minutes=5), chat_id=CHAIRMAN_CHAT_ID,
                      schedule_id=sid)

        age = schedule_id_age_seconds(sid, now=NOW, history_dir=tmp_path)
        assert age is not None
        assert 290 <= age <= 310  # ~5min ± 10s

    def test_threshold_boundary_just_under(self, tmp_path: Path):
        """Threshold = 60min → 59m59s 전 record 는 여전히 FRESH."""
        sid = "BNDRY001"
        log = tmp_path / f"{sid}.log"
        _write_record(log, ts=NOW - timedelta(minutes=59, seconds=59),
                      chat_id=CHAIRMAN_CHAT_ID, schedule_id=sid)
        assert classify_freshness(sid, now=NOW, history_dir=tmp_path) == "FRESH"


# ===========================================================================
# 2. STALE (90분 전 응답)
# ===========================================================================

class TestStaleRecord:
    """회귀 #2: 마지막 chairman record 가 90분 전이면 STALE."""

    def test_classify_stale(self, tmp_path: Path):
        sid = "STALE001"
        log = tmp_path / f"{sid}.log"
        _write_record(log, ts=NOW - timedelta(minutes=90), chat_id=CHAIRMAN_CHAT_ID,
                      schedule_id=sid)

        state = classify_freshness(sid, now=NOW, history_dir=tmp_path)
        assert state == "STALE"

        is_fresh, token = is_schedule_id_fresh(sid, NOW, history_dir=tmp_path)
        assert is_fresh is False
        assert token == "STALE"

    def test_threshold_boundary_exactly_60min(self, tmp_path: Path):
        """정확히 60분 → STALE (≥ threshold 는 stale 로 박제)."""
        sid = "BNDRY060"
        log = tmp_path / f"{sid}.log"
        _write_record(log, ts=NOW - timedelta(minutes=60), chat_id=CHAIRMAN_CHAT_ID,
                      schedule_id=sid)
        assert classify_freshness(sid, now=NOW, history_dir=tmp_path) == "STALE"

    def test_constant_value(self):
        """SCHEDULE_FRESHNESS_THRESHOLD_MIN 은 회장 §명시 60."""
        assert SCHEDULE_FRESHNESS_THRESHOLD_MIN == 60


# ===========================================================================
# 3. MISSING (history 부재)
# ===========================================================================

class TestMissingRecord:
    """회귀 #3: schedule_history/{ID}.log 부재 → MISSING."""

    def test_classify_missing_when_log_absent(self, tmp_path: Path):
        sid = "NONEXIST"
        # 로그 파일을 만들지 않음
        state = classify_freshness(sid, now=NOW, history_dir=tmp_path)
        assert state == "MISSING"

        is_fresh, token = is_schedule_id_fresh(sid, NOW, history_dir=tmp_path)
        assert is_fresh is False
        assert token == "MISSING"
        assert schedule_id_age_seconds(sid, now=NOW, history_dir=tmp_path) is None

    def test_empty_schedule_id_returns_missing(self, tmp_path: Path):
        assert classify_freshness("", now=NOW, history_dir=tmp_path) == "MISSING"

    def test_log_with_no_chairman_records_is_missing(self, tmp_path: Path):
        """다른 chat 만 있는 log 는 MISSING (chairman 격리)."""
        sid = "OTHERCHT"
        log = tmp_path / f"{sid}.log"
        _write_record(log, ts=NOW - timedelta(minutes=10), chat_id=9999999,
                      schedule_id=sid)
        assert classify_freshness(sid, now=NOW, history_dir=tmp_path) == "MISSING"


# ===========================================================================
# 4. lifecycle stuck STALE_SCHEDULE_ID 분류
# ===========================================================================

class TestLifecycleStuckClassification:
    """회귀 #4: lifecycle_reconciliation_manager.detect_stuck_cases 가
    STALE schedule_id + timer running → STALE_SCHEDULE_ID 박제."""

    def test_stuck_case_emitted_for_stale_running_timer(self):
        ev = _make_evidence(
            task_id="task-9501",
            timer_status="running",
            schedule_id="ABCDEF12",
            schedule_id_freshness="STALE",
            schedule_id_age_seconds=5400.0,  # 90 min
        )
        cases = detect_stuck_cases(ev)
        reasons = {c.reason for c in cases}
        assert StuckReason.STALE_SCHEDULE_ID in reasons

    def test_no_stuck_when_fresh(self):
        ev = _make_evidence(
            task_id="task-9502",
            timer_status="running",
            schedule_id="FRESH001",
            schedule_id_freshness="FRESH",
            schedule_id_age_seconds=300.0,
        )
        cases = detect_stuck_cases(ev)
        reasons = {c.reason for c in cases}
        assert StuckReason.STALE_SCHEDULE_ID not in reasons

    def test_no_stuck_when_timer_not_running(self):
        ev = _make_evidence(
            task_id="task-9503",
            timer_status="completed",
            schedule_id="STALE001",
            schedule_id_freshness="STALE",
            schedule_id_age_seconds=5400.0,
        )
        cases = detect_stuck_cases(ev)
        reasons = {c.reason for c in cases}
        assert StuckReason.STALE_SCHEDULE_ID not in reasons

    def test_no_stuck_when_pr_already_merged(self):
        """PR MERGED 인 task 는 다른 stuck reason 이 처리 — STALE_SCHEDULE_ID 박제 X."""
        ev = _make_evidence(
            task_id="task-9504",
            pr_state="MERGED",
            merged_into_main=True,
            timer_status="running",
            schedule_id="STALE002",
            schedule_id_freshness="STALE",
            schedule_id_age_seconds=5400.0,
        )
        cases = detect_stuck_cases(ev)
        reasons = {c.reason for c in cases}
        assert StuckReason.STALE_SCHEDULE_ID not in reasons

    def test_missing_freshness_does_not_emit_stuck(self):
        """schedule_id_freshness=MISSING 은 보류 (stuck 박제 X — 매핑 부재만으로 stuck X)."""
        ev = _make_evidence(
            task_id="task-9505",
            timer_status="running",
            schedule_id=None,
            schedule_id_freshness="MISSING",
            schedule_id_age_seconds=None,
        )
        cases = detect_stuck_cases(ev)
        reasons = {c.reason for c in cases}
        assert StuckReason.STALE_SCHEDULE_ID not in reasons


# ===========================================================================
# 5. chat=6937032012 격리
# ===========================================================================

class TestChairmanChatIsolation:
    """회귀 #5: chat=6937032012 외 record 는 freshness 판정에 포함되지 않음."""

    def test_chairman_chat_id_constant(self):
        assert CHAIRMAN_CHAT_ID == 6937032012

    def test_other_chat_records_ignored(self, tmp_path: Path):
        """동일 schedule_id 에 다른 chat 의 fresh record 가 있어도 chairman record 가
        STALE 이면 STALE 로 박제 (다른 chat record 는 격리되어 freshness 판정 미참여).

        시나리오:
          - schedule_id 같은 log 파일에
          - chairman chat (6937032012) 의 90분 전 record
          - 다른 chat (9999999) 의 5분 전 record (포함되면 안 됨)
          → 기대: STALE (chairman 만 보면 90분 전 → STALE)
        """
        sid = "ISOL0001"
        log = tmp_path / f"{sid}.log"
        # 다른 chat 의 fresh record 를 먼저 쓴다 (이게 사용되면 잘못된 FRESH 판정)
        _write_record(log, ts=NOW - timedelta(minutes=5), chat_id=9999999,
                      schedule_id=sid)
        # chairman 의 stale record
        _write_record(log, ts=NOW - timedelta(minutes=90), chat_id=CHAIRMAN_CHAT_ID,
                      schedule_id=sid)

        state = classify_freshness(sid, now=NOW, history_dir=tmp_path)
        assert state == "STALE", "다른 chat record 가 포함되면 안 됨 — chairman 만 사용"

    def test_only_other_chat_records_yield_missing(self, tmp_path: Path):
        sid = "ISOL0002"
        log = tmp_path / f"{sid}.log"
        # 다른 chat 의 record 만 5건
        for offset in (5, 10, 20, 30, 40):
            _write_record(log, ts=NOW - timedelta(minutes=offset), chat_id=9999999,
                          schedule_id=sid)

        # chairman record 는 0 → MISSING
        assert classify_freshness(sid, now=NOW, history_dir=tmp_path) == "MISSING"

    def test_chairman_record_string_chat_id_handled(self, tmp_path: Path):
        """chat_id 가 문자열 '6937032012' 로 직렬화돼도 정상 인식 (cokacdir 호환)."""
        sid = "STR_CHID"
        log = tmp_path / f"{sid}.log"
        record = {
            "ts": (NOW - timedelta(minutes=10)).isoformat(),
            "chat_id": "6937032012",  # ★ str 형
            "schedule_id": sid,
            "status": "ok",
            "response": "ok",
            "prompt": "p",
        }
        log.write_text(json.dumps(record, ensure_ascii=False) + "\n", encoding="utf-8")
        assert classify_freshness(sid, now=NOW, history_dir=tmp_path) == "FRESH"


# ===========================================================================
# 6. token raw 0
# ===========================================================================

class TestTokenRawZero:
    """회귀 #6: validator 출력 / stuck detail 에 token raw prefix 0건."""

    def test_classify_freshness_does_not_leak_response_body(self, tmp_path: Path):
        """log 의 response 본문에 token 이 있어도 freshness 판정 결과에는 노출되지 않음."""
        sid = "TOKEN001"
        log = tmp_path / f"{sid}.log"
        # 일부러 token-like 문자열을 response 에 끼워 넣음
        _write_record(log, ts=NOW - timedelta(minutes=10), chat_id=CHAIRMAN_CHAT_ID,
                      schedule_id=sid,
                      response="this should not leak: ghs_FAKE_TOKEN_ABCDEFGH")

        state = classify_freshness(sid, now=NOW, history_dir=tmp_path)
        # 분류 토큰은 단순 문자열이며 token prefix 부재
        assert state == "FRESH"
        for prefix in TOKEN_PREFIXES:
            assert prefix not in state

    def test_stuck_detail_contains_no_token_prefix(self):
        ev = _make_evidence(
            task_id="task-2535",
            timer_status="running",
            schedule_id="ABCDEF12",
            schedule_id_freshness="STALE",
            schedule_id_age_seconds=5400.0,
        )
        cases = detect_stuck_cases(ev)
        stale_cases = [c for c in cases if c.reason == StuckReason.STALE_SCHEDULE_ID]
        assert len(stale_cases) == 1
        detail = stale_cases[0].detail
        for prefix in TOKEN_PREFIXES:
            assert prefix not in detail, f"token prefix {prefix} leaked in detail"

    def test_module_source_no_token_strings(self):
        """모듈 소스 자체에 token raw prefix 가 없는지 회귀 (회장 §금지 token 박제)."""
        src_path = _WORKTREE_ROOT / "utils" / "schedule_id_freshness.py"
        text = src_path.read_text(encoding="utf-8")
        for prefix in TOKEN_PREFIXES:
            assert prefix not in text, f"{prefix} found in {src_path.name}"


if __name__ == "__main__":
    sys.exit(pytest.main([__file__, "-v"]))