"""tests/regression/test_cron_timers_upsert_2533.py

회귀 테스트 — task-2533 cron --session timers upsert hook (신호등 sync fix A).

회장 §본질 (2026-05-10 신호등 sync fix A):
  cron ``--session`` 발사 직후 ``memory/task-timers.json`` 에 task entry 가 자동 갱신되지
  않아 대시보드 신호등 100% gap. 본 hook 은 cron 발사 직후 timers entry 를 atomic+idempotent
  하게 upsert. 본 회귀는 7개 시나리오를 박제하여 동일 사고가 다시 발생하지 못하도록 강제한다.

회귀 7 (정확히 7개):
  1. cron 발사 성공 → timers entry 신규 생성 (status=running, schedule_id 명시)
  2. 동일 task 재발사 → entry 1건만 유지 (idempotent — 중복 X)
  3. opt-out prompt(read_only/analysis_only/report_only) 도 upsert 호출 시 entry 생성
     (활성 표시 필수)
  4. cron 발사 실패 (rc!=0) → timers entry 미생성 (atomic)
  5. team_id 추출 정확성 — task md / explicit / prompt fallback 우선순위
  6. schedule_id 발사에서 description 에 raw token / raw uuid / raw hex key 누수 0
  7. chat=6937032012 격리 — entry 에 chat_id 명시 저장 (다른 chat entry 와 충돌 X)

회장 §보안:
  - production timers JSON 절대 건드리지 않음 (모든 테스트가 ``tmp_path`` 사용)
  - subprocess 실행 X (runner 가 stub 으로 주입)
  - raw cron prompt 가 description 에 그대로 저장되지 않음을 정적 검사
"""
from __future__ import annotations

import json
import sys
from pathlib import Path
from typing import List, Sequence, Tuple

import pytest

# ---------------------------------------------------------------------------
# Worktree root → sys.path (force position 0) — 패턴 task-2526/task-2528/task-2529 정합
# ---------------------------------------------------------------------------
_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 scripts.safe_cron_dispatch import (  # noqa: E402  # pyright: ignore[reportMissingImports]
    parse_schedule_id_from_cokacdir_stdout,
    run_cron_with_timers_upsert,
)
from utils.cron_timers_upsert import (  # noqa: E402  # pyright: ignore[reportMissingImports]
    DEFAULT_TIMERS_PATH,
    DESCRIPTION_MAX_LEN,
    TIMERS_TASK_KEY,
    extract_task_id_from_prompt,
    extract_team_id_from_task_md,
    sanitize_description,
)


# ===========================================================================
# Fixtures — 신호등 sync fix A 박제
# ===========================================================================

CHAIR_CHAT = "6937032012"
OTHER_CHAT = "9999999999"  # 다른 chat — 격리 검증용
SCHEDULE_ID = "5C9995CCB"  # 본 사건 cron schedule id (PR #74 fixture)
RAW_HEX_KEY = "0b94683120a691cf"  # dev3-dagda raw bot key (절대 description 에 가면 안 됨)
RAW_UUID = "5eee7634-b0be-4594-b84e-311ae64e557b"  # 본 사건 self-session UUID
RAW_GHP_TOKEN = "ghp_AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"  # 임의 PAT (테스트 dummy)


def _make_runner(rc: int, stdout: str) -> "Tuple[List[Sequence[str]], object]":
    """지정 rc/stdout 을 반환하는 stub runner 와 호출 기록 list 를 함께 만든다."""
    calls: List[Sequence[str]] = []

    def _runner(argv: Sequence[str]) -> Tuple[int, str]:
        calls.append(tuple(argv))
        return rc, stdout

    return calls, _runner


def _ok_stdout(schedule_id: str = SCHEDULE_ID) -> str:
    """cokacdir cron 등록 정상 응답 stdout."""
    return json.dumps(
        {
            "status": "ok",
            "id": schedule_id,
            "prompt": "...",
            "schedule": "0 9 * * *",
        },
        ensure_ascii=False,
    )


