"""#108 i2i 시네마틱 변환 스테이지 — 서비스(걷기) 경계 회귀.

resolver 단위 시험(test_cine_transform_stage)이 못 보는 것을 잠근다
(Codex R1 리뷰 반영):
- BLOCK-1: 변환 실패 = 원본 fallback 으로 샷은 영속되지만 걷기 끝에
  StillCineTransformIncomplete — 스텝이 completed 로 봉인되지 않고,
  resume 재진입이 실패 변환만 다시 산다(성공 시 새 primary 영속).
- BLOCK-3: 변환 적용 자산의 직접 lineage = grok 이 실제 본 원본 sel
  1장(sha256) — nb2 롤 참조를 "직접 첨부"로 기록하지 않는다.
  generation_call_id 는 cine exact 만, miss=None(롤 fallback 금지).
하네스는 test_still_jit_verify 와 같은 방식. 시나리오 의존 0.
"""
from __future__ import annotations

import json
from types import SimpleNamespace
from unittest.mock import MagicMock, patch

import pytest

PID, EID = "SAMPLE_P", "SAMPLE_E"


def _write_cp(tmp_path, step_id, data):
    d = tmp_path / PID / "checkpoints" / "episodes" / EID / step_id
    d.mkdir(parents=True, exist_ok=True)
    (d / "manifest.json").write_text(
        json.dumps({"status": "completed", "data": data},
                   ensure_ascii=False), encoding="utf-8")


def _prior_asset(path, asset_id="PRIOR_ID", **kw):
    """옛 대표 자산 대역 — **진짜 모델로** 만든다.

    ★`SimpleNamespace` 흉내는 칸 이름이 틀려도 안 막는다. 실제로
     `is_manual_upload` 가 `prompt_used` 를 읽게 되자 이 파일의 시험
     여섯이 `AttributeError` 로 죽었다 — 코드가 아니라 **대역이** 틀린
     것이었다. 모델을 import 해서 만들면 칸이 없으면 만들 때 터진다.
    """
    from app.models.project import ImageAsset

    return ImageAsset(id=asset_id, file_path=str(path), **kw)


def _still(sid, si, shi):
    return {
        "id": sid, "still_index": 0, "scene_index": si, "shot_index": shi,
        "screenplay_scene_heading": f"S#{si}. SAMPLE",
        "beat_title": "", "still_frame_prompt": "SAMPLE still prompt",
        "camera_json": "{}", "lighting_json": "{}",
        "visible_entities_json": "[]", "dependent_scene_id": None,
    }


def _db_mock(first_result=None):
    q = MagicMock()
    for m in ("filter", "filter_by", "join", "order_by", "options"):
        getattr(q, m).return_value = q
    q.all.return_value = []
    q.first.return_value = first_result
    q.count.return_value = 0
    db = MagicMock()
    db.query.return_value = q
    return db


_FLAG_PATCHES = [
    ("still_bgfirst_enabled", False),
    ("still_bgfirst_full_enabled", False),
    ("still_variants_enabled", False),
    ("still_plate_select_enabled", False),
    ("still_conti_ab_enabled", False),
    ("multiroll_fix_rejudge_enabled", False),
    ("multiroll_gpt_composition_enabled", False),
    ("still_recipe_camera_frame_enabled", False),
    ("still_recipe_lighting_enabled", False),
    ("still_recipe_conduct_enabled", False),
    ("outdoor_lane_pipe_enabled", False),
    ("outdoor_lane_plan_enabled", False),
    ("background_share_plan_enabled", False),
    ("still_lane_prev_bgfirst_enabled", False),
    ("still_cine_transform_enabled", True),
    # 참조 선별(2026-08-19) — 기준선을 꺼짐으로 못박는다. 이 줄이 없으면
    # 하네스가 기계의 .env 값을 그대로 물어(지금 켜짐) 시험 결과가 환경에
    # 따라 흔들린다. 켜는 시험은 settings_over 로 이 값을 덮는다.
    ("still_fix_ref_gate_enabled", False),
    # 2026-08-20: 이 둘이 없으면 하네스가 기계의 .env 를 물어(둘 다 켜짐)
    # 걷기가 **실제 유료 LLM 호출**을 내보낸다 — 시험을 돌릴 때마다 돈이
    # 나간다(실측: 전체 시험 한 바퀴에 Gemini 168건). 이 파일의 시험들은
    # 간판·시대를 겨누지 않으므로 기준선에서 끈다.
    # 감시: .venv/bin/python -m pytest ... -p tests.netprobe
    ("signage_author_enabled", False),
    ("era_research_enabled", False),
    # 2026-08-25: **provider 도 기준선을 못박는다.** 이 줄이 없으면 하네스가
    # 기계의 .env(지금 reve)를 물어 **실제 fal 호출**이 나간다 — 위 두 줄이
    # 막은 것과 같은 함정이고, 실제로 이 파일 10건이 netprobe 에 걸렸다.
    # 이 파일의 stub 은 `GrokImageClient` 자리에 꽂히므로 provider 가 grok
    # 이어야 그 자리를 탄다. reve 경로는 test_cine_provider_swap 이 잰다.
    ("still_cine_provider", "grok"),
    # 연출 재료 이동도 같이 — 이 파일은 그것을 겨누지 않는다(켜지면 팩
    # selector 가 v24 로 옮겨 가 지문 단정이 흔들린다).
    ("still_cine_stage_direction_enabled", False),
]


