"""D6 T5 (+ T5-fix) — consumer hash stamp tests.

각 consumer step 이 master_plan cp 의 sibling field (`bg_catalog_hash` /
`shot_binding_hash`) 를 read 후 자체 cp 의 return dict 에 `consumed_*_hash` stamp.

T5-fix (review iter7) 후 모든 D6 consumer 가 두 hash 양쪽 stamp:
- floor_plan_prompt: catalog + binding (applied_shots → user prompt `{shots_block}` inject).
- background_prompt: catalog + binding (둘 다 의존하는 유일한 consumer — T5b 부터 양쪽).
- background_render: catalog + binding (background_prompt 가 binding 의존 + ImageAsset
  shot_ids 저장 → transitively).
- scene_detail: catalog + binding (binding 위주, catalog 보조).

iter6 minor note 의 "render = catalog only" 분리는 실제 코드 의존성과 안 맞아 폐기 —
preflight (T8) 가 어느 한쪽이라도 변하면 catch 하려면 모든 consumer 양쪽 stamp.

spec: docs/superpowers/specs/2026-05-09-deterministic-bg-id-and-catalog-lineage.md §4.6
plan: T5 (B1 + I6 — split into 4 sub-tasks) + T5-fix (review iter7 B1+B2)
"""
from __future__ import annotations

from typing import Any, Dict
from unittest.mock import MagicMock, patch

import pytest


def _master_plan_cp_with_hashes(
    bg_catalog_hash: str = "catalog_hash_abc",
    shot_binding_hash: str = "binding_hash_xyz",
    plans: Dict[str, Any] = None,
) -> Dict[str, Any]:
    return {
        "schema_version": 2,
        "data": {
            "plans": plans or {},
            "background_catalog": {"L09B01": {"bg_id": "L09B01"}},
            "shot_background_map": {"S1_Shot1": "L09B01"},
            "bg_catalog_hash": bg_catalog_hash,
            "shot_binding_hash": shot_binding_hash,
        },
    }


# ──────────────────────────────────────────────────────────────────────
# T5a + T5-fix B1 — floor_plan_prompt_step: catalog + binding 양쪽 stamp
# ──────────────────────────────────────────────────────────────────────


def test_floor_plan_prompt_step_stamps_bg_catalog_hash(tmp_path, monkeypatch):
    """floor_plan_prompt cp 에 consumed_bg_catalog_hash 가 master_plan cp 의 hash 그대로 복사."""
    from app.core.steps.floor_plan_prompt_step import FloorPlanPromptStep

    monkeypatch.setattr("app.core.config.settings.background_mode", "on")
    monkeypatch.setattr("app.core.config.settings.projects_dir", str(tmp_path))

    step = FloorPlanPromptStep.__new__(FloorPlanPromptStep)
    step.project_id = "p-d6-t5a"
    step.episode_id = "ep1"
    step.project_config = {}
    step.update_progress = lambda *a, **kw: None
    step.build_opik_metadata = lambda *a, **kw: {}

    plans_cp = _master_plan_cp_with_hashes(
        bg_catalog_hash="catalog_t5a",
        plans={
            "g1": {"status": "ok", "plan": {
                "floor_plans": [],
                "backgrounds": [],
            }},
        },
    )

    def _load_cp(name):
        if name == "background_master_plan":
            return plans_cp
        return None

    step._load_prev_checkpoint = _load_cp
    result = step._execute()
    assert "consumed_bg_catalog_hash" in result["data"], (
        f"floor_plan_prompt cp 에 consumed_bg_catalog_hash 누락 — got keys={list(result['data'].keys())}"
    )
    assert result["data"]["consumed_bg_catalog_hash"] == "catalog_t5a"


