# -*- coding: utf-8 -*-
"""tests/regression/test_dispatch_spawn_verification_2942.py — task-2942 회귀.

dispatch spawn 검증 결선(false-OK 감지) 보호:
  1. marker 있으면 SPAWNED → status 무손상
  2. marker 없고 child 없으면 DISPATCH_FALSE_OK → status 명시 + handoff marker 박제
  3. TIMEOUT_BOT_ALIVE_BUT_NO_MARKER → 경고 필드만, status 유지
  4. ★ 무손상: verify_fn None(import 실패) / 예외 / 비-dict / kill-switch OFF
  5. spawn-confirmed marker 생성 + _verify_bot_spawn 과의 round-trip
  6. _verify_bot_spawn timeout_sec 하위호환
  7. marker 생성 경로 = 봇 부팅 훅 (task-2946: 프롬프트 주입 → 부팅 훅 이전)

대상 모듈은 stdlib 전용이므로 파일 경로로 직접 로드한다(무거운 dispatch 패키지
__init__ 로드 회피). 외부 네트워크 / 실제 cokacdir 호출 없음.
"""
from __future__ import annotations

import importlib.util
import json
import pathlib
import sys

import pytest

_ROOT = pathlib.Path(__file__).resolve().parents[2]


def _load(name: str, relpath: str):
    """workspace 상대경로의 .py 를 독립 모듈로 로드."""
    path = _ROOT / relpath
    spec = importlib.util.spec_from_file_location(name, path)
    assert spec is not None and spec.loader is not None, f"spec 실패: {path}"
    mod = importlib.util.module_from_spec(spec)
    sys.modules[name] = mod
    spec.loader.exec_module(mod)
    return mod


sv = _load("_t2942_spawn_verification", "dispatch/spawn_verification.py")
tsc = _load("_t2942_terminal_state_classifier", "scripts/harness/v36/terminal_state_classifier.py")


@pytest.fixture(autouse=True)
def _clean_env(monkeypatch):
    """검증 관련 env 를 매 테스트마다 초기화(호스트 설정 누출 방지)."""
    for key in (sv.ENABLED_ENV, sv.TOTAL_WAIT_ENV, sv.MARKER_TIMEOUT_ENV, sv.EVENTS_DIR_ENV):
        monkeypatch.delenv(key, raising=False)
    return monkeypatch


def _base_result() -> dict:
    return {"status": "dispatched", "task_id": "task-9999", "team": "dev1-team"}


# ---------------------------------------------------------------------------
# 1. marker 있으면 SPAWNED → status 무손상
# ---------------------------------------------------------------------------
def test_spawned_keeps_status_untouched():
    result = sv.verify_and_annotate(
        _base_result(), "task-9999", "dev1-team", verify_fn=lambda *a, **k: sv.SPAWNED
    )
    assert result["status"] == "dispatched"
    assert result[sv.SPAWN_VERIFICATION_KEY] == sv.SPAWNED
    assert sv.SPAWN_VERIFICATION_REASON_KEY not in result


def test_spawned_via_real_verify_fn_with_marker(tmp_path):
    """실제 _verify_bot_spawn + 실제 marker 로 SPAWNED 판정 (round-trip)."""
    marker = sv.write_spawn_confirmed_marker("task-9999", "dev1-team", pid=1234, events_dir=str(tmp_path))
    assert marker is not None

    verdict = tsc._verify_bot_spawn(
        "task-9999", "dev1-team", None, events_dir=str(tmp_path), timeout_sec=1
    )
    assert verdict == "SPAWNED"