@pytest.fixture
def empty_timers(tmp_path: Path) -> Path:
    """비어있는 task-timers.json fixture (production 오염 X)."""
    p = tmp_path / "task-timers.json"
    p.write_text(
        json.dumps({"tasks": {}, "counter": 0}, ensure_ascii=False, indent=2),
        encoding="utf-8",
    )
    return p


@pytest.fixture
def empty_tasks_dir(tmp_path: Path) -> Path:
    """비어있는 tasks/ 디렉토리 fixture."""
    d = tmp_path / "tasks"
    d.mkdir()
    return d


# ===========================================================================
# 회귀 1 — 신규 cron 발사 시 entry 생성 (status=running, schedule_id 명시)
# ===========================================================================

def test_2533_new_cron_dispatch_creates_running_entry(
    empty_timers: Path, empty_tasks_dir: Path
) -> None:
    """cron 발사 성공 → tasks[task-2533] entry 신규 생성."""
    # task md 에 team_id 명시 (extract 검증).
    md_path = empty_tasks_dir / "task-2533.md"
    md_path.write_text(
        "# task-2533 — cron timers upsert hook\n\ndev1-team 헤르메스 단독.\n",
        encoding="utf-8",
    )

    prompt = "task-2533 진행. 본질=cron --session 발사 hook 이 task-timers.json upsert."
    calls, runner = _make_runner(rc=0, stdout=_ok_stdout())

    rc, _, entry = run_cron_with_timers_upsert(
        argv=("/usr/local/bin/cokacdir", "--cron", prompt, "--chat", CHAIR_CHAT),
        prompt=prompt,
        chat=CHAIR_CHAT,
        explicit_team=None,
        timers_path=empty_timers,
        tasks_dir=empty_tasks_dir,
        runner=runner,
    )

    assert rc == 0
    assert len(calls) == 1, "runner 는 정확히 1번 호출되어야 한다"
    assert entry is not None, "성공한 dispatch 직후 entry 가 반환되어야 한다"
    assert entry["task_id"] == "task-2533"
    assert entry["team_id"] == "dev1-team"
    assert entry["status"] == "running"
    assert entry["schedule_id"] == SCHEDULE_ID
    assert entry["chat_id"] == CHAIR_CHAT
    assert entry["start_time"] is not None
    assert entry["end_time"] is None

    persisted = json.loads(empty_timers.read_text(encoding="utf-8"))
    assert "task-2533" in persisted[TIMERS_TASK_KEY]
    saved = persisted[TIMERS_TASK_KEY]["task-2533"]
    assert saved["status"] == "running"
    assert saved["schedule_id"] == SCHEDULE_ID


# ===========================================================================
# 회귀 2 — 동일 task 재발사 시 idempotent (중복 X / start_time 보존)
# ===========================================================================

def test_2533_redispatch_same_task_is_idempotent(
    empty_timers: Path, empty_tasks_dir: Path
) -> None:
    """동일 task_id 두 번 발사 → tasks dict 길이 1 유지, start_time 보존, schedule_id 갱신."""
    md_path = empty_tasks_dir / "task-2533.md"
    md_path.write_text("# task-2533\n\ndev1-team 단독.\n", encoding="utf-8")

    prompt = "task-2533 진행. retry."

    # 1차 발사.
    _, runner1 = _make_runner(rc=0, stdout=_ok_stdout("FIRST_ID"))
    rc1, _, entry1 = run_cron_with_timers_upsert(
        argv=("/usr/local/bin/cokacdir",),
        prompt=prompt,
        chat=CHAIR_CHAT,
        timers_path=empty_timers,
        tasks_dir=empty_tasks_dir,
        runner=runner1,
    )
    assert rc1 == 0 and entry1 is not None
    first_start = entry1["start_time"]

    # 2차 발사 — 같은 task_id, 다른 schedule_id.
    _, runner2 = _make_runner(rc=0, stdout=_ok_stdout("SECOND_ID"))
    rc2, _, entry2 = run_cron_with_timers_upsert(
        argv=("/usr/local/bin/cokacdir",),
        prompt=prompt,
        chat=CHAIR_CHAT,
        timers_path=empty_timers,
        tasks_dir=empty_tasks_dir,
        runner=runner2,
    )
    assert rc2 == 0 and entry2 is not None

    persisted = json.loads(empty_timers.read_text(encoding="utf-8"))
    tasks = persisted[TIMERS_TASK_KEY]
    # 동일 key 갱신 — dict 길이 1.
    assert len(tasks) == 1, f"중복 entry 생기면 안 된다, got: {list(tasks.keys())}"
    assert "task-2533" in tasks

    saved = tasks["task-2533"]
    # status 는 여전히 running, schedule_id 는 최신.
    assert saved["status"] == "running"
    assert saved["schedule_id"] == "SECOND_ID"
    # idempotent — running 상태 유지 시 start_time 보존.
    assert saved["start_time"] == first_start
    # last_dispatch_at 은 최신.
    assert saved["last_dispatch_at"] >= first_start


