"""SceneImagePipelineStep 표적 씬 슬라이스 결정론 테스트 (2026-07-23).

Codex 설계 합의 잠금: OFF=byte-identical / config_hash 는 recipe OFF 조기
return 이전 스탬프(NARROW-6) / target+force fail-closed(BLOCKING-5) /
카운트·verify=실행 audit 의 effective 동일 소비(HIGH-3·4).
생성(LLM/이미지)은 전부 mock — 시나리오 의존 0.
"""
from __future__ import annotations

from unittest.mock import MagicMock, patch

import pytest


@pytest.fixture
def _no_drift_ack():
    """운영 .env 격리 — opik pytest 플러그인이 backend/.env 를 os.environ
    에 실어(2026-08-17 실측: 플러그인 autoload bisect), #77-A 운영 잔재
    STEP_CONFIG_DRIFT_ACK=scene_image_pipeline 이 새면 기본 BLOCK 검증이
    config_drift_ack 갈래로 새서 거짓 실패한다. 이 파일은 #77-A(08-09)
    이전 작성이라 ack 기본값을 명시로 잠근다."""
    with patch("app.core.config.settings.step_config_drift_ack", "",
               create=True):
        yield


def _make_step():
    from app.core.steps.image_steps import SceneImagePipelineStep

    step = SceneImagePipelineStep.__new__(SceneImagePipelineStep)
    step.project_id = "SAMPLE_P"
    step.episode_id = "SAMPLE_E"
    step.project_config = {}
    step.db = MagicMock()
    step._setup_image_tracer_context = MagicMock()
    step._get_system_actor_id = MagicMock(return_value="SAMPLE_ACTOR")
    return step


def _write_audit(tmp_path, effective, requested=None, added=None):
    from app.modules.pipeline.scene_image_scope import (
        build_scope_audit,
        save_scope_audit,
        scope_audit_path,
    )

    requested = requested if requested is not None else effective
    added = added if added is not None else [
        x for x in effective if x not in set(requested)]
    p = scope_audit_path(tmp_path, "SAMPLE_P", "SAMPLE_E")
    save_scope_audit(p, build_scope_audit(
        target_scenes=(1,), requested_ids=requested,
        dependency_added_ids=added, path_kind="recipe"))
    return p


# ── config_hash (NARROW-6) ──────────────────────────────────────────


def test_config_hash_target_stamped_before_recipe_off_return():
    """★zoom 축을 **고정한다** (2026-08-27, #92).

    이 시험이 재는 것은 「표적 설정이 없으면 base 그대로」다.
    zoom bbox 모델 스탬프도 조기 return 앞에 있어서, 그 플래그를
    안 고정하면 **이 시험이 재려던 축과 무관한 이유로** 빨강이 된다.
    계약은 그대로다 — 재는 축만 고정한다.
    """
    step = _make_step()
    with patch("app.core.config.settings.zoom_continuity_anchor_enabled",
               False, create=True), \
            patch("app.core.config.settings.still_recipe_mode", "off",
                  create=True):
        with patch("app.core.config.settings.scene_image_target_scenes",
                   "", create=True):
            base = step._config_hash()
        with patch("app.core.config.settings.scene_image_target_scenes",
                   "1,10", create=True):
            targeted = step._config_hash()
        with patch("app.core.config.settings.scene_image_target_scenes",
                   "10,1", create=True):
            targeted_reordered = step._config_hash()
    # legacy(recipe OFF) 경로에서도 표적 설정=stale 유발
    assert targeted != base
    # 빈 설정=기존 base byte-identical (조기 return 값 그대로)
    from app.core.step_runner import compute_config_hash

    assert base == compute_config_hash({})
    # 정규화(정렬·중복 제거)로 순서 무관 동일
    assert targeted == targeted_reordered


def test_config_hash_target_stamped_in_recipe_mode_too():
    step = _make_step()
    with patch("app.core.config.settings.still_recipe_mode", "v1",
               create=True):
        with patch("app.core.config.settings.scene_image_target_scenes",
                   "", create=True):
            recipe_base = step._config_hash()
        with patch("app.core.config.settings.scene_image_target_scenes",
                   "1,10", create=True):
            recipe_targeted = step._config_hash()
    assert recipe_targeted != recipe_base


