"""ScenePersistenceService 단위 테스트 — W5 F22 Phase B.9.1.

SceneImageService에서 분리된 ImageAsset 저장 로직의 실제 동작을 검증한다.
scene_image_service 쪽에는 delegation smoke test만 남긴다.
"""
from __future__ import annotations

from pathlib import Path
from unittest.mock import MagicMock

import pytest

from app.services.scene_persistence_service import ScenePersistenceService


@pytest.fixture
def svc() -> ScenePersistenceService:
    instance = ScenePersistenceService.__new__(ScenePersistenceService)
    instance._db = MagicMock()
    instance._project_id = "p1"
    return instance


def _make_var(vid: str = "a1", file_path: str = "/tmp/a.png", **extra):
    base = {
        "id": vid,
        "asset_type": "scene",
        "entity_id": None,
        "still_id": "s1",
        "episode_id": "e1",
        "file_path": file_path,
        "prompt_used": "prompt",
        "generation_model": "gemini",
        "width": None,
        "height": None,
        "status": "generated",
        "review_notes": "",
        "created_at": "2026-04-22T00:00:00+00:00",
    }
    base.update(extra)
    return base


def _make_lineage():
    return {
        "prompt_type": "cinematic",
        "code_version": "0.6.1",
        "prompt_file_version": "v5",
        "reference_image_ids": "[]",
    }


def test_save_scene_variations_empty_list(svc):
    """빈 input → 빈 결과 + commit 호출."""
    ids, paths = svc.save_scene_variations([], _make_lineage())
    assert ids == []
    assert paths == []
    svc._db.commit.assert_called_once()
    svc._db.add.assert_not_called()


def test_save_scene_variations_adds_each_asset(svc):
    """N개 variation → N개 ImageAsset add + 한 번의 commit."""
    vars_in = [
        _make_var("a1", "/tmp/a.png"),
        _make_var("a2", "/tmp/b.png", variant_type="angle"),
        _make_var("a3", "/tmp/c.png", theme_label="close-up"),
    ]
    lineage = _make_lineage()

    ids, paths = svc.save_scene_variations(vars_in, lineage)

    assert ids == ["a1", "a2", "a3"]
    assert paths == [Path("/tmp/a.png"), Path("/tmp/b.png"), Path("/tmp/c.png")]
    assert svc._db.add.call_count == 3
    svc._db.commit.assert_called_once()


def test_save_scene_variations_persists_actual_attached_lineage(svc):
    """★ P0 (2026-07-01): actual_attached_image_ids → input_image_ids(정확히 그 UUID),
    generation_call_id 저장, pipeline_metadata_json 에 actual/unresolved refs 보존,
    reference_image_ids 는 scene_lineage 그대로(오염 X)."""
    import json

    v = _make_var(
        "a1",
        actual_attached_image_ids=["uuid-bg", "uuid-face", "uuid-guide"],
        actual_attached_refs=[
            {"asset_id": "uuid-bg", "role": "background_render", "label": "bg"},
            {"asset_id": "uuid-face", "role": "reference_face", "label": "face"},
            {"asset_id": "uuid-guide", "role": "registered_pose_guide", "label": "guide"},
        ],
        unresolved_attached_refs=[{"role": "previous_shot_continuity", "label": "prev"}],
        generation_call_id="call-123",
    )
    lineage = {
        "prompt_type": "cinematic",
        "code_version": "0.6.1",
        "prompt_file_version": "v5",
        "reference_image_ids": "[\"face-lineage\"]",
    }
    svc.save_scene_variations([v], lineage)

    asset = svc._db.add.call_args_list[0].args[0]
    assert json.loads(asset.input_image_ids) == ["uuid-bg", "uuid-face", "uuid-guide"]
    assert asset.generation_call_id == "call-123"
    meta = json.loads(asset.pipeline_metadata_json)
    assert len(meta["actual_attached_refs"]) == 3
    assert meta["unresolved_attached_refs"][0]["role"] == "previous_shot_continuity"
    # reference_image_ids 는 scene_lineage 그대로 — input_image_ids 와 분리 유지.
    assert asset.reference_image_ids == "[\"face-lineage\"]"


def test_save_scene_variations_falls_back_to_pose_guide_ids(svc):
    """actual_attached_image_ids 부재 시 legacy pose_guide_asset_ids fallback."""
    import json

    v = _make_var("a1", pose_guide_asset_ids=["uuid-guide"], generation_call_id="c1")
    svc.save_scene_variations([v], _make_lineage())
    asset = svc._db.add.call_args_list[0].args[0]
    assert json.loads(asset.input_image_ids) == ["uuid-guide"]


def _svc_with_query_returns(all_returns):
    """_resolve_registered_guide_asset_ids 용 — .filter chain + .all() side_effect 목."""
    instance = ScenePersistenceService.__new__(ScenePersistenceService)
    db = MagicMock()
    q = MagicMock()
    db.query.return_value = q
    q.filter.return_value = q
    q.all.side_effect = all_returns
    instance._db = db
    instance._project_id = "p1"
    return instance


def _reg_guide_unresolved(**over):
    base = {
        "role": "immobilized_pose_guide",
        "label": "POSE / SUPPORT GUIDE",
        "pipeline_role": "registered_pose_guide",
        "group_id": "vca-s12-c04-immobilized",
        "bg_key": "L04B06",
        "guide_hash": "abc123",
    }
    base.update(over)
    return base