def test_floor_plan_prompt_step_stamps_shot_binding_hash(tmp_path, monkeypatch):
    """T5-fix B1 (review iter7): floor_plan_prompt 도 shot_binding_hash stamp.

    이전 가정 (review iter6 minor) — render 계열은 catalog only — 은 floor_plan_prompt
    의 실제 input 흐름과 안 맞음. floor_plan_prompt 가 applied_shots 를 user prompt
    `{shots_block}` 으로 inject (`floor_plan_prompt_step.py:110-120` + `floor_plan_prompt.py:54`).
    shot_background_map 변경 시 prompt 입력이 바뀌므로 shot_binding_hash 도 추적 의무.
    """
    from app.core.steps.floor_plan_prompt_step import FloorPlanPromptStep

    monkeypatch.setattr("app.core.config.settings.background_mode", "on")
    monkeypatch.setattr("app.core.config.settings.projects_dir", str(tmp_path))

    step = FloorPlanPromptStep.__new__(FloorPlanPromptStep)
    step.project_id = "p-d6-t5a-binding"
    step.episode_id = "ep1"
    step.project_config = {}
    step.update_progress = lambda *a, **kw: None
    step.build_opik_metadata = lambda *a, **kw: {}

    plans_cp = _master_plan_cp_with_hashes(
        bg_catalog_hash="cat_t5fix_a",
        shot_binding_hash="bind_t5fix_a",
        plans={"g1": {"status": "ok", "plan": {"floor_plans": [], "backgrounds": []}}},
    )

    def _load_cp(name):
        if name == "background_master_plan":
            return plans_cp
        return None

    step._load_prev_checkpoint = _load_cp
    result = step._execute()
    assert result["data"].get("consumed_bg_catalog_hash") == "cat_t5fix_a"
    assert result["data"].get("consumed_shot_binding_hash") == "bind_t5fix_a", (
        f"floor_plan_prompt 도 shot_binding_hash stamp 의무 — applied_shots 가 prompt 입력. "
        f"got data={result['data']}"
    )


def test_floor_plan_prompt_step_legacy_cp_no_hash_stamp_zero(tmp_path, monkeypatch):
    """legacy master_plan cp (D6 sibling 부재) 면 consumed_bg_catalog_hash 비워둠."""
    from app.core.steps.floor_plan_prompt_step import FloorPlanPromptStep

    monkeypatch.setattr("app.core.config.settings.background_mode", "on")
    monkeypatch.setattr("app.core.config.settings.projects_dir", str(tmp_path))

    step = FloorPlanPromptStep.__new__(FloorPlanPromptStep)
    step.project_id = "p-d6-t5a-legacy"
    step.episode_id = "ep1"
    step.project_config = {}
    step.update_progress = lambda *a, **kw: None
    step.build_opik_metadata = lambda *a, **kw: {}

    legacy_cp = {
        "data": {"plans": {"g1": {"status": "ok", "plan": {"floor_plans": [], "backgrounds": []}}}},
    }

    def _load_cp(name):
        if name == "background_master_plan":
            return legacy_cp
        return None

    step._load_prev_checkpoint = _load_cp
    result = step._execute()
    # legacy 는 stamp 빈 문자열 (None 보다 명시 — preflight 가 빈 string 으로 mismatch 처리)
    assert result["data"].get("consumed_bg_catalog_hash", "") == ""


# ──────────────────────────────────────────────────────────────────────
# T5b — background_prompt_step: catalog + binding 둘 다
# ──────────────────────────────────────────────────────────────────────


def _make_bg_prompt_step(monkeypatch, tmp_path, plans_cp):
    from app.core.steps.background_prompt_step import BackgroundPromptStep

    monkeypatch.setattr("app.core.config.settings.background_mode", "on")
    monkeypatch.setattr("app.core.config.settings.projects_dir", str(tmp_path))

    step = BackgroundPromptStep.__new__(BackgroundPromptStep)
    step.project_id = "p-d6-t5b"
    step.episode_id = "ep1"
    step.project_config = {}
    step.update_progress = lambda *a, **kw: None
    step.build_opik_metadata = lambda *a, **kw: {}

    def _load_cp(name):
        if name == "background_master_plan":
            return plans_cp
        return None

    step._load_prev_checkpoint = _load_cp
    return step


def test_background_prompt_step_stamps_both_hashes(tmp_path, monkeypatch):
    """background_prompt cp 에 consumed_bg_catalog_hash + consumed_shot_binding_hash 모두 stamp."""
    plans_cp = _master_plan_cp_with_hashes(
        bg_catalog_hash="cat_t5b",
        shot_binding_hash="bind_t5b",
        plans={"g1": {"status": "ok", "plan": {"floor_plans": [], "backgrounds": []}}},
    )
    step = _make_bg_prompt_step(monkeypatch, tmp_path, plans_cp)
    result = step._execute()
    assert result["data"].get("consumed_bg_catalog_hash") == "cat_t5b", (
        f"bg_catalog_hash stamp 누락 — data={result['data']}"
    )
    assert result["data"].get("consumed_shot_binding_hash") == "bind_t5b", (
        f"shot_binding_hash stamp 누락 — data={result['data']}"
    )