# ===========================================================================
# 회귀 3 — opt-out prompt (read_only/analysis_only/report_only) 도 upsert 됨
# ===========================================================================

@pytest.mark.parametrize(
    "opt_out_token",
    [
        "read_only: true",
        "analysis_only: true",
        "report_only: true",
        "finalize_policy: no_pr",
    ],
)
def test_2533_optout_prompt_still_upserts(
    empty_timers: Path,
    empty_tasks_dir: Path,
    opt_out_token: str,
) -> None:
    """opt-out 토큰이 prompt 에 있어도 timers upsert 는 호출되어 활성 봇 표시가 보장된다."""
    md_path = empty_tasks_dir / "task-2533.md"
    md_path.write_text("dev1-team\n", encoding="utf-8")

    prompt = f"task-2533 audit only. {opt_out_token}"

    _, runner = _make_runner(rc=0, stdout=_ok_stdout())
    rc, _, entry = run_cron_with_timers_upsert(
        argv=("/usr/local/bin/cokacdir",),
        prompt=prompt,
        chat=CHAIR_CHAT,
        timers_path=empty_timers,
        tasks_dir=empty_tasks_dir,
        runner=runner,
    )

    assert rc == 0
    assert entry is not None, (
        f"opt-out 토큰 ({opt_out_token}) 이 있어도 timers upsert 는 활성 표시를 위해 호출돼야 한다"
    )
    assert entry["status"] == "running"

    persisted = json.loads(empty_timers.read_text(encoding="utf-8"))
    assert "task-2533" in persisted[TIMERS_TASK_KEY]


# ===========================================================================
# 회귀 4 — cron 발사 실패 (rc!=0) → entry 미생성 (atomic)
# ===========================================================================

def test_2533_failed_dispatch_does_not_upsert(
    empty_timers: Path, empty_tasks_dir: Path
) -> None:
    """runner rc != 0 → upsert hook skip. timers 파일 변경 0."""
    md_path = empty_tasks_dir / "task-2533.md"
    md_path.write_text("dev1-team\n", encoding="utf-8")

    before = empty_timers.read_text(encoding="utf-8")
    _, runner = _make_runner(rc=2, stdout="boom: dispatch failed")
    rc, _, entry = run_cron_with_timers_upsert(
        argv=("/usr/local/bin/cokacdir",),
        prompt="task-2533 진행.",
        chat=CHAIR_CHAT,
        timers_path=empty_timers,
        tasks_dir=empty_tasks_dir,
        runner=runner,
    )

    assert rc == 2
    assert entry is None
    after = empty_timers.read_text(encoding="utf-8")
    assert before == after, "dispatch 실패 시 timers JSON 은 한 바이트도 바뀌면 안 된다"


def test_2533_dispatch_ok_but_no_schedule_id_skips_upsert(
    empty_timers: Path, empty_tasks_dir: Path
) -> None:
    """rc=0 인데 stdout 에 id 가 없으면 skip (atomic — schedule_id 없는 entry 안 만듦)."""
    md_path = empty_tasks_dir / "task-2533.md"
    md_path.write_text("dev1-team\n", encoding="utf-8")

    before = empty_timers.read_text(encoding="utf-8")
    bad_stdout = json.dumps({"status": "ok"}, ensure_ascii=False)  # id 누락
    _, runner = _make_runner(rc=0, stdout=bad_stdout)
    rc, _, entry = run_cron_with_timers_upsert(
        argv=("/usr/local/bin/cokacdir",),
        prompt="task-2533 진행.",
        chat=CHAIR_CHAT,
        timers_path=empty_timers,
        tasks_dir=empty_tasks_dir,
        runner=runner,
    )

    assert rc == 0
    assert entry is None
    assert empty_timers.read_text(encoding="utf-8") == before