class _StubCineClient:
    """GrokImageClient 대체 — 클래스 변수로 실패/성공·호출 수 통제."""

    fail = False
    calls: list = []

    def set_context(self, **kw):
        return self

    def generate_image(self, prompt, reference_images=None,
                       aspect_ratio="16:9", labeled_references=None):
        type(self).calls.append({
            "prompt": prompt, "labeled_references": labeled_references})
        if type(self).fail:
            raise RuntimeError("SAMPLE cine backend down")
        return b"CINE-BYTES", 10


def _resolve_by_op(still_id, episode_id, operation_type="",
                   multiroll_tag=""):
    return {"still_recipe_roll": "ROLL_CALL",
            "still_cine_transform": "CINE_CALL"}.get(operation_type)


def _run(tmp_path, *, stills, already_done, records=None, db=None,
         gen=None, persistence=None, scene_cp=None, settings_over=()):
    _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": {}})
    recipe_dir = tmp_path / "scene" / "recipe"
    recipe_dir.mkdir(parents=True, exist_ok=True)
    if records is not None:
        (recipe_dir / "records.json").write_text(
            json.dumps(records, ensure_ascii=False), encoding="utf-8")

    from app.services.still_recipe_service import (
        run_still_recipe_generation,
    )

    scene_cp = scene_cp if scene_cp is not None else MagicMock()
    persistence = persistence if persistence is not None else MagicMock()
    # safety 사다리 미발화 선언 — MagicMock 반환값(truthy)이 provenance
    # 정정 분기를 오발화시켜 review_notes JSON 직렬화가 깨진다
    persistence.safety_ladder_call_provenance.return_value = None
    patches = [
        patch("app.core.config.settings.projects_dir", str(tmp_path)),
        patch("app.modules.llm.grok_image_client.GrokImageClient",
              _StubCineClient),
        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.{name}", val, create=True)
        for name, val in _FLAG_PATCHES
    ] + [
        patch(f"app.core.config.settings.{name}", val, create=True)
        for name, val in settings_over
    ]
    from contextlib import ExitStack

    with ExitStack() as stack:
        for pch in patches:
            stack.enter_context(pch)
        generated = run_still_recipe_generation(
            db=db if db is not None else _db_mock(),
            project_id=PID, episode_id=EID,
            stills=stills, 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=already_done,
            target_scenes=None,
        )
    return generated, persistence, scene_cp


def _gen_writing(path, data):
    def _fake(**kwargs):
        path.write_bytes(data)
        return path, {"selected": "a", "totals": {}}
    return _fake


def _gen_echo(path, data):
    def _fake(**kwargs):
        path.write_bytes(data)
        rec = dict(kwargs["records"].data.get(kwargs["rec_key"]) or {})
        return path, rec
    return _fake


def test_cine_failure_persists_fallback_but_does_not_seal_then_resume_retries(
        tmp_path):
    """BLOCK-1 통합 회귀: 첫 run 변환 실패 → fallback asset 존재+스텝
    미완료(raise) → resume 이 실패 변환만 재호출 → 성공 후 정상 종료."""
    from app.services.still_recipe_service import (
        StillCineTransformIncomplete,
    )

    sel = tmp_path / "made_sel.png"

    # ── run 1: 신규 샷, 변환 실패 ──
    _StubCineClient.fail = True
    _StubCineClient.calls = []
    persistence1 = MagicMock()
    persistence1._resolve_generation_call_id.side_effect = _resolve_by_op
    scene_cp1 = MagicMock()
    with pytest.raises(StillCineTransformIncomplete) as ei:
        _run(tmp_path, stills=[_still("st_1", 1, 1)], already_done=set(),
             gen=_gen_writing(sel, b"SEL-BYTES"),
             persistence=persistence1, scene_cp=scene_cp1)
    assert "S1sh1" in str(ei.value)
    # 샷 자체는 원본으로 영속됐다(체인 안전 — fallback asset 실재)
    assert persistence1.save_single_scene_asset.call_count == 1
    scene_cp1.mark_completed.assert_called_once()
    asset_payload = persistence1.save_single_scene_asset.call_args.args[0]
    notes = json.loads(asset_payload["review_notes"])
    assert notes["cine_transform"]["applied"] is False
    assert "SAMPLE cine backend down" in notes["cine_transform"]["error"]
    # fallback 자산의 직접 lineage 는 롤 그대로(변환 미적용)
    assert asset_payload["generation_call_id"] == "ROLL_CALL"
    # 실패 기록이 디스크에 남아 재개 재시도의 근거가 된다
    stored = json.loads(
        (tmp_path / "scene" / "recipe" / "records.json").read_text(
            encoding="utf-8"))
    assert stored["S1sh1::cine"]["applied"] is False
    assert not (tmp_path / "scene" / "recipe" / "S1sh1_cine.png").exists()

    # ── run 2 (resume): 완료 샷 재진입, 변환만 재시도 → 성공 ──
    _StubCineClient.fail = False
    _StubCineClient.calls = []
    prior = tmp_path / "prior_primary.png"
    prior.write_bytes(b"SEL-BYTES")  # run1 이 영속한 fallback primary
    persistence2 = MagicMock()
    persistence2._resolve_generation_call_id.side_effect = _resolve_by_op
    scene_cp2 = MagicMock()
    generated, _, _ = _run(
        tmp_path, stills=[_still("st_1", 1, 1)], already_done={"st_1"},
        records=stored,
        db=_db_mock(_prior_asset(prior)),
        gen=_gen_echo(sel, b"SEL-BYTES"),
        persistence=persistence2, scene_cp=scene_cp2)
    # 변환 1회만 다시 샀다 (롤은 echo 재사용)
    assert len(_StubCineClient.calls) == 1
    assert (tmp_path / "scene" / "recipe" / "S1sh1_cine.png").read_bytes() \
        == b"CINE-BYTES"
    # 변환본으로 새 primary 영속 + 이번엔 raise 없음(정상 반환)
    assert persistence2.save_single_scene_asset.call_count == 1
    payload2 = persistence2.save_single_scene_asset.call_args.args[0]
    assert json.loads(payload2["review_notes"])["cine_transform"][
        "applied"] is True


