"""tests/regression/test_worktree_timer_reconcile_2528.py — 회귀 7건.

회장 명시 task-2528 P0:
  worktree에서 완료된 task가 task-timers.json에 reconcile 안 되는 lifecycle bug fix.
  본 회귀는 task-2527 audit (2026-05-10)에서 식별된 root cause를 박제한다.

회귀 7건:
  1. happy path: worktree 완료 → timer entry 추가 PASS
  2. task-timers.json 부재 fixture (5/9 14:32 8 task 사고): 8개 task 부재 → reconcile 후 8개 모두 entry 존재
  3. idempotent: reconcile 2회 실행 → entry 1개만 존재 (중복 X)
  4. archived 충돌: archived에 이미 있는 task → active에 중복 추가 X
  5. mtime fallback 회귀: reconcile 후 helpers.py:450-454 mtime fallback 미발동 검증
  6. chat=6937032012 격리: 다른 chat record fixture → reconcile 결과에 포함 X
  7. token raw 0: reconcile 결과 JSON에 ghs_/ghp_/github_pat_ prefix 부재
"""
from __future__ import annotations

import json
import sys
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.lifecycle_reconciliation_manager import (  # noqa: E402  # pyright: ignore[reportMissingImports]
    LifecycleEvidence,
    _build_reconciled_timer_entry,
    _read_timers_file,
    _reconcile_worktree_completion_to_timers,
    _timer_entry_present,
    reconcile,
)


# ---------------------------------------------------------------------------
# Fixture helpers
# ---------------------------------------------------------------------------

EIGHT_STUCK_TASKS = (
    "task-2514",
    "task-2515",
    "task-2516",
    "task-2516+1",
    "task-2517",
    "task-2518-self-host",
    "task-2519",
    "task-2519-self",
)

TOKEN_PREFIXES = ("ghs_", "ghp_", "github_pat_")


def _make_evidence(**overrides) -> 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)


def _setup_workspace(tmp_path: Path, *, with_active: bool = True) -> Path:
    """Create memory/ tree under tmp_path. Returns workspace root."""
    (tmp_path / "memory" / "events").mkdir(parents=True, exist_ok=True)
    if with_active:
        (tmp_path / "memory" / "task-timers.json").write_text(
            json.dumps({"tasks": {}}, ensure_ascii=False),
            encoding="utf-8",
        )
    return tmp_path


def _write_done_json(workspace: Path, task_id: str, *, end_time: str, team_id: str = "dev3-team",
                     duration_seconds: float = 1234.5, qc_result: str = "PASS") -> None:
    p = workspace / "memory" / "events" / f"{task_id}.done"
    p.write_text(
        json.dumps(
            {
                "task_id": task_id,
                "team_id": team_id,
                "end_time": end_time,
                "duration_seconds": duration_seconds,
                "qc_result": qc_result,
                "status": "done",
                "completed_at": end_time,
            },
            ensure_ascii=False,
        ),
        encoding="utf-8",
    )


def _evidence_completed(task_id: str, *, has_done: bool = True, has_merge_done: bool = True,
                        pr_state: str | None = "MERGED", merged_into_main: bool = True) -> LifecycleEvidence:
    return _make_evidence(
        task_id=task_id,
        pr_number=70,
        pr_state=pr_state,
        merge_commit="aaaa1111bbbb2222cccc3333dddd4444eeee5555",
        merged_into_main=merged_into_main,
        ci_status="SUCCESS",
        smoke_status="PASS",
        has_done=has_done,
        has_done_acked=True,
        has_merge_done=has_merge_done,
        has_qc_result=True,
    )


# ===========================================================================
# 1. Happy path
# ===========================================================================

