"""task-2534 회귀 테스트: data_loader 자동머지 시그널 병합 (traffic_light_fix_b)

회장 §결정(2026-05-10) 신호등 sync fix B — dashboard/data_loader.py가 task-timers.json만이
아니라 cron schedule_history + events/*.merge-queue.json 시그널을 병합해 활성/유휴 판정.

검증 포인트 (회귀 8건):
    1. timers running → 작업중 (heartbeat fresh + PID alive)
    2. cron 발사 후 timers stale → 작업중 (cron 시그널이 idle 덮어씀)
    3. merge-queue HEAD 있음 → 작업중
    4. 모든 시그널 부재 → 유휴
    5. heartbeat 가장 최근 source 우선 (cron > mq > heartbeat priority 검증)
    6. chat=6937032012 격리 (다른 chat schedule_history 무시)
    7. token raw 0 (prompt/response/schedule_id 절대 결과 dict에 노출 X)
    8. task-2528+1 / task-2530 / task-2531 fixture 회귀 (회장 발견 사례 3건 박제)
"""

from __future__ import annotations

import json
import os
import sys
import tempfile
import time
from pathlib import Path

import pytest

# project import — worktree 안전: 테스트 파일 기준으로 상대 경로 등록
_PROJECT_ROOT = str(Path(__file__).resolve().parents[2])
if _PROJECT_ROOT not in sys.path:
    sys.path.insert(0, _PROJECT_ROOT)
from dashboard.data_loader import (  # type: ignore[import-not-found]  # noqa: E402
    DASHBOARD_CHAT_ID,
    HEARTBEAT_FRESH_SECONDS,
    SIGNAL_FRESHNESS_SECONDS,
    _PidLivenessProvider,
    _collect_cron_schedule_signals,
    _collect_merge_queue_signals,
    _compute_member_status,
)


# ── fixtures ────────────────────────────────────────────────────────────────


@pytest.fixture
def tmp_heartbeat_dir():
    with tempfile.TemporaryDirectory() as tmp:
        yield Path(tmp)


@pytest.fixture
def tmp_schedule_history_dir():
    with tempfile.TemporaryDirectory() as tmp:
        yield Path(tmp)


@pytest.fixture
def tmp_events_dir():
    with tempfile.TemporaryDirectory() as tmp:
        yield Path(tmp)


def _touch_fresh(dir_: Path, name: str) -> Path:
    f = dir_ / name
    f.touch()
    return f


def _touch_with_age(dir_: Path, name: str, age_sec: float) -> Path:
    f = dir_ / name
    f.touch()
    past = time.time() - age_sec
    os.utime(f, (past, past))
    return f


def _write_schedule_log(
    dir_: Path,
    schedule_id: str,
    chat_id: int,
    prompt: str,
    workspace_path: str = "",
    age_sec: float = 0.0,
) -> Path:
    """schedule_history JSONL 1라인 파일 생성."""
    f = dir_ / f"{schedule_id}.log"
    rec = {
        "schedule_id": schedule_id,
        "chat_id": chat_id,
        "prompt": prompt,
        "response": "(redacted)",
        "workspace_path": workspace_path,
        "ts": "2026-05-10T03:00:00.000+09:00",
        "status": "ok",
        "duration_ms": 1000,
    }
    f.write_text(json.dumps(rec, ensure_ascii=False) + "\n", encoding="utf-8")
    if age_sec > 0:
        past = time.time() - age_sec
        os.utime(f, (past, past))
    return f


def _write_merge_queue(dir_: Path, task_id: str, age_sec: float = 0.0) -> Path:
    f = dir_ / f"{task_id}.merge-queue.json"
    payload = {
        "decision": "WAITING_FOR_PREDECESSOR",
        "task_id": task_id,
        "pr_number": 99,
        "timestamp": "2026-05-10T03:10:26Z",
    }
    f.write_text(json.dumps(payload, ensure_ascii=False), encoding="utf-8")
    if age_sec > 0:
        past = time.time() - age_sec
        os.utime(f, (past, past))
    return f