def test_resolve_registered_guide_exact_triple():
    """★ P0: (group_id,bg_key,guide_hash) exact 1개 → resolve + unresolved 제거."""
    svc = _svc_with_query_returns([[("guide-uuid",)]])  # exact match 1건
    ids, refs, remaining = svc._resolve_registered_guide_asset_ids(
        [_reg_guide_unresolved()], "e1",
    )
    assert ids == ["guide-uuid"]
    assert refs[0]["asset_id"] == "guide-uuid"
    assert refs[0]["resolved_from_unresolved"] is True
    assert remaining == []


def test_resolve_registered_guide_pair_fallback():
    """guide_hash 부재 → (group_id,bg_key) 정확히 1개 → resolve."""
    svc = _svc_with_query_returns([[("guide-uuid2",)]])  # pair match 1건 (exact 스킵됨)
    ids, refs, remaining = svc._resolve_registered_guide_asset_ids(
        [_reg_guide_unresolved(guide_hash=None)], "e1",
    )
    assert ids == ["guide-uuid2"]
    assert remaining == []


def test_resolve_registered_guide_ambiguous_pair_kept_unresolved():
    """(group_id,bg_key) 2+ → wrong edge 회피, unresolved 유지 + reason."""
    # exact(triple) → [] , pair → 2건
    svc = _svc_with_query_returns([[], [("a",), ("b",)]])
    ids, refs, remaining = svc._resolve_registered_guide_asset_ids(
        [_reg_guide_unresolved()], "e1",
    )
    assert ids == []
    assert refs == []
    assert len(remaining) == 1
    assert remaining[0]["reason"] == "registered_pose_guide_ambiguous_pair"


def test_resolve_registered_guide_non_guide_passthrough():
    """registered_pose_guide 아닌 unresolved(prev-shot)는 그대로 통과."""
    svc = _svc_with_query_returns([])
    prev = {"role": "previous_shot_continuity", "label": "prev"}
    ids, refs, remaining = svc._resolve_registered_guide_asset_ids([prev], "e1")
    assert ids == []
    assert remaining == [prev]


def test_save_scene_variations_defaults_variant_type_to_base(svc):
    """variant_type 누락 → 'base' 기본값 적용."""
    from app.models.project import ImageAsset

    v = _make_var("a1")
    v.pop("variant_type", None)
    svc.save_scene_variations([v], _make_lineage())

    added = svc._db.add.call_args_list[0].args[0]
    assert isinstance(added, ImageAsset)
    assert added.variant_type == "base"
    assert added.is_primary == 0


def test_save_scene_variations_applies_lineage_uniformly(svc):
    """모든 asset에 동일 lineage 적용."""
    vars_in = [_make_var("a1"), _make_var("a2")]
    lineage = {
        "prompt_type": "cinematic",
        "code_version": "9.9.9",
        "prompt_file_version": "v99",
        "reference_image_ids": "[\"ref1\"]",
    }

    svc.save_scene_variations(vars_in, lineage)

    for call in svc._db.add.call_args_list:
        asset = call.args[0]
        assert asset.prompt_type == "cinematic"
        assert asset.code_version == "9.9.9"
        assert asset.prompt_file_version == "v99"
        assert asset.reference_image_ids == "[\"ref1\"]"


def test_scan_completed_scene_stills_union_db_primary_and_checkpoint(svc, tmp_path):
    """DB primary + checkpoint 완료 항목 union, 파일 존재 확인 (W5 F22 Phase B.15)."""
    # DB: s1은 primary+file, s2는 primary+no-file (skip), s3는 primary 없음 (skip)
    img1 = tmp_path / "s1.png"; img1.write_bytes(b"X")
    missing = tmp_path / "missing.png"

    prim_s1 = MagicMock(); prim_s1.file_path = str(img1)
    prim_s2 = MagicMock(); prim_s2.file_path = str(missing)

    distinct_query = MagicMock()
    distinct_query.filter.return_value = distinct_query
    distinct_query.distinct.return_value = distinct_query
    distinct_query.all.return_value = [("s1",), ("s2",), ("s3",), (None,)]

    primary_query = MagicMock()
    primary_query.filter.return_value = primary_query
    primary_query.first.side_effect = [prim_s1, prim_s2, None]

    # ImageAsset.still_id 첫 호출 → distinct_query, 이후 3회는 primary_query
    svc._db.query.side_effect = [distinct_query, primary_query, primary_query, primary_query]

    # 체크포인트: s4 포함 (파일 존재), s5 (파일 없음, skip)
    cp_img = tmp_path / "s4.png"; cp_img.write_bytes(b"Y")
    scene_cp = MagicMock()
    scene_cp.get_completed_ids.return_value = ["s4", "s5"]
    scene_cp._data = {"completed": {
        "s4": {"primary_path": str(cp_img)},
        "s5": {"primary_path": str(tmp_path / "nope.png")},
    }}

    out = svc.scan_completed_scene_stills("ep1", scene_cp)

    assert out == {"s1", "s4"}


def test_scan_completed_scene_stills_empty_inputs(svc):
    """DB 비어있고 체크포인트 비어있으면 빈 set."""
    distinct_query = MagicMock()
    distinct_query.filter.return_value = distinct_query
    distinct_query.distinct.return_value = distinct_query
    distinct_query.all.return_value = []
    svc._db.query.return_value = distinct_query

    scene_cp = MagicMock()
    scene_cp.get_completed_ids.return_value = []
    scene_cp._data = {"completed": {}}

    assert svc.scan_completed_scene_stills("ep1", scene_cp) == set()