# ---------------------------------------------------------------------------
# 2. marker 없고 child 없으면 DISPATCH_FALSE_OK → status 명시 + handoff 박제
# ---------------------------------------------------------------------------
def test_false_ok_sets_status_and_emits_handoff(tmp_path):
    calls = []

    def _fake_handoff(task_id, terminal_state, **kw):
        calls.append((task_id, terminal_state, kw))
        return {"path": str(tmp_path / "handoff.json")}

    result = sv.verify_and_annotate(
        _base_result(),
        "task-9999",
        "dev1-team",
        events_dir=str(tmp_path),
        verify_fn=lambda *a, **k: sv.DISPATCH_FALSE_OK,
        handoff_fn=_fake_handoff,
    )

    assert result["status"] == sv.DISPATCH_FALSE_OK
    assert result[sv.SPAWN_VERIFICATION_KEY] == sv.DISPATCH_FALSE_OK
    assert result[sv.SPAWN_VERIFICATION_REASON_KEY] == sv.SPAWN_FALSE_OK_FAILURE_KIND
    assert result["spawn_verification_handoff"]

    # handoff marker 는 INFRA_DEFECT + spawn_false_ok 사유로 박제된다
    assert len(calls) == 1
    task_id, terminal_state, kw = calls[0]
    assert task_id == "task-9999"
    assert terminal_state == sv.SPAWN_FALSE_OK_TERMINAL_STATE == "INFRA_DEFECT"
    assert kw["failure_kind"] == "spawn_false_ok"
    assert kw["events_dir"] == str(tmp_path)


def test_false_ok_via_real_verify_fn_without_marker(tmp_path):
    """실제 _verify_bot_spawn: marker 없음 + child 없음 → DISPATCH_FALSE_OK."""
    verdict = tsc._verify_bot_spawn(
        "task-9999", "dev1-team", None, events_dir=str(tmp_path), timeout_sec=0
    )
    assert verdict == "DISPATCH_FALSE_OK"


def test_false_ok_survives_handoff_failure(tmp_path):
    """handoff 박제가 실패해도 status 표면화는 유지되고 예외는 전파되지 않는다."""

    def _boom(*a, **k):
        raise RuntimeError("disk full")

    result = sv.verify_and_annotate(
        _base_result(),
        "task-9999",
        "dev1-team",
        events_dir=str(tmp_path),
        verify_fn=lambda *a, **k: sv.DISPATCH_FALSE_OK,
        handoff_fn=_boom,
    )
    assert result["status"] == sv.DISPATCH_FALSE_OK
    assert "spawn_verification_handoff" not in result


# ---------------------------------------------------------------------------
# 3. TIMEOUT_BOT_ALIVE_BUT_NO_MARKER → 경고만, status 유지
# ---------------------------------------------------------------------------
def test_timeout_bot_alive_keeps_status():
    result = sv.verify_and_annotate(
        _base_result(),
        "task-9999",
        "dev1-team",
        verify_fn=lambda *a, **k: sv.TIMEOUT_BOT_ALIVE_BUT_NO_MARKER,
    )
    assert result["status"] == "dispatched"
    assert result[sv.SPAWN_VERIFICATION_WARNING_KEY] == sv.TIMEOUT_BOT_ALIVE_BUT_NO_MARKER
    assert sv.SPAWN_VERIFICATION_REASON_KEY not in result


# ---------------------------------------------------------------------------
# 4. ★ 무손상 (비차단) 계약
# ---------------------------------------------------------------------------
def test_import_none_keeps_original_ok(monkeypatch):
    """_verify_bot_spawn 미탑재(import 실패) → 기존 ok 무손상."""
    monkeypatch.setattr(sv, "_load_verify_fn", lambda: None)
    original = _base_result()
    result = sv.verify_and_annotate(original, "task-9999", "dev1-team")
    assert result is original
    assert result["status"] == "dispatched"
    assert sv.SPAWN_VERIFICATION_KEY not in result


def test_verify_exception_keeps_original_ok():
    """검증 중 예외 → 기존 ok 무손상 (예외 전파 금지)."""

    def _boom(*a, **k):
        raise RuntimeError("verify exploded")

    original = _base_result()
    result = sv.verify_and_annotate(original, "task-9999", "dev1-team", verify_fn=_boom)
    assert result is original
    assert result["status"] == "dispatched"
    assert sv.SPAWN_VERIFICATION_KEY not in result