def test_cine_applied_asset_lineage_is_cine_stage(tmp_path):
    """BLOCK-3: 변환 적용 자산 — 직접 입력=원본 sel(sha256, unresolved
    채널), 롤 참조는 review_notes.base_recipe 로 분리, prompt_used=변환
    문안, generation_call_id=cine exact."""
    import hashlib

    _StubCineClient.fail = False
    _StubCineClient.calls = []
    sel = tmp_path / "made_sel.png"
    persistence = MagicMock()
    persistence._resolve_generation_call_id.side_effect = _resolve_by_op
    # ★변환 자산의 모델 SOT = `generation_call_id` 가 가리키는 실제 호출
    #  (2026-08-26 감사 0-B). mock 이 그대로 흘러 모델 이름 자리에 박히지
    #  않도록 이 시험이 값을 못 박는다.
    persistence.call_model_name.return_value = "reve/2.1/edit"
    generated, persistence, scene_cp = _run(
        tmp_path, stills=[_still("st_1", 1, 1)], already_done=set(),
        gen=_gen_writing(sel, b"SEL-BYTES"), persistence=persistence)
    assert generated == 1
    payload = persistence.save_single_scene_asset.call_args.args[0]
    meta = persistence.save_single_scene_asset.call_args.args[1]

    # 직접 채널 = grok 이 실제 본 원본 sel 1장 (sha 신원)
    assert payload["actual_attached_image_ids"] == []
    assert payload["actual_attached_refs"] == []
    [src] = payload["unresolved_attached_refs"]
    assert src["role"] == "cine_source_sel"
    assert src["sha256"] == hashlib.sha256(b"SEL-BYTES").hexdigest()
    # 생성 프롬프트 = 변환 문안 (조립 전문 아님)
    from app.modules.pipeline.still_recipe import (
        build_cine_transform_prompt,
    )

    assert payload["prompt_used"] == build_cine_transform_prompt()
    # 생성 호출 링크 = cine exact
    assert payload["generation_call_id"] == "CINE_CALL"
    # 최종 bytes 모델 = grok / 팩 provenance = 변환 팩
    from app.core.config import settings as _s

    assert payload["generation_model"] == "reve/2.1/edit"
    from app.modules.pipeline.still_recipe import (
        CINE_TRANSFORM_PROMPT_VERSION,
        resolve_prompt_version,
    )

    assert meta["prompt_file_version"] == resolve_prompt_version(
        CINE_TRANSFORM_PROMPT_VERSION)
    # 롤 단계 provenance 는 분리 보존
    notes = json.loads(payload["review_notes"])
    assert notes["base_recipe"]["generation_call_id"] == "ROLL_CALL"
    assert "prompt_file_version" in notes["base_recipe"]
    # 최종 파일 bytes = 변환본
    assert (tmp_path / "scene").glob("*.png")
    final = payload["file_path"]
    from pathlib import Path as _P

    assert _P(final).read_bytes() == b"CINE-BYTES"


def test_lineage_annotation_respects_explicit_none_call_id():
    """Codex R2 BLOCK-1: 호출자가 generation_call_id 키를 명시했으면
    (명시 None 포함) persistence 가 broad resolve 로 되살리지 않는다 —
    과거 무관 single_scene 호출이 최종본에 거짓 연결되는 창."""
    from app.services.scene_persistence_service import (
        ScenePersistenceService,
    )

    svc = object.__new__(ScenePersistenceService)
    svc._db = MagicMock()
    svc._project_id = "P"
    svc._resolve_generation_call_id = MagicMock(return_value="BROAD_CALL")
    svc._resolve_registered_guide_asset_ids = MagicMock(
        side_effect=lambda u, e: ([], [], u))
    svc._resolve_prev_frame_asset_ids = MagicMock(
        side_effect=lambda u, e: ([], [], u))

    base = {"still_id": "S", "actual_attached_image_ids": [],
            "actual_attached_refs": [], "unresolved_attached_refs": []}

    # 키 명시(None) → None 그대로, broad resolve 미호출
    _, call_id, _ = svc._build_lineage_annotation(
        {**base, "generation_call_id": None}, "E", "single_scene_image_gen")
    assert call_id is None
    svc._resolve_generation_call_id.assert_not_called()

    # 키 명시(값) → 그 값
    _, call_id, _ = svc._build_lineage_annotation(
        {**base, "generation_call_id": "CINE_CALL"}, "E",
        "single_scene_image_gen")
    assert call_id == "CINE_CALL"
    svc._resolve_generation_call_id.assert_not_called()

    # 키 부재 → legacy broad fallback 유지
    _, call_id, _ = svc._build_lineage_annotation(
        dict(base), "E", "single_scene_image_gen")
    assert call_id == "BROAD_CALL"