def test_delete_existing_scene_assets_is_hard_fenced(svc):
    """feedback_never_delete_images: 일괄 DELETE 영구 fence (RuntimeError)."""
    import pytest
    with pytest.raises(RuntimeError, match="hard-fenced"):
        svc.delete_existing_scene_assets("ep1")


def test_delete_orphan_scene_assets_filters_by_missing_still(svc):
    """parent scene_still 사라진 image_asset 만 삭제. live still 보존."""
    query = MagicMock()
    query.filter.return_value = query
    query.delete.return_value = 3
    svc._db.query.return_value = query

    count = svc.delete_orphan_scene_assets("ep1")

    assert count == 3
    query.delete.assert_called_once_with(synchronize_session="fetch")
    svc._db.commit.assert_called_once()


def test_delete_orphan_scene_assets_returns_zero_when_no_orphans(svc):
    """live still 만 있으면 삭제 0 + commit (defensive net 정상 무동작)."""
    query = MagicMock()
    query.filter.return_value = query
    query.delete.return_value = 0
    svc._db.query.return_value = query

    count = svc.delete_orphan_scene_assets("ep1")

    assert count == 0
    svc._db.commit.assert_called_once()


def test_delete_orphan_entity_assets_filters_by_missing_canon(svc):
    """parent entity_canon 사라진 reference/composite 만 삭제. live entity 보존."""
    query = MagicMock()
    query.filter.return_value = query
    query.delete.return_value = 5
    svc._db.query.return_value = query

    count = svc.delete_orphan_entity_assets()

    assert count == 5
    query.delete.assert_called_once_with(synchronize_session="fetch")
    svc._db.commit.assert_called_once()


def test_set_primary_asset_updates_and_commits(svc):
    """is_primary=1 업데이트 + commit."""
    query_chain = MagicMock()
    svc._db.query.return_value = query_chain
    query_chain.filter.return_value = query_chain

    svc.set_primary_asset("asset123")

    query_chain.update.assert_called_once_with({"is_primary": 1})
    svc._db.commit.assert_called_once()


def test_save_single_scene_asset_maps_all_fields(svc, monkeypatch):
    """단일 scene asset 저장 + auto_set_primary 호출 + commit."""
    from app.models.project import ImageAsset
    import app.services.scene_persistence_service as pers_mod

    calls = []

    def _fake_auto(db, project_id, asset):
        calls.append((db, project_id, asset))

    monkeypatch.setattr(pers_mod, "auto_set_primary", _fake_auto)

    scene_result = {
        "id": "a1",
        "asset_type": "scene",
        "entity_id": None,
        "still_id": "s1",
        "episode_id": "e1",
        "file_path": "/tmp/s.png",
        "prompt_used": "prompt",
        "generation_model": "gemini",
        "width": None,
        "height": None,
        "status": "generated",
        "review_notes": "",
        "sanitization_strategy": "paraphrase",
        "sanitization_note": "note",
        "original_prompt": "orig",
        "created_at": "2026-04-22T00:00:00+00:00",
    }
    lineage = {
        "prompt_type": "original",
        "code_version": "0.6.1",
        "prompt_file_version": "v5",
        "reference_image_ids": "[\"r1\"]",
    }

    returned = svc.save_single_scene_asset(scene_result, lineage)

    assert isinstance(returned, ImageAsset)
    assert returned.id == "a1"
    assert returned.sanitization_strategy == "paraphrase"
    assert returned.sanitization_note == "note"
    assert returned.original_prompt == "orig"
    assert returned.prompt_type == "original"
    assert returned.reference_image_ids == "[\"r1\"]"
    # auto_set_primary 호출 검증
    assert len(calls) == 1
    assert calls[0][1] == "p1"
    assert calls[0][2] is returned
    svc._db.add.assert_called_once()
    svc._db.commit.assert_called_once()


def test_save_single_scene_asset_missing_optional_fields(svc, monkeypatch):
    """sanitization_*/original_prompt 누락 → None으로 저장."""
    import app.services.scene_persistence_service as pers_mod
    monkeypatch.setattr(pers_mod, "auto_set_primary", lambda *a, **k: None)

    scene_result = {
        "id": "a1",
        "asset_type": "scene",
        "entity_id": None,
        "still_id": "s1",
        "episode_id": "e1",
        "file_path": "/tmp/s.png",
        "prompt_used": "prompt",
        "generation_model": "gemini",
        "width": None,
        "height": None,
        "status": "generated",
        "review_notes": "",
        "created_at": "2026-04-22T00:00:00+00:00",
    }
    lineage = {
        "prompt_type": "original",
        "code_version": "x",
        "prompt_file_version": "x",
        "reference_image_ids": "[]",
    }

    returned = svc.save_single_scene_asset(scene_result, lineage)
    assert returned.sanitization_strategy is None
    assert returned.sanitization_note is None
    assert returned.original_prompt is None


def test_build_resume_state_empty_done_returns_empty_maps(svc):
    """already_done_stills가 비어있으면 4개 빈 dict 반환, DB 조회 없음."""
    out = svc.build_resume_state([{"id": "s1"}], set(), {})
    assert out == {
        "scene_paths_by_index": {},
        "scene_paths_by_index_by_id": {},
        "scene_results_by_index": {},
        "location_scene_history": {},
    }
    svc._db.query.assert_not_called()