# ── 회귀 1: timers running → 작업중 ──────────────────────────────────────────


def test_signal_merge_1_timers_running_returns_working(tmp_heartbeat_dir):
    """timers running + heartbeat fresh + PID alive → working (기존 경로 회귀)."""
    _touch_fresh(tmp_heartbeat_dir, "task-A.heartbeat")
    p = _PidLivenessProvider()
    p.set_override({"sched-A": True})
    result = _compute_member_status(
        "task-A",
        {"status": "running", "schedule_id": "sched-A"},
        tmp_heartbeat_dir,
        p,
        cron_signals={},
        merge_queue_signals={},
    )
    assert result == "working"


# ── 회귀 2: cron 발사 후 timers stale → 작업중 (cron 시그널) ──────────────────


def test_signal_merge_2_cron_overrides_stale_timers(tmp_heartbeat_dir):
    """timers status=completed(또는 hb stale)이지만 cron 시그널 fresh → 작업중.

    회장 발견 사례 핵심 — cron --session으로 봇이 실 작업 중이지만 timers는 'idle'.
    """
    p = _PidLivenessProvider()
    p.set_override({})
    cron_signals = {"task-2528+1": time.time() - 60.0}  # 1분 전
    result = _compute_member_status(
        "task-2528+1",
        {"status": "completed"},  # timers는 idle 상태
        tmp_heartbeat_dir,
        p,
        cron_signals=cron_signals,
        merge_queue_signals={},
    )
    assert result == "working"


# ── 회귀 3: merge-queue HEAD 있음 → 작업중 ───────────────────────────────────


def test_signal_merge_3_merge_queue_head_returns_working(tmp_heartbeat_dir):
    """merge-queue HEAD에 fresh 진입 → 작업중 (timers 무관)."""
    p = _PidLivenessProvider()
    p.set_override({})
    mq_signals = {"task-2530": time.time() - 30.0}
    result = _compute_member_status(
        "task-2530",
        {},  # timers 미존재 (phantom task)
        tmp_heartbeat_dir,
        p,
        cron_signals={},
        merge_queue_signals=mq_signals,
    )
    assert result == "working"


# ── 회귀 4: 모든 시그널 부재 → 유휴 ──────────────────────────────────────────


def test_signal_merge_4_no_signals_returns_idle(tmp_heartbeat_dir):
    """cron 부재 + merge-queue 부재 + timers stale/completed → idle."""
    p = _PidLivenessProvider()
    p.set_override({})
    result = _compute_member_status(
        "task-XYZ",
        {"status": "completed", "schedule_id": "none"},
        tmp_heartbeat_dir,
        p,
        cron_signals={},
        merge_queue_signals={},
    )
    assert result == "idle"


# ── 회귀 5: heartbeat 가장 최근 source 우선 ──────────────────────────────────