# ===========================================================================
# 회귀 5 — team_id 추출 정확성 (task md / explicit / prompt fallback 우선순위)
# ===========================================================================

def test_2533_team_id_extraction_priority(
    empty_timers: Path, empty_tasks_dir: Path, tmp_path: Path
) -> None:
    """우선순위:
       1. ``--team`` (explicit) > 2. task md 의 ``devN-team`` > 3. prompt body fallback
    """
    # 1. explicit > task md
    md = empty_tasks_dir / "task-2533.md"
    md.write_text("# task-2533\n\ndev2-team\n", encoding="utf-8")
    _, runner = _make_runner(rc=0, stdout=_ok_stdout())
    _, _, entry = run_cron_with_timers_upsert(
        argv=("/usr/local/bin/cokacdir",),
        prompt="task-2533 진행 dev9-team",
        chat=CHAIR_CHAT,
        explicit_team="dev1-team",  # explicit 가 win
        timers_path=empty_timers,
        tasks_dir=empty_tasks_dir,
        runner=runner,
    )
    assert entry is not None and entry["team_id"] == "dev1-team"

    # 2. task md > prompt — explicit 없을 때.
    timers2 = tmp_path / "t2.json"
    timers2.write_text(json.dumps({"tasks": {}}), encoding="utf-8")
    _, runner = _make_runner(rc=0, stdout=_ok_stdout())
    _, _, entry = run_cron_with_timers_upsert(
        argv=("/usr/local/bin/cokacdir",),
        prompt="task-2533 진행 dev9-team",  # prompt 에 dev9
        chat=CHAIR_CHAT,
        explicit_team=None,
        timers_path=timers2,
        tasks_dir=empty_tasks_dir,  # md 에는 dev2-team
        runner=runner,
    )
    assert entry is not None and entry["team_id"] == "dev2-team"

    # 3. prompt fallback — task md 없을 때.
    empty_tasks_dir2 = tmp_path / "tasks2"
    empty_tasks_dir2.mkdir()
    timers3 = tmp_path / "t3.json"
    timers3.write_text(json.dumps({"tasks": {}}), encoding="utf-8")
    _, runner = _make_runner(rc=0, stdout=_ok_stdout())
    _, _, entry = run_cron_with_timers_upsert(
        argv=("/usr/local/bin/cokacdir",),
        prompt="task-2533 진행 dev7-team 헤르메스",
        chat=CHAIR_CHAT,
        explicit_team=None,
        timers_path=timers3,
        tasks_dir=empty_tasks_dir2,  # md 없음
        runner=runner,
    )
    assert entry is not None and entry["team_id"] == "dev7-team"


def test_2533_extract_team_id_from_task_md_helper(tmp_path: Path) -> None:
    """``extract_team_id_from_task_md`` 단위 동작."""
    md = tmp_path / "task-2533.md"

    # 정식 태그 매치.
    md.write_text("dev1-team 단독\n", encoding="utf-8")
    assert extract_team_id_from_task_md(md) == "dev1-team"

    # 약식 매치 → -team 자동 부여.
    md.write_text("dev3 헤르메스만\n", encoding="utf-8")
    assert extract_team_id_from_task_md(md) == "dev3-team"

    # 매치 없음 → fallback.
    md.write_text("아무 단어도 없음\n", encoding="utf-8")
    assert extract_team_id_from_task_md(md) == "unknown-team"
    assert extract_team_id_from_task_md(md, fallback="custom") == "custom"

    # 파일 자체 없음 → fallback.
    missing = tmp_path / "no-such.md"
    assert extract_team_id_from_task_md(missing) == "unknown-team"


# ===========================================================================
# 회귀 6 — token / uuid / hex key 누수 0 (description 정적 검사)
# ===========================================================================