def test_non_dict_result_returned_as_is():
    for bad in (None, "dispatched", 42, ["x"]):
        assert sv.verify_and_annotate(bad, "task-9999", "dev1-team") is bad
        assert sv.annotate_spawn_verification(bad, sv.DISPATCH_FALSE_OK) is bad


def test_kill_switch_disables_verification(monkeypatch):
    """DISPATCH_SPAWN_VERIFY_ENABLED=0 → 검증 스킵, status 무손상."""
    monkeypatch.setenv(sv.ENABLED_ENV, "0")

    def _must_not_run(*a, **k):  # pragma: no cover - 호출되면 실패
        raise AssertionError("kill-switch OFF 인데 검증이 실행됨")

    result = sv.verify_and_annotate(
        _base_result(), "task-9999", "dev1-team", verify_fn=_must_not_run
    )
    assert result["status"] == "dispatched"
    assert result[sv.SPAWN_VERIFICATION_KEY] == sv.VERIFY_DISABLED


def test_kill_switch_default_on(monkeypatch):
    assert sv.is_enabled() is True
    for falsey in ("0", "false", "FALSE", "no", "off", ""):
        monkeypatch.setenv(sv.ENABLED_ENV, falsey)
        assert sv.is_enabled() is False
    for truthy in ("1", "true", "yes"):
        monkeypatch.setenv(sv.ENABLED_ENV, truthy)
        assert sv.is_enabled() is True


def test_verdict_none_keeps_original():
    original = _base_result()
    assert sv.annotate_spawn_verification(original, None) is original
    assert sv.SPAWN_VERIFICATION_KEY not in original


def test_legacy_verify_fn_without_timeout_sec_is_supported():
    """구버전 _verify_bot_spawn(timeout_sec 미지원) → TypeError 흡수 후 재시도."""
    seen = {}

    def _legacy(task_id, expected_bot_id, child_pid=None, events_dir="memory/events"):
        seen["called"] = True
        return sv.SPAWNED

    result = sv.verify_and_annotate(
        _base_result(), "task-9999", "dev1-team", verify_fn=_legacy
    )
    assert seen.get("called") is True
    assert result[sv.SPAWN_VERIFICATION_KEY] == sv.SPAWNED
    assert result["status"] == "dispatched"


# ---------------------------------------------------------------------------
# 5. spawn-confirmed marker 생성
# ---------------------------------------------------------------------------
def test_write_spawn_confirmed_marker_contents(tmp_path):
    path = sv.write_spawn_confirmed_marker("task-2942", "dev1-team", pid=4321, events_dir=str(tmp_path))
    assert path is not None

    p = pathlib.Path(path)
    assert p.exists()
    # _verify_bot_spawn 의 glob 패턴과 파일명이 일치해야 한다
    assert p.name.startswith("task-2942.spawn-confirmed-")
    assert p.name.endswith(".json")

    payload = json.loads(p.read_text(encoding="utf-8"))
    assert payload["task_id"] == "task-2942"
    assert payload["bot_id"] == "dev1-team"
    assert payload["pid"] == 4321
    assert payload["ts"]


def test_write_spawn_confirmed_marker_creates_dir(tmp_path):
    nested = tmp_path / "a" / "b" / "events"
    path = sv.write_spawn_confirmed_marker("task-2942", "dev2-team", events_dir=str(nested))
    assert path is not None and pathlib.Path(path).exists()


def test_write_spawn_confirmed_marker_defaults_pid(tmp_path):
    import os as _os

    path = sv.write_spawn_confirmed_marker("task-2942", "dev1-team", events_dir=str(tmp_path))
    payload = json.loads(pathlib.Path(path).read_text(encoding="utf-8"))
    assert payload["pid"] == _os.getpid()