def test_fresh_paid_cine_success_persists_even_when_bytes_identical(
        tmp_path):
    """Codex R2 BLOCK-2: 유료 신규 변환 성공이 이전 primary 와 bytes
    동일해도(계약 bump 후 동일 산출) spent-same 생략으로 넘기면 primary
    메타가 옛 단계 것으로 남은 채 봉인된다 — 반드시 영속한다.
    (실패→실패·재사용 바퀴는 기존대로 생략 — 별도 시험이 잠근다.)"""
    _StubCineClient.fail = False
    _StubCineClient.calls = []
    sel = tmp_path / "made_sel.png"
    recipe_dir = tmp_path / "scene" / "recipe"
    recipe_dir.mkdir(parents=True, exist_ok=True)
    prior = tmp_path / "prior_primary.png"
    prior.write_bytes(b"CINE-BYTES")  # 옛 변환본과 동일 산출이 될 상황

    records = {
        "S1sh1": {"selected": "a", "totals": {}},
        # 지문 stale → 재변환 강제 (계약/문안/모델 bump 모사)
        "S1sh1::cine": {"applied": True, "fingerprint": "STALE",
                        "file": "S1sh1_cine.png", "model": "m", "pack": "p"},
    }
    persistence = MagicMock()
    persistence._resolve_generation_call_id.side_effect = _resolve_by_op
    generated, persistence, scene_cp = _run(
        tmp_path, stills=[_still("st_1", 1, 1)], already_done={"st_1"},
        records=records,
        db=_db_mock(_prior_asset(prior)),
        gen=_gen_echo(sel, b"SEL-BYTES"), persistence=persistence)

    assert len(_StubCineClient.calls) == 1  # 유료 재변환이 실제 일어남
    assert persistence.save_single_scene_asset.call_count == 1, \
        "bytes 동일이어도 신규 유료 성공 변환은 영속해야 한다"
    payload = persistence.save_single_scene_asset.call_args.args[0]
    notes = json.loads(payload["review_notes"])
    assert notes["cine_transform"]["applied"] is True
    assert payload["generation_call_id"] == "CINE_CALL"


def test_settings_put_requires_admin_dependency():
    """Codex R2 BLOCK-3: 전역 유료 토글 PUT 은 admin 전용 — 라우트
    의존성에 require_admin 이 실려 있어야 한다."""
    from app.api.deps import require_admin
    from app.api.v1.settings import router

    put_routes = [
        r for r in router.routes
        if "PUT" in (getattr(r, "methods", None) or set())
    ]
    assert put_routes, "PUT 라우트가 있어야 한다"
    for r in put_routes:
        dep_calls = [d.call for d in r.dependant.dependencies]
        assert require_admin in dep_calls


def _gen_recording(path, data, sink):
    """run_branch_select 자리를 대신하면서 받은 인자를 그대로 적어 둔다.

    바깥 세계(그림 모델 호출)만 흉내다 — 켤지 말지 판별하고 팩 문안을
    읽고 지문을 조립하는 일은 전부 실물 서비스 코드가 한다.
    """
    def _fake(**kwargs):
        sink.append(kwargs)
        path.write_bytes(data)
        return path, {"selected": "a", "totals": {}}
    return _fake


@pytest.mark.parametrize("gate_on", [True, False], ids=["on", "off"])
def test_fix_ref_gate_reaches_the_branch_and_the_fingerprint(
        tmp_path, gate_on):
    """★참조 선별이 **서비스를 실제로 돌렸을 때** 끝까지 가는가 (양방향).

    잠그는 것은 셋이다.
      ① 실행 몸통이 받는 `fix_ref_gate` — 이 값이 안 가면 편집 호출에
         참조가 종전대로 전부 붙는다.
      ② 팩 v13 문안 둘이 **실제로 읽혀** 실린다 — 빈 채로 가면 「없는
         것을 새로 넣어라」 절이 통째로 사라진다.
      ③ 샷 지문에 `fix_ref_gate`·`fix_ref_contract` 가 접힌다 — 없으면
         선별 없이 만든 옛 그림이 「선별 적용됨」으로 재사용된다
         (2026-08-08 스테일 재사용 사고와 같은 부류).

    꺼짐도 함께 본다 — 켜짐만 보면 「원래 늘 그랬던 것」과 구분되지
    않는다. 꺼짐에서 셋 다 없어야 완성된 에피소드의 지문이 안 흔들린다.
    (run_branch_select 가 아래 실행 몸통으로 넘길 때 키 자체를 생략하는
    계약은 tests/unit/test_still_recipe.py 의 생략 목록이 잠근다.)
    """
    _StubCineClient.fail = False
    _StubCineClient.calls = []
    sel = tmp_path / "made_sel.png"
    seen: list = []
    persistence = MagicMock()
    persistence._resolve_generation_call_id.side_effect = _resolve_by_op

    _run(tmp_path, stills=[_still("st_1", 1, 1)], already_done=set(),
         gen=_gen_recording(sel, b"SEL-BYTES", seen),
         persistence=persistence,
         settings_over=[("still_fix_ref_gate_enabled", gate_on)])

    assert seen, "run_branch_select 가 한 번도 안 불렸다"
    kw = seen[0]
    fp = kw["extra_fingerprint"]

    if not gate_on:
        assert not kw.get("fix_ref_gate")
        assert not kw.get("fix_missing_texts")
        assert "fix_ref_gate" not in fp
        assert "fix_ref_contract" not in fp
        return

    # ① 켜짐 표식이 그대로 간다
    assert kw["fix_ref_gate"] is True

    # ② 팩 v13 문안 둘 — 실제 파일에서 읽힌 값과 같고, 비어 있지 않다
    from app.modules.pipeline.multiroll_gemini import (
        FIX_MISSING_PACK_VERSION,
        fix_ref_contract_sha,
        load_fix_missing_texts,
    )

    texts = load_fix_missing_texts(FIX_MISSING_PACK_VERSION)
    assert set(texts) == {"missing_head", "missing_tail"}
    assert all(v.strip() for v in texts.values()), "팩 문안이 비었다"
    assert kw["fix_missing_texts"] == texts

    # ③ 샷 지문 — 켜짐 표식 + 팩 밖 코드 문안의 지문(오늘 새로 넣은 키)
    assert fp["fix_ref_gate"] is True
    assert fp["fix_ref_contract"] == fix_ref_contract_sha()
    # 팩 쪽 지문도 함께 있어야 문안 개정이 반영된다
    assert fp["fix_missing_pack_content"]


