"""shot_continuity — 자세 정본·중의어 교정·carried 결정론 로직 테스트."""
from __future__ import annotations

from typing import Any, Dict, List

import pytest

from app.core.errors import AppError
from app.modules.pipeline.shot_continuity_author import run_shot_continuity


SHOTS = [
    {"scene_index": 1, "shot_index": 1, "description": "쓰러진 남자, 손에 사진"},
    {"scene_index": 1, "shot_index": 2, "description": "방을 둘러보는 여자"},
]
SCENE_TEXTS = {1: "씬1 원문 전문..."}
SCENE_HEADINGS = {1: "S#1. 방 (밤)"}


class FakeLLM:
    def __init__(self, pose_canon=None, carried=None, pose_fix=None):
        self.calls: List[Dict[str, Any]] = []
        self._pose_canon = pose_canon or {
            "immobile": [
                {
                    "character_ko": "남자",
                    "pose_en": "A man slumped against the wall, head tilted.",
                    "shots": ["S1sh1"],
                    "basis_ko": "원문",
                }
            ]
        }
        self._carried = carried or {
            "items": [
                {"shot": "S1sh1", "carried_en": "He grips a photograph.", "basis_ko": "원문"},
                {"shot": "S1sh2", "carried_en": "", "basis_ko": "없음"},
            ]
        }
        self._pose_fix = pose_fix or {
            "movement_en": "",
            "figures_en": "",
            "carried_en": "He grips a photograph in his fixed pose.",
        }

    def __call__(self, step_tag, system, user, schema, **kw):
        self.calls.append({"step_tag": step_tag, "user": user})
        if step_tag == "shot_continuity_pose_canon":
            return self._pose_canon
        if step_tag == "shot_continuity_carried":
            return self._carried
        if step_tag.startswith("shot_continuity_pose_fix"):
            return self._pose_fix
        raise AssertionError(step_tag)


def run(llm, **over):
    kw = dict(
        shots=SHOTS,
        scene_texts=SCENE_TEXTS,
        scene_headings=SCENE_HEADINGS,
        char_name_to_short={"남자": "C01"},
        call_structured_fn=llm,
    )
    kw.update(over)
    return run_shot_continuity(**kw)


def test_payload_shape():
    llm = FakeLLM()
    out = run(llm)
    assert out["pose_canon"][0]["character_short_id"] == "C01"
    assert out["pose_canon"][0]["pose_en"].startswith("A man slumped")
    assert out["carried"]["S1sh1"]["carried_en"] == "He grips a photograph."
    assert out["carried"]["S1sh2"]["carried_en"] == ""
    # pose_fix 는 정본 샷(S1sh1)에만
    assert set(out["pose_fix"]) == {"S1sh1"}
    assert out["pose_fix"]["S1sh1"]["carried_en"].endswith("fixed pose.")


def test_pose_canon_unknown_short_id_none():
    llm = FakeLLM()
    out = run(llm, char_name_to_short={})
    assert out["pose_canon"][0]["character_short_id"] is None


def test_pose_canon_v2_roster_injected_v1_unchanged():
    """v2(E2E6 ④): pose_canon 입력에 정본 명단 블록 — v1 은 불변."""
    llm2 = FakeLLM()
    run(llm2, prompt_version="2",
        char_name_to_short={"강민숙": "C07", "수리영": "C09"})
    pc_user2 = next(
        c["user"] for c in llm2.calls
        if c["step_tag"] == "shot_continuity_pose_canon"
    )
    assert "등장인물 정본 명단:" in pc_user2
    assert "- 강민숙" in pc_user2 and "- 수리영" in pc_user2

    llm1 = FakeLLM()
    run(llm1, prompt_version="1",
        char_name_to_short={"강민숙": "C07"})
    pc_user1 = next(
        c["user"] for c in llm1.calls
        if c["step_tag"] == "shot_continuity_pose_canon"
    )
    assert "등장인물 정본 명단" not in pc_user1


def test_pose_canon_v2_pack_contract_roster_clause():
    """v2 팩 계약: 명단 표기 그대로(닫힌 집합) 강제 문구 존재+중립성."""
    from pathlib import Path

    from app.modules.pipeline.shot_continuity_author import (
        resolve_prompt_version,
    )

    repo = Path(__file__).resolve().parents[3]
    ver = resolve_prompt_version("2")
    text = (repo / "prompts" / "_base" / "shot_continuity" / ver
            / "pose_canon_system.md").read_text(encoding="utf-8")
    assert "등장인물 정본 명단" in text
    assert "그대로" in text
    # 시나리오 고유명 하드코딩 0
    for word in ("민숙", "수리영", "옥탑", "rooftop"):
        assert word not in text


def test_pose_canon_unknown_shot_dropped():
    llm = FakeLLM(
        pose_canon={
            "immobile": [
                {
                    "character_ko": "남자",
                    "pose_en": "pose",
                    "shots": ["S1sh1", "S9sh9"],
                    "basis_ko": "",
                }
            ]
        }
    )
    out = run(llm)
    assert out["pose_canon"][0]["shots"] == ["S1sh1"]  # 미지 샷 제거


def test_no_immobile_no_pose_fix_calls():
    llm = FakeLLM(pose_canon={"immobile": []})
    out = run(llm)
    assert out["pose_canon"] == []
    assert out["pose_fix"] == {}
    fix_calls = [c for c in llm.calls if "pose_fix" in c["step_tag"]]
    assert fix_calls == []


def test_missing_carried_fails():
    llm = FakeLLM(
        carried={"items": [{"shot": "S1sh1", "carried_en": "", "basis_ko": ""}]}
    )
    with pytest.raises(AppError):
        run(llm)


def test_carried_input_has_full_scene_text():
    llm = FakeLLM()
    run(llm)
    carried_call = next(
        c for c in llm.calls if c["step_tag"] == "shot_continuity_carried"
    )
    assert "씬1 원문 전문..." in carried_call["user"]


def test_step_registered():
    from app.core.step_catalog import STEP_CATALOG
    from app.core.steps import STEP_CLASSES

    assert "shot_continuity" in STEP_CLASSES
    entry = STEP_CATALOG["shot_continuity"]
    assert entry.applicability == "if_still_recipe"


# ──────────────────────────────────────────────────────────────────────
# carried 출력 샤딩 (2026-09-18 컨트리로드 실측)
#
# 242샷을 한 답에 쓰게 했더니 15분 1초 만에 제공자 게이트웨이가 504 로 끊었다
# (같은 주행 pose_canon 은 100초 · 출력 4,834 토큰). 입력(씬 원문 전문 + 전체
# 시간순 샷 목록)은 그대로 두고 **답할 샷만** 나눈다 — 팩 규칙 4 의 양방향
# 소급이 전역 문맥을 요구하기 때문이다.
# ──────────────────────────────────────────────────────────────────────

import app.modules.pipeline.shot_continuity_author as sca


def _many_shots(n):
    return [{"scene_index": 1, "shot_index": i, "description": f"샷 {i}"} for i in range(1, n + 1)]


class ChunkLLM(FakeLLM):
    """대상 샷 목록에 실린 샷만 답한다 (프로덕션이 시키는 대로 하는 대역)."""

    def __call__(self, step_tag, system, user, schema, **kw):
        if step_tag == "shot_continuity_carried":
            self.calls.append({"step_tag": step_tag, "user": user})
            head = "이번 답의 대상 샷"
            if head in user:
                body = user.split(head, 1)[1]
                tags = re.findall(r"(S\d+sh\d+):", body)
            else:
                tags = re.findall(r"(S\d+sh\d+):", user)
            return {"items": [{"shot": t, "carried_en": f"c-{t}", "basis_ko": "원문"} for t in tags]}
        return super().__call__(step_tag, system, user, schema, **kw)


import re  # noqa: E402


def test_carried_is_sharded_and_every_shot_is_answered_once(monkeypatch):
    monkeypatch.setattr(sca, "CARRIED_CHUNK_SIZE", 30)
    llm = ChunkLLM(pose_canon={"immobile": []})
    out = run(llm, shots=_many_shots(100))
    carried_calls = [c for c in llm.calls if c["step_tag"] == "shot_continuity_carried"]
    assert len(carried_calls) == 4, len(carried_calls)
    assert set(out["carried"]) == {f"S1sh{i}" for i in range(1, 101)}
    for c in carried_calls:
        # 입력은 안 자른다 — 원문 전문 + 전체 시간순 목록이 묶음마다 그대로 실린다
        assert SCENE_TEXTS[1] in c["user"]
        assert len(re.findall(r"S1sh\d+:", c["user"].split("이번 답의 대상 샷", 1)[0])) == 100
        # 답할 샷은 그 묶음만
        assert len(re.findall(r"S1sh\d+:", c["user"].split("이번 답의 대상 샷", 1)[1])) <= 30


def test_one_chunk_input_is_byte_identical_to_the_old_assembly(monkeypatch):
    """묶음이 하나면 조립이 옛날과 한 글자도 다르지 않다 — 짧은 화의 재현이 안 흔들린다."""
    monkeypatch.setattr(sca, "CARRIED_CHUNK_SIZE", 30)
    llm = ChunkLLM()
    run(llm)
    user = [c for c in llm.calls if c["step_tag"] == "shot_continuity_carried"][0]["user"]
    assert "이번 답의 대상 샷" not in user
    assert user.endswith("S1sh2: 방을 둘러보는 여자")


def test_a_chunk_only_owns_its_own_shots(monkeypatch):
    """맥락으로 딸려 온 남의 샷 답은 쓰지 않는다 — 묶음마다 제 샷만."""
    monkeypatch.setattr(sca, "CARRIED_CHUNK_SIZE", 2)

    class Greedy(FakeLLM):
        def __call__(self, step_tag, system, user, schema, **kw):
            if step_tag == "shot_continuity_carried":
                self.calls.append({"step_tag": step_tag, "user": user})
                body = user.split("이번 답의 대상 샷", 1)[1]
                mine = re.findall(r"(S\d+sh\d+):", body)
                items = [{"shot": t, "carried_en": f"mine-{t}", "basis_ko": "원문"} for t in mine]
                if "S1sh1" not in mine:   # 제 몫이 아닌 샷까지 답해 본다
                    items.append({"shot": "S1sh1", "carried_en": "남의 답", "basis_ko": "원문"})
                return {"items": items}
            return super().__call__(step_tag, system, user, schema, **kw)

    llm = Greedy(pose_canon={"immobile": []})
    out = run(llm, shots=_many_shots(4))
    assert out["carried"]["S1sh1"]["carried_en"] == "mine-S1sh1"


def test_workers_carry_stop_check_budget_and_trace(monkeypatch):
    """★정지 표·조사 예산·Opik trace 는 스레드마다 따로다 — 안 나르면 팬아웃에서 끊긴다."""
    import threading

    from app.core import image_call_budget as icb
    from app.core import research_call_budget as rcb
    from app.modules.llm import opik_trace as ot

    monkeypatch.setattr(sca, "CARRIED_CHUNK_SIZE", 2)
    stop_sentinel, trace_sentinel = object(), object()
    budget = rcb.ResearchCallBudget.__new__(rcb.ResearchCallBudget)
    seen = []

    class Spy(ChunkLLM):
        def __call__(self, step_tag, system, user, schema, **kw):
            if step_tag == "shot_continuity_carried":
                seen.append((threading.get_ident(), icb.get_current_stop_check(),
                             rcb.get_current_budget(), ot.current_trace()))
            return super().__call__(step_tag, system, user, schema, **kw)

    llm = Spy(pose_canon={"immobile": []})
    main = threading.get_ident()
    prev_stop, prev_budget = icb.get_current_stop_check(), rcb.get_current_budget()
    icb.install_stop_check(stop_sentinel)
    rcb.install_budget(budget)
    token = ot._trace_ctx.set(trace_sentinel)
    try:
        run(llm, shots=_many_shots(6))
    finally:
        ot._trace_ctx.reset(token)
        icb.install_stop_check(prev_stop)
        if prev_budget is not None:
            rcb.install_budget(prev_budget)
        else:
            rcb.uninstall_budget()
    assert any(tid != main for tid, *_ in seen), "스레드 풀 갈래를 안 탔다 — 시험이 무의미"
    for _tid, stop, bud, tr in seen:
        assert stop is stop_sentinel and bud is budget and tr is trace_sentinel


def test_a_long_episode_is_split_by_default():
    """★기본값 그대로 — 컨트리로드 크기(242샷)면 한 답에 다 맡기지 않는다.

    값이 아니라 관계를 잠근다: 긴 화는 여러 번에 나눠 묻고, 모든 샷이 답을 받는다.
    """
    llm = ChunkLLM(pose_canon={"immobile": []})
    out = run(llm, shots=_many_shots(242))
    n = len([c for c in llm.calls if c["step_tag"] == "shot_continuity_carried"])
    assert n >= 4, f"242샷을 {n}번에 맡긴다 — 한 답이 너무 크다"
    assert len(out["carried"]) == 242


def test_a_failing_chunk_stops_the_rest_before_they_are_sent(monkeypatch):
    """★한 묶음이 죽으면 **아직 안 보낸 묶음은 보내지 않는다** (Codex BLOCK 2026-09-18).

    제출 순서로 기다리면 뒤 묶음의 실패를 늦게 보고, 빈 일꾼이 다음 묶음을 계속 산다.
    실패한 일꾼이 직접 중단 표를 세우고, 부모는 끝나는 순서로 본다.
    """
    import threading
    import time

    monkeypatch.setattr(sca, "CARRIED_CHUNK_SIZE", 2)
    monkeypatch.setattr(sca, "CARRIED_CHUNK_WORKERS", 3)
    sent, lock = [], threading.Lock()

    class Failing(ChunkLLM):
        def __call__(self, step_tag, system, user, schema, **kw):
            if step_tag == "shot_continuity_carried":
                body = user.split("이번 답의 대상 샷", 1)[1]
                tags = re.findall(r"(S\d+sh\d+):", body)
                with lock:
                    sent.append(tags[0])
                if tags[0] == "S1sh1":
                    time.sleep(0.5)          # 첫 묶음은 느리다
                if tags[0] == "S1sh3":
                    raise RuntimeError("boom")
                return {"items": [{"shot": t, "carried_en": f"c-{t}", "basis_ko": "원문"} for t in tags]}
            return super().__call__(step_tag, system, user, schema, **kw)

    llm = Failing(pose_canon={"immobile": []})
    with pytest.raises(RuntimeError, match="boom"):
        run(llm, shots=_many_shots(10))     # 2개씩 5묶음
    assert "S1sh9" not in sent, f"실패 뒤에도 뒤 묶음을 보냈다: {sent}"
    assert len(sent) <= 3, f"동시 3인데 {len(sent)}묶음이 나갔다: {sent}"


def test_one_chunk_assembly_equals_the_old_string(monkeypatch):
    """묶음이 하나면 조립이 옛 문자열과 **완전히 같다** (부분 일치가 아니라 동치)."""
    monkeypatch.setattr(sca, "CARRIED_CHUNK_SIZE", 30)
    llm = ChunkLLM()
    run(llm)
    user = [c for c in llm.calls if c["step_tag"] == "shot_continuity_carried"][0]["user"]
    shots_txt = "\n".join(f"S1sh{s['shot_index']}: {s['description']}" for s in SHOTS)
    expected = (f"씬 원문 전문:\n\n[씬 1] {SCENE_HEADINGS[1]}\n{SCENE_TEXTS[1]}"
                f"\n\n샷 목록(시간순 — 각각 carried state 판정):\n{shots_txt}")
    assert user == expected


def test_v3_people_survive_the_fan_out(monkeypatch):
    """★실제 스텝이 쓰는 v3(사람별 상태)에서도 묶음을 건너 사람·short_id 가 남는다."""
    monkeypatch.setattr(sca, "CARRIED_CHUNK_SIZE", 2)

    class V3(FakeLLM):
        def __call__(self, step_tag, system, user, schema, **kw):
            if step_tag == "shot_continuity_carried":
                self.calls.append({"step_tag": step_tag, "user": user})
                assert "등장인물 정본 명단" in user, "명단이 묶음 입력에서 빠졌다"
                body = user.split("이번 답의 대상 샷", 1)[1] if "이번 답의 대상 샷" in user else user
                tags = re.findall(r"(S\d+sh\d+):", body)
                return {"items": [{
                    "shot": t, "objects_en": "A lamp stays lit.",
                    "people": [{"character_ko": "남자", "state_en": "He keeps the photograph.",
                                "basis_ko": "원문"}],
                    "offscreen_effects_en": "", "basis_ko": "원문",
                } for t in tags]}
            return super().__call__(step_tag, system, user, schema, **kw)

    llm = V3(pose_canon={"immobile": []})
    out = run(llm, shots=_many_shots(6), prompt_version="3", char_name_to_short={"남자": "C01"})
    assert len(out["carried"]) == 6
    for tag, row in out["carried"].items():
        assert row[sca.CARRIED_CONTRACT_KEY] == sca.CARRIED_CONTRACT_SCOPED
        assert row["people"][0]["character_short_id"] == "C01", (tag, row["people"])


def test_chunked_calls_get_retries_but_a_single_chunk_does_not(monkeypatch):
    """★나눈 묶음만 재시도 — 제공자 500 한 번에 완료분까지 잃지 않게 (2026-09-18 실측).

    한 묶음일 때는 옛 계약(재시도 0) 그대로다.
    """
    seen = []

    class Retry(ChunkLLM):
        def __call__(self, step_tag, system, user, schema, **kw):
            if step_tag == "shot_continuity_carried":
                seen.append(kw.get("num_retries"))
            return super().__call__(step_tag, system, user, schema, **kw)

    monkeypatch.setattr(sca, "CARRIED_CHUNK_SIZE", 2)
    run(Retry(pose_canon={"immobile": []}), shots=_many_shots(6))
    assert seen and all(n == sca.CARRIED_CHUNK_RETRIES for n in seen), seen
    assert sca.CARRIED_CHUNK_RETRIES >= 1

    seen.clear()
    monkeypatch.setattr(sca, "CARRIED_CHUNK_SIZE", 30)
    run(Retry(), shots=_many_shots(2))
    assert seen == [sca.GLOBAL_AUTHOR_RETRIES], seen