"""의존 대기는 **풀리는 길**이 있어야 한다 (2026-09-20 Codex BLOCK).

앞 샷이 붙잡혀 뒤 샷을 `blocked_dependency` 로 적어 두는 것까지는 했는데,
**부모가 풀린 다음 방문에 그 대기를 벗기는 코드가 없었다.** 입력 지문이
그대로라 `multiroll_select` 가 기존 record 를 그대로 돌려주고, 기록의
`gate` 는 여전히 의존 대기라 소비자 문이 **영원히** 막는다.

여기서 재는 것:
  ① A→B→C 연쇄 + 독립 D — 한 번 방문에 B·C 가 **대기로 남고** D 는 간다
  ② 두 번째 방문에서 **A 의 판정만** 복구되면 B·C 의 대기가 풀리고
     원래 판정으로 돌아간다 (새 그림을 **사지 않는다**)
  ③ 원래 판정이 `unresolved`·`incomplete` 였으면 **clean 으로 올리지
     않는다** — 대기 한 겹만 벗긴다
  ④ 같은 대기를 되풀이 기록해도 `prior` 를 **겹쳐 쌓지 않는다**

유료 경계만 대역이고 서비스 걷기를 그대로 태운다.
"""
from __future__ import annotations

import json
from contextlib import ExitStack
from pathlib import Path
from unittest.mock import MagicMock, patch

import pytest

from app.services.still_recipe_service import (
    GATE_DEPENDENCY,
    gate_unresolved,
    release_dependency_hold,
)

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

A, B, C, D = "S1sh1", "S1sh2", "S1sh3", "S2sh1"


@pytest.fixture(autouse=True)
def _gate_on(monkeypatch):
    from app.core import config

    monkeypatch.setattr(config.settings, "still_winner_gate_enabled", True,
                        raising=False)


class _FakeRecords:
    """`_Records` 와 같은 모양의 최소 대역 — 저장 횟수를 센다."""

    def __init__(self, data):
        self.data = data
        self.saves = 0

    def save(self):
        self.saves += 1


# ── ③④ 해제 함수 자체 ─────────────────────────────────────────────

def test_the_release_restores_the_prior_verdict_not_clean():
    """★대기 **한 겹만** 벗긴다 — 원래 판정이 미해결이면 미해결로 돌아간다.

    여기서 `clean` 으로 올려 주면, 앞 샷이 풀렸다는 이유로 **이 샷 자신의
    결함**이 사라진다. 그건 판정이 아니라 사면이다.
    """
    for prior in ({"outcome": "unresolved"}, {"outcome": "incomplete"},
                  {"outcome": "clean"}):
        recs = _FakeRecords({B: {"gate": {
            "outcome": GATE_DEPENDENCY, "blocked_by": A, "prior": dict(prior)}}})
        assert release_dependency_hold(recs, B) is True
        assert recs.data[B]["gate"] == prior
        assert recs.saves == 1


def test_a_shot_that_never_had_a_verdict_goes_back_to_having_none():
    """`prior` 가 없었으면 **대기 이전과 똑같이** 판정 칸을 지운다.

    없던 판정을 만들어 내지 않는다 — 판정이 없는 것은 `incomplete` 로
    새로 발명할 일이 아니라 **대기 이전 상태**다.
    """
    recs = _FakeRecords({B: {"selected": "a", "gate": {
        "outcome": GATE_DEPENDENCY, "blocked_by": A, "prior": None}}})
    assert release_dependency_hold(recs, B) is True
    assert "gate" not in recs.data[B]
    assert recs.data[B]["selected"] == "a", "다른 기록까지 건드렸다"


def test_the_release_does_not_touch_a_quality_verdict():
    """이 샷 **자신이** 실격인 것은 못 푼다 — 앞 샷과 무관하다."""
    for g in ({"outcome": "unresolved"}, {"outcome": "incomplete"},
              {"outcome": "clean"}, None):
        recs = _FakeRecords({B: {"gate": g}})
        assert release_dependency_hold(recs, B) is False
        assert recs.saves == 0