def test_write_spawn_confirmed_marker_never_raises(tmp_path):
    """실패 시 None 반환 (봇 작업을 죽이지 않는다)."""
    assert sv.write_spawn_confirmed_marker("", "dev1-team", events_dir=str(tmp_path)) is None
    # 파일을 디렉토리 자리에 놓아 mkdir 실패 유도
    blocker = tmp_path / "blocked"
    blocker.write_text("not a dir", encoding="utf-8")
    assert sv.write_spawn_confirmed_marker("task-2942", "dev1-team", events_dir=str(blocker)) is None


# ---------------------------------------------------------------------------
# 6. 대기 예산 (cron 발사 offset 반영) + timeout_sec 하위호환
# ---------------------------------------------------------------------------
def test_resolve_wait_sec_includes_cron_delay(monkeypatch):
    """기본 = cron 발사 offset(10s) + 마커 예산(15s). 봇이 존재할 수 없는 구간 오판 방지."""
    assert sv.resolve_wait_sec(10) == 25
    assert sv.resolve_wait_sec(0) == 15


def test_resolve_wait_sec_env_overrides(monkeypatch):
    monkeypatch.setenv(sv.MARKER_TIMEOUT_ENV, "30")
    assert sv.resolve_wait_sec(10) == 40
    monkeypatch.setenv(sv.TOTAL_WAIT_ENV, "3")
    assert sv.resolve_wait_sec(10) == 3


def test_resolve_wait_sec_bad_env_falls_back(monkeypatch):
    monkeypatch.setenv(sv.MARKER_TIMEOUT_ENV, "not-a-number")
    assert sv.resolve_wait_sec(10) == 25
    monkeypatch.setenv(sv.TOTAL_WAIT_ENV, "garbage")
    assert sv.resolve_wait_sec(10) == 25


def test_verify_bot_spawn_timeout_sec_backward_compatible(tmp_path, monkeypatch):
    """timeout_sec 미지정 시 기존 env 경로 유지 (§5.2.1 하위호환)."""
    monkeypatch.setenv("FAILURE_CALLBACK_2712_SPAWN_TIMEOUT_SEC", "0")
    assert (
        tsc._verify_bot_spawn("task-9999", "dev1-team", None, events_dir=str(tmp_path))
        == "DISPATCH_FALSE_OK"
    )
    sv.write_spawn_confirmed_marker("task-9999", "dev1-team", events_dir=str(tmp_path))
    monkeypatch.setenv("FAILURE_CALLBACK_2712_SPAWN_TIMEOUT_SEC", "1")
    assert (
        tsc._verify_bot_spawn("task-9999", "dev1-team", None, events_dir=str(tmp_path))
        == "SPAWNED"
    )


def test_marker_checked_at_least_once_even_with_zero_timeout(tmp_path):
    """★ 회귀: timeout_sec=0 이어도 marker 는 최소 1회 확인해야 한다.

    기존 while 선판정 구조는 timeout_sec=0 일 때 glob 을 한 번도 돌지 않아, 이미
    marker 가 있는(=봇이 빠르게 뜬) 경우까지 DISPATCH_FALSE_OK 로 오판했다.
    """
    sv.write_spawn_confirmed_marker("task-9999", "dev1-team", events_dir=str(tmp_path))
    verdict = tsc._verify_bot_spawn(
        "task-9999", "dev1-team", None, events_dir=str(tmp_path), timeout_sec=0
    )
    assert verdict == "SPAWNED"


def test_zero_wait_end_to_end_does_not_false_flag(tmp_path):
    """marker 존재 + 대기 0 → status 무손상 (false positive 금지)."""
    sv.write_spawn_confirmed_marker("task-9999", "dev1-team", events_dir=str(tmp_path))
    result = sv.verify_and_annotate(
        _base_result(),
        "task-9999",
        "dev1-team",
        events_dir=str(tmp_path),
        verify_fn=tsc._verify_bot_spawn,
        dispatch_delay_sec=0,
    )
    assert result["status"] == "dispatched"
    assert result[sv.SPAWN_VERIFICATION_KEY] == sv.SPAWNED