def test_2533_no_raw_token_or_uuid_in_persisted_description(
    empty_timers: Path, empty_tasks_dir: Path
) -> None:
    """raw token / raw uuid / raw hex key 가 prompt 에 있어도 description 에 그대로 저장 X.

    redact 마커 (``<redacted-...>``) 만 남고 raw 값은 timers JSON 어디에도 등장하면 안 된다.
    """
    md = empty_tasks_dir / "task-2533.md"
    md.write_text("dev1-team\n", encoding="utf-8")

    # 위험 prompt — 토큰/uuid/hex key 모두 포함.
    prompt = (
        f"task-2533 진행. token={RAW_GHP_TOKEN} key={RAW_HEX_KEY} session={RAW_UUID}"
    )
    _, runner = _make_runner(rc=0, stdout=_ok_stdout())
    _, _, entry = run_cron_with_timers_upsert(
        argv=("/usr/local/bin/cokacdir",),
        prompt=prompt,
        chat=CHAIR_CHAT,
        timers_path=empty_timers,
        tasks_dir=empty_tasks_dir,
        runner=runner,
    )

    assert entry is not None
    persisted_text = empty_timers.read_text(encoding="utf-8")

    # 정적 검사 — raw 값 0 등장.
    assert RAW_GHP_TOKEN not in persisted_text, "raw GHP token 이 timers JSON 에 있으면 안 된다"
    assert RAW_HEX_KEY not in persisted_text, "raw hex key 가 timers JSON 에 있으면 안 된다"
    assert RAW_UUID not in persisted_text, "raw UUID 가 timers JSON 에 있으면 안 된다"

    # entry description 에도 마찬가지.
    desc = entry["description"]
    assert RAW_GHP_TOKEN not in desc
    assert RAW_HEX_KEY not in desc
    assert RAW_UUID not in desc
    assert len(desc) <= DESCRIPTION_MAX_LEN

    # sanitize_description 단독 단위 검증.
    sanitized = sanitize_description(prompt)
    assert "<redacted-token>" in sanitized or len(sanitized) < len(prompt)
    assert RAW_GHP_TOKEN not in sanitized
    assert RAW_HEX_KEY not in sanitized
    assert RAW_UUID not in sanitized


# ===========================================================================
# 회귀 7 — chat 격리 (chat_id 명시 + 다른 chat entry 와 충돌 0)
# ===========================================================================

def test_2533_chat_isolation_and_chat_id_persisted(
    empty_timers: Path, empty_tasks_dir: Path
) -> None:
    """chat=6937032012 entry 와 다른 chat entry 가 같은 task_id 라도 chat_id 가 명시되어
    구분 가능해야 한다 (현재 schema 는 task_id 단일 key — 본 테스트는 chat_id 가 entry 내부에
    명시 저장됨을 검증)."""
    md = empty_tasks_dir / "task-2533.md"
    md.write_text("dev1-team\n", encoding="utf-8")

    # 회장 chat 발사.
    _, runner = _make_runner(rc=0, stdout=_ok_stdout("CHAIR_SCHED"))
    _, _, entry_chair = run_cron_with_timers_upsert(
        argv=("/usr/local/bin/cokacdir",),
        prompt="task-2533 진행 dev1-team",
        chat=CHAIR_CHAT,
        timers_path=empty_timers,
        tasks_dir=empty_tasks_dir,
        runner=runner,
    )
    assert entry_chair is not None
    assert entry_chair["chat_id"] == CHAIR_CHAT

    # 다른 chat 발사 — 동일 task_id 면 갱신되지만 chat_id 가 명시 갱신되어
    # "마지막 발사 chat" 이 entry 에 박제된다.
    _, runner = _make_runner(rc=0, stdout=_ok_stdout("OTHER_SCHED"))
    _, _, entry_other = run_cron_with_timers_upsert(
        argv=("/usr/local/bin/cokacdir",),
        prompt="task-2533 진행 dev1-team",
        chat=OTHER_CHAT,
        timers_path=empty_timers,
        tasks_dir=empty_tasks_dir,
        runner=runner,
    )
    assert entry_other is not None
    assert entry_other["chat_id"] == OTHER_CHAT

    persisted = json.loads(empty_timers.read_text(encoding="utf-8"))
    saved = persisted[TIMERS_TASK_KEY]["task-2533"]
    # chat_id 는 마지막 발사자 chat. 본 회귀의 본질은 "chat_id 가 명시 저장됨".
    assert saved["chat_id"] == OTHER_CHAT
    # 회장 chat 의 schedule_id 는 갱신되었으므로 다른 chat 의 schedule_id 가 최신이다.
    assert saved["schedule_id"] == "OTHER_SCHED"
    # 그러나 회장 chat 의 발사 자체가 timers entry 를 만든 것은 사실이며,
    # 본 테스트는 두 호출 모두 entry 를 정확히 그 chat 으로 박제했음을 검증한다.