def test_background_prompt_step_legacy_cp_empty_hash(tmp_path, monkeypatch):
    """legacy master_plan cp → 둘 다 빈 문자열 (preflight 가 catch)."""
    legacy_cp = {
        "data": {"plans": {"g1": {"status": "ok", "plan": {"floor_plans": [], "backgrounds": []}}}},
    }
    step = _make_bg_prompt_step(monkeypatch, tmp_path, legacy_cp)
    result = step._execute()
    assert result["data"].get("consumed_bg_catalog_hash", "") == ""
    assert result["data"].get("consumed_shot_binding_hash", "") == ""


# ──────────────────────────────────────────────────────────────────────
# T5c + T5-fix B2 — background_render_step: catalog + binding 양쪽 stamp + schema_version bump
# ──────────────────────────────────────────────────────────────────────


def _make_bg_render_step(monkeypatch, tmp_path, plans_cp, prompts_cp=None):
    from unittest.mock import MagicMock as _MagicMock
    from app.core.steps.background_render_step import BackgroundRenderStep

    monkeypatch.setattr("app.core.config.settings.background_mode", "on")
    monkeypatch.setattr("app.core.config.settings.projects_dir", str(tmp_path))

    step = BackgroundRenderStep.__new__(BackgroundRenderStep)
    step.project_id = "p-d6-t5c"
    step.episode_id = "ep1"
    step.project_config = {}
    step.db = _MagicMock()

    def _load_cp(name):
        if name == "background_master_plan":
            return plans_cp
        if name == "background_prompt":
            return prompts_cp
        return None

    step._load_prev_checkpoint = _load_cp
    return step


def test_background_render_step_stamps_bg_catalog_hash(tmp_path, monkeypatch):
    """background_render cp 에 consumed_bg_catalog_hash stamp."""
    plans_cp = _master_plan_cp_with_hashes(
        bg_catalog_hash="cat_t5c",
        plans={"g1": {"status": "ok", "plan": {"floor_plans": [], "backgrounds": []}}},
    )
    step = _make_bg_render_step(monkeypatch, tmp_path, plans_cp)
    result = step._execute()
    assert result["data"].get("consumed_bg_catalog_hash") == "cat_t5c", (
        f"render cp 에 catalog hash stamp 누락 — {result['data']}"
    )


def test_background_render_step_stamps_shot_binding_hash(tmp_path, monkeypatch):
    """T5-fix B2 (review iter7): background_render 도 shot_binding_hash stamp.

    이전 가정 — render 는 catalog only — 은 transitively 안 맞음:
    - render 가 background_prompt 의 t2i_prompt / shot_guides 소비
      (`background_render_step.py:277` + `background_prompt.py:99` 의 applies_to_shots
      → shot_guides 정합).
    - render 가 ImageAsset 의 shot_ids 도 applies_to_shots 에서 저장
      (`background_render_step.py:281`).
    binding 변경 시 background_prompt 가 갱신되면 render 의 input 도 변함.
    consumer 단위로 추적 의무.
    """
    plans_cp = _master_plan_cp_with_hashes(
        bg_catalog_hash="cat_t5fix_c",
        shot_binding_hash="bind_t5fix_c",
        plans={"g1": {"status": "ok", "plan": {"floor_plans": [], "backgrounds": []}}},
    )
    step = _make_bg_render_step(monkeypatch, tmp_path, plans_cp)
    result = step._execute()
    assert result["data"].get("consumed_bg_catalog_hash") == "cat_t5fix_c"
    assert result["data"].get("consumed_shot_binding_hash") == "bind_t5fix_c", (
        f"background_render 도 shot_binding_hash stamp 의무 — got data={result['data']}"
    )


def test_background_render_step_schema_version_bumped_for_d6():
    """T5c — schema_version bump (D6 의 consumed_bg_catalog_hash 추가)."""
    from app.core.steps.background_render_step import SCHEMA_VERSION
    assert SCHEMA_VERSION >= 3, (
        f"background_render schema_version D6 (≥3) 으로 bump 의무. 현재: {SCHEMA_VERSION}"
    )


def test_background_render_step_manifest_schema_version_matches():
    """STEP_MANIFEST 의 background_render schema_version 이 step body 와 일치."""
    from app.core.step_manifest import STEP_MANIFEST
    from app.core.steps.background_render_step import SCHEMA_VERSION
    entry = STEP_MANIFEST.get("background_render", {})
    assert entry.get("schema_version") == SCHEMA_VERSION, (
        f"manifest.background_render.schema_version={entry.get('schema_version')!r} != "
        f"step SCHEMA_VERSION={SCHEMA_VERSION!r}. resume cp_mismatch BLOCK 위험."
    )