class TestHappyPath:
    """회귀 #1: worktree 완료 → timer entry 추가 PASS."""

    def test_happy_path_inserts_entry(self, tmp_path: Path):
        ws = _setup_workspace(tmp_path)
        tid = "task-2999"
        end_time = "2026-05-09T03:02:43+00:00"
        _write_done_json(ws, tid, end_time=end_time, team_id="dev3-team", duration_seconds=2490.2)

        ev = _evidence_completed(tid)

        actions_taken: list[str] = []
        actions_planned: list[str] = []

        meta = _reconcile_worktree_completion_to_timers(
            tid,
            "run-1",
            ev,
            workspace_root=ws,
            apply=True,
            actions_taken=actions_taken,
            actions_planned=actions_planned,
        )

        assert meta["entry_inserted"] is True, f"meta={meta}"
        assert "reconciled_worktree_timer" in actions_taken
        assert actions_planned == []

        timers = _read_timers_file(ws / "memory" / "task-timers.json")
        assert tid in timers["tasks"], f"timers={timers}"
        entry = timers["tasks"][tid]
        assert entry["task_id"] == tid
        assert entry["end_time"] == end_time
        assert entry["status"] == "completed"
        assert entry["team_id"] == "dev3-team"
        assert entry["duration_seconds"] == pytest.approx(2490.2)
        assert entry["reconciled_from"] == "done_json"
        assert entry["ended_by"] == "lifecycle_reconciliation_manager"
        assert entry["reconcile_run_id"] == "run-1"

    def test_dry_run_does_not_insert(self, tmp_path: Path):
        ws = _setup_workspace(tmp_path)
        tid = "task-2999"
        _write_done_json(ws, tid, end_time="2026-05-09T03:02:43+00:00")

        ev = _evidence_completed(tid)

        actions_taken: list[str] = []
        actions_planned: list[str] = []

        meta = _reconcile_worktree_completion_to_timers(
            tid,
            "run-dry",
            ev,
            workspace_root=ws,
            apply=False,
            actions_taken=actions_taken,
            actions_planned=actions_planned,
        )

        assert meta["entry_inserted"] is False
        assert "reconciled_worktree_timer" in actions_planned
        assert actions_taken == []
        timers = _read_timers_file(ws / "memory" / "task-timers.json")
        assert tid not in timers.get("tasks", {})


# ===========================================================================
# 2. task-timers.json 부재 fixture (8 task 사고 박제)
# ===========================================================================

class TestEightStuckTaskFixture:
    """회귀 #2: 5/9 14:32 사고 fixture — 8개 task 모두 부재 → reconcile 후 8개 모두 entry 존재."""

    def test_eight_stuck_tasks_all_get_entries(self, tmp_path: Path):
        ws = _setup_workspace(tmp_path)
        # archived 빈 파일도 함께 만들어 collision 검사 통과 보장
        (ws / "memory" / "task-timers-archived.json").write_text(
            json.dumps({"tasks": {}}, ensure_ascii=False),
            encoding="utf-8",
        )

        # 모든 8개 task에 .done 만 두고 timer entry는 없음 (회장 §사고 재현)
        end_times = {
            "task-2514": "2026-05-09T03:02:43+00:00",
            "task-2515": "2026-05-09T04:18:03+00:00",
            "task-2516": "2026-05-09T05:30:00+00:00",
            "task-2516+1": "2026-05-09T06:00:00+00:00",
            "task-2517": "2026-05-09T09:11:17+00:00",
            "task-2518-self-host": "2026-05-09T11:00:00+00:00",
            "task-2519": "2026-05-09T12:00:00+00:00",
            "task-2519-self": "2026-05-09T13:00:00+00:00",
        }
        for tid, et in end_times.items():
            _write_done_json(ws, tid, end_time=et)

        # reconcile 8개 모두
        for tid in EIGHT_STUCK_TASKS:
            ev = _evidence_completed(tid)
            actions_taken: list[str] = []
            actions_planned: list[str] = []
            _reconcile_worktree_completion_to_timers(
                tid,
                f"run-{tid}",
                ev,
                workspace_root=ws,
                apply=True,
                actions_taken=actions_taken,
                actions_planned=actions_planned,
            )

        timers = _read_timers_file(ws / "memory" / "task-timers.json")
        present = set(timers.get("tasks", {}).keys())
        missing = set(EIGHT_STUCK_TASKS) - present
        assert not missing, f"미부착 task 존재: {missing}"
        assert len(present) == 8

        # end_time 보존 검증 (mtime 균등화 회귀의 핵심: end_time이 .done JSON에서 와야 함)
        for tid in EIGHT_STUCK_TASKS:
            entry = timers["tasks"][tid]
            assert entry["end_time"] == end_times[tid], (
                f"{tid}: end_time 손실 (got={entry['end_time']}, expected={end_times[tid]})"
            )
            assert entry["reconciled_from"] == "done_json"


# ===========================================================================
# 3. Idempotent
# ===========================================================================