def test_build_resume_state_populates_maps_for_done_still(svc, tmp_path):
    """already_done still → primary asset 조회 후 4개 map populate."""
    img = tmp_path / "scene.png"
    img.write_bytes(b"SCENE_BYTES")

    primary = MagicMock()
    primary.id = "a1"
    primary.file_path = str(img)
    primary.prompt_used = "p"
    primary.status = "generated"
    primary.review_notes = ""

    query = MagicMock()
    query.filter.return_value = query
    query.order_by.return_value = query
    query.first.return_value = primary
    svc._db.query.return_value = query

    stills = [{
        "id": "s1",
        "scene_index": 5,
        "shot_index": 2,
        "visible_entities_json": '[{"id":"L01"}]',
    }]
    entity_lookup = {"L01": {"id": "L01", "entity_type": "location"}}

    out = svc.build_resume_state(stills, {"s1"}, entity_lookup)

    assert out["scene_paths_by_index"] == {0: Path(str(img))}
    assert out["scene_paths_by_index_by_id"] == {"s1": Path(str(img))}
    assert out["scene_results_by_index"][0]["id"] == "a1"
    assert out["scene_results_by_index"][0]["status"] == "generated"
    # location은 bytes + still_data
    bytes_, still = out["location_scene_history"]["L01"]
    assert bytes_ == b"SCENE_BYTES"
    assert still == stills[0]


def test_build_resume_state_falls_back_to_latest_when_no_primary(svc, tmp_path):
    """primary 없으면 latest asset 선택."""
    img = tmp_path / "scene.png"
    img.write_bytes(b"X")

    latest = MagicMock()
    latest.id = "a2"
    latest.file_path = str(img)
    latest.prompt_used = None
    latest.status = "needs_fix"
    latest.review_notes = None

    query = MagicMock()
    query.filter.return_value = query
    query.order_by.return_value = query
    # 첫 호출(primary): None, 둘째 호출(latest): latest
    query.first.side_effect = [None, latest]
    svc._db.query.return_value = query

    stills = [{
        "id": "s1",
        "scene_index": 1,
        "shot_index": 1,
        "visible_entities_json": "[]",
    }]
    out = svc.build_resume_state(stills, {"s1"}, {})

    assert out["scene_results_by_index"][0]["id"] == "a2"
    # None prompt/review_notes → '' coalesce
    assert out["scene_results_by_index"][0]["prompt_used"] == ""
    assert out["scene_results_by_index"][0]["review_notes"] == ""


def test_build_resume_state_skips_missing_file(svc, tmp_path):
    """existing_asset의 file_path가 없으면 populate 하지 않음."""
    primary = MagicMock()
    primary.file_path = str(tmp_path / "nope.png")
    query = MagicMock()
    query.filter.return_value = query
    query.order_by.return_value = query
    query.first.return_value = primary
    svc._db.query.return_value = query

    out = svc.build_resume_state(
        [{"id": "s1", "scene_index": 0, "shot_index": 0, "visible_entities_json": "[]"}],
        {"s1"}, {},
    )
    assert out["scene_paths_by_index"] == {}


def test_save_fal_angle_asset_applies_fixed_contract(svc):
    """fal.ai 전용 필드(model, variant_type, asset_type, status)는 고정."""
    from app.models.project import ImageAsset

    svc.save_fal_angle_asset(
        asset_id="fal1",
        file_path="/tmp/fal.png",
        still_id="s1",
        episode_id="e1",
        prompt_used="[fal.ai angle] H=10 V=-5 Z=1.2",
        source_image_id="src1",
        created_at="2026-04-22T00:00:00+00:00",
    )

    added = svc._db.add.call_args_list[0].args[0]
    assert isinstance(added, ImageAsset)
    assert added.id == "fal1"
    assert added.project_id == "p1"
    assert added.asset_type == "scene"
    assert added.file_path == "/tmp/fal.png"
    assert added.still_id == "s1"
    assert added.episode_id == "e1"
    assert added.prompt_used == "[fal.ai angle] H=10 V=-5 Z=1.2"
    assert added.generation_model == "fal-ai/qwen-image-edit-2511-multiple-angles"
    assert added.variant_type == "angle_fal"
    assert added.status == "generated"
    assert added.is_primary == 0
    assert added.source_image_id == "src1"
    svc._db.commit.assert_called_once()


# ──────────────────────────────────────────────────────────────────────
# resolve_world_guide — W5 F22 Phase B.19
# ──────────────────────────────────────────────────────────────────────


def _make_episode(fulltext="SCENE 1\nA scene.", source_filename="ep1.txt"):
    ep = MagicMock()
    ep.fulltext = fulltext
    ep.source_filename = source_filename
    return ep


class _FakeProvOp:
    def __enter__(self):
        return self
    def __exit__(self, *a):
        return False
    def set_input(self, d):
        self.last_input = d
    def set_output(self, d):
        self.last_output = d


class _FakeProvenance:
    def __init__(self):
        self.ops: list = []
    def start_operation(self, *a, **k):
        op = _FakeProvOp()
        self.ops.append(op)
        return op