def test_verify_bot_spawn_timeout_zero_is_instant(tmp_path):
    """timeout_sec=0 → 대기 없이 즉시 판정 (테스트가 15초 blocking 되지 않음)."""
    import time as _time

    t0 = _time.monotonic()
    tsc._verify_bot_spawn("task-9999", "dev1-team", None, events_dir=str(tmp_path), timeout_sec=0)
    assert _time.monotonic() - t0 < 1.0


# ---------------------------------------------------------------------------
# 7. marker 생성 경로 = 봇 부팅 훅 (task-2946 에서 프롬프트 주입에서 이전)
# ---------------------------------------------------------------------------
def test_team_prompts_no_longer_injects_spawn_block():
    """★ task-2946: 마커 지시는 프롬프트에서 제거됐다 (4096자 절벽 회귀 방지).

    프롬프트로 되돌리면 dev1~7 이 다시 절벽 밖으로 밀려나므로 source 로 고정한다.
    """
    src = (_ROOT / "prompts" / "team_prompts.py").read_text(encoding="utf-8")
    assert "_build_spawn_confirmed_block" not in src
    assert "dispatch/spawn_verification.py --task-id" not in src


def test_boot_hook_writes_marker_for_dispatch_prompt(tmp_path, monkeypatch):
    """부팅 훅이 dispatch 프롬프트를 받으면 marker 를 만든다 (프롬프트 지시 대체)."""
    hook = _load("_t2946_boot_hook", "hooks/spawn_confirmed_boot_hook.py")
    prompt = "memory/tasks/task-2946.md 를 읽으세요\ncollector_role=ANU\n"
    assert hook.extract_task_id(prompt) == "task-2946"

    monkeypatch.setenv("FAILURE_CALLBACK_2712_EVENTS_DIR", str(tmp_path))
    path = sv.write_spawn_confirmed_marker("task-2946", "dev1-team", events_dir=str(tmp_path))
    assert path is not None
    # 훅이 만든 marker 로 _verify_bot_spawn 이 SPAWNED 를 판정해야 한다
    verdict = tsc._verify_bot_spawn("task-2946", "dev1-team", None, events_dir=str(tmp_path), timeout_sec=0)
    assert verdict == "SPAWNED"


def test_boot_hook_ignores_non_dispatch_prompt():
    """★ false-positive 방지: 일반 대화에서 task id 를 언급해도 marker 를 쓰면 안 된다."""
    hook = _load("_t2946_boot_hook_neg", "hooks/spawn_confirmed_boot_hook.py")
    # doctrine 지문 없음 → None
    assert hook.extract_task_id("task-2946 진행 상황 알려줘") is None
    assert hook.extract_task_id("memory/tasks/task-2946.md 좀 봐줘") is None
    # 작업파일 경로 없음 → None
    assert hook.extract_task_id("collector_role=ANU 규칙이 뭐야?") is None


def test_boot_hook_is_wired_into_registered_hook():
    """훅 등록 지점(~/.claude/settings.json)은 workspace 밖이므로, 이미 등록된
    workspace 소유 훅에서 호출되는지를 source 로 고정한다."""
    src = (_ROOT / "scripts" / "ensure-bot-memory.sh").read_text(encoding="utf-8")
    assert "hooks/spawn_confirmed_boot_hook.py" in src
    # 비차단 계약: 실패해도 봇 세션을 죽이지 않는다
    assert "|| true" in src


# ---------------------------------------------------------------------------
# 8. dispatch 결선 위치 (호출부 0건 회귀 방지)
# ---------------------------------------------------------------------------
def test_dispatch_wires_spawn_verification():
    """dispatch/__init__.py 의 성공 반환 경로 3곳에 결선돼 있어야 한다."""
    src = (_ROOT / "dispatch" / "__init__.py").read_text(encoding="utf-8")
    assert "def _verify_spawn_or_annotate" in src
    # 정상 / fallback-key / 복합업무 3 경로
    assert src.count("_result = _verify_spawn_or_annotate(") == 3
    assert "from dispatch.spawn_verification import verify_and_annotate" in src