def test_cine_exact_miss_leaves_generation_call_none(tmp_path):
    """BLOCK-3: cine exact resolve miss → None — 롤 호출로 fallback 하면
    grok 산출이 nb2 호출에 이어지는 거짓 링크다."""
    _StubCineClient.fail = False
    _StubCineClient.calls = []
    sel = tmp_path / "made_sel.png"
    persistence = MagicMock()

    def _roll_only(still_id, episode_id, operation_type="",
                   multiroll_tag=""):
        return {"still_recipe_roll": "ROLL_CALL"}.get(operation_type)

    persistence._resolve_generation_call_id.side_effect = _roll_only
    _run(tmp_path, stills=[_still("st_1", 1, 1)], already_done=set(),
         gen=_gen_writing(sel, b"SEL-BYTES"), persistence=persistence)
    payload = persistence.save_single_scene_asset.call_args.args[0]
    assert payload["generation_call_id"] is None


# ── 포기(declined) 계약 — 걷기 층 (2026-08-19/20) ────────────────
#
# 변환이 검열로 막혀 「포기」로 돌아온 샷은 실패와 다루는 법이 반대다.
#   · 실패 = 미완  → 스텝을 안 닫는다. resume 이 그 변환만 다시 산다.
#   · 포기 = 결말  → 스텝을 닫는다. 원본이 최종본이고, 자산에 표식을
#                    남겨 나중에 「일시 실패」와 구분되게 한다.
# 아래 넷이 그 갈림을 걷기 층에서 잠근다. 바깥 세계(그림 모델 호출)만
# 흉내고 — 검열인지 가르는 일, 포기 셈, 지문 계산, 자산 조립, 걷기 끝
# 집계는 전부 실물 코드가 돈다.

# 제공자가 돌려주는 거부 문구를 그대로 본뜬 문장. 이 문구를 보고 검열
# 여부를 정하는 것은 시험이 아니라 실물 판별(is_moderation_error)이다.
_MODERATION_TEXT = "SAMPLE Generated image rejected by content moderation."


def _cine_refuses(text=_MODERATION_TEXT):
    """변환 호출만 이 문구로 거부시키는 컨텍스트 관리자.

    기존 하네스의 `fail` 스위치는 검열이 아닌 오류(백엔드 다운)를 뜻해
    그대로 두고, 여기서는 호출 하나만 갈아 끼운다 — 빠져나오면 원래대로
    돌아오므로 뒤 시험에 문구가 새지 않는다.
    """
    def _boom(self, prompt, reference_images=None, aspect_ratio="16:9",
              labeled_references=None):
        type(self).calls.append({
            "prompt": prompt, "labeled_references": labeled_references})
        raise RuntimeError(text)

    return patch.object(_StubCineClient, "generate_image", _boom)


def _cine_fingerprint_of(sel_bytes):
    """이번 걷기가 계산할 변환 지문 — 실물 함수·실물 팩으로 그대로 구한다.

    문자열을 지어내면 「지문이 같다」는 전제 자체가 흉내가 된다.
    """
    from app.core.config import settings as _s
    from app.modules.pipeline.cine_transform import (
        CINE_TRANSFORM_STEM,
        cine_fingerprint,
    )
    from app.modules.pipeline.still_recipe import (
        CINE_TRANSFORM_PROMPT_VERSION,
        recipe_stem_content_hash,
    )

    return cine_fingerprint(
        stem_content_hash=recipe_stem_content_hash(
            CINE_TRANSFORM_PROMPT_VERSION, CINE_TRANSFORM_STEM),
        model=_s.grok_image_model, sel_bytes=sel_bytes)


@pytest.fixture(autouse=True)
def _no_moderation_fallback(monkeypatch):
    """★[2026-09-09] 이 파일의 검열 시험들은 **대체가 없던 시절의 계약**을
    잰다 — 「주 제공자가 declined 면 원본을 최종본으로 확정한다」.

    대체 체인(grok → seedream)이 생기면서 그 시험들이 대체를 진짜로 부르려
    했다. 대체 자체는 `test_cine_moderation_fallback` 이 따로 잰다.
    """
    from app.core.config import settings
    monkeypatch.setattr(settings, "still_cine_moderation_fallback_providers",
                        "", raising=False)