class TestIdempotent:
    """회귀 #3: reconcile 2회 실행 → entry 1개만 존재 (중복 X)."""

    def test_idempotent_two_runs(self, tmp_path: Path):
        ws = _setup_workspace(tmp_path)
        tid = "task-2517"
        _write_done_json(ws, tid, end_time="2026-05-09T09:11:17+00:00")

        ev = _evidence_completed(tid)

        for run_id in ("run-A", "run-B"):
            actions_taken: list[str] = []
            actions_planned: list[str] = []
            _reconcile_worktree_completion_to_timers(
                tid,
                run_id,
                ev,
                workspace_root=ws,
                apply=True,
                actions_taken=actions_taken,
                actions_planned=actions_planned,
            )

        timers = _read_timers_file(ws / "memory" / "task-timers.json")
        # 정확히 1개 entry, run-A의 reconcile_run_id 보존
        assert tid in timers["tasks"]
        entry = timers["tasks"][tid]
        assert entry["reconcile_run_id"] == "run-A", "두 번째 run이 첫 entry를 덮어쓰면 안 됨"

        # 2회차 실행 결과는 skipped + reason=timer_entry_present_in_active
        actions_taken2: list[str] = []
        actions_planned2: list[str] = []
        meta2 = _reconcile_worktree_completion_to_timers(
            tid,
            "run-C",
            ev,
            workspace_root=ws,
            apply=True,
            actions_taken=actions_taken2,
            actions_planned=actions_planned2,
        )
        assert meta2["skipped"] is True
        assert meta2["reason"] == "timer_entry_present_in_active"
        assert "reconciled_worktree_timer" not in actions_taken2


# ===========================================================================
# 4. Archived 충돌
# ===========================================================================

class TestArchiveCollision:
    """회귀 #4: archived에 이미 있는 task → active에 중복 추가 X."""

    def test_archive_present_blocks_active_insert(self, tmp_path: Path):
        ws = _setup_workspace(tmp_path)
        tid = "task-2515"
        _write_done_json(ws, tid, end_time="2026-05-09T04:18:03+00:00")

        # archived에 이미 entry가 있는 상태 (회장 §3건전 ledger)
        (ws / "memory" / "task-timers-archived.json").write_text(
            json.dumps(
                {
                    "tasks": {
                        tid: {
                            "task_id": tid,
                            "team_id": "dev3-team",
                            "end_time": "2026-05-09T04:18:03+00:00",
                            "status": "completed",
                            "duration_seconds": 3981.2,
                        }
                    }
                },
                ensure_ascii=False,
            ),
            encoding="utf-8",
        )

        ev = _evidence_completed(tid)

        actions_taken: list[str] = []
        actions_planned: list[str] = []
        meta = _reconcile_worktree_completion_to_timers(
            tid,
            "run-collision",
            ev,
            workspace_root=ws,
            apply=True,
            actions_taken=actions_taken,
            actions_planned=actions_planned,
        )

        assert meta["skipped"] is True
        assert meta["reason"] == "timer_entry_present_in_archive"
        assert meta["entry_inserted"] is False
        assert "reconciled_worktree_timer" not in actions_taken

        # active에는 여전히 비어있음
        active = _read_timers_file(ws / "memory" / "task-timers.json")
        assert tid not in active.get("tasks", {})

        # _timer_entry_present 도 archive를 우선 보고
        present, source = _timer_entry_present(
            tid,
            active_path=ws / "memory" / "task-timers.json",
            archive_path=ws / "memory" / "task-timers-archived.json",
        )
        assert present is True
        assert source == "archive"


# ===========================================================================
# 5. mtime fallback 회귀 (helpers.py:450-454 미발동)
# ===========================================================================