def test_config_hash_stamps_scope_contract_version():
    """재리뷰 HIGH-3: SCOPE_CONTRACT_VERSION 도 hash 실질 입력 — 계약
    개정 시 표적 CP 자동 stale."""
    step = _make_step()
    with patch("app.core.config.settings.still_recipe_mode", "off",
               create=True), patch(
        "app.core.config.settings.scene_image_target_scenes", "1,10",
        create=True,
    ):
        h1 = step._config_hash()
        with patch(
            "app.modules.pipeline.scene_image_scope.SCOPE_CONTRACT_VERSION",
            999,
        ):
            h2 = step._config_hash()
    assert h1 != h2


# ── target+force fail-closed (BLOCKING-5 + 재리뷰 BLOCKING-1) ───────


def test_target_with_force_fail_closed():
    from app.core.errors import AppError

    step = _make_step()
    with patch("app.core.config.settings.scene_image_target_scenes",
               "1,10", create=True):
        with pytest.raises(AppError) as exc:
            step._execute(mode="force")
    assert "target_force_forbidden" in exc.value.code
    # 표적 미설정 force 는 이 가드에 걸리지 않는다 (이후 게이트에서
    # 진행 — 여기서는 가드 통과만 확인하기 위해 게이트를 실패시킴)
    with patch("app.core.config.settings.scene_image_target_scenes",
               "", create=True), patch(
        "app.core.pipeline_gate.check_scene_images_ready",
        side_effect=RuntimeError("gate reached"),
    ):
        with pytest.raises(RuntimeError, match="gate reached"):
            step._execute(mode="force")


def test_public_run_force_guard_before_any_destruction():
    """재리뷰 BLOCKING-1: public run(force) 가 claim/cleanup_artifacts/
    invalidate_downstream/clear_checkpoint 어느 것도 밟기 전에 거부."""
    from app.core.errors import AppError

    step = _make_step()
    step.check_gate = MagicMock()
    step.check_applicability = MagicMock(return_value=True)
    step.cleanup_artifacts = MagicMock()
    step.invalidate_downstream = MagicMock()
    step.clear_checkpoint = MagicMock()
    step._try_claim_running = MagicMock(return_value=True)
    with patch("app.core.config.settings.scene_image_target_scenes",
               "1,10", create=True):
        with pytest.raises(AppError) as exc:
            step.run(mode="force")
    assert "target_force_forbidden" in exc.value.code
    step.check_gate.assert_not_called()
    step._try_claim_running.assert_not_called()
    step.cleanup_artifacts.assert_not_called()
    step.invalidate_downstream.assert_not_called()
    step.clear_checkpoint.assert_not_called()


# ── target A→B 변경 = 비파괴 RERUN_SELF (재리뷰 BLOCKING-2) ─────────