def _stored_records(tmp_path):
    return json.loads(
        (tmp_path / "scene" / "recipe" / "records.json").read_text(
            encoding="utf-8"))


def test_cine_declined_closes_the_step_and_stamps_the_asset(tmp_path):
    """(1)+(2) 포기는 실패로 세지 않는다 — 스텝이 닫히고 표식이 남는다.

    잠그는 것 둘.
      ① 걷기가 StillCineTransformIncomplete 없이 끝난다. 포기를 실패로
         세면 스텝이 영영 안 닫히고, 재개마다 앞 단계가 다시 돌아 연쇄
         재생성이 난다(08-19 실측: 한 바퀴에 그림 66장).
      ② 그 샷 자산의 review_notes.cine_transform 에 declined=True 와
         사유가 남는다. 이 표식이 없으면 자산에는 applied=false 와 오류
         문구만 남아, 결말인지 일시 실패인지 가릴 길이 글자 대조뿐이다.
    """
    _StubCineClient.calls = []
    sel = tmp_path / "made_sel.png"
    persistence = MagicMock()
    persistence._resolve_generation_call_id.side_effect = _resolve_by_op
    scene_cp = MagicMock()

    with _cine_refuses():
        generated, _, _ = _run(
            tmp_path, stills=[_still("st_1", 1, 1)], already_done=set(),
            gen=_gen_writing(sel, b"SEL-BYTES"),
            persistence=persistence, scene_cp=scene_cp,
            settings_over=[("still_cine_moderation_give_up_after", 1)])

    # ① 미완 예외 없이 반환됐다 = 스텝이 닫힌다
    assert generated == 1
    scene_cp.mark_completed.assert_called_once()
    scene_cp.mark_failed.assert_not_called()
    assert len(_StubCineClient.calls) == 1  # 거부는 한 번만 산다

    # 원본이 최종본 — 변환본 파일은 아예 없다
    assert not (tmp_path / "scene" / "recipe" / "S1sh1_cine.png").exists()
    payload = persistence.save_single_scene_asset.call_args.args[0]
    from pathlib import Path as _P

    assert _P(payload["file_path"]).read_bytes() == b"SEL-BYTES"
    # 변환이 안 붙었으니 직접 lineage 는 롤 그대로
    assert payload["generation_call_id"] == "ROLL_CALL"

    # ② 자산 기록의 포기 표식
    cine = json.loads(payload["review_notes"])["cine_transform"]
    assert cine["applied"] is False
    assert cine["declined"] is True
    assert cine["declined_reason"] == "moderation"
    assert "moderation" in (cine["error"] or "").lower()

    # 디스크 기록도 같다 — 다음 방문이 이 기록을 보고 돈을 안 쓴다
    stored = _stored_records(tmp_path)["S1sh1::cine"]
    assert stored["declined"] is True
    assert stored["moderation_refusals"] == 1


def test_cine_moderation_below_give_up_still_blocks_the_step(tmp_path):
    """대조: 포기 기준에 못 미친 검열 거부는 아직 미완이다.

    이 시험이 없으면 위 (1)은 「검열이면 무조건 통과」와 구분되지 않는다.
    기준(2회)에 못 미친 첫 거부까지 스텝을 닫아 버리면, 다시 보내면 될
    샷까지 미변환인 채로 봉인된다.
    """
    from app.services.still_recipe_service import (
        StillCineTransformIncomplete,
    )

    _StubCineClient.calls = []
    sel = tmp_path / "made_sel.png"
    persistence = MagicMock()
    persistence._resolve_generation_call_id.side_effect = _resolve_by_op

    with _cine_refuses(), pytest.raises(StillCineTransformIncomplete) as ei:
        _run(tmp_path, stills=[_still("st_1", 1, 1)], already_done=set(),
             gen=_gen_writing(sel, b"SEL-BYTES"), persistence=persistence,
             settings_over=[("still_cine_moderation_give_up_after", 2)])
    assert "S1sh1" in str(ei.value)

    # 샷은 원본으로 영속되지만 포기 표식은 아직 없다
    payload = persistence.save_single_scene_asset.call_args.args[0]
    cine = json.loads(payload["review_notes"])["cine_transform"]
    assert cine["applied"] is False
    assert "declined" not in cine
    stored = _stored_records(tmp_path)["S1sh1::cine"]
    assert stored["moderation_refusals"] == 1
    assert "declined" not in stored


def _revisit_walk(tmp_path, records, persistence, prior, sel, settings_over):
    """완료 샷 재방문 한 바퀴 — 옛 primary 를 든 채 같은 샷을 다시 밟는다."""
    _StubCineClient.calls = []
    with _cine_refuses():  # 변환을 부르면 검열 거부가 돌아온다
        _run(tmp_path, stills=[_still("st_1", 1, 1)],
             already_done={"st_1"}, records=records,
             db=_db_mock(_prior_asset(prior)),
             gen=_gen_echo(sel, b"SEL-BYTES"), persistence=persistence,
             settings_over=settings_over)
    return _stored_records(tmp_path)