class TestMtimeFallbackRegression:
    """회귀 #5: reconcile 후 dashboard helpers.py:450-454 mtime fallback 미발동."""

    def test_end_time_present_after_reconcile_so_mtime_fallback_skipped(self, tmp_path: Path):
        """helpers.py:450-454 등가 로직: task-timers.json의 end_time 우선, 없으면 mtime."""
        ws = _setup_workspace(tmp_path)
        tid = "task-2514"
        true_end_time = "2026-05-09T03:02:43+00:00"
        _write_done_json(ws, tid, end_time=true_end_time)

        ev = _evidence_completed(tid)
        _reconcile_worktree_completion_to_timers(
            tid, "run-1", ev,
            workspace_root=ws, apply=True,
            actions_taken=[], actions_planned=[],
        )

        timers = _read_timers_file(ws / "memory" / "task-timers.json")
        entry = timers["tasks"][tid]
        end_time_from_timers = entry.get("end_time")
        assert end_time_from_timers, "reconcile 후 end_time 부재 → mtime fallback 발동 위험"

        # helpers.py:450-454 등가 로직 시뮬레이션:
        #   end_time = task_info.get("end_time")
        #   if not end_time: end_time = datetime.fromtimestamp(stat.st_mtime).isoformat()
        # 본 회귀는 task_info["end_time"]이 truthy → mtime fallback 진입 X 보장
        task_info_end_time = entry.get("end_time")
        used_mtime_fallback = not task_info_end_time
        assert used_mtime_fallback is False, (
            "task-timers.json end_time이 있는데도 mtime fallback이 발동되면 안 됨 "
            "(helpers.py:450-454 회귀 게이트)"
        )

        # end_time이 .done JSON의 정확한 timestamp인지 확인 (mtime 균등화 회귀 방지의 핵심)
        assert task_info_end_time == true_end_time

    def test_no_completion_evidence_skips(self, tmp_path: Path):
        """완료 evidence 없는 in-progress task는 reconcile하지 않음."""
        ws = _setup_workspace(tmp_path)
        ev = _make_evidence(task_id="task-9999", pr_state="OPEN", timer_status="running")

        actions_taken: list[str] = []
        actions_planned: list[str] = []
        meta = _reconcile_worktree_completion_to_timers(
            "task-9999", "run-x", ev,
            workspace_root=ws, apply=True,
            actions_taken=actions_taken, actions_planned=actions_planned,
        )
        assert meta["skipped"] is True
        assert meta["reason"] == "no_completion_evidence"

        timers = _read_timers_file(ws / "memory" / "task-timers.json")
        assert "task-9999" not in timers.get("tasks", {})


# ===========================================================================
# 6. chat=6937032012 격리
# ===========================================================================

class TestChatIsolation:
    """회귀 #6: 다른 chat record fixture는 reconcile 결과에 포함되지 않음."""

    def test_other_chat_record_not_in_reconcile_output(self, tmp_path: Path):
        """task-2528 fix는 task-timers.json만 건드림. 다른 chat의 schedule history /
        cron record는 별도 namespace이며 reconcile 결과 entry에 포함되어선 안 됨."""
        ws = _setup_workspace(tmp_path)
        tid = "task-2517"
        _write_done_json(ws, tid, end_time="2026-05-09T09:11:17+00:00")

        # 다른 chat (e.g. 9999999999) 의 schedule_history 위치를 흉내낸 fixture
        other_chat_dir = tmp_path / "fake_schedule_history_other_chat"
        other_chat_dir.mkdir(parents=True, exist_ok=True)
        (other_chat_dir / "OTHER.log").write_text(
            json.dumps({
                "ts": "2026-05-10T14:31:35+09:00",
                "chat_id": 9999999999,  # NOT 6937032012
                "schedule_id": "OTHER1234",
                "prompt": "task-2517 무관",
                "status": "ok",
                "response": "irrelevant",
            }) + "\n",
            encoding="utf-8",
        )

        ev = _evidence_completed(tid)
        meta = _reconcile_worktree_completion_to_timers(
            tid, "run-iso", ev,
            workspace_root=ws, apply=True,
            actions_taken=[], actions_planned=[],
        )
        assert meta["entry_inserted"] is True

        timers = _read_timers_file(ws / "memory" / "task-timers.json")
        entry = timers["tasks"][tid]
        # Reconciled entry 직렬화에 다른 chat의 schedule_id / chat_id 흔적 0
        entry_str = json.dumps(entry, ensure_ascii=False)
        assert "9999999999" not in entry_str, "다른 chat id 노출"
        assert "OTHER1234" not in entry_str, "다른 chat schedule_id 노출"
        assert "fake_schedule_history_other_chat" not in entry_str

        # 6937032012 chat record도 entry 본문에 직접 들어가선 안 됨 (timer는 chat-agnostic)
        assert "6937032012" not in entry_str


# ===========================================================================
# 7. Token raw 0
# ===========================================================================