def test_contract_drift_target_only_change_rerun_self(_no_drift_ack):
    """base hash 동일+config_hash mismatch = 표적 목록만 바뀐 drift →
    RERUN_SELF(cleanup 없는 resume). base 자체 변경=기본 BLOCK 유지."""
    from app.core.step_runner import ResumeAction

    step = _make_step()
    step.step_id = "scene_image_pipeline"
    step._config_hash_base = MagicMock(return_value="BASE_X")
    step.load_checkpoint = MagicMock(
        return_value={"target_scope_base_hash": "BASE_X"})
    d = step._evaluate_contract_drift("config_hash mismatch: cp=aa now=bb")
    assert d.action == ResumeAction.RERUN_SELF
    assert d.origin == "contract_drift"
    assert "target-scope drift" in d.reason

    # base 가 다르면(팩·모델 등 실질 변경) 기본 BLOCK
    step.load_checkpoint = MagicMock(
        return_value={"target_scope_base_hash": "BASE_OLD"})
    d2 = step._evaluate_contract_drift("config_hash mismatch: cp=aa now=bb")
    assert d2.action == ResumeAction.BLOCK

    # 저장 base 부재(구 CP, config_hash 도 없음) = 판별 불가 → 기본 BLOCK
    step.load_checkpoint = MagicMock(return_value={})
    d3 = step._evaluate_contract_drift("config_hash mismatch: cp=aa now=bb")
    assert d3.action == ResumeAction.BLOCK

    # 무표적→표적 전이 fallback (재재리뷰 HIGH-2): 무표적 CP 는 표적
    # 필드를 영속하지 않음 — 저장 config_hash(표적 fold 없음)==현재 base
    # 면 표적-only drift
    step.load_checkpoint = MagicMock(
        return_value={"config_hash": "BASE_X"})
    d4 = step._evaluate_contract_drift("config_hash mismatch: cp=aa now=bb")
    assert d4.action == ResumeAction.RERUN_SELF

    # fallback config_hash 도 base 와 다르면(구 md5 fallback CP 포함) BLOCK
    step.load_checkpoint = MagicMock(
        return_value={"config_hash": "LEGACY_MD5"})
    d5 = step._evaluate_contract_drift("config_hash mismatch: cp=aa now=bb")
    assert d5.action == ResumeAction.BLOCK

    # config_hash 외 mismatch = 기본 정책 그대로
    d6 = step._evaluate_contract_drift("schema_version mismatch: 1 != 2")
    assert d6.action == ResumeAction.BLOCK


def test_execute_off_result_shape_byte_identical(tmp_path):
    """재재리뷰 HIGH-2: 표적 미설정 default 경로=result/CP shape 불변 —
    표적 필드(config_hash/target_scope_base_hash) 미포함 exact-shape."""
    step = _make_step()
    step.db.query.side_effect = lambda *m: _selected_query(["st_1"])
    with patch("app.core.config.settings.scene_image_target_scenes",
               "", create=True), patch(
        "app.core.pipeline_gate.check_scene_images_ready",
    ), patch(
        "app.services.scene_image_service.SceneImageService",
    ):
        result = step._execute(mode="resume")
    assert set(result) == {
        "completed_count", "applicable_count", "failed_count", "data"}


# ── _execute 카운트 = audit effective (HIGH-3·4) ─────────────────────


def _selected_query(ids):
    q = MagicMock()
    q.filter.return_value = q
    q.all.return_value = [(x,) for x in ids]
    q.count.return_value = len(ids)
    return q


def _query_router(selected_ids, asset_q, universe=None):
    """db.query 라우팅 — ImageAsset/selected id(1컬럼)/universe(2컬럼)."""
    from app.models.project import ImageAsset

    selected_q = _selected_query(selected_ids)
    uni_q = MagicMock()
    uni_q.filter.return_value = uni_q
    uni_q.all.return_value = [
        (sid, si) for sid, si in (universe or {}).items()]

    def _query(*models):
        if models and models[0] is ImageAsset:
            return asset_q
        if len(models) == 2:
            return uni_q
        return selected_q

    return _query


def test_execute_counts_use_audit_effective(tmp_path):
    """비표적 완료 still 은 카운트에 영향 0 — effective 교집합만."""
    step = _make_step()
    _write_audit(tmp_path, ["st_1", "st_2"], requested=["st_1"],
                 added=["st_2"])

    # DB: 에피소드 전체 selected=4 (표적 밖 st_3/st_4 포함) — 표적 씬 1
    # 의 selected 는 st_1 뿐 (audit.requested 와 동등)
    asset_q = MagicMock()
    asset_q.filter.return_value = asset_q
    asset_q.count.return_value = 2
    step.db.query.side_effect = _query_router(
        ["st_1", "st_2", "st_3", "st_4"], asset_q,
        universe={"st_1": 1, "st_2": 2, "st_3": 3, "st_4": 4})

    with patch("app.core.config.settings.scene_image_target_scenes",
               "1", create=True), patch(
        "app.core.config.settings.projects_dir", str(tmp_path),
    ), patch(
        "app.core.pipeline_gate.check_scene_images_ready",
    ), patch(
        "app.services.scene_image_service.SceneImageService",
    ) as svc_cls:
        result = step._execute(mode="resume")

    svc_cls.return_value.generate_images.assert_called_once()
    # applicable=effective(2) — 전체 selected(4) 아님
    assert result["applicable_count"] == 2
    assert result["data"]["scene_total"] == 2
    # 표적 설정 시=base 영속 (BLOCKING-2 — 표적 drift 비교 기준)
    with patch("app.core.config.settings.scene_image_target_scenes",
               "1", create=True):
        assert result["target_scope_base_hash"] == step._config_hash_base()