def test_fresh_decline_persists_even_when_output_bytes_identical(
        tmp_path, caplog):
    """(3) 이번 걷기에 **새로 확정된** 포기는 산출이 같아도 영속한다.

    완료 샷 재방문 갈래다. 포기하면 최종 원본이 원본 sel 그대로라 옛
    primary 와 bytes 가 같다 — 「같으니 건너뛴다」로 넘기면 자산은
    「applied=false + 오류 문구」인 채로 스텝이 닫혀 봉인되고, 결말인지
    일시 실패인지 나중에 가릴 길이 없어진다.

    어느 갈래를 밟았는지는 로그로 가른다 — 「영속 생략」 문구는 생략
    갈래에만 있다. 그 문구가 없다 + 영속이 일어났다, 두 신호가 함께여야
    이 시험이 새 갈래를 짚었다고 말할 수 있다.
    """
    import logging

    sel = tmp_path / "made_sel.png"
    prior = tmp_path / "prior_primary.png"
    prior.write_bytes(b"SEL-BYTES")  # 옛 primary = 변환 미적용 원본
    records = {
        "S1sh1": {"selected": "a", "totals": {}},
        # 지난 걷기의 일시 실패 기록(지문 다름) — 이번엔 다시 시도한다
        "S1sh1::cine": {"applied": False, "fingerprint": "STALE",
                        "error": "RuntimeError: SAMPLE 일시 오류"},
    }
    persistence = MagicMock()
    persistence._resolve_generation_call_id.side_effect = _resolve_by_op

    with caplog.at_level(logging.WARNING,
                         logger="app.services.still_recipe_service"):
        stored = _revisit_walk(
            tmp_path, records, persistence, prior, sel,
            [("still_cine_moderation_give_up_after", 1)])

    assert len(_StubCineClient.calls) == 1  # 이번 걷기에 실제로 시도했다
    assert persistence.save_single_scene_asset.call_count == 1, (
        "새로 확정된 포기는 산출 bytes 가 같아도 영속해야 한다")
    payload = persistence.save_single_scene_asset.call_args.args[0]
    from pathlib import Path as _P

    # 전제 확인 — 산출이 옛 primary 와 bytes 가 같다. 다르면 생략 갈래가
    # 아예 열리지 않아 이 시험이 헛돈다.
    assert _P(payload["file_path"]).read_bytes() == prior.read_bytes()
    # 생략 갈래를 안 탔다
    assert "영속 생략" not in caplog.text
    cine = json.loads(payload["review_notes"])["cine_transform"]
    assert cine["declined"] is True
    assert cine["declined_reason"] == "moderation"
    assert stored["S1sh1::cine"]["declined"] is True


def test_settled_decline_pays_nothing_and_skips_persist(tmp_path, caplog):
    """대조: 지난 걷기에 이미 확정된 포기는 호출도 영속도 없다.

    이 갈래가 조용히 건너뛰기 때문에 바로 위 시험의 「이번에 새로
    확정됐는가」 판별이 필요하다. 여기서 옛 primary·산출 bytes 는 위와
    똑같이 두므로, 이 시험이 통과한다는 것은 위 상황에서도 bytes 가
    실제로 같았다는 뜻이 된다.

    두 바퀴를 돈다 — 첫 바퀴가 이 샷의 곁다리 기록(간판 등)을 채우고,
    둘째 바퀴가 실제 재개에 해당한다. 실제 재개는 그 기록이 이미 있는
    상태로 오므로, 그래야 「기록이 하나도 안 움직인다」를 잴 수 있다.
    """
    import logging

    sel = tmp_path / "made_sel.png"
    prior = tmp_path / "prior_primary.png"
    prior.write_bytes(b"SEL-BYTES")
    settled = {
        "applied": False, "declined": True, "declined_reason": "moderation",
        "moderation_refusals": 2,
        # 지문은 실물 계산 그대로 — 지어내면 「같다」는 전제가 흉내가 된다
        "fingerprint": _cine_fingerprint_of(b"SEL-BYTES"),
        "model": "m", "pack": "p", "source_file": "made_sel.png",
        "error": f"RuntimeError: {_MODERATION_TEXT}",
    }

    def _mock_persistence():
        p = MagicMock()
        p._resolve_generation_call_id.side_effect = _resolve_by_op
        return p

    warm = _revisit_walk(
        tmp_path, {"S1sh1": {"selected": "a", "totals": {}},
                   "S1sh1::cine": dict(settled)},
        _mock_persistence(), prior, sel, ())

    persistence = _mock_persistence()
    caplog.clear()  # 첫 바퀴 로그를 지운다 — 둘째 바퀴만 본다
    with caplog.at_level(logging.WARNING,
                         logger="app.services.still_recipe_service"):
        after = _revisit_walk(tmp_path, warm, persistence, prior, sel, ())

    assert _StubCineClient.calls == [], "확정된 포기에 다시 돈을 쓰면 안 된다"
    assert persistence.save_single_scene_asset.call_count == 0
    # 기록이 1비트라도 움직이면 지출로 읽혀 재생성 상한 제동이 거짓
    # 발동한다 — 조용한 재사용이어야 한다
    assert after == warm
    assert after["S1sh1::cine"] == settled
    assert "영속 생략" not in caplog.text
    assert "입력이 낡아 재생성" not in caplog.text