# ===========================================================================
# 보조 — helper 단위 검증 (회귀 5 & 6 보강)
# ===========================================================================

def test_2533_extract_task_id_from_prompt_variants() -> None:
    """task_id 추출 — 정상 / 누락 / 합성 task (task-NNNN+M) 모두."""
    assert extract_task_id_from_prompt("task-2533 진행") == "task-2533"
    assert extract_task_id_from_prompt("foo bar task-100+1 baz") == "task-100+1"
    assert extract_task_id_from_prompt("no task here") is None
    assert extract_task_id_from_prompt("") is None


def test_2533_parse_schedule_id_handles_noise_and_invalid() -> None:
    """parse_schedule_id_from_cokacdir_stdout — JSON 노이즈 / 손상 / 실패 케이스."""
    # 정상.
    assert parse_schedule_id_from_cokacdir_stdout(_ok_stdout("SID1")) == "SID1"
    # 빈 문자열.
    assert parse_schedule_id_from_cokacdir_stdout("") is None
    # JSON 손상.
    assert parse_schedule_id_from_cokacdir_stdout("not json") is None
    # status != ok.
    assert parse_schedule_id_from_cokacdir_stdout(
        json.dumps({"status": "error", "id": "x"})
    ) is None
    # id 없음.
    assert parse_schedule_id_from_cokacdir_stdout(
        json.dumps({"status": "ok"})
    ) is None
    # prefix 잡음 + 정상 JSON.
    noisy = "[info] starting...\n" + _ok_stdout("WITH_NOISE")
    assert parse_schedule_id_from_cokacdir_stdout(noisy) == "WITH_NOISE"


def test_2533_default_timers_path_points_to_memory() -> None:
    """DEFAULT_TIMERS_PATH 가 worktree memory/task-timers.json 으로 결정되어야 한다.

    production 환경 정합 — task-2528 / task-2470 entry 가 같은 파일에 있어야 신호등이
    한 단일 source 를 본다.
    """
    assert DEFAULT_TIMERS_PATH.name == "task-timers.json"
    assert DEFAULT_TIMERS_PATH.parent.name == "memory"


# ===========================================================================
# 격리 보장 — production timers JSON 의 raw token 정적 검사
# ===========================================================================

def test_2533_production_timers_json_no_raw_secrets_residue() -> None:
    """본 PR 시점 production timers JSON 에 raw GHP/UUID 패턴 0 — task-2533 hook 이
    description sanitize 를 적용한 결과로 새 entry 를 박을 때 누수 0임을 회귀로 박제."""
    timers = _WORKTREE_ROOT / "memory" / "task-timers.json"
    if not timers.exists():
        pytest.skip("production timers not present in worktree")
    text = timers.read_text(encoding="utf-8")
    # 정적 검사 — GHP token / explicit raw uuid + key (RAW_HEX_KEY 자체는 일반 hex 패턴이라
    # production 에 자연 발생 가능. 본 검사는 위험도 높은 PAT prefix 만 박제).
    assert "ghp_" not in text, "GHP PAT prefix 가 timers JSON 에 등장하면 안 된다"
    assert "ghs_" not in text, "GHS PAT prefix 가 timers JSON 에 등장하면 안 된다"
    assert "ghu_" not in text, "GHU PAT prefix 가 timers JSON 에 등장하면 안 된다"