def test_execute_target_audit_missing_fail_closed(tmp_path):
    """표적 설정인데 실행 audit 부재=fail-closed (독자 필터 재구현 금지)."""
    step = _make_step()
    step.db.query.side_effect = lambda model: _selected_query(["st_1"])
    with patch("app.core.config.settings.scene_image_target_scenes",
               "1", create=True), patch(
        "app.core.config.settings.projects_dir", str(tmp_path),
    ), patch(
        "app.core.pipeline_gate.check_scene_images_ready",
    ), patch(
        "app.services.scene_image_service.SceneImageService",
    ):
        with pytest.raises(ValueError, match="audit 부재"):
            step._execute(mode="resume")


# ── verify_completion = 동일 audit scope + 의존 파일 결손 partial ────


def _make_verify_step(tmp_path, selected_ids, disk_ids, universe=None):
    step = _make_step()
    asset_q = MagicMock()
    asset_q.filter.return_value = asset_q
    assets = []
    for i, sid in enumerate(disk_ids):
        a = MagicMock()
        a.still_id = sid
        p = tmp_path / f"img_{i}.png"
        p.write_bytes(b"PNG")
        a.file_path = str(p)
        assets.append(a)
    asset_q.all.return_value = assets
    step.db.query.side_effect = _query_router(
        selected_ids, asset_q,
        universe=universe if universe is not None
        else {sid: i + 1 for i, sid in enumerate(selected_ids)})
    return step


def test_verify_completion_dependency_missing_partial(tmp_path):
    """의존 추가 스틸(st_2)의 파일 결손도 partial — requested 만 검사 금지
    (Codex HIGH-4)."""
    step = _make_verify_step(
        tmp_path, selected_ids=["st_1", "st_2", "st_3"],
        disk_ids=["st_1"])  # requested st_1 만 존재, 의존 st_2 결손
    _write_audit(tmp_path, ["st_1", "st_2"], requested=["st_1"],
                 added=["st_2"])
    with patch("app.core.config.settings.scene_image_target_scenes",
               "1", create=True), patch(
        "app.core.config.settings.projects_dir", str(tmp_path),
    ), patch(
        "app.core.file_paths.resolve_image_path",
        side_effect=lambda p: __import__("pathlib").Path(p),
    ):
        report = step.verify_completion()
    assert report.is_complete is False
    assert report.metadata["expected"] == 2  # effective 기준 (st_3 제외)
    assert report.metadata["found"] == 1


def test_verify_completion_target_complete_ignores_nontarget(tmp_path):
    """비표적 st_3 미생성이어도 effective 전부 존재=clean."""
    step = _make_verify_step(
        tmp_path, selected_ids=["st_1", "st_2", "st_3"],
        disk_ids=["st_1", "st_2"])
    _write_audit(tmp_path, ["st_1", "st_2"], requested=["st_1"],
                 added=["st_2"])
    with patch("app.core.config.settings.scene_image_target_scenes",
               "1", create=True), patch(
        "app.core.config.settings.projects_dir", str(tmp_path),
    ), patch(
        "app.core.file_paths.resolve_image_path",
        side_effect=lambda p: __import__("pathlib").Path(p),
    ):
        report = step.verify_completion()
    assert report.is_complete is True
    assert report.metadata["expected"] == 2


def test_verify_completion_target_audit_missing_fail_closed(tmp_path):
    step = _make_verify_step(tmp_path, selected_ids=["st_1"],
                             disk_ids=["st_1"])
    with patch("app.core.config.settings.scene_image_target_scenes",
               "1", create=True), patch(
        "app.core.config.settings.projects_dir", str(tmp_path),
    ):
        report = step.verify_completion()
    assert report.is_complete is False
    assert report.severity == "missing"
    assert any("audit 부재" in m for m in report.missing)