# ──────────────────────────────────────────────────────────────────────
# T5d — scene_detail (detail_steps.py): shot_binding hash 위주 + SCHEMA_VERSION bump
# ──────────────────────────────────────────────────────────────────────


def test_scene_detail_schema_version_bumped_for_d6():
    """T5d (B1 정정 — detail_steps.py 가 진짜 위치): SCENE_DETAIL_SCHEMA_VERSION bump.

    `consumed_shot_binding_hash` + `consumed_bg_catalog_hash` field 추가 → cp shape
    변경. 기존 v7 cp 는 stale 처리 (allowlist 외라 BLOCK, 운영자 force 의무).
    """
    from app.core.steps.detail_steps import SCENE_DETAIL_SCHEMA_VERSION
    assert SCENE_DETAIL_SCHEMA_VERSION >= 8, (
        f"SCENE_DETAIL_SCHEMA_VERSION D6 (≥8) 으로 bump 의무. 현재: {SCENE_DETAIL_SCHEMA_VERSION}"
    )


def test_scene_detail_manifest_schema_version_matches():
    """STEP_MANIFEST 의 scene_detail schema_version 이 step body 와 일치."""
    from app.core.step_manifest import STEP_MANIFEST
    from app.core.steps.detail_steps import SCENE_DETAIL_SCHEMA_VERSION
    entry = STEP_MANIFEST.get("scene_detail", {})
    assert entry.get("schema_version") == SCENE_DETAIL_SCHEMA_VERSION, (
        f"manifest.scene_detail.schema_version={entry.get('schema_version')!r} != "
        f"step SCENE_DETAIL_SCHEMA_VERSION={SCENE_DETAIL_SCHEMA_VERSION!r}."
    )


def test_scene_detail_load_d6_consumed_hashes_helper_returns_both():
    """SceneDetailStep helper `_load_d6_consumed_hashes` 가 master_plan cp 의
    bg_catalog_hash + shot_binding_hash 를 dict 로 반환.

    review iter6 minor note: scene_detail 은 binding hash 위주 (어느 shot 이 어느 bg
    에 매핑인지). catalog hash 도 보조 stamp (drift 추적).
    """
    from app.core.steps.detail_steps import SceneDetailStep

    step = SceneDetailStep.__new__(SceneDetailStep)
    step.project_id = "p-d6-t5d-helper"
    step.episode_id = "ep1"

    plans_cp = _master_plan_cp_with_hashes(
        bg_catalog_hash="cat_t5d_helper",
        shot_binding_hash="bind_t5d_helper",
    )
    step._load_prev_checkpoint = lambda name: plans_cp if name == "background_master_plan" else None

    result = step._load_d6_consumed_hashes()
    assert result == {
        "consumed_bg_catalog_hash": "cat_t5d_helper",
        "consumed_shot_binding_hash": "bind_t5d_helper",
    }


def test_scene_detail_load_d6_consumed_hashes_legacy_cp_empty():
    """legacy master_plan cp (D6 sibling 부재) → 둘 다 빈 문자열."""
    from app.core.steps.detail_steps import SceneDetailStep

    step = SceneDetailStep.__new__(SceneDetailStep)
    step.project_id = "p-d6-t5d-legacy"
    step.episode_id = "ep1"

    legacy_cp = {"data": {"plans": {}}}
    step._load_prev_checkpoint = lambda name: legacy_cp if name == "background_master_plan" else None

    result = step._load_d6_consumed_hashes()
    assert result == {
        "consumed_bg_catalog_hash": "",
        "consumed_shot_binding_hash": "",
    }


def test_scene_detail_load_d6_consumed_hashes_missing_cp_empty():
    """master_plan cp 자체 부재 → 둘 다 빈 문자열."""
    from app.core.steps.detail_steps import SceneDetailStep

    step = SceneDetailStep.__new__(SceneDetailStep)
    step.project_id = "p-d6-t5d-missing"
    step.episode_id = "ep1"
    step._load_prev_checkpoint = lambda name: None

    result = step._load_d6_consumed_hashes()
    assert result == {
        "consumed_bg_catalog_hash": "",
        "consumed_shot_binding_hash": "",
    }