def _setup_db_chain(svc, *, world_guide_first, proj_settings):
    """Set up db.query(...).filter(...).order_by(...).first() chain.

    Two separate queries happen in resolve_world_guide:
      1) WorldGuide query (only if mode != "full")
      2) ProjectSettings query (always)

    `world_guide_first` is what WorldGuide .first() returns.
    `proj_settings` is what ProjectSettings .first() returns.
    """
    # query() gets called with a model class. Return different mocks per class.
    from app.models.project import ProjectSettings, WorldGuide

    wg_query = MagicMock()
    wg_query.filter.return_value = wg_query
    wg_query.order_by.return_value = wg_query
    wg_query.first.return_value = world_guide_first

    ps_query = MagicMock()
    ps_query.filter.return_value = ps_query
    ps_query.first.return_value = proj_settings

    def _query(model):
        if model is WorldGuide:
            return wg_query
        if model is ProjectSettings:
            return ps_query
        return MagicMock()
    svc._db.query.side_effect = _query


def test_resolve_world_guide_reuses_existing_when_hash_matches(monkeypatch, svc):
    """mode!=full + hash match → existing guide_json 파싱해서 반환."""
    import hashlib
    episode = _make_episode(fulltext="ABC" * 200)
    expected_hash = hashlib.md5(
        f"{episode.fulltext[:500]}:0:0".encode()
    ).hexdigest()
    existing = MagicMock()
    existing.source_hash = expected_hash
    existing.guide_json = '{"world_setting_summary": "REUSED"}'
    _setup_db_chain(svc, world_guide_first=existing, proj_settings=None)

    result = svc.resolve_world_guide(
        episode_id="e1", episode=episode, entities=[], stills=[],
        mode="resume", provenance=_FakeProvenance(), language="ko",
    )

    assert result == {"world_setting_summary": "REUSED"}
    # 재생성 경로 안 탔으므로 add/flush 없음
    svc._db.add.assert_not_called()


def test_resolve_world_guide_regenerates_when_hash_mismatches(monkeypatch, svc):
    """hash 불일치 → WorldGuideGenerator 호출 + DB add + flush (commit 없음)."""
    episode = _make_episode()
    existing = MagicMock()
    existing.source_hash = "different_hash"
    existing.guide_json = '{"old": true}'
    _setup_db_chain(svc, world_guide_first=existing, proj_settings=None)

    gen_called = {"count": 0, "kwargs": None}

    class _FakeGen:
        def __init__(self, llm_client):
            pass
        def generate(self, **kw):
            gen_called["count"] += 1
            gen_called["kwargs"] = kw
            return {"world_setting_summary": "NEW"}

    import app.services.scene_persistence_service as sps
    monkeypatch.setattr(sps, "WorldGuideGenerator", _FakeGen)
    monkeypatch.setattr(sps, "OpenAIClient", lambda: MagicMock())

    prov = _FakeProvenance()
    result = svc.resolve_world_guide(
        episode_id="e1", episode=episode, entities=[], stills=[],
        mode="resume", provenance=prov, language="ko",
    )

    assert result == {"world_setting_summary": "NEW"}
    assert gen_called["count"] == 1
    assert gen_called["kwargs"]["fulltext"] == episode.fulltext
    assert gen_called["kwargs"]["language"] == "ko"
    assert gen_called["kwargs"]["source_file"] == "ep1.txt"
    svc._db.add.assert_called_once()
    svc._db.flush.assert_called_once()
    svc._db.commit.assert_not_called()  # caller 책임
    assert len(prov.ops) == 1


def test_resolve_world_guide_force_full_skips_existing_query(monkeypatch, svc):
    """mode="full"이면 기존 조회 스킵하고 무조건 재생성."""
    episode = _make_episode()
    _setup_db_chain(svc, world_guide_first=MagicMock(source_hash="any"), proj_settings=None)

    class _FakeGen:
        def __init__(self, llm_client):
            pass
        def generate(self, **kw):
            return {"world_setting_summary": "FORCED"}

    import app.services.scene_persistence_service as sps
    monkeypatch.setattr(sps, "WorldGuideGenerator", _FakeGen)
    monkeypatch.setattr(sps, "OpenAIClient", lambda: MagicMock())

    result = svc.resolve_world_guide(
        episode_id="e1", episode=episode, entities=[], stills=[],
        mode="full", provenance=_FakeProvenance(), language="ko",
    )

    assert result == {"world_setting_summary": "FORCED"}
    svc._db.add.assert_called_once()


def test_resolve_world_guide_injects_project_style_when_no_must_maintain(monkeypatch, svc):
    """기존 style_rules에 must_maintain 없으면 project style을 style_rules에 삽입."""
    import hashlib
    episode = _make_episode(fulltext="XYZ")
    expected_hash = hashlib.md5(f"{'XYZ'}:0:0".encode()).hexdigest()
    existing = MagicMock()
    existing.source_hash = expected_hash
    existing.guide_json = '{"style_rules": {"other_key": 1}}'

    ps = MagicMock()
    ps.style_rules_json = '{"palette": "warm"}'
    _setup_db_chain(svc, world_guide_first=existing, proj_settings=ps)

    result = svc.resolve_world_guide(
        episode_id="e1", episode=episode, entities=[], stills=[],
        mode="resume", provenance=_FakeProvenance(), language="ko",
    )

    assert result["style_rules"] == {"palette": "warm"}
    assert "project_style" not in result


