"""재롤의 **자산 경계** — 저장 payload 에서 잰다 (2026-09-20 Codex BLOCK 2).

## 왜 여기서 재는가

「재롤이 지면 원래 생성자가 유지된다」를 `record` 안에서만 확인하면, 그
값을 실제로 쓰는 자리(`still_recipe_service` 의 자산 저장)까지 닿는지는
**모른다**. `shot_run_produced` 가 True 면 그 자산에 **이번 방문 uid** 가
붙는다 — 라벨만 바뀐 방문이 저자를 가로채면 계보가 거짓이 된다.

그래서 서비스 걷기와 `run_multiroll_select` 는 **그대로 태우고**, 이미지
생성·판정·DB·자산 저장만 대역으로 둬서 **저장 payload 를 잡는다**.
`produced=True` 를 미리 심는 대역은 이 결함을 숨긴다.
"""
from __future__ import annotations

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

import pytest

from app.modules.pipeline.multiroll_select import roll_labels

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

L2 = roll_labels(2)
L3 = roll_labels(3)
RR = roll_labels(3)[-1]
TAG = "S1sh1"


def _judge(violations: Dict[str, List[str]], *, ranking=None):
    calls: List[Any] = []

    def judge_fn(tag, prompt, labeled_refs, cand_paths, labels):
        calls.append(list(labels))
        rank = list(ranking or 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": violations.get(l, [])}
                         for l in labels],
        }

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


def _walk(tmp_path, *, uid, records=None, initial=None, rejudge=None,
          gen_fn=None):
    """서비스 걷기 한 번. 돌려주는 것: (저장 payload 목록, 생성 라벨, 기록)."""
    from app.services.still_recipe_service import (
        run_still_recipe_generation,
    )

    _write_cp(tmp_path, "shot_ref_classify",
              {"shots": {TAG: {"person_visible": True}}, "scenes": {},
               "world_anchor_en": ""})
    _write_cp(tmp_path, "shot_continuity", {"pose_canon": []})
    _write_cp(tmp_path, "shot_conti_light", {"contis": {}})
    recipe = tmp_path / "scene" / "recipe"
    recipe.mkdir(parents=True, exist_ok=True)
    if records is not None:
        (recipe / "records.json").write_text(
            json.dumps(records, ensure_ascii=False), encoding="utf-8")

    made: List[str] = []

    def _gen_fn(tag, prompt, labeled_refs, out_path: Path):
        made.append(out_path.stem.rsplit("_", 1)[-1])
        if gen_fn is not None:
            return gen_fn(tag, prompt, labeled_refs, out_path)
        out_path.parent.mkdir(parents=True, exist_ok=True)
        out_path.write_bytes(f"img:{out_path.stem}".encode())
        return out_path

    saved: List[Dict[str, Any]] = []
    per = MagicMock()
    per.safety_ladder_call_provenance.return_value = None
    per._resolve_generation_call_id.return_value = None

    def _save(*args, **kwargs):
        # ★위치 인자로도 온다 — 대역이 호출 모양을 틀리면 잡을 것을 못 잡는다.
        payload = next(
            (a for a in args if isinstance(a, dict)), None)
        if payload is None:
            payload = next(
                (v for v in kwargs.values() if isinstance(v, dict)), {})
        saved.append(dict(payload))
        return MagicMock(id="G", is_primary=1, file_path="/x/g.png")

    per.save_single_scene_asset.side_effect = _save

    patches = [
        patch("app.core.config.settings.projects_dir", str(tmp_path)),
        # ★생성·판정만 대역 — 걷기와 `run_multiroll_select` 는 진짜다.
        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: (
                  rejudge if kw.get("step_tag") ==
                  "still_recipe_gate_rejudge" else initial)),
        patch("app.modules.pipeline.multiroll_gemini.make_gemini_critique_fn",
              return_value=MagicMock()),
        patch("app.modules.llm.opik_trace.current_shot_uid",
              return_value=uid),
    ] + [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",
              False, create=True),
        patch("app.core.config.settings.still_winner_gate_enabled",
              True, create=True),
        patch("app.core.config.settings.still_gate_reroll_enabled",
              True, create=True),
    ]
    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=[_still("a", 1, 1)], 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)
    recs = json.loads((recipe / "records.json").read_text(encoding="utf-8"))
    return saved, made, recs


def _uid_of(payload: Dict[str, Any]):
    return payload.get("shot_run_uid")


def test_a_losing_reroll_does_not_hand_the_asset_this_visits_uid(tmp_path):
    """★★재롤이 **져서** 옛 초기 후보가 최종이면, 다음 방문은 저자가 아니다.

    U1 이 초기 롤 + 재롤을 사고 B 가 이긴다 → U1 이 저자.
    U2 는 아무것도 안 만든다 → **자산에 U2 를 붙이면 안 된다**.
    """
    # ★**같은 디렉토리**를 두 방문이 쓴다 — 롤 파일이 남아 있어야
    #  「재개가 안 산다」를 잴 수 있다(다른 디렉토리면 당연히 다시 산다).
    saved1, made1, recs1 = _walk(
        tmp_path, uid="U1",
        initial=_judge({"A": ["x"], "B": ["y"]}),
        rejudge=_judge({"A": ["x"], "B": [], RR: ["z"]},
                       ranking=[RR, "A", "B"]))
    assert RR.lower() in made1, "재롤을 안 샀다"
    assert recs1[TAG]["selected"] == "B"
    assert saved1 and _uid_of(saved1[-1]) == "U1"

    saved2, made2, recs2 = _walk(
        tmp_path, uid="U2", records=recs1,
        initial=_judge({}), rejudge=_judge({}))
    assert made2 == [], "재개가 그림을 샀다"
    assert _uid_of(saved2[-1]) is None, (
        "아무것도 안 만든 방문이 자산 계보를 가로챘다")


def test_the_visit_that_actually_bought_the_reroll_keeps_the_lineage(tmp_path):
    """★이번에 **실제로 산** 재롤이 최종이면 그 방문이 저자다."""
    saved, made, recs = _walk(
        tmp_path, uid="U1",
        initial=_judge({l: ["x"] for l in L2}),
        rejudge=_judge({l: (["x"] if l in L2 else []) for l in L3},
                       ranking=[RR, "A", "B"]))
    assert RR.lower() in made
    assert recs[TAG]["selected"] == RR
    assert _uid_of(saved[-1]) == "U1"


def test_the_reroll_prompt_reaches_the_asset(tmp_path):
    """★★재롤이 이기면 자산에 **실제로 나간 문안**이 저장된다.

    `roll_prompts[C]` 가 없으면 `effective_prompt_used` 가 base_prompt 로
    떨어져 **안 쓴 문안**이 자산에 남는다(Codex BLOCK 4).
    """
    saved, made, recs = _walk(
        tmp_path, uid="U1",
        initial=_judge({"A": ["팔이 셋이다"], "B": ["떠 있다"]}),
        rejudge=_judge({l: (["x"] if l in L2 else []) for l in L3},
                       ranking=[RR, "A", "B"]))
    assert recs[TAG]["selected"] == RR
    used = str(saved[-1].get("prompt_used") or "")
    assert used == recs[TAG]["roll_prompts"][RR], (
        "자산에 재롤 전문이 아닌 다른 문안이 저장됐다")
    assert "팔이 셋이다" in used
    for bad in ("storey count", "massing", "footprint"):
        assert bad not in used, f"구조물 문안이 자산에 남았다: {bad}"


def test_a_visit_that_only_made_part_of_the_rolls_is_not_the_author(tmp_path):
    """★★「하나라도 만들었나」와 「**최종 후보**를 만들었나」는 다르다.

    Codex 반례 그대로:
      · U1 이 A 를 만들고 B 에서 죽는다 → A 파일과 지문은 남는다
      · U2 가 B 를 만든다 → 초기 판정 둘 다 실격 → C 재롤 → 재판정에서
        **기존 A 만 허용**되어 A 채택
      · 최종 bytes 는 **U1 의 A** 인데 bool 로 보면 U2 가 저자로 찍힌다
    """
    # ── U1: A 만 만들고 B 에서 죽는다
    def _dying(tag, prompt, labeled_refs, out_path: Path):
        if out_path.stem.endswith("b"):
            raise RuntimeError("U1 에서 B 생성 실패")
        out_path.parent.mkdir(parents=True, exist_ok=True)
        out_path.write_bytes(f"img:{out_path.stem}".encode())
        return out_path

    # ★걷기는 샷 단위로 실패를 격리한다 — 예외가 밖으로 안 나온다.
    #  그래서 「터졌나」가 아니라 **디스크에 무엇이 남았나**로 본다.
    _walk(tmp_path, uid="U1", initial=_judge({}), rejudge=_judge({}),
          gen_fn=_dying)
    assert (tmp_path / "scene" / "recipe" / f"{TAG}_a.png").exists()
    assert not (tmp_path / "scene" / "recipe" / f"{TAG}_b.png").exists()

    # ── U2: B 를 만들고, 재판정에서 **기존 A 만** 허용된다
    saved, made, recs = _walk(
        tmp_path, uid="U2",
        initial=_judge({l: ["x"] for l in L2}),
        rejudge=_judge({"A": [], "B": ["x"], RR: ["z"]},
                       ranking=["A", RR, "B"]))
    assert "b" in made and RR.lower() in made, "U2 가 B·C 를 만들어야 한다"
    assert "a" not in made, "A 는 U1 것이라 다시 만들면 안 된다"
    assert recs[TAG]["selected"] == "A", "최종은 U1 이 만든 A 여야 한다"
    assert _uid_of(saved[-1]) is None, (
        "U2 가 A 를 만들지 않았는데 자산 계보를 가로챘다")


def test_a_rejudge_only_resume_does_not_claim_the_lineage(tmp_path):
    """★U1 이 C 를 사고 재판정에서 죽고, U2 가 **판정만** 성공한 경우.

    생성자는 U1 이다 — U2 는 아무것도 안 만들었다 (Codex (r)).
    """
    class _Boom:
        owns_order = True
        emits_readings = True

        def __call__(self, *a, **k):
            raise RuntimeError("재판정 실패")

    _walk(tmp_path, uid="U1",
          initial=_judge({l: ["x"] for l in L2}), rejudge=_Boom())
    recipe = tmp_path / "scene" / "recipe"
    assert (recipe / f"{TAG}_{RR.lower()}.png").exists()
    bought = (recipe / f"{TAG}_{RR.lower()}.png").read_bytes()

    saved, made, recs = _walk(
        tmp_path, uid="U2",
        initial=_judge({l: ["x"] for l in L2}),
        rejudge=_judge({l: (["x"] if l in L2 else []) for l in L3},
                       ranking=[RR, "A", "B"]))
    assert made == [], "재개가 그림을 샀다"
    assert recs[TAG]["selected"] == RR
    assert recs[TAG]["gate_reroll"]["produced_by_visit"] == "U1"
    assert (recipe / f"{TAG}_sel.png").read_bytes() == bought
    assert _uid_of(saved[-1]) is None, (
        "판정만 한 방문이 자산 계보를 가로챘다")