def test_signal_merge_5_priority_cron_over_mq_over_heartbeat(tmp_heartbeat_dir):
    """priority: cron > merge-queue > timers heartbeat.

    검증 1: cron stale + mq stale + hb fresh+alive → working (heartbeat path)
    검증 2: cron fresh + 나머지 stale → working (cron path)
    검증 3: cron stale + mq fresh + hb stale → working (mq path)
    검증 4: 모두 stale → idle
    """
    p = _PidLivenessProvider()
    p.set_override({"sched-X": True})

    # ① heartbeat만 fresh — 가장 최근 source가 hb
    _touch_fresh(tmp_heartbeat_dir, "task-X.heartbeat")
    result1 = _compute_member_status(
        "task-X",
        {"status": "running", "schedule_id": "sched-X"},
        tmp_heartbeat_dir,
        p,
        cron_signals={"task-X": time.time() - SIGNAL_FRESHNESS_SECONDS - 60},
        merge_queue_signals={"task-X": time.time() - SIGNAL_FRESHNESS_SECONDS - 60},
    )
    assert result1 == "working"

    # ② cron만 fresh — 가장 최근 source가 cron
    _touch_with_age(tmp_heartbeat_dir, "task-Y.heartbeat", HEARTBEAT_FRESH_SECONDS + 60)
    result2 = _compute_member_status(
        "task-Y",
        {"status": "running", "schedule_id": "sched-Y"},
        tmp_heartbeat_dir,
        p,
        cron_signals={"task-Y": time.time() - 30.0},
        merge_queue_signals={"task-Y": time.time() - SIGNAL_FRESHNESS_SECONDS - 60},
    )
    assert result2 == "working"

    # ③ merge-queue만 fresh
    _touch_with_age(tmp_heartbeat_dir, "task-Z.heartbeat", HEARTBEAT_FRESH_SECONDS + 60)
    result3 = _compute_member_status(
        "task-Z",
        {"status": "running", "schedule_id": "sched-Z"},
        tmp_heartbeat_dir,
        p,
        cron_signals={"task-Z": time.time() - SIGNAL_FRESHNESS_SECONDS - 60},
        merge_queue_signals={"task-Z": time.time() - 60.0},
    )
    assert result3 == "working"

    # ④ 모두 stale
    _touch_with_age(tmp_heartbeat_dir, "task-W.heartbeat", HEARTBEAT_FRESH_SECONDS + 60)
    result4 = _compute_member_status(
        "task-W",
        {"status": "running", "schedule_id": "sched-W"},
        tmp_heartbeat_dir,
        p,
        cron_signals={"task-W": time.time() - SIGNAL_FRESHNESS_SECONDS - 60},
        merge_queue_signals={"task-W": time.time() - SIGNAL_FRESHNESS_SECONDS - 60},
    )
    assert result4 == "idle"


# ── 회귀 6: chat=6937032012 격리 ─────────────────────────────────────────────


def test_signal_merge_6_chat_isolation(tmp_schedule_history_dir):
    """다른 chat의 schedule_history 라인은 결과에 포함 X.

    회장 §명시: chat=6937032012 외 라인은 무시.
    """
    # 회장 chat 라인 (task-100 포함)
    _write_schedule_log(
        tmp_schedule_history_dir,
        "ABCD0001",
        chat_id=DASHBOARD_CHAT_ID,
        prompt="task-100 진행",
    )
    # 다른 chat 라인 (task-200 포함, 격리되어야 함)
    _write_schedule_log(
        tmp_schedule_history_dir,
        "ABCD0002",
        chat_id=9999999999,
        prompt="task-200 진행",
    )

    signals = _collect_cron_schedule_signals(
        tmp_schedule_history_dir,
        chat_id=DASHBOARD_CHAT_ID,
    )
    assert "task-100" in signals
    assert "task-200" not in signals  # 격리


# ── 회귀 7: token raw 0 ──────────────────────────────────────────────────────


def test_signal_merge_7_token_raw_zero(tmp_schedule_history_dir):
    """반환 dict에는 task_id/mtime만 — prompt/response/schedule_id raw 절대 노출 X."""
    _write_schedule_log(
        tmp_schedule_history_dir,
        "FF00FF00",
        chat_id=DASHBOARD_CHAT_ID,
        prompt="task-300 비밀 토큰 ABC123 포함",
        workspace_path="/home/jay/.cokacdir/workspace/FF00FF00",
    )

    signals = _collect_cron_schedule_signals(
        tmp_schedule_history_dir,
        chat_id=DASHBOARD_CHAT_ID,
    )
    # value는 float (epoch seconds) 만
    assert all(isinstance(v, float) for v in signals.values())
    # 결과 직렬화 시에도 토큰 raw / schedule_id 노출 0
    serialized = json.dumps(signals)
    assert "ABC123" not in serialized
    assert "FF00FF00" not in serialized
    assert "비밀 토큰" not in serialized


# ── 회귀 8: task-2528+1 / task-2530 / task-2531 fixture (회장 발견 사례 3건) ──