def test_resolve_world_guide_preserves_must_maintain_adds_project_style(monkeypatch, svc):
    """기존 style_rules.must_maintain 있으면 project_style로 추가 (style_rules 보존)."""
    import hashlib
    episode = _make_episode(fulltext="X")
    expected_hash = hashlib.md5(f"{'X'}:0:0".encode()).hexdigest()
    existing = MagicMock()
    existing.source_hash = expected_hash
    existing.guide_json = '{"style_rules": {"must_maintain": ["grain"]}}'

    ps = MagicMock()
    ps.style_rules_json = '{"palette": "cold"}'
    _setup_db_chain(svc, world_guide_first=existing, proj_settings=ps)

    result = svc.resolve_world_guide(
        episode_id="e1", episode=episode, entities=[], stills=[],
        mode="resume", provenance=_FakeProvenance(), language="ko",
    )

    assert result["style_rules"] == {"must_maintain": ["grain"]}  # 원본 보존
    assert result["project_style"] == {"palette": "cold"}


def test_resolve_world_guide_no_project_style_when_absent(monkeypatch, svc):
    """ProjectSettings 없거나 style_rules_json 비어 있으면 주입 없음."""
    import hashlib
    episode = _make_episode(fulltext="X")
    expected_hash = hashlib.md5(f"{'X'}:0:0".encode()).hexdigest()
    existing = MagicMock()
    existing.source_hash = expected_hash
    existing.guide_json = '{}'
    _setup_db_chain(svc, world_guide_first=existing, proj_settings=None)

    result = svc.resolve_world_guide(
        episode_id="e1", episode=episode, entities=[], stills=[],
        mode="resume", provenance=_FakeProvenance(), language="ko",
    )
    assert result == {}


def test_resolve_world_guide_empty_project_style_skipped(monkeypatch, svc):
    """ProjectSettings 있지만 style_rules_json=None이면 주입 스킵."""
    import hashlib
    episode = _make_episode(fulltext="X")
    expected_hash = hashlib.md5(f"{'X'}:0:0".encode()).hexdigest()
    existing = MagicMock()
    existing.source_hash = expected_hash
    existing.guide_json = '{"style_rules": {"existing": 1}}'

    ps = MagicMock()
    ps.style_rules_json = None
    _setup_db_chain(svc, world_guide_first=existing, proj_settings=ps)

    result = svc.resolve_world_guide(
        episode_id="e1", episode=episode, entities=[], stills=[],
        mode="resume", provenance=_FakeProvenance(), language="ko",
    )
    assert result["style_rules"] == {"existing": 1}  # 변함 없음


# ──────────────────────────────────────────────────────────────────────
# load_episode_entity_dicts — W5 F22 Phase B.22.10
# ──────────────────────────────────────────────────────────────────────


def test_load_episode_entity_dicts_no_links_returns_empty(svc):
    """EntityEpisodeLink 빈 리스트면 EntityCanon 쿼리 스킵 + []."""
    svc._db.reset_mock()
    # first query (EntityEpisodeLink) → 빈 리스트
    svc._db.query.return_value.filter.return_value.all.return_value = []

    result = svc.load_episode_entity_dicts("ep-1")

    assert result == []
    # query는 EntityEpisodeLink 1회만 호출 (EntityCanon은 canon_ids가 빈 리스트라 스킵)
    assert svc._db.query.call_count == 1


def test_load_episode_entity_dicts_maps_fields_with_fallbacks(svc):
    """short_id/description/t2i_prompt None → ''. stable_traits None → '{}'."""
    link1 = MagicMock(canon_id="c1")
    link2 = MagicMock(canon_id="c2")

    ent1 = MagicMock()
    ent1.id = "c1"; ent1.name = "Char1"; ent1.short_id = "C01"
    ent1.entity_type = "character"; ent1.description = "desc1"
    ent1.stable_traits = '{"age": 30}'; ent1.t2i_prompt = "prompt1"

    ent2 = MagicMock()
    ent2.id = "c2"; ent2.name = "Char2"; ent2.short_id = None
    ent2.entity_type = "character"; ent2.description = None
    ent2.stable_traits = None; ent2.t2i_prompt = None

    call_count = {"n": 0}
    def _query_side_effect(model):
        q = MagicMock()
        call_count["n"] += 1
        if call_count["n"] == 1:
            q.filter.return_value.all.return_value = [link1, link2]
        else:
            q.filter.return_value.all.return_value = [ent1, ent2]
        return q
    svc._db.query.side_effect = _query_side_effect

    result = svc.load_episode_entity_dicts("ep-1")

    assert len(result) == 2
    r1, r2 = result
    assert r1["id"] == "c1"
    assert r1["short_id"] == "C01"
    assert r1["description"] == "desc1"
    assert r1["stable_traits"] == '{"age": 30}'
    # 두 번째: None 필드들 fallback
    assert r2["short_id"] == ""
    assert r2["description"] == ""
    assert r2["stable_traits"] == "{}"
    assert r2["t2i_prompt"] == ""


# ──────────────────────────────────────────────────────────────────────
# load_episode_still_dicts — W5 F22 Phase B.22.10
# ──────────────────────────────────────────────────────────────────────


def test_load_episode_still_dicts_empty_returns_empty_tuple(svc):
    """SceneStill 없음 → ([], []) tuple."""
    svc._db.reset_mock()
    svc._db.query.return_value.filter.return_value.order_by.return_value.all.return_value = []

    stills, stills_orm = svc.load_episode_still_dicts("ep-1")

    assert stills == []
    assert stills_orm == []