class TestTokenRawZero:
    """회귀 #7: reconcile 결과 JSON에 ghs_/ghp_/github_pat_ prefix 부재."""

    def test_reconciled_entry_no_token_prefixes(self, tmp_path: Path):
        ws = _setup_workspace(tmp_path)
        tid = "task-2519"
        _write_done_json(ws, tid, end_time="2026-05-09T12:00:00+00:00")

        ev = _evidence_completed(tid)
        meta = _reconcile_worktree_completion_to_timers(
            tid, "run-token", ev,
            workspace_root=ws, apply=True,
            actions_taken=[], actions_planned=[],
        )
        assert meta["entry_inserted"] is True

        # entry 직렬화
        timers_text = (ws / "memory" / "task-timers.json").read_text(encoding="utf-8")
        meta_text = json.dumps(meta, ensure_ascii=False)

        for prefix in TOKEN_PREFIXES:
            assert prefix not in timers_text, f"task-timers.json에 token prefix '{prefix}' 노출"
            assert prefix not in meta_text, f"reconcile meta에 token prefix '{prefix}' 노출"

        # 일반적 token raw 패턴(40자 hex 등) 추가 검사 — merge_commit (40 hex)는 정상이므로 제외
        # github_pat_, ghs_, ghp_ 접두 외에도 환경변수 GITHUB_TOKEN 등 raw 노출 차단
        assert "GITHUB_TOKEN" not in timers_text or "GITHUB_TOKEN=" not in timers_text

    def test_build_reconciled_entry_no_token_in_dict(self, tmp_path: Path):
        ws = _setup_workspace(tmp_path)
        tid = "task-2515"
        _write_done_json(ws, tid, end_time="2026-05-09T04:18:03+00:00")

        ev = _evidence_completed(tid)
        entry = _build_reconciled_timer_entry(tid, "run-z", ev, workspace_root=ws)
        assert entry is not None

        # entry의 모든 string-shaped value 안전 검증
        def _walk(obj):
            if isinstance(obj, dict):
                for v in obj.values():
                    yield from _walk(v)
            elif isinstance(obj, (list, tuple)):
                for v in obj:
                    yield from _walk(v)
            elif isinstance(obj, str):
                yield obj

        for s in _walk(entry):
            for prefix in TOKEN_PREFIXES:
                assert prefix not in s, f"reconciled entry value에 token prefix 노출: {s[:20]}..."


# ===========================================================================
# Bonus: full reconcile() 통합 — task-2518 회귀와 충돌 없음 검증
# ===========================================================================

class TestReconcileIntegration:
    """reconcile() 호출 시 task-2528 helper가 정상 동작하고 task-2518 stuck 처리와 충돌 없음."""

    def test_reconcile_finalized_task_with_no_timer_entry_gets_one(self, tmp_path: Path):
        ws = _setup_workspace(tmp_path)
        tid = "task-2517"
        end_time = "2026-05-09T09:11:17+00:00"
        _write_done_json(ws, tid, end_time=end_time)

        # FINALIZED 조건 충족 마커 모두 생성
        events = ws / "memory" / "events"
        (events / f"{tid}.done.acked").write_text("{}", encoding="utf-8")
        (events / f"{tid}.merge-done").write_text("{}", encoding="utf-8")

        # gh CLI / git 호출을 short-circuit하기 위한 stub injection
        def fake_pr_lookup(task_id: str) -> dict:
            del task_id
            return {
                "number": 70,
                "state": "MERGED",
                "mergeCommit": {"oid": "aaaa1111bbbb2222cccc3333dddd4444eeee5555"},
            }

        def fake_runner(args, *, cwd=None, **_kw):  # pyright: ignore[reportUnusedParameter]
            del cwd, _kw  # protocol parameters
            class _R:
                returncode = 0
                stdout = ""
                stderr = ""
            if args[:3] == ["git", "merge-base", "--is-ancestor"]:
                return _R()
            if args[:2] == ["git", "ls-remote"]:
                r = _R()
                r.stdout = ""
                return r
            return _R()

        def fake_timer_loader(task_id: str) -> dict:
            del task_id
            return {}

        report = reconcile(
            tid,
            apply=True,
            workspace_root=ws,
            runner=fake_runner,
            pr_lookup=fake_pr_lookup,
            timer_loader=fake_timer_loader,
        )

        # task-2528 helper가 entry를 만들어줘야 함
        timers = _read_timers_file(ws / "memory" / "task-timers.json")
        assert tid in timers.get("tasks", {}), (
            f"reconcile() integrated path에서 entry 누락: timers={timers}, "
            f"actions_taken={report.actions_taken}"
        )
        assert report.backfill_metadata.get("worktree_timer_reconcile", {}).get("entry_inserted") is True