def test_human_rejected_cine_keeps_original_and_buys_nothing(tmp_path):
    """사람이 변환본을 거절한 샷은 **원본이 최종본** — 변환을 사지 않는다
    (2026-09-19).

    실측(컨트리로드 2판, 사용자 육안): 변환이 로봇 얼굴을 사람 얼굴로,
    파란 홀로그램을 불투명한 실물로 바꿨다. `::cine` 기록에 rejected 를 손으로
    심으면 검사가 꺼진 동안 **복권된다** — 그래서 샷 단위 기록을 따로 둔다.

    잠그는 것: 변환 호출 0 · 최종 자산 = 원본 bytes · 자산 기록에 사람의 거절 ·
    스텝이 닫힌다(실패로 세지 않는다) · 거절 기록은 디스크에 그대로.
    """
    from app.modules.pipeline.cine_transform import (
        cine_human_key,
        record_human_keep_original,
    )

    class _Rec:
        def __init__(self):
            self.data = {}

        def save(self):
            return None

    pre = _Rec()
    record_human_keep_original(pre, "S1sh1", reason="SAMPLE 사람이 본 결함")
    _StubCineClient.calls = []
    sel = tmp_path / "made_sel.png"
    persistence = MagicMock()
    persistence._resolve_generation_call_id.side_effect = _resolve_by_op
    scene_cp = MagicMock()

    generated, _, _ = _run(
        tmp_path, stills=[_still("st_1", 1, 1)], already_done=set(),
        records=pre.data, gen=_gen_writing(sel, b"SEL-BYTES"),
        persistence=persistence, scene_cp=scene_cp)

    assert generated == 1
    assert _StubCineClient.calls == []            # 변환 호출 0
    scene_cp.mark_completed.assert_called_once()   # 실패로 세지 않는다
    scene_cp.mark_failed.assert_not_called()
    payload = persistence.save_single_scene_asset.call_args.args[0]
    from pathlib import Path as _P

    assert _P(payload["file_path"]).read_bytes() == b"SEL-BYTES"
    cine = json.loads(payload["review_notes"])["cine_transform"]
    assert cine["applied"] is False and cine["rejected"] is True
    assert cine["rejected_by"] == "human"
    assert cine["finish_owner"] == "none"
    stored = _stored_records(tmp_path)
    assert stored[cine_human_key("S1sh1")]["decision"] == "keep_original"
    assert "S1sh1::cine" not in stored             # 변환 기록 자체가 없다


def test_human_rejection_of_a_finished_cine_switches_to_original_once(
        tmp_path, caplog):
    """운영 경로 그대로 — **이미 변환본이 최종인 샷**을 사람이 거절한 뒤 재개
    (Codex 2026-09-19 비차단 1).

    첫 재개: 변환 호출 0 · 원본(SEL)을 새 최종으로 영속 · 자산에 사람의 거절.
    둘째 재개: 호출 0 · 영속 0 · 기록 1비트도 안 움직임(거짓 지출 없음).
    """
    import logging

    from app.modules.pipeline.cine_transform import (
        cine_human_key,
        record_human_keep_original,
    )

    sel = tmp_path / "made_sel.png"
    recipe_dir = tmp_path / "scene" / "recipe"
    recipe_dir.mkdir(parents=True, exist_ok=True)
    (recipe_dir / "S1sh1_cine.png").write_bytes(b"CINE-BYTES")
    prior_cine = tmp_path / "prior_primary_cine.png"
    prior_cine.write_bytes(b"CINE-BYTES")   # 지금 최종 = 변환본

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

        def save(self):
            return None

    recs = _Rec({
        "S1sh1": {"selected": "a", "totals": {}},
        "S1sh1::cine": {
            "applied": True, "file": "S1sh1_cine.png",
            "fingerprint": _cine_fingerprint_of(b"SEL-BYTES"),
            "model": "m", "pack": "p", "source_file": "made_sel.png"},
    })
    record_human_keep_original(recs, "S1sh1", reason="SAMPLE 사람이 본 결함",
                               rejected_cine=recs.data["S1sh1::cine"])

    def _mock_persistence():
        p = MagicMock()
        p._resolve_generation_call_id.side_effect = _resolve_by_op
        return p

    # ① 첫 재개 — 원본으로 바뀐다
    p1 = _mock_persistence()
    warm = _revisit_walk(tmp_path, recs.data, p1, prior_cine, sel, ())
    assert _StubCineClient.calls == []
    assert p1.save_single_scene_asset.call_count == 1
    payload = p1.save_single_scene_asset.call_args.args[0]
    from pathlib import Path as _P

    assert _P(payload["file_path"]).read_bytes() == b"SEL-BYTES"
    cine = json.loads(payload["review_notes"])["cine_transform"]
    assert cine["rejected_by"] == "human" and cine["applied"] is False
    assert warm[cine_human_key("S1sh1")]["decision"] == "keep_original"

    # ② 둘째 재개 — 최종이 이미 원본이다: 호출·영속·기록 변화 0
    prior_sel = tmp_path / "prior_primary_sel.png"
    prior_sel.write_bytes(b"SEL-BYTES")
    p2 = _mock_persistence()
    caplog.clear()
    with caplog.at_level(logging.WARNING,
                         logger="app.services.still_recipe_service"):
        after = _revisit_walk(tmp_path, warm, p2, prior_sel, sel, ())
    assert _StubCineClient.calls == []
    assert p2.save_single_scene_asset.call_count == 0
    assert after == warm