"""스텝 락 — 죽음을 추측하지 않고 확인하는가.

이 파일이 지키는 계약:

1. **등록부**: 진 contender 의 퇴장이 산 소유자를 지우지 않는다
2. **판정**: 확정 사망만 DEAD, 나머지는 ALIVE 또는 UNKNOWN
3. **하트비트 만료는 자동으로 풀기가 아니다** (기본값)
4. **신원이 있는 UNKNOWN 은 경과 시간으로 안 뺏는다**
5. **기동 풀기**는 이 호스트 + 확인된 사망에만
"""

from datetime import datetime, timedelta, timezone

import pytest

from app.core import step_lock
from app.core.step_lock import (
    HeldLockRegistry,
    LockOwner,
    OwnerVerdict,
    is_pid_alive,
    judge_owner,
    parse_iso,
    process_identity,
)


KEY = ("proj", "epi", "step")


@pytest.fixture
def reg():
    return HeldLockRegistry()


# ── 1. 등록부 ──


def test_등록부는_같은_키에_여러_run_id를_담는다(reg):
    reg.register(KEY, "A")
    reg.register(KEY, "B")
    assert reg.holds(KEY, "A")
    assert reg.holds(KEY, "B")


def test_진_contender_의_퇴장이_산_소유자를_안_지운다(reg):
    """claim 에 진 B 가 물러날 때 A 의 항목까지 지우면, 그 뒤 판정이
    「등록부에 없다 → 죽었다」로 **살아있는 A 를 뺏는다.**"""
    reg.register(KEY, "A")       # A 가 먼저 잡았다
    reg.register(KEY, "B")       # B 도 노렸다 (claim 은 실패할 것)
    reg.unregister(KEY, "B")     # B 가 물러난다

    assert reg.holds(KEY, "A"), "A 가 등록부에서 사라졌다 — 락을 빼앗기게 된다"
    assert not reg.holds(KEY, "B")


def test_마지막_소유자가_나가면_키가_사라진다(reg):
    reg.register(KEY, "A")
    reg.unregister(KEY, "A")
    assert not reg.holds(KEY)


# ── 2. 판정 ──


def _owner(**kw) -> LockOwner:
    return LockOwner(**kw)


def test_이_프로세스가_등록부에_있으면_살아있다(reg):
    _host, _pid, boot_id = process_identity()
    reg.register(KEY, "run-1")
    verdict, _ = judge_owner(
        _owner(boot_id=boot_id, run_id="run-1"), key=KEY, registry=reg
    )
    assert verdict is OwnerVerdict.ALIVE


def test_이_프로세스인데_등록부에_없으면_죽었다(reg):
    """프로세스 안의 일은 프로세스가 안다 — 이건 확정 사망이다."""
    _host, _pid, boot_id = process_identity()
    verdict, reason = judge_owner(
        _owner(boot_id=boot_id, run_id="run-1"), key=KEY, registry=reg
    )
    assert verdict is OwnerVerdict.DEAD
    assert "등록부에 없다" in reason


def test_같은_호스트인데_PID_가_없으면_죽었다(reg, monkeypatch):
    monkeypatch.setattr(step_lock, "is_pid_alive", lambda pid: False)
    verdict, reason = judge_owner(
        _owner(host=step_lock.PROCESS_HOST, pid=999999, boot_id="다른-부팅"),
        key=KEY, registry=reg,
    )
    assert verdict is OwnerVerdict.DEAD
    assert "PID 가 없다" in reason


def test_하트비트_만료는_기본적으로_자동_풀기가_아니다(reg, monkeypatch):
    """하트비트가 멈춘 것은 죽음의 **증거가 아니라 정황**이다.
    결과물 쓰기를 막는 울타리가 얇은 동안은 자동으로 안 뺏는다."""
    monkeypatch.setattr(step_lock, "is_pid_alive", lambda pid: True)
    old = datetime.now(timezone.utc) - timedelta(seconds=600)
    verdict, reason = judge_owner(
        _owner(
            host=step_lock.PROCESS_HOST, pid=4242,
            boot_id="다른-부팅", heartbeat_at=old.isoformat(),
        ),
        key=KEY, registry=reg, lease_seconds=90,
    )
    assert verdict is OwnerVerdict.UNKNOWN
    assert "자동으로 풀기 안 함" in reason


