"""발송 장부의 **귀속** — 어느 샷·어느 일의 발송인가 (2026-09-20 Codex).

## 무엇이 틀렸나

장부가 「몇 번 보냈나」는 세는데 **누구 것인지**가 두 자리에서 어긋났다.

    ① 방문 신원을 **샷 scope 에 들어가기 전에** 읽었다.
       `current_shot_uid()` 는 이름과 달리 「지금 열려 있는 trace」의 uid 다.
       샷 trace 는 `_shot_capture_scope` 가 연다 — 그 앞에서 읽으면 **부모
       스텝의 uid** 가 잡혀 **모든 샷이 같은 신원**으로 적힌다.
    ② 본체 생성과 **뒤의 변환**이 같은 이름으로 남았다. 변환 기록 키는
       `tag::cine[::슬롯]` 인데 장부에는 `tag` 로 적혀, 「본체에서 몇 번·
       변환에서 몇 번」을 못 가른다.

## 이 시험이 잠그는 것

소스 문자열 검사가 **아니다**. 진짜 걷기를 태우고, 제공자 대역이 호출
경계에서 `record_send` 를 부른다 — 배선이 빠지면 여기서 걸린다.

★★**범위**(Codex NON-BLOCK): 적히는 자리는 **이미지 생성·변환 대역**
 뿐이다. 판정 대역(`_judge_fn`)은 `record_send` 를 부르지 않으므로,
 이 파일이 **LLM 슬롯·실제 전송까지** 재는 시험이라고 넓혀 쓰지 않는다.

★반례가 살아 있어야 한다: 바깥 uid 를 `P`, 샷 안에서 `U1`·`U2` 로 두고
 장부가 `P` 를 쓰면 실패한다. `current_shot_uid` 를 **하나로 고정**해
 patch 하면 이 순서 결함을 **못 잡는다**.
"""
from __future__ import annotations

import json
from contextlib import ExitStack, contextmanager
from pathlib import Path
from typing import Any, Dict, List, Tuple
from unittest.mock import MagicMock, patch

from tests.services.test_still_cine_stage import (
    EID, PID, _FLAG_PATCHES, _db_mock, _still, _write_cp,
)

T1, T2 = "S1sh1", "S1sh2"


def _png() -> bytes:
    import io as _io

    from PIL import Image

    buf = _io.BytesIO()
    Image.new("RGB", (8, 8), (7, 7, 7)).save(buf, format="PNG")
    return buf.getvalue()


def _judge_fn():
    def judge_fn(tag, prompt, labeled_refs, cand_paths, labels):
        rank = list(labels)
        return {
            "winner": rank[0], "ranking": rank,
            "verdicts": [{"label": l, "score": 10 - rank.index(l),
                          "verdict_ko": "ok"} for l in labels],
            "readings": [{"label": l, "hard_violations": []} for l in labels],
        }

    judge_fn.owns_order = True
    judge_fn.emits_readings = True
    return judge_fn


class _CaptureLedger:
    """설치된 장부를 시험이 집을 수 있게 **인스턴스를 모아 둔다**."""

    made: List[Any] = []


def _walk(tmp_path, *, stills, cine=None, fallbacks=(), give_up=1,
          conti_ab=False):
    """진짜 걷기를 태우고 (장부 시도 목록, 샷별 uid) 를 돌려준다.

    ★uid 는 **샷 scope 안에서만** 바뀐다 — 바깥은 `P` 다. 이것이 반례다.
    """
    from app.core.send_ledger import GRAIN_RAW, SendLedger, record_send

    for st in stills:
        _write_cp(tmp_path, "shot_ref_classify",
                  {"shots": {f"S{s['scene_index']}sh{s['shot_index']}":
                             {"person_visible": True} for s in stills},
                   "scenes": {}, "world_anchor_en": ""})
    _write_cp(tmp_path, "shot_continuity", {"pose_canon": []})
    if conti_ab:
        # A/B 는 콘티 + LOCATION 권위(플레이트)가 있어야 갈래를 짓는다
        conti_png = tmp_path / "conti.png"
        conti_png.write_bytes(b"\x89PNG\r\n\x1a\n" + b"CONTI")
        plate_png = tmp_path / "plate.png"
        plate_png.write_bytes(b"\x89PNG\r\n\x1a\n" + b"PLATE")
        _write_cp(tmp_path, "background_render", {"groups": {"BG1": {
            "status": "ok", "png_path": str(plate_png),
            "shot_ids": ["S1_Shot1"]}}})
        _write_cp(tmp_path, "shot_conti_light", {
            "contis": {T1: {"image_path": str(conti_png)}}})
    else:
        _write_cp(tmp_path, "shot_conti_light", {"contis": {}})
    (tmp_path / "scene" / "recipe").mkdir(parents=True, exist_ok=True)

    # ── trace 대역: scope 안에서만 uid 가 선다 ────────────────────────
    stack: List[str] = []
    uid_of = {s["id"]: f"U{i + 1}" for i, s in enumerate(stills)}

    @contextmanager
    def _scope(project_id, episode_id, *, still_id, scene_index, shot_index):
        stack.append(uid_of[still_id])
        try:
            yield
        finally:
            stack.pop()

    def _cur_uid():
        return stack[-1] if stack else "P"      # ★바깥은 부모 uid

    def _gen_fn(tag, prompt, labeled_refs, out_path: Path):
        # ★**호출 경계 대역** — 진짜 제공자가 적는 자리에서 적는다.
        record_send(kind="image", granularity=GRAIN_RAW, source="fake.gen")
        out_path.parent.mkdir(parents=True, exist_ok=True)
        out_path.write_bytes(f"img:{out_path.stem}".encode())
        return out_path

    class _Captured(SendLedger):
        def __init__(self, *a, **kw):
            super().__init__(*a, **kw)
            _CaptureLedger.made.append(self)

    per = MagicMock()
    per.safety_ladder_call_provenance.return_value = None
    per._resolve_generation_call_id.return_value = None
    per.save_single_scene_asset.side_effect = lambda *a, **k: MagicMock(
        id="G", is_primary=1, file_path="/x/g.png")

    _CaptureLedger.made.clear()
    patches = [
        patch("app.core.config.settings.projects_dir", str(tmp_path)),
        patch("app.core.send_ledger.SendLedger", _Captured),
        patch("app.services.still_recipe_service._shot_capture_scope", _scope),
        patch("app.modules.llm.opik_trace.current_shot_uid", _cur_uid),
        patch("app.modules.pipeline.multiroll_gemini.make_nb2_gen_fn",
              return_value=_gen_fn),
        patch("app.modules.pipeline.multiroll_gemini.make_gemini_judge_fn",
              side_effect=lambda **kw: _judge_fn()),
        patch("app.modules.pipeline.multiroll_gemini.make_gemini_critique_fn",
              return_value=MagicMock()),
        patch("app.modules.pipeline.cine_provider.build_cine_client",
              return_value=cine),
        patch("app.services.still_recipe_service._cine_moderation_fallbacks",
              return_value=list(fallbacks)),
        patch("app.core.config.settings.still_cine_moderation_give_up_after",
              give_up, create=True),
    ] + [patch(f"app.core.config.settings.{n}", v, create=True)
         for n, v in _FLAG_PATCHES] + [
        patch("app.core.config.settings.still_cine_transform_enabled",
              cine is not None, create=True),
        patch("app.core.config.settings.still_winner_gate_enabled",
              False, create=True),
        patch("app.core.config.settings.still_conti_ab_enabled",
              conti_ab, create=True),
    ] + ([patch("app.modules.pipeline.conti_ab.resolve_or_run_outer",
                return_value={"fingerprint": "fp", "winner": "A",
                              "outer": {}, "judged": False})]
         if conti_ab else [])
    from app.services.still_recipe_service import run_still_recipe_generation

    with ExitStack() as st:
        for p in patches:
            st.enter_context(p)
        run_still_recipe_generation(
            db=_db_mock(), project_id=PID, episode_id=EID,
            stills=list(stills), stills_orm=[],
            entity_lookup={}, ref_image_map={}, reference_svc=MagicMock(),
            scene_ref_image_map={}, scene_ref_asset_id_map={},
            staging_map={}, scene_cp=MagicMock(), persistence_svc=per,
            progress=MagicMock(), project_config=None,
            scene_dir=tmp_path / "scene", already_done_stills=set(),
            target_scenes=None)
    assert _CaptureLedger.made, "장부가 아예 안 설치됐다"
    led = _CaptureLedger.made[-1]
    return [(a.work, a.visit) for a in led.attempts]


# ── ① 방문 신원 — 샷 scope **안에서** 읽는다 ────────────────────────

def test_the_visit_is_read_inside_the_shot_scope(tmp_path):
    """★★부모 스텝 uid 로 적히면 **모든 샷이 같은 신원**이 된다.

    바깥은 `P`, 첫 샷 안은 `U1`, 둘째 샷 안은 `U2` 다. 미리 계산한 값을
    넘기는 판은 전부 `P` 가 되어 여기서 걸린다.
    """
    seen = _walk(tmp_path, stills=[_still("a", 1, 1), _still("b", 1, 2)])
    assert seen, "발송이 한 건도 안 잡혔다(이 반례가 무의미)"
    assert not [w for w, v in seen if v == "P"], (
        "★샷 scope **밖에서** 방문 신원을 읽었다 — 부모 스텝 uid 다")
    got: Dict[str, set] = {}
    for w, v in seen:
        got.setdefault(w.split("::", 1)[0], set()).add(v)
    assert got.get(T1) == {"U1"}, f"첫 샷 신원이 틀렸다: {got}"
    assert got.get(T2) == {"U2"}, f"둘째 샷 신원이 틀렸다: {got}"


def test_two_shots_do_not_share_one_visit(tmp_path):
    """★샷마다 **다른** 신원이어야 한다 — 같으면 귀속이 뭉개진다."""
    seen = _walk(tmp_path, stills=[_still("a", 1, 1), _still("b", 1, 2)])
    visits = {v for _, v in seen}
    assert len(visits) >= 2, f"두 샷이 한 신원을 공유한다: {visits}"


# ── ② 변환은 본체와 **다른 일**이다 ─────────────────────────────────

def _cine_client(calls: List[str], *, refuse=False):
    png = _png()

    class _C:
        def set_context(self, **kw):
            pass

        def set_submit_hook(self, *a, **kw):
            pass

        def generate_image(self, prompt, labeled_references=None, **kw):
            from app.core.send_ledger import GRAIN_RAW, record_send

            record_send(kind="image", granularity=GRAIN_RAW,
                        source="fake.cine")
            calls.append(prompt)
            if refuse:
                raise RuntimeError("moderation_blocked")
            return png, 10

    return _C()


def test_the_cine_transform_is_not_recorded_as_the_body(tmp_path):
    """★★「본체 생성」과 「뒤의 변환」이 같은 이름으로 남으면 안 된다."""
    calls: List[str] = []
    seen = _walk(tmp_path, stills=[_still("a", 1, 1)],
                 cine=_cine_client(calls))
    assert calls, "변환이 안 돌았다(이 반례가 무의미)"
    works = [w for w, _ in seen]
    assert T1 in works, "본체 발송이 없다"
    assert f"{T1}::cine" in works, (
        "★변환 발송이 **본체 이름으로** 적혔다 — 어느 쪽이 몇 번인지 못 센다")
    # ★같은 샷이므로 방문 신원은 **하나**다(변환이 새로 캐지 않는다)
    assert len({v for _, v in seen}) == 1, "변환이 방문 신원을 갈아 끼웠다"


def test_the_body_context_comes_back_after_the_transform(tmp_path):
    """★앞 샷의 변환이 **뒤 샷 본체까지 덮지 않는다**.

    ★★**이 시험이 잠그는 범위**(Codex NON-BLOCK): 둘째 샷은 제 소유자
     scope 가 `work` 를 **다시 설치**한다. 그래서 이것만으로는 「같은 샷
     안에서 변환 scope 를 나온 직후 부모 맥락이 복원된다」까지 증명되지
     않는다 — 복원 자체는 `ledger_scope`·`send_context` 의 `__exit__` 에
     있고, 같은 샷 안의 복원은 후속 보강 자리다. **넓혀 쓰지 않는다.**
    """
    calls: List[str] = []
    seen = _walk(tmp_path, stills=[_still("a", 1, 1), _still("b", 1, 2)],
                 cine=_cine_client(calls))
    assert f"{T1}::cine" in [w for w, _ in seen]
    assert T2 in [w for w, _ in seen], (
        "★앞 샷의 변환 맥락이 **복원되지 않아** 뒤 샷 본체까지 덮였다")


def test_the_moderation_fallback_gets_its_own_slot(tmp_path):
    """★대체 제공자는 **슬롯이 갈린 기록**이다 — 장부도 갈려야 한다."""
    main_calls: List[str] = []
    fb_calls: List[str] = []
    fb = ("grok", _cine_client(fb_calls),
          {"model": "m", "provider": "grok"})
    with patch("app.modules.pipeline.cine_transform.is_moderation_error",
               return_value=True):
        seen = _walk(tmp_path, stills=[_still("a", 1, 1)],
                     cine=_cine_client(main_calls, refuse=True),
                     fallbacks=[fb], give_up=1)
    assert fb_calls, "대체 제공자가 안 돌았다(이 반례가 무의미)"
    works = [w for w, _ in seen]
    assert f"{T1}::cine::grok" in works, (
        f"★대체 발송이 슬롯 없이 적혔다: {sorted(set(works))}")


# ── 규칙이 **한 자리**에 있는가 ─────────────────────────────────────

def test_the_cine_key_rule_lives_in_one_place(tmp_path):
    """★장부 쪽이 키 규칙을 **다시 짐작하지 않는다**.

    소유자가 쓰는 `cine_record_key` 와 기록부가 실제로 쓰는 키가 같아야
    한다 — 두 곳에 적으면 한쪽만 고쳐진다. 문자열 검사가 아니라 **진짜로
    쓴 키**와 견준다.
    """
    from app.modules.pipeline.cine_transform import cine_record_key

    calls: List[str] = []
    recipe = tmp_path / "scene" / "recipe"
    seen = _walk(tmp_path, stills=[_still("a", 1, 1)],
                 cine=_cine_client(calls))
    recs = json.loads(
        (recipe / "records.json").read_text(encoding="utf-8"))
    assert cine_record_key(T1) in recs, (
        f"기록부가 쓴 키와 다르다: {[k for k in recs if 'cine' in k]}")
    # ★장부에 적힌 이름이 **진짜 기록 키**여야 한다 — 어느 하나가 제
    #  규칙을 따로 지으면 여기서 어긋난다.
    for w, _ in seen:
        assert w in recs, f"장부의 {w!r} 은 기록에 없는 이름이다"


# ── 자식 갈래는 **제 이름**으로 적힌다 ──────────────────────────────

def test_a_child_branch_is_recorded_under_its_own_key(tmp_path):
    """★★conti A/B 두 갈래는 기록 키가 `tag::ab_conti`·`tag::ab_noconti` 다.

    씌우지 않으면 둘 다 본체 `tag` 로 남아 「어느 갈래가 몇 번 나갔나」가
    뭉개진다. 표준 갈래(`rec_key == tag`)에서는 안 드러나는 결함이라
    **A/B 를 켜야** 잡힌다.
    """
    seen = _walk(tmp_path, stills=[_still("a", 1, 1)], conti_ab=True)
    works = {w for w, _ in seen}
    assert f"{T1}::ab_conti" in works and f"{T1}::ab_noconti" in works, (
        f"★자식 갈래가 **본체 이름으로** 적혔다: {sorted(works)}")
    assert len({v for _, v in seen}) == 1, "자식이 방문 신원을 갈아 끼웠다"