def test_load_episode_still_dicts_maps_fields_with_fallbacks(svc):
    """JSON 필드 None fallback: camera/lighting='{}', visible_entities='[]'. 문자열='' fallback."""
    s1 = MagicMock()
    s1.id = "s1"; s1.still_index = 0; s1.scene_index = 1; s1.shot_index = 1
    s1.screenplay_scene_heading = "INT. ROOM - NIGHT"
    s1.beat_title = "beat A"
    s1.still_frame_prompt = "prompt A"
    s1.camera_json = '{"angle": "wide"}'
    s1.lighting_json = '{"mood": "dark"}'
    s1.visible_entities_json = '["c1", "c2"]'
    s1.dependent_scene_id = "dep-uuid"

    s2 = MagicMock()
    s2.id = "s2"; s2.still_index = 1; s2.scene_index = 2; s2.shot_index = 1
    s2.screenplay_scene_heading = None
    s2.beat_title = None
    s2.still_frame_prompt = None
    s2.camera_json = None
    s2.lighting_json = None
    s2.visible_entities_json = None
    s2.dependent_scene_id = None

    svc._db.reset_mock()
    svc._db.query.return_value.filter.return_value.order_by.return_value.all.return_value = [s1, s2]

    stills, stills_orm = svc.load_episode_still_dicts("ep-1")

    assert len(stills) == 2
    assert stills_orm == [s1, s2]  # ORM 리스트 그대로 반환

    r1, r2 = stills
    assert r1["id"] == "s1"
    assert r1["screenplay_scene_heading"] == "INT. ROOM - NIGHT"
    assert r1["camera_json"] == '{"angle": "wide"}'
    assert r1["dependent_scene_id"] == "dep-uuid"

    # s2 모든 None 필드 fallback
    assert r2["screenplay_scene_heading"] == ""
    assert r2["beat_title"] == ""
    assert r2["still_frame_prompt"] == ""
    assert r2["camera_json"] == "{}"
    assert r2["lighting_json"] == "{}"
    assert r2["visible_entities_json"] == "[]"
    assert r2["dependent_scene_id"] is None  # None 그대로 전달


def test_load_episode_still_dicts_preserves_ordering(svc):
    """stills_orm 순서가 dict 리스트에도 유지."""
    s_low = MagicMock()
    s_low.id = "s_low"; s_low.still_index = 0
    s_low.scene_index = 1; s_low.shot_index = 1
    s_low.screenplay_scene_heading = "x"
    s_low.beat_title = "x"; s_low.still_frame_prompt = "x"
    s_low.camera_json = "{}"; s_low.lighting_json = "{}"
    s_low.visible_entities_json = "[]"; s_low.dependent_scene_id = None

    s_high = MagicMock()
    s_high.id = "s_high"; s_high.still_index = 5
    s_high.scene_index = 2; s_high.shot_index = 1
    s_high.screenplay_scene_heading = "y"
    s_high.beat_title = "y"; s_high.still_frame_prompt = "y"
    s_high.camera_json = "{}"; s_high.lighting_json = "{}"
    s_high.visible_entities_json = "[]"; s_high.dependent_scene_id = None

    svc._db.reset_mock()
    svc._db.query.return_value.filter.return_value.order_by.return_value.all.return_value = [s_low, s_high]

    stills, stills_orm = svc.load_episode_still_dicts("ep-1")

    assert [s["id"] for s in stills] == ["s_low", "s_high"]
    assert stills_orm == [s_low, s_high]


# ──────────────────────────────────────────────────────────────────────
# validate_episode_ready — W5 F22 Phase B.22.11
# ──────────────────────────────────────────────────────────────────────


def test_validate_episode_ready_pipeline_gate_failure_propagates(svc, monkeypatch):
    """pipeline_gate가 예외 raise하면 그대로 전파 (catch 안 함)."""
    from app.core.errors import AppError as _AppError

    import app.core.pipeline_gate as _gate
    def _gate_raises(db, pid, eid):
        raise _AppError(code="pipeline.not_ready", message="x", status_code=400)
    monkeypatch.setattr(_gate, "check_scene_images_ready", _gate_raises)

    with pytest.raises(_AppError) as exc:
        svc.validate_episode_ready("ep-1")
    assert exc.value.code == "pipeline.not_ready"


def test_validate_episode_ready_zero_scenes_raises_no_scenes(svc, monkeypatch):
    """pipeline_gate 통과 + scene_count==0 → image.no_scenes AppError."""
    from app.core.errors import AppError as _AppError

    import app.core.pipeline_gate as _gate
    monkeypatch.setattr(_gate, "check_scene_images_ready", lambda db, pid, eid: None)

    svc._db.reset_mock()
    svc._db.query.return_value.filter.return_value.count.return_value = 0

    with pytest.raises(_AppError) as exc:
        svc.validate_episode_ready("ep-1")
    assert exc.value.code == "image.no_scenes"
    assert exc.value.status_code == 400
    assert exc.value.message == "씬이 없습니다. 분석을 먼저 완료하세요."


def test_validate_episode_ready_positive_count_returns_none(svc, monkeypatch):
    """scene_count>0 → 정상 (None 반환)."""
    import app.core.pipeline_gate as _gate
    monkeypatch.setattr(_gate, "check_scene_images_ready", lambda db, pid, eid: None)

    svc._db.reset_mock()
    svc._db.query.return_value.filter.return_value.count.return_value = 3

    # 예외 없이 정상 종료
    result = svc.validate_episode_ready("ep-1")
    assert result is None


# ──────────────────────────────────────────────────────────────────────
# fetch_still_for_generation — W5 F22 Phase B.23.1
# ──────────────────────────────────────────────────────────────────────


def test_fetch_still_pipeline_gate_failure_propagates(svc, monkeypatch):
    """still_check 조회 성공 시 pipeline_gate 예외 그대로 전파."""
    from app.core.errors import AppError as _AppError

    fake_still = MagicMock()
    fake_still.episode_id = "ep-1"
    svc._db.reset_mock()
    svc._db.query.return_value.filter.return_value.first.return_value = fake_still

    import app.core.pipeline_gate as _gate
    def _gate_raise(db, pid, eid):
        raise _AppError(code="pipeline.not_ready", message="x", status_code=400)
    monkeypatch.setattr(_gate, "check_scene_images_ready", _gate_raise)

    with pytest.raises(_AppError) as exc:
        svc.fetch_still_for_generation("still-1")
    assert exc.value.code == "pipeline.not_ready"


def test_fetch_still_gemini_key_missing_raises(svc, monkeypatch):
    """still_check 있어도 gemini_api_key 없고 pool 비면 image.gemini_key_missing."""
    from app.core.errors import AppError as _AppError

    fake_still = MagicMock()
    fake_still.episode_id = "ep-1"
    svc._db.reset_mock()
    svc._db.query.return_value.filter.return_value.first.return_value = fake_still

    import app.core.pipeline_gate as _gate
    monkeypatch.setattr(_gate, "check_scene_images_ready", lambda db, pid, eid: None)
    # settings.gemini_api_key 빈 문자열 + key_count=0 시나리오
    import app.services.scene_persistence_service as sps
    monkeypatch.setattr(sps.settings, "gemini_api_key", "")
    monkeypatch.setattr(sps, "gemini_key_count", lambda: 0)

    with pytest.raises(_AppError) as exc:
        svc.fetch_still_for_generation("still-1")
    assert exc.value.code == "image.gemini_key_missing"


def test_fetch_still_not_found_raises(svc, monkeypatch):
    """still_check None + gemini_key 정상 → 두 번째 쿼리도 None → still.not_found."""
    from app.core.errors import AppError as _AppError

    svc._db.reset_mock()
    svc._db.query.return_value.filter.return_value.first.return_value = None

    # pipeline_gate는 still_check가 None이면 호출 안 됨
    import app.core.pipeline_gate as _gate
    _gate_called = {"n": 0}
    def _gate_noop(db, pid, eid):
        _gate_called["n"] += 1
    monkeypatch.setattr(_gate, "check_scene_images_ready", _gate_noop)

    import app.services.scene_persistence_service as sps
    monkeypatch.setattr(sps.settings, "gemini_api_key", "fake")
    monkeypatch.setattr(sps, "gemini_key_count", lambda: 0)

    with pytest.raises(_AppError) as exc:
        svc.fetch_still_for_generation("missing-still")
    assert exc.value.code == "still.not_found"
    assert exc.value.status_code == 404
    # still_check가 None이었으므로 pipeline_gate 호출 안 됐어야 함
    assert _gate_called["n"] == 0


def test_fetch_still_happy_path_returns_still(svc, monkeypatch):
    """모든 검증 통과 → still 반환."""
    fake_still = MagicMock()
    fake_still.episode_id = "ep-1"
    svc._db.reset_mock()
    # 두 쿼리 모두 동일 still 반환 (literal 중복 보존)
    svc._db.query.return_value.filter.return_value.first.return_value = fake_still

    import app.core.pipeline_gate as _gate
    monkeypatch.setattr(_gate, "check_scene_images_ready", lambda db, pid, eid: None)

    import app.services.scene_persistence_service as sps
    monkeypatch.setattr(sps.settings, "gemini_api_key", "fake-key")

    result = svc.fetch_still_for_generation("still-1")
    assert result is fake_still


def test_fetch_still_query_count_preserves_literal_duplicate(svc, monkeypatch):
    """원본 generate_single_scene_image처럼 쿼리 2회 실행 (중복 literal 보존)."""
    fake_still = MagicMock()
    fake_still.episode_id = "ep-1"
    svc._db.reset_mock()
    svc._db.query.return_value.filter.return_value.first.return_value = fake_still

    import app.core.pipeline_gate as _gate
    monkeypatch.setattr(_gate, "check_scene_images_ready", lambda db, pid, eid: None)

    import app.services.scene_persistence_service as sps
    monkeypatch.setattr(sps.settings, "gemini_api_key", "fake-key")

    svc.fetch_still_for_generation("still-1")
    # SceneStill 모델로 2회 쿼리 실행 (first query + 본 쿼리)
    assert svc._db.query.call_count == 2


# ──────────────────────────────────────────────────────────────────────
# mark_asset_as_original — W5 F22 Phase B.24.3
# ──────────────────────────────────────────────────────────────────────


def test_mark_asset_as_original_updates_and_flushes(svc):
    """asset 존재 시 variant_type='original' + flush."""
    fake_asset = MagicMock()
    svc._db.reset_mock()
    svc._db.query.return_value.filter.return_value.first.return_value = fake_asset

    svc.mark_asset_as_original("asset-1")

    assert fake_asset.variant_type == "original"
    svc._db.flush.assert_called_once()


def test_mark_asset_as_original_missing_is_noop(svc):
    """asset 없으면 no-op (AttributeError 방지)."""
    svc._db.reset_mock()
    svc._db.query.return_value.filter.return_value.first.return_value = None

    # 예외 없이 정상 종료
    svc.mark_asset_as_original("missing")
    svc._db.flush.assert_not_called()