def test_verify_completion_off_unchanged(tmp_path):
    """표적 미설정=기존 전체 검증 byte-identical (audit 미소비)."""
    step = _make_verify_step(
        tmp_path, selected_ids=["st_1", "st_2"], disk_ids=["st_1", "st_2"])
    with patch("app.core.config.settings.scene_image_target_scenes",
               "", create=True), patch(
        "app.core.file_paths.resolve_image_path",
        side_effect=lambda p: __import__("pathlib").Path(p),
    ):
        report = step.verify_completion()
    assert report.is_complete is True
    assert report.metadata["expected"] == 2


def test_verify_completion_stale_audit_after_still_regeneration(tmp_path):
    """재재리뷰 HIGH-1(a): 스틸 세대 교체 후 같은 씬 번호의 stale audit
    = 교집합 0 이어도 CLEAN 금지 — 우주 결합 검증이 missing 판정."""
    step = _make_verify_step(
        tmp_path, selected_ids=["new_still"], disk_ids=["new_still"],
        universe={"new_still": 1})
    _write_audit(tmp_path, ["old_still"])  # requested_scenes=[1], old id
    with patch("app.core.config.settings.scene_image_target_scenes",
               "1", create=True), patch(
        "app.core.config.settings.projects_dir", str(tmp_path),
    ):
        report = step.verify_completion()
    assert report.is_complete is False
    assert any("우주와 불일치" in m for m in report.missing)


# ── pose-guide allowlist (재리뷰 HIGH-4 / 재재리뷰 TEST GAP-1) ───────


def test_indoor_pose_guide_execution_allowlist(tmp_path):
    """selected_keys(그룹·연속성 신호)=전체 유지, 생성 제외(skip)=
    already_done+allowlist 밖 — 컨텍스트 SOT 유지·실행만 allowlist."""
    from app.core.steps.indoor_shared_pose_guide_context import (
        load_indoor_shared_pose_guides,
    )

    stills = [
        {"id": "st_1", "scene_index": 1, "shot_index": 1},
        {"id": "st_2", "scene_index": 1, "shot_index": 2},
        {"id": "st_3", "scene_index": 2, "shot_index": 1},
    ]
    captured = {}

    def fake_builder(**kwargs):
        captured.update(kwargs)
        return {"guide_by_shot": {}, "diagnostics": []}

    mod = "app.core.steps.indoor_shared_pose_guide_context"
    with patch(
        "app.core.config.settings.indoor_shared_pose_guide_enabled",
        True, create=True,
    ), patch(
        "app.core.config.settings.projects_dir", str(tmp_path),
    ), patch(f"{mod}.build_indoor_shared_pose_context",
             side_effect=fake_builder), patch(
        f"{mod}._load_bg_loc_by_id", return_value={},
    ), patch(
        f"{mod}._load_indoor_outdoor_locs", return_value=(set(), set()),
    ), patch(
        f"{mod}._load_bg_asset_by_id", return_value={},
    ), patch(
        f"{mod}._load_name_by_short_id", return_value={},
    ), patch(
        f"{mod}._resolve_guide_asset_ids",
    ), patch(
        "app.services.image_capture.context.generation_context",
    ):
        load_indoor_shared_pose_guides(
            project_id="SAMPLE_P", episode_id="SAMPLE_E", stills=stills,
            staging_map={}, background_chain_bg_map={}, zoom_ctx=None,
            db=MagicMock(),
            already_done_still_ids={"st_3"},
            execution_allowlist_still_ids={"st_1"},
        )

    # 컨텍스트 신호=전체 3키 유지
    assert set(captured["selected_keys"]) == {(1, 1), (1, 2), (2, 1)}
    # 생성 제외=allowlist 밖(st_2)+already_done(st_3), allowlist(st_1)=생성
    assert captured["skip_shot_keys"] == {(1, 2), (2, 1)}


def test_indoor_pose_guide_allowlist_none_byte_identical(tmp_path):
    """allowlist None=기존 skip(already_done 만) byte-identical."""
    from app.core.steps.indoor_shared_pose_guide_context import (
        load_indoor_shared_pose_guides,
    )

    stills = [
        {"id": "st_1", "scene_index": 1, "shot_index": 1},
        {"id": "st_2", "scene_index": 1, "shot_index": 2},
    ]
    captured = {}

    def fake_builder(**kwargs):
        captured.update(kwargs)
        return {"guide_by_shot": {}, "diagnostics": []}

    mod = "app.core.steps.indoor_shared_pose_guide_context"
    with patch(
        "app.core.config.settings.indoor_shared_pose_guide_enabled",
        True, create=True,
    ), patch(
        "app.core.config.settings.projects_dir", str(tmp_path),
    ), patch(f"{mod}.build_indoor_shared_pose_context",
             side_effect=fake_builder), patch(
        f"{mod}._load_bg_loc_by_id", return_value={},
    ), patch(
        f"{mod}._load_indoor_outdoor_locs", return_value=(set(), set()),
    ), patch(
        f"{mod}._load_bg_asset_by_id", return_value={},
    ), patch(
        f"{mod}._load_name_by_short_id", return_value={},
    ), patch(
        f"{mod}._resolve_guide_asset_ids",
    ), patch(
        "app.services.image_capture.context.generation_context",
    ):
        load_indoor_shared_pose_guides(
            project_id="SAMPLE_P", episode_id="SAMPLE_E", stills=stills,
            staging_map={}, background_chain_bg_map={}, zoom_ctx=None,
            db=MagicMock(),
            already_done_still_ids={"st_2"},
        )
    assert captured["skip_shot_keys"] == {(1, 2)}


def test_untargeted_cp_then_target_transition_rerun_self(
        tmp_path, _no_drift_ack):
    """3차 리뷰 HIGH: **실제 save_checkpoint 경로**로 recipe ON 무표적
    CP 를 만든 뒤 표적 설정 → _check_cp_mismatch→_evaluate_contract_drift
    가 비파괴 RERUN_SELF — 수동 dict fixture 가 놓치던 저장/비교 비대칭
    (md5 fallback vs step-local sha) 잠금."""
    import json as _json

    from app.core.step_runner import ResumeAction

    step = _make_step()
    step.step_id = "scene_image_pipeline"
    step.run_id = "SAMPLE_RUN"
    step.manifest = {}
    step._cp_dir = tmp_path / "cp"
    step._cp_dir.mkdir(parents=True)
    step._resolve_model = MagicMock(return_value="SAMPLE_MODEL")

    with patch("app.core.config.settings.still_recipe_mode", "v1",
               create=True):
        # 1) 무표적 완료 CP — override 가 step-local hash(=base) 저장
        with patch("app.core.config.settings.scene_image_target_scenes",
                   "", create=True):
            step.save_checkpoint({"status": "completed"})
            stored = _json.loads(
                (step._cp_dir / "manifest.json").read_text())
            assert stored["config_hash"] == step._config_hash()
            assert stored["config_hash"] == step._config_hash_base()
            from app.core.step_runner import compute_config_hash

            # md5 fallback 이 아님 (recipe ON=sha 페이로드)
            assert stored["config_hash"] != compute_config_hash({})
        # 2) 표적 설정 → mismatch 는 감지되되 표적-only drift 로 분류
        with patch("app.core.config.settings.scene_image_target_scenes",
                   "1,10", create=True):
            cp = step.load_checkpoint()
            reason = step._check_cp_mismatch(cp)
            assert reason and "config_hash mismatch" in reason
            d = step._evaluate_contract_drift(reason)
            assert d.action == ResumeAction.RERUN_SELF
            assert "target-scope drift" in d.reason
        # 3) base 실질 변경(다른 project_config)이면 BLOCK 유지
        with patch("app.core.config.settings.scene_image_target_scenes",
                   "1,10", create=True):
            step.project_config = {"SAMPLE": "changed"}
            d2 = step._evaluate_contract_drift(
                "config_hash mismatch: cp=x now=y")
            assert d2.action == ResumeAction.BLOCK