def test_signal_merge_8_chairman_fixture_2528_plus_1(
    tmp_heartbeat_dir, tmp_schedule_history_dir, tmp_events_dir
):
    """회장 §발견 fixture: task-2528+1 dev1 — cron+merge-queue 모두 fresh, timers entry 0.

    audit fix B 본질 — task-timers는 entry 0이지만 cron schedule_history와
    merge-queue.json이 모두 살아있음 → 'working' 판정 보장.
    """
    _write_schedule_log(
        tmp_schedule_history_dir,
        "B9792045",
        chat_id=DASHBOARD_CHAT_ID,
        prompt="[task-2528+1] worktree timer reconcile",
        workspace_path="/home/jay/workspace/.worktrees/task-2528-dev1",
    )
    _write_merge_queue(tmp_events_dir, "task-2528+1")

    cron_signals = _collect_cron_schedule_signals(
        tmp_schedule_history_dir,
        chat_id=DASHBOARD_CHAT_ID,
    )
    mq_signals = _collect_merge_queue_signals(tmp_events_dir)

    assert "task-2528+1" in cron_signals
    assert "task-2528+1" in mq_signals

    p = _PidLivenessProvider()
    p.set_override({})
    result = _compute_member_status(
        "task-2528+1",
        {},  # timers entry 0
        tmp_heartbeat_dir,
        p,
        cron_signals=cron_signals,
        merge_queue_signals=mq_signals,
    )
    assert result == "working"


def test_signal_merge_8_chairman_fixture_2530(
    tmp_heartbeat_dir, tmp_schedule_history_dir, tmp_events_dir
):
    """회장 §발견 fixture: task-2530 dev7 — cron 발사 26m43s 진행, timers idle 표시.

    audit fix B 본질 — cron 시그널만으로도 working 판정.
    """
    _write_schedule_log(
        tmp_schedule_history_dir,
        "5AFBFD82",
        chat_id=DASHBOARD_CHAT_ID,
        prompt="[task-2530] composite ③ cross-cutting agent 페르소나 합성",
        workspace_path="/home/jay/workspace/.worktrees/task-2530-dev7",
    )
    # merge-queue는 없는 케이스 (cron만)

    cron_signals = _collect_cron_schedule_signals(
        tmp_schedule_history_dir,
        chat_id=DASHBOARD_CHAT_ID,
    )
    mq_signals = _collect_merge_queue_signals(tmp_events_dir)

    assert "task-2530" in cron_signals
    assert "task-2530" not in mq_signals

    p = _PidLivenessProvider()
    p.set_override({})
    result = _compute_member_status(
        "task-2530",
        {"status": "completed"},  # timers는 idle (status박제)
        tmp_heartbeat_dir,
        p,
        cron_signals=cron_signals,
        merge_queue_signals=mq_signals,
    )
    assert result == "working"


def test_signal_merge_8_chairman_fixture_2531(
    tmp_heartbeat_dir, tmp_schedule_history_dir, tmp_events_dir
):
    """회장 §발견 fixture: task-2531 dev4 — 표시=유휴(cron 미발견).

    audit fix B 본질 — cron 미발견이면 적법하게 idle 판정 (over-trigger 금지).
    이 경계 케이스를 fixture로 박제해 회귀 보호.
    """
    # schedule_history 비어있음 (혹은 다른 task만)
    _write_schedule_log(
        tmp_schedule_history_dir,
        "298A007B",
        chat_id=DASHBOARD_CHAT_ID,
        prompt="[task-2999] 다른 작업",  # task-2531 미언급
    )

    cron_signals = _collect_cron_schedule_signals(
        tmp_schedule_history_dir,
        chat_id=DASHBOARD_CHAT_ID,
    )
    mq_signals = _collect_merge_queue_signals(tmp_events_dir)

    assert "task-2531" not in cron_signals
    assert "task-2531" not in mq_signals

    p = _PidLivenessProvider()
    p.set_override({})
    result = _compute_member_status(
        "task-2531",
        {"status": "completed"},
        tmp_heartbeat_dir,
        p,
        cron_signals=cron_signals,
        merge_queue_signals=mq_signals,
    )
    assert result == "idle"