def test_하트비트_풀기를_켜면_만료가_죽음이_된다(reg, monkeypatch):
    monkeypatch.setattr(step_lock, "is_pid_alive", lambda pid: True)
    old = datetime.now(timezone.utc) - timedelta(seconds=600)
    verdict, _ = judge_owner(
        _owner(
            host=step_lock.PROCESS_HOST, pid=4242,
            boot_id="다른-부팅", heartbeat_at=old.isoformat(),
        ),
        key=KEY, registry=reg, lease_seconds=90,
        allow_heartbeat_steal=True,
    )
    assert verdict is OwnerVerdict.DEAD


def test_신선한_하트비트는_살아있다(reg, monkeypatch):
    monkeypatch.setattr(step_lock, "is_pid_alive", lambda pid: True)
    now = datetime.now(timezone.utc)
    verdict, _ = judge_owner(
        _owner(
            host="다른-호스트", pid=4242,
            boot_id="다른-부팅", heartbeat_at=now.isoformat(),
        ),
        key=KEY, registry=reg, lease_seconds=90,
    )
    assert verdict is OwnerVerdict.ALIVE


def test_신원도_하트비트도_없으면_모른다(reg):
    """구 행 — 호출자가 경과 시간으로 판단하게 넘긴다."""
    verdict, _ = judge_owner(_owner(), key=KEY, registry=reg)
    assert verdict is OwnerVerdict.UNKNOWN


def test_PermissionError_는_살아있음이다(monkeypatch):
    """프로세스가 있는데 남의 것이라 신호를 못 보내는 것이다.
    죽음으로 읽으면 남의 일을 빼앗는다."""
    def _kill(pid, sig):
        raise PermissionError()
    monkeypatch.setattr(step_lock.os, "kill", _kill)
    assert is_pid_alive(12345) is True


def test_없는_PID_는_사망이다(monkeypatch):
    def _kill(pid, sig):
        raise ProcessLookupError()
    monkeypatch.setattr(step_lock.os, "kill", _kill)
    assert is_pid_alive(12345) is False


# ── 3. 시각 파싱 ──


def test_parse_iso_는_문자열과_datetime_을_둘_다_받는다():
    """`heartbeat_at` 은 timestamptz 라 드라이버가 datetime 을 준다.
    `started_at` 은 text 다. 둘 다 읽혀야 한다."""
    dt = datetime(2026, 8, 26, 1, 0, tzinfo=timezone.utc)
    assert parse_iso(dt) == dt
    assert parse_iso("2026-08-26T01:00:00+00:00") == dt
    assert parse_iso("2026-08-26T01:00:00Z") == dt
    assert parse_iso(None) is None
    assert parse_iso("말이 안 되는 값") is None


def test_tz_없는_값은_UTC_로_읽는다():
    parsed = parse_iso("2026-08-26T01:00:00")
    assert parsed is not None and parsed.tzinfo is timezone.utc


# ── 4. fork 대비 ──


def test_PID_가_바뀌면_신원과_등록부를_새로_만든다(monkeypatch):
    """pre-fork 로 띄우면 자식이 부모의 boot_id 와 등록부를 물려받는다.
    그대로 두면 자식이 부모의 락을 「내 것」으로 읽는다."""
    _h, _p, boot_before = process_identity()
    step_lock.REGISTRY.register(KEY, "부모-run")

    monkeypatch.setattr(step_lock.os, "getpid", lambda: 999_999)
    _h, pid_after, boot_after = process_identity()

    assert pid_after == 999_999
    assert boot_after != boot_before
    assert not step_lock.REGISTRY.holds(KEY), "자식이 부모의 등록부를 물려받았다"


# ── 5. resume 판정 — 「확정 사망만 자동으로 풀기」가 실제로 참인가 ──
#
# 하트비트 만료를 UNKNOWN 으로 낮춰도, 그 UNKNOWN 이 기존 경과 시간 경로로
# 흘러가면 3600초 뒤에 결국 뺏는다 — 계약이 시간이 지나면 무너지는 것이다.
# 신원이 있는 행은 경과 시간으로 안 뺏고, 신원이 통째로 없는 옛 행만
# 예전 동작(3600초)을 유지해야 한다.


def _runner(step_id: str = "scene_image_pipeline"):
    """DB 없이 판정만 부르는 최소 러너."""
    from app.core.step_runner import StepRunner

    r = object.__new__(StepRunner)
    r.step_id = step_id
    r.project_id = "proj"
    r.episode_id = "epi"
    r.run_id = "내-run"
    r.project_config = {}
    return r


def _row(**kw):
    base = {
        "status": "running",
        "run_id": "남의-run",
        "started_at": None,
        "owner_host": None,
        "owner_pid": None,
        "owner_boot_id": None,
        "heartbeat_at": None,
        "cancel_requested_at": None,
    }
    base.update(kw)
    return base


def test_신원_있는_UNKNOWN_은_경과_시간으로_안_뺏는다(monkeypatch):
    from app.core.step_runner import ResumeAction

    monkeypatch.setattr(step_lock, "is_pid_alive", lambda pid: True)
    아주_오래전 = (datetime.now(timezone.utc) - timedelta(seconds=99_999)).isoformat()
    멈춘_하트비트 = (datetime.now(timezone.utc) - timedelta(seconds=600)).isoformat()

    decision = _runner()._evaluate_running_state(_row(
        started_at=아주_오래전,
        owner_host=step_lock.PROCESS_HOST,
        owner_pid=4242,
        owner_boot_id="다른-부팅",
        heartbeat_at=멈춘_하트비트,
    ))

    assert decision.action is ResumeAction.BLOCK, (
        "신원이 있는데 경과 시간만으로 뺏었다 — 「확정 사망만 풀기」가 거짓이 된다"
    )
    assert "release" in decision.reason, "풀기하는 길을 안 알려 준다"


def test_신원_없는_옛_행은_예전대로_경과_시간으로_풀린다():
    """오늘까지의 동작 — 새 위험을 안 들인다."""
    from app.core.step_runner import ResumeAction

    아주_오래전 = (datetime.now(timezone.utc) - timedelta(seconds=99_999)).isoformat()
    decision = _runner()._evaluate_running_state(_row(started_at=아주_오래전))
    assert decision.action is ResumeAction.STALE_RUNNING_RECOVERY


def test_신원_없는_옛_행이_아직_안_지났으면_막는다():
    from app.core.step_runner import ResumeAction

    방금 = datetime.now(timezone.utc).isoformat()
    decision = _runner()._evaluate_running_state(_row(started_at=방금))
    assert decision.action is ResumeAction.BLOCK


def test_확정_사망은_즉시_풀린다(monkeypatch):
    """새벽에 잃은 40분이 여기서 사라진다 — 서버를 죽였고, 그 PID 는 없다."""
    from app.core.step_runner import ResumeAction

    monkeypatch.setattr(step_lock, "is_pid_alive", lambda pid: False)
    방금 = datetime.now(timezone.utc).isoformat()
    decision = _runner()._evaluate_running_state(_row(
        started_at=방금,
        owner_host=step_lock.PROCESS_HOST,
        owner_pid=999_999,
        owner_boot_id="죽은-부팅",
    ))
    assert decision.action is ResumeAction.STALE_RUNNING_RECOVERY, (
        "소유 프로세스가 없는데 풀기를 안 한다 — 40분 손실이 그대로다"
    )


def test_cancelled_는_재개하면_다시_돈다():
    """운영자가 세운 것은 실패가 아니다. 표를 내리고 resume 하면 이어서 간다."""
    from app.core.step_runner import ResumeAction

    r = _runner()
    r._get_step_run = lambda sid: _row(status="cancelled")
    decision = r._evaluate_resume_decision("resume")
    assert decision.action is ResumeAction.RERUN_SELF
    assert decision.origin == "prior_state"


# ── 6. 정지 기록이 다른 예외 흐름을 안 가리는가 ──
#
# 정지를 별도 `except AppError` 절로 빼면, `_execute` 가 올린 **다른** AppError
# 까지 그 절이 가로챈다. 거기서 re-raise 하면 형제 except 절은 안 도므로
# `status='failed'` 기록이 통째로 사라진다. 같은 절 안에서 갈라야 한다.


def _finalize_runner(raising):
    """_execute 가 주어진 예외를 던지는 최소 러너."""
    from unittest.mock import MagicMock
    from app.core.step_runner import StepRunner

    r = object.__new__(StepRunner)
    r.step_id = "text_cleanup"
    r.project_id = "p"
    r.episode_id = "e"
    r.run_id = "r"
    r.project_config = {}
    r.opik_context = {}
    r.manifest = {}
    r.db = MagicMock()
    r._execute = lambda mode: (_ for _ in ()).throw(raising)
    r.기록 = []
    r._update_step_run_strict = lambda status, **kw: r.기록.append((status, kw))
    return r


def test_정지는_실패가_아니라_cancelled_로_적힌다():
    from app.core.errors import AppError

    r = _finalize_runner(
        AppError(code="step.cancelled", message="멈추라는 말이 왔다", status_code=409)
    )
    with pytest.raises(AppError) as caught:
        r._execute_and_finalize_inner("resume")

    assert caught.value.code == "step.cancelled"
    assert [s for s, _ in r.기록] == ["cancelled"], (
        f"정지가 {r.기록} 로 적혔다 — 운영자가 세운 것과 깨진 것이 구별 안 된다"
    )


def test_정지가_아닌_AppError_는_여전히_failed_로_적힌다():
    """정지 분기가 다른 AppError 를 가로채 failed 기록을 건너뛰면 안 된다."""
    from app.core.errors import AppError

    r = _finalize_runner(
        AppError(code="step.something_else", message="진짜 깨졌다", status_code=500)
    )
    with pytest.raises(AppError):
        r._execute_and_finalize_inner("resume")

    assert [s for s, _ in r.기록] == ["failed"], (
        f"AppError 가 {r.기록} 로 적혔다 — 실패 기록이 사라졌다"
    )


def test_일반_예외도_failed_로_적힌다():
    r = _finalize_runner(RuntimeError("그냥 터졌다"))
    with pytest.raises(RuntimeError):
        r._execute_and_finalize_inner("resume")
    assert [s for s, _ in r.기록] == ["failed"]


# ── 7. 유료 호출 길목의 정지 ──
#
# 씬 진입점 하나만 막으면 새는 곳이 남는다 — 한 씬 안에서 roll 여러 번,
# 재시도, cine 변환이 각각 돈을 쓴다. 모든 이미지 호출이 지나는
# `reserve_current_call` 에서 봐야 어느 갈래로 가도 선다.


def _멈춰라():
    from app.core.errors import AppError
    raise AppError(code="step.cancelled", message="세워라", status_code=409)


def test_정지_표가_걸리면_유료_호출_길목에서_멈춘다():
    from app.core.errors import AppError
    from app.core.image_call_budget import (
        install_stop_check, uninstall_stop_check, reserve_current_call,
    )

    install_stop_check(_멈춰라)
    try:
        with pytest.raises(AppError) as caught:
            reserve_current_call(source="테스트")
        assert caught.value.code == "step.cancelled"
    finally:
        uninstall_stop_check()


def test_정지_표가_없으면_길목은_그대로_통과한다():
    """예산도 표도 없는 경로의 동작이 안 바뀌어야 한다."""
    from app.core.image_call_budget import reserve_current_call, uninstall_stop_check

    uninstall_stop_check()
    reserve_current_call(source="테스트")  # 예외 없이 통과


def test_정지_표는_예산_없는_경로에서도_들린다():
    """`budget is None` early return 뒤에 두면 이 경로가 통째로 빠진다."""
    from app.core.errors import AppError
    from app.core.image_call_budget import (
        get_current_budget, install_stop_check, uninstall_stop_check,
        reserve_current_call, uninstall_budget,
    )

    uninstall_budget()
    assert get_current_budget() is None
    install_stop_check(_멈춰라)
    try:
        with pytest.raises(AppError):
            reserve_current_call(source="테스트")
    finally:
        uninstall_stop_check()


def test_정지_표가_pool_worker_로_실려_간다():
    """스레드 지역이라 안 나르면 팬아웃 안에서 정지가 통째로 안 들린다."""
    from concurrent.futures import ThreadPoolExecutor
    from app.core.errors import AppError
    from app.core.image_call_budget import (
        bind_current_budget, install_stop_check, uninstall_stop_check,
        reserve_current_call,
    )

    def _worker():
        reserve_current_call(source="worker")
        return "돈을 썼다"

    install_stop_check(_멈춰라)
    try:
        bound = bind_current_budget(_worker)
        with ThreadPoolExecutor(max_workers=1) as pool:
            future = pool.submit(bound)
            with pytest.raises(AppError) as caught:
                future.result()
        assert caught.value.code == "step.cancelled"
    finally:
        uninstall_stop_check()


def test_글_LLM_도_같은_표를_본다():
    """이미지만 막으면 `scene_detail` 같은 글 팬아웃은 정지를 못 듣는다.

    ★**`call_structured` 로 태운다.** 처음엔 검사을 `router_completion` 에
     걸고 그 함수를 직접 불러 시험했는데, 실제 글 호출부는 그 함수를 안 쓰고
     `_completion` 을 직접 부른다 — 조립하는 자리를 재고 「나가는 것을 쟀다」고
     말한 셈이었다. 프로덕션이 쓰는 문으로 들어가야 검사을 실제로 잰다.
    """
    from app.core.errors import AppError
    from app.core.image_call_budget import install_stop_check, uninstall_stop_check
    from app.modules.llm import llm_client

    install_stop_check(_멈춰라)
    try:
        for 문 in ("call_structured", "call_text", "call_multiturn"):
            with pytest.raises(AppError) as caught:
                getattr(llm_client, 문)(
                    "테스트", "sys", [{"type": "text", "text": "본문"}],
                    {"type": "object"})
            assert caught.value.code == "step.cancelled", f"{문} 이 안 멈췄다"
    finally:
        uninstall_stop_check()


def test_router_completion_도_같은_표를_본다():
    """드물게 이 문으로 들어오는 호출부도 있다 — 아래 `_completion` 이 잡는다."""
    from app.core.errors import AppError
    from app.core.image_call_budget import install_stop_check, uninstall_stop_check
    from app.modules.llm import llm_client

    install_stop_check(_멈춰라)
    try:
        with pytest.raises(AppError) as caught:
            llm_client.router_completion(model="gemini-pro", messages=[])
        assert caught.value.code == "step.cancelled"
    finally:
        uninstall_stop_check()


# ── 8. 에피소드 하나에 배치 하나 ────────────────────────────────────
#
# `run-all` 은 category(analysis/image/all) 별로 다른 키로 등록한다. 앞에서
# 모든 category 를 훑긴 하지만 훑은 뒤 등록까지 preflight·설정 읽기·스텝
# 고르기가 끼어 있어, 요청 둘이 그 틈에 나란히 훑으면 둘 다 통과한다.


def test_에피소드_자리는_한_번만_잡힌다():
    from app.core.task_registry import (
        claim_episode_run, release_episode_run, episode_run_holder,
    )

    키 = "run_all:p:e"
    release_episode_run(키)
    try:
        assert claim_episode_run(키, holder="analysis") is True
        assert claim_episode_run(키, holder="image") is False, (
            "category 가 다르다고 두 배치가 같이 돌면 정지 표 하나로 둘 다 못 세운다"
        )
        assert episode_run_holder(키) == "analysis"
    finally:
        release_episode_run(키)


def test_자리를_놓으면_다음_요청이_들어온다():
    from app.core.task_registry import claim_episode_run, release_episode_run

    키 = "run_all:p:e2"
    release_episode_run(키)
    assert claim_episode_run(키, holder="analysis") is True
    release_episode_run(키)
    try:
        assert claim_episode_run(키, holder="image") is True
    finally:
        release_episode_run(키)


def test_동시에_들어와도_하나만_이긴다():
    """검사와 잡기가 한 자물쇠 안에 있어야 한다 — 나누면 그 사이가 틈이다."""
    from concurrent.futures import ThreadPoolExecutor
    from app.core.task_registry import claim_episode_run, release_episode_run

    키 = "run_all:p:e3"
    release_episode_run(키)
    try:
        with ThreadPoolExecutor(max_workers=8) as pool:
            결과 = list(pool.map(lambda i: claim_episode_run(키, holder=str(i)),
                                range(8)))
        assert sum(결과) == 1, f"여덟이 달려들어 {sum(결과)}개가 잡았다"
    finally:
        release_episode_run(키)