def test_a_repeated_hold_does_not_nest_the_prior():
    """★같은 대기를 두 번 적어도 **원래 판정을 잃지 않는다**.

    대기를 대기 안에 또 넣으면 몇 바퀴 뒤에 `prior.prior.prior…` 가 되어
    원래 판정을 아무도 못 찾는다.
    """
    src = Path(__file__).resolve().parents[2] / (
        "app/services/still_recipe_service.py")
    text = src.read_text(encoding="utf-8")
    i = text.find('"outcome": GATE_DEPENDENCY,')
    assert i > 0
    seg = text[max(0, i - 700):i]
    assert "_prior_gate.get(\"prior\")" in seg, (
        "이미 대기인 기록 위에 또 대기를 씌운다 — 원래 판정이 묻힌다")


# ── ①② 실제 걷기: A→B→C + 독립 D ──────────────────────────────────

def _persistence():
    p = MagicMock()
    p.safety_ladder_call_provenance.return_value = None
    # 변환이 켜진 갈래는 롤 호출 링크를 자산 기록에 넣는다 — MagicMock 이
    # 그대로 들어가면 JSON 직렬화에서 터진다(대역의 결함, 코드가 아니다).
    p._resolve_generation_call_id.return_value = None
    p.save_single_scene_asset.side_effect = lambda *a, **kw: MagicMock(
        id="G", is_primary=1, file_path="/x/g.png")
    return p


def _walk(tmp_path, *, verdicts, persistence, scene_cp, seed_records=None,
          reuse_tags=frozenset(), already_done=frozenset()):
    """네 샷을 한 번 걷는다. `verdicts` = {tag: outcome}.

    `reuse_tags` = **입력 지문이 그대로라 캐시를 그대로 돌려주는** 샷.
    프로덕션에서 이것이 바로 대기가 안 풀리던 자리다 — 다시 안 그리니
    record 가 갱신되지 않아 옛 `blocked_dependency` 가 그대로 남는다.
    이 모양을 대역이 흉내 내지 않으면 시험이 결함을 **정답으로 잠근다**
    (실제로 첫 판이 그랬다: 해제를 지워도 통과했다).
    """
    from app.services.still_recipe_service import run_still_recipe_generation

    _write_cp(tmp_path, "shot_ref_classify", {
        "shots": {
            A: {"person_visible": True},
            B: {"person_visible": True, "prev": A},
            C: {"person_visible": True, "prev": B},
            D: {"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)
    for t in (A, B, C, D):
        (recipe / f"{t}_sel.png").write_bytes(b"SEL")
    if seed_records is not None:
        (recipe / "records.json").write_text(
            json.dumps(seed_records, ensure_ascii=False), encoding="utf-8")

    generated: list = []
    reused: list = []

    def _gen(**kwargs):
        # ★프로덕션의 `run_branch_select._persist` 와 **같은 일**을 한다 —
        #  선정이 끝나면 record 를 그 자리에서 records 에 쓰고 저장한다.
        #  이걸 빠뜨리면 대역이 프로덕션과 다른 것을 재게 된다(내 첫 판이
        #  그랬다: 앞 샷 판정이 안 남아 연쇄가 안 잡히는 것처럼 보였다).
        key = kwargs["rec_key"]
        _existing = kwargs["records"].data.get(key)
        if key in reuse_tags and isinstance(_existing, dict) and _existing:
            # 지문 일치 = **아무것도 안 산다**. record 도 안 건드린다.
            reused.append(key)
            return recipe / f"{key}_sel.png", dict(_existing)
        generated.append(key)
        rec = dict(_existing or {})
        rec["gate"] = {"outcome": verdicts[key], "policy": "p"}
        kwargs["records"].data[key] = rec
        kwargs["records"].save()
        return recipe / f"{key}_sel.png", rec

    patches = [
        patch("app.core.config.settings.projects_dir", str(tmp_path)),
        patch("app.modules.pipeline.multiroll_gemini.make_nb2_gen_fn",
              return_value=MagicMock()),
        patch("app.modules.pipeline.multiroll_gemini.make_gemini_judge_fn",
              return_value=MagicMock()),
        patch("app.modules.pipeline.multiroll_gemini.make_gemini_critique_fn",
              return_value=MagicMock()),
        patch("app.modules.pipeline.still_recipe.run_branch_select",
              side_effect=_gen),
    ] + [patch(f"app.core.config.settings.{n}", v, create=True)
         for n, v in _FLAG_PATCHES] + [
        # 이 시험이 재는 것은 **게이트 문**이지 변환이 아니다 — 안 끄면
        # 실제 provider 를 부른다(실측 전례 있음).
        patch("app.core.config.settings.still_cine_transform_enabled",
              False, 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), _still("b", 1, 2),
                    _still("c", 1, 3), _still("d", 2, 1)],
            stills_orm=[], entity_lookup={}, ref_image_map={},
            reference_svc=MagicMock(), scene_ref_image_map={},
            scene_ref_asset_id_map={}, staging_map={},
            scene_cp=scene_cp, persistence_svc=persistence,
            progress=MagicMock(), project_config=None,
            scene_dir=tmp_path / "scene",
            already_done_stills=set(already_done),
            target_scenes=None)
    records = json.loads(
        (recipe / "records.json").read_text(encoding="utf-8"))
    return records, generated, reused


def test_the_chain_holds_and_then_releases(tmp_path):
    """★A→B→C + 독립 D — **한 번 붙잡고, 부모가 풀리면 놓는다**.

    앞선 주행에서 B·C 는 이미 `clean` 으로 끝나 있었다(지문 그대로).
    첫 방문: A 가 미해결 → B·C 는 **의존 대기**로 남고 D 만 간다.
      ★B·C 는 생성을 아예 안 탄다 — 대기는 새 그림을 사서 푸는 게 아니다.
    둘째 방문: A 의 **판정만** 복구(그림·지문 그대로) → B·C 의 대기가
      벗겨지고, **새 그림 없이** 기존 clean 산출이 그대로 쓰인다.
    """
    seed = {t: {"gate": {"outcome": "clean", "policy": "p"},
                "selected": "a", "critique_skipped": True}
            for t in (B, C)}
    per1, cp1 = _persistence(), MagicMock()
    recs1, gen1, _ = _walk(
        tmp_path,
        verdicts={A: "incomplete", B: "clean", C: "clean", D: "clean"},
        persistence=per1, scene_cp=cp1, seed_records=seed,
        reuse_tags={B, C})

    assert recs1[B]["gate"]["outcome"] == GATE_DEPENDENCY
    assert recs1[B]["gate"]["blocked_by"] == A
    assert recs1[B]["gate"]["prior"]["outcome"] == "clean", (
        "B 자신의 판정을 대기가 덮어썼다")
    assert recs1[C]["gate"]["outcome"] == GATE_DEPENDENCY
    assert recs1[C]["gate"]["blocked_by"] == B, (
        "C 가 B 때문이라고 안 적혔다 — 연쇄가 한 샷 뒤에서 뚫린다")
    assert B not in gen1 and C not in gen1, (
        "대기 샷에 새 그림을 샀다 — 대기는 지출이 아니다")
    assert D in gen1, "독립 샷까지 세웠다"
    assert per1.save_single_scene_asset.call_count == 1, (
        "붙잡힌 샷을 저장했다 (독립 D 한 장만이어야 한다)")

    # ── 둘째 방문: A 의 판정만 복구. 그림·지문은 그대로.
    #    ★B·C 는 지문이 같아 **다시 안 그린다** — 그래서 record 가 저절로
    #     갱신되지 않는다. 푸는 코드가 없으면 여기서 영구 보류가 된다.
    per2, cp2 = _persistence(), MagicMock()
    recs2, gen2, reused2 = _walk(
        tmp_path,
        verdicts={A: "clean", B: "clean", C: "clean", D: "clean"},
        persistence=per2, scene_cp=cp2, reuse_tags={B, C})

    for t in (B, C):
        assert recs2[t]["gate"]["outcome"] != GATE_DEPENDENCY, (
            f"{t} 가 **영구 보류**됐다 — 부모가 풀렸는데 못 푼다")
        assert gate_unresolved(recs2[t]) is False
        assert t in reused2 and t not in gen2, (
            f"{t} 를 다시 그려서 풀었다 — 해제는 유료가 아니다")
    # ★이 수는 **이 시험의 통과 경로**를 확인하는 것이지, 「기존 DB 자산
    #  재저장 0」의 증거가 아니다 (Codex NON-BLOCK). 이 하네스는
    #  `already_done_stills=set()` 이라 JIT 의 「이미 완료된 샷 skip」
    #  갈래는 아예 안 탄다.
    assert per2.save_single_scene_asset.call_count == 4, (
        "부모가 풀렸는데 연쇄가 안 살아났다")


def test_a_held_shot_keeps_its_own_verdict_through_the_hold(tmp_path):
    """★대기가 **원래 판정을 덮지 않는다** — 풀려도 그 판정으로 돌아간다.

    B 가 앞선 주행에서 스스로 `unresolved` 였다면, A 가 풀린다고 해서
    B 가 통과되면 안 된다. 대기는 **겹**이지 판정이 아니다.
    """
    seed = {B: {"gate": {"outcome": "unresolved", "policy": "p"},
                "selected": "a"}}
    per1, cp1 = _persistence(), MagicMock()
    recs1, _, _ = _walk(
        tmp_path,
        verdicts={A: "incomplete", B: "unresolved", C: "clean", D: "clean"},
        persistence=per1, scene_cp=cp1, seed_records=seed, reuse_tags={B})
    assert recs1[B]["gate"]["outcome"] == GATE_DEPENDENCY
    assert recs1[B]["gate"]["prior"]["outcome"] == "unresolved", (
        "B 자신의 판정을 대기가 덮어썼다")

    per2, cp2 = _persistence(), MagicMock()
    recs2, _, _ = _walk(
        tmp_path,
        verdicts={A: "clean", B: "unresolved", C: "clean", D: "clean"},
        persistence=per2, scene_cp=cp2, reuse_tags={B})
    assert gate_unresolved(recs2[B]) is True, (
        "앞 샷이 풀렸다는 이유로 B 자신의 실격이 사면됐다")
    assert recs2[C]["gate"]["outcome"] == GATE_DEPENDENCY, (
        "B 가 아직 붙잡혀 있는데 C 가 풀렸다")


def test_the_hold_is_not_a_spend_trace(tmp_path):
    """★의존 대기는 **지출 흔적이 아니다** (Codex NON-BLOCK).

    `_jit_tag_snapshot` 은 지출 탐지용 직렬화라 `gate` 를 통째로 넣는다.
    대기가 붙었다 풀리는 것만으로 스냅샷이 움직이면, **메타만 바뀐
    방문**이 「지출」로 읽혀 JIT 재생성 계수가 거짓으로 오른다.
    """
    from app.services.still_recipe_service import _jit_tag_snapshot

    plain = _FakeRecords({B: {"gate": {"outcome": "clean", "policy": "p"}}})
    held = _FakeRecords({B: {"gate": {
        "outcome": GATE_DEPENDENCY, "blocked_by": A,
        "policy": "p", "prior": {"outcome": "clean", "policy": "p"}}}})
    assert (_jit_tag_snapshot(plain, B, {B})
            == _jit_tag_snapshot(held, B, {B})), (
        "대기 한 겹이 지출로 읽힌다")

    # ★**판정 종착 자체도** 지출이 아니다 (Codex). 옛 기록에 판정만 채우는
    #  무료 소급이 「지출」로 읽히면 반대 방향으로 틀린다.
    other = _FakeRecords({B: {"gate": {"outcome": "unresolved",
                                       "policy": "p"}}})
    none_yet = _FakeRecords({B: {}})
    assert (_jit_tag_snapshot(plain, B, {B})
            == _jit_tag_snapshot(other, B, {B})
            == _jit_tag_snapshot(none_yet, B, {B})), (
        "판정 이름표를 붙이는 것이 지출로 읽힌다")

    # ★**자식 기록의 판정 이름표도** 지출이 아니다 (Codex NON-BLOCK).
    #  conti A/B 는 `tag::ab_conti`·`tag::ab_noconti` 두 자식에도 소급이
    #  `not_applicable` 을 새로 쓴다 — 본체만 빼면 그 자식 메타가 「유료
    #  방문」으로 읽힌다.
    child_plain = _FakeRecords({B: {"selected": "a"},
                                f"{B}::ab_conti": {"selected": "a"}})
    child_gated = _FakeRecords({B: {"selected": "a"},
                                f"{B}::ab_conti": {
                                    "selected": "a",
                                    "gate": {"outcome": "not_applicable",
                                             "policy": "p"}}})
    assert (_jit_tag_snapshot(child_plain, B, {B})
            == _jit_tag_snapshot(child_gated, B, {B})), (
        "자식 기록의 판정 이름표가 지출로 읽힌다")

    # ★그런데 자식의 **진짜 유료 흔적**은 그대로 잡혀야 한다.
    child_spent = _FakeRecords({B: {"selected": "a"},
                                f"{B}::ab_conti": {"selected": "a",
                                                   "critique": {"issues": []}}})
    assert (_jit_tag_snapshot(child_plain, B, {B})
            != _jit_tag_snapshot(child_spent, B, {B}))

    # 그런데 **정말 돈이 나간 흔적**은 여전히 움직여야 한다.
    spent = _FakeRecords({B: {"gate": {"outcome": "clean", "policy": "p"},
                              "shot_run_spend_attempt_count": 1}})
    assert _jit_tag_snapshot(plain, B, {B}) != _jit_tag_snapshot(spent, B, {B})


# ── 문이 **유료 경계**에서 막는가 (Codex ④) ────────────────────────

def test_the_door_stops_the_cine_purchase_at_the_provider(tmp_path):
    """★변환을 **켜 놓고** 재야 문이 막는 것을 증명한다.

    변환을 꺼 놓고 「구매 0」이라고 말하면, 그건 문이 아니라 레버가
    막은 것이다 — 내가 그렇게 세던 자리다. 여기서는 제공자 경계
    (`cine_provider.build_cine_client` 가 주는 client 의 `generate_image`)
    에서 **직접** 센다.

      · 붙잡힌 샷 → 호출 **0**
      · 정상 샷  → 호출 **1**
    """
    import io

    from PIL import Image

    from app.services.still_recipe_service import run_still_recipe_generation

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

    calls: list = []

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

        def generate_image(self, prompt, labeled_references=None, **kw):
            calls.append(prompt)
            return png, 10

    def _run(outcome):
        calls.clear()
        _write_cp(tmp_path, "shot_ref_classify",
                  {"shots": {A: {"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)
        for f in recipe.glob("*"):
            if f.is_file():
                f.unlink()
        (recipe / f"{A}_sel.png").write_bytes(png)

        def _gen(**kwargs):
            key = kwargs["rec_key"]
            rec = dict(kwargs["records"].data.get(key) or {})
            rec["gate"] = {"outcome": outcome, "policy": "p"}
            kwargs["records"].data[key] = rec
            kwargs["records"].save()
            return recipe / f"{key}_sel.png", rec

        patches = [
            patch("app.core.config.settings.projects_dir", str(tmp_path)),
            patch("app.modules.pipeline.multiroll_gemini.make_nb2_gen_fn",
                  return_value=MagicMock()),
            patch("app.modules.pipeline.multiroll_gemini.make_gemini_judge_fn",
                  return_value=MagicMock()),
            patch(
                "app.modules.pipeline.multiroll_gemini.make_gemini_critique_fn",
                return_value=MagicMock()),
            patch("app.modules.pipeline.still_recipe.run_branch_select",
                  side_effect=_gen),
            # ★대역은 **제공자 조립부**에 건다 — 특정 client 클래스에 걸면
            #  제공자가 바뀌는 순간 시험이 아무것도 안 막는다.
            patch("app.modules.pipeline.cine_provider.build_cine_client",
                  return_value=_Client()),
        ] + [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",
                  True, create=True),
            patch("app.core.config.settings.still_cine_verify_enabled",
                  False, create=True),
        ]
        per = _persistence()
        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)
        return len(calls)

    assert _run("clean") == 1, "정상 샷이 변환을 안 탔다 — 시험이 아무것도 안 잰다"
    assert _run("unresolved") == 0, "붙잡힌 샷이 변환을 **샀다**"
    assert _run("incomplete") == 0, "판정을 못 읽은 샷이 변환을 **샀다**"


# ── 게이트 ON 재개는 JIT 검증을 요구한다 (Codex (a)) ───────────────

def test_the_gate_on_run_requires_jit_verify(tmp_path):
    """★JIT 가 꺼지면 완료 샷이 **대기를 푸는 자리에 못 온다**.

    그러면 「ON 으로 재개하면 저절로 복구된다」가 성립하지 않고, 앞 샷이
    풀렸는데도 뒤 샷이 영영 붙잡힌다. 새 그림을 사서 푸는 것이 아니라
    **멈춰 세워** 사람이 레버를 맞추게 한다.
    """
    from app.services.still_recipe_service import run_still_recipe_generation

    _write_cp(tmp_path, "shot_ref_classify",
              {"shots": {}, "scenes": {}, "world_anchor_en": ""})
    _write_cp(tmp_path, "shot_continuity", {"pose_canon": []})
    _write_cp(tmp_path, "shot_conti_light", {"contis": {}})

    patches = [
        patch("app.core.config.settings.projects_dir", str(tmp_path)),
    ] + [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_jit_verify_enabled",
              False, create=True),
    ]
    with ExitStack() as st:
        for p in patches:
            st.enter_context(p)
        with pytest.raises(RuntimeError, match="JIT"):
            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=_persistence(),
                progress=MagicMock(), project_config=None,
                scene_dir=tmp_path / "scene", already_done_stills=set(),
                target_scenes=None)


def test_the_gate_off_run_does_not_require_jit_verify(tmp_path, monkeypatch):
    """★꺼진 판에는 이 조건이 **없다** — 레버 계약이다."""
    from app.core import config
    from app.services.still_recipe_service import gate_is_on

    monkeypatch.setattr(config.settings, "still_winner_gate_enabled", False,
                        raising=False)
    monkeypatch.setattr(config.settings, "still_jit_verify_enabled", False,
                        raising=False)
    assert gate_is_on() is False


def test_a_failed_save_does_not_leave_a_released_hold_in_memory():
    """★저장이 터지면 메모리도 되돌린다 (Codex NON-BLOCK 1).

    먼저 메모리를 바꾸고 저장이 실패하면, 로그는 「대기가 남는다」인데
    이번 방문의 소비자들은 **이미 풀린 것**을 본다 — 기록과 실상이 갈린다.
    """
    held = {"gate": {"outcome": GATE_DEPENDENCY, "blocked_by": A,
                     "prior": {"outcome": "clean"}}}

    class _Boom(_FakeRecords):
        def save(self):
            raise OSError("디스크 가득")

    recs = _Boom({B: dict(held)})
    assert release_dependency_hold(recs, B) is False, (
        "저장이 터졌는데 「풀었다」고 말한다")
    assert recs.data[B] == held, "저장이 터졌는데 메모리는 풀려 있다"


# ── 판정 없는 앞 샷은 앵커가 아니다 (2026-09-20 Codex) ─────────────

def test_a_prev_without_a_verdict_is_not_an_anchor():
    """★칸이 비어 있는 것은 「비대상」이 아니라 **「아직 안 봤다」**.

    켜진 판에서는 **모든 갈래가** 판정을 적는다 — 비대상도
    `not_applicable` 로 적는다. 그러니 칸이 없다는 것은 아직 안 본 것이다.
    verify 는 그것을 미판정으로 붙잡는데 앵커 쪽만 통과시키면 두 소비자가
    어긋난다.
    """
    from app.services.still_recipe_service import gate_blocks_prev_anchor

    class _R:
        def __init__(self, data):
            self.data = data

    assert gate_blocks_prev_anchor(_R({A: {"selected": "a"}}), A) is True
    assert gate_blocks_prev_anchor(_R({}), A) is True
    assert gate_blocks_prev_anchor(
        _R({A: {"gate": {"outcome": "not_applicable"}}}), A) is False
    assert gate_blocks_prev_anchor(
        _R({A: {"gate": {"outcome": "clean"}}}), A) is False


def test_the_gate_off_run_still_uses_an_unjudged_prev(monkeypatch):
    """★꺼진 판은 **종전 그대로** — 판정이 없어도 앵커로 쓴다."""
    from app.core import config
    from app.services.still_recipe_service import gate_blocks_prev_anchor

    monkeypatch.setattr(config.settings, "still_winner_gate_enabled", False,
                        raising=False)

    class _R:
        def __init__(self, data):
            self.data = data

    assert gate_blocks_prev_anchor(_R({}), A) is False


def test_a_record_less_finished_prev_does_not_become_an_anchor(tmp_path):
    """★★`record` 없는 완료 샷은 **몸통에 안 들어온다** — 그 그림이 뒤 샷의
    권위가 되면 아무도 안 본 것이 연쇄로 번진다.

    A 는 완료로 표시돼 있고 기록이 없다(구세대 산출·기록 손상). 걷기는
    지문을 잴 근거가 없어 A 를 건너뛴다 — 그래서 A 에는 판정이 안 붙는다.
    B 는 그 A 를 앵커로 쓰면 안 된다.
    """
    per, cp = _persistence(), MagicMock()
    recs, gen, _ = _walk(
        tmp_path,
        verdicts={A: "clean", B: "clean", C: "clean", D: "clean"},
        persistence=per, scene_cp=cp,
        already_done={"a"})           # A 만 완료 표시 · 기록 없음

    assert A not in gen, "기록 없는 완료 샷이 몸통에 들어왔다"
    assert recs[B]["gate"]["outcome"] == GATE_DEPENDENCY, (
        "판정 없는 앞 샷을 앵커로 썼다")
    assert recs[B]["gate"]["blocked_by"] == A
    assert recs[C]["gate"]["outcome"] == GATE_DEPENDENCY, "연쇄가 안 잡혔다"
    assert D in gen, "독립 샷까지 세웠다"


# ── 재롤 레버가 **초기 후보 생성 지문**에 닿으면 안 된다 (Codex BLOCK 1)

def _fingerprint_extras(tmp_path, *, gate_on, reroll_on):
    """서비스가 `run_branch_select` 에 넘기는 `extra_fingerprint` 를 잡는다.

    ★**끝점에서 잰다** — 「안 넣었다」를 소스로 읽지 않고, 실제로 그
     갈래에 무엇이 닿는지 본다.
    """
    from app.core import config
    from app.services.still_recipe_service import (
        run_still_recipe_generation,
    )

    seen: Dict[str, Any] = {}

    def _gen(**kwargs):
        seen.update(kwargs.get("extra_fingerprint") or {})
        recipe = tmp_path / "scene" / "recipe"
        recipe.mkdir(parents=True, exist_ok=True)
        f = recipe / f"{kwargs['rec_key']}_sel.png"
        f.write_bytes(b"SEL")
        rec = {"gate": {"outcome": "clean", "policy": "p"}}
        kwargs["records"].data[kwargs["rec_key"]] = rec
        kwargs["records"].save()
        return f, rec

    _write_cp(tmp_path, "shot_ref_classify",
              {"shots": {A: {"person_visible": True}}, "scenes": {},
               "world_anchor_en": ""})
    _write_cp(tmp_path, "shot_continuity", {"pose_canon": []})
    _write_cp(tmp_path, "shot_conti_light", {"contis": {}})

    patches = [
        patch("app.core.config.settings.projects_dir", str(tmp_path)),
        patch("app.modules.pipeline.multiroll_gemini.make_nb2_gen_fn",
              return_value=MagicMock()),
        patch("app.modules.pipeline.multiroll_gemini.make_gemini_judge_fn",
              return_value=MagicMock()),
        patch("app.modules.pipeline.multiroll_gemini.make_gemini_critique_fn",
              return_value=MagicMock()),
        patch("app.modules.pipeline.still_recipe.run_branch_select",
              side_effect=_gen),
    ] + [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",
              gate_on, create=True),
        patch("app.core.config.settings.still_gate_reroll_enabled",
              reroll_on, 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=_persistence(), progress=MagicMock(),
            project_config=None, scene_dir=tmp_path / "scene",
            already_done_stills=set(), target_scenes=None)
    return seen


def test_the_reroll_lever_does_not_touch_the_generation_fingerprint(
        tmp_path, monkeypatch):
    """★★재롤 레버만 바꿔도 **초기 후보를 다시 사면 안 된다** (Codex BLOCK 1).

    공용 `extra_fingerprint` 는 **비대상 갈래까지** 간다. 거기 재롤 신원을
    넣으면 레버를 켜는 것만으로 지문이 달라져 `_clear_outputs` 가 초기
    후보를 지우고 다시 산다 — 재롤 상한은 그보다 뒤라 막지도 못한다.
    §15.5 에서 고친 것과 **같은 병**이다.
    """
    off = _fingerprint_extras(tmp_path / "a", gate_on=True, reroll_on=False)
    on = _fingerprint_extras(tmp_path / "b", gate_on=True, reroll_on=True)
    assert on == off, (
        "재롤 레버가 생성 지문을 움직였다 — 켜는 것만으로 초기 후보를 "
        f"다시 산다: {set(on) ^ set(off)}")
    assert not [k for k in on if k.startswith("gate_")], (
        f"게이트 신원이 생성 지문에 들어갔다: {sorted(on)}")