"""OutdoorStructureSeedStep 결정론 테스트 (Stage D) — 게이트/재검증/격리/hash.

생성(LLM/이미지)은 monkeypatch 로 차단 — 스텝의 게이트·plan 재검증·그룹
집계 분기만 검증한다. fixture 는 전부 시나리오 중립 SAMPLE 데이터.
"""

import hashlib
import tempfile
from pathlib import Path
from unittest.mock import MagicMock, patch

import pytest

from app.core.steps.outdoor_structure_seed_step import (
    SCHEMA_VERSION,
    OutdoorStructureSeedStep,
    collect_lane2_groups,
)

# ── 검색 선택 형태 참조 fixture (2026-08-01) ────────────────────────
# form_reference v3 이후 필수 그룹은 실제 참조 파일 + sha 가 있어야 한다.
# 소비자가 소비 직전 재검증하므로 경로만 흉내 내면 통과하지 못한다.
_FIXTURE_REF_DIR = Path(tempfile.mkdtemp(prefix="sample_fixture_form_ref_"))
_FIXTURE_REF = _FIXTURE_REF_DIR / "cand_01.png"
_FIXTURE_REF.write_bytes(b"SAMPLE_FIXTURE_FORM_REF_BYTES")
_FIXTURE_REF_SHA = hashlib.sha256(_FIXTURE_REF.read_bytes()).hexdigest()


def _formref_cp(gids=("sample_site",), no_spec=(), status="completed",
                typology=None):
    """선행 form_reference 체크포인트 — 기본은 정상 완료·전 그룹 필수.

    소비자가 정책·팩·우주·분할을 exact 검증하므로 producer 가 실제로 쓰는
    메타를 그대로 담는다(2026-08-01 Codex 재리뷰 BLOCKING 2).
    """
    from app.modules.pipeline.search_grounded_ref import (
        TARGET_POLICY_VERSION,
        resolve_ref_pack_version,
    )

    return {
        "status": status,
        "data": {
            "groups": {
                g: {
                    "status": "ok",
                    "group_id": g,
                    "form_ref_path": str(_FIXTURE_REF),
                    "form_ref_sha256": _FIXTURE_REF_SHA,
                    "form_ref_asset_id": f"asset-{g}",
                    # 사전조사는 이 스텝이 아니라 **앞 스텝**이 한다
                    # (2026-08-03) — 씨드는 그 기록을 읽기만 한다.
                    **({"typology": typology} if typology else {}),
                } for g in gids
            },
            "mandatory_group_ids": list(gids),
            "target": {
                "policy_version": TARGET_POLICY_VERSION,
                "pack_version": resolve_ref_pack_version(),
                "universe_group_ids": list(gids) + list(no_spec),
                "no_spec_group_ids": list(no_spec),
            },
        },
    }


def _fake_db():
    """form_reference 자산이 **실재하는** 상태 — 실제 선행 실행 뒤와 같다.

    [2026-08-01 Codex 2차 재리뷰 BLOCKING 2] 소비자가 계보를 파일·sha 뿐
    아니라 **자산 실존·소유·경로**까지 검증한다. 통짜 MagicMock 은
    `filter_by().first()` 가 항상 truthy 를 돌려주고 `file_path` 도
    MagicMock 이라, 실제로는 있을 수 없는 상태를 시험하게 된다.

    form_ref 조회만 실제처럼 답하고 나머지 조회는 기존 MagicMock 동작을
    유지한다(seed 자산 upsert 경로의 기존 계약 불변).
    """
    from types import SimpleNamespace

    from app.core.file_paths import ImagePathType

    # ★ORM 이 실제로 돌려주는 값을 쓴다. 소비자와 **같은 함수**로 fake 를 만들면
    # 구현이 틀려도 같이 틀린다 — 실제로 그렇게 상대/절대 불일치 결함을 놓쳤다.
    _col = ImagePathType()
    ref_path = _col.process_result_value(
        _col.process_bind_param(str(_FIXTURE_REF), None), None)

    def _filter_by(**kw):
        q = MagicMock()
        if kw.get("asset_type") == "structure_form_ref":
            matches = (
                kw.get("id") == f"asset-{kw.get('entity_id')}"
                and kw.get("project_id") == "SAMPLE_PROJECT"
                and kw.get("episode_id") == "SAMPLE_EPISODE"
                and kw.get("variant_type") == "form_ref"
            )
            q.first.return_value = (
                SimpleNamespace(id=kw.get("id"), file_path=ref_path)
                if matches else None)
        return q

    db = MagicMock()
    db.query.return_value.filter_by.side_effect = _filter_by
    return db


def _asset_id_by_kind(**kw) -> str:
    """자산 대역 — 몇 번 불릴지에 기대지 않는다.

    고정 목록을 주면 파이프라인에 자산이 하나 늘 때마다 그것과 무관한
    테스트까지 StopIteration 으로 죽는다(2026-08-03 도해 자산 추가 때 실측).
    """
    return f"a-{kw.get('variant_type') or 'seed'}"


def _make_step():
    step = OutdoorStructureSeedStep.__new__(OutdoorStructureSeedStep)
    step.project_id = "SAMPLE_PROJECT"
    step.episode_id = "SAMPLE_EPISODE"
    step.project_config = {}
    step.build_opik_metadata = MagicMock(return_value={})
    step.db = _fake_db()
    return step


def _lane_plan(lane="structure_plate", quote="대문 안 구조물"):
    return {
        "segments": [
            {"segment_id": "seg1", "label_en": "sample structure front",
             "dominant_mode": "structure",
             "evidence": [{"scene_index": 3, "quote_ko": quote}],
             "confidence": "high"},
        ],
        "shot_bindings": [
            {"scene_index": 3, "shot_index": 1, "segment_id": "seg1",
             "lane": lane, "confidence": "high",
             "rationale_ko": "구조물 지배",
             "evidence": {"scene_index": 3, "quote_ko": quote}},
        ],
    }


def _cps(lane="structure_plate", quote="대문 안 구조물"):
    return {
        "outdoor_lane_plan": {"data": {"groups": {
            "sample_site": {
                "status": "ok",
                "plan": _lane_plan(lane, quote),
                "scene_indices": [3],
                "outdoor_loc_ids": ["L01"],
            },
        }}},
        "outdoor_place_spec": {"data": {"groups": {
            "sample_site": {
                "outdoor_loc_ids": ["L01"],
                "scene_indices": [3],
                "spec": {
                "layout_narration_en": "A SAMPLE path reaches the gate.",
                "zone_labels_en": ["Structure Front"],
                "items": [{"code": "Y1", "kind": "building",
                           "name_en": "SAMPLE main building",
                           "placement_en": "central mass",
                           "temporal_scope": "persistent_site"}],
                "excluded_transient_elements": [
                    {"name_en": "SAMPLE movable cart"}],
            }},
        }}},
        "scene_save": {"data": {"segments": [
            {"scene_index": 3, "heading": "EXT. SAMPLE 구조물 - 낮",
             "text": "대문 안 구조물 앞에 선다."},
        ]}},
        "background_classify": {"data": {"building_groups": [
            {"group_id": "sample_site", "members": [
                {"loc_id": "L02", "is_indoor": True}]},
        ]}},
        "entity_merge": {"data": {"locations": [
            {"short_id": "L02", "name": "SAMPLE 내부",
             "description": "거실과 방"},
        ]}},
        "shot_ref_classify": {"data": {
            "world_anchor_en": "contemporary SAMPLE region"}},
        # BLOCKING-2: authoritative 그룹 샷 재구성 입력
        "shot_validator": {"data": {"scenes": [
            {"scene_index": 3, "shots": [
                {"shot_index": 1, "location_id": "L01",
                 "description": "구조물 앞 인물"}]},
        ]}},
        "shot_staging": {"data": {"shots": [
            {"scene_index": 3, "shot_index": 1,
             "camera_direction": "정면"}]}},
        # 소비자가 소비 직전 검증하는 선행 계약 (2026-08-01 BLOCKING 3)
        "outdoor_structure_form_reference": _formref_cp(),
    }


def _patches(pipe=True, plan=True):
    return (
        patch("app.core.config.settings.outdoor_lane_pipe_enabled",
              pipe, create=True),
        patch("app.core.config.settings.outdoor_lane_plan_enabled",
              plan, create=True),
    )


def test_collect_lane2_groups_filters_structure_plate():
    data = {"groups": {
        "a": {"status": "ok", "plan": _lane_plan("structure_plate")},
        "b": {"status": "ok", "plan": _lane_plan("map_marker")},
        "c": {"status": "failed", "plan": _lane_plan("structure_plate")},
    }}
    assert collect_lane2_groups(data) == ["a"]


def test_flag_off_noop():
    for pipe, plan in ((False, True), (True, False), (False, False)):
        p1, p2 = _patches(pipe, plan)
        with p1, p2:
            result = _make_step()._execute()
        assert result["applicable_count"] == 0
        assert result["data"] == {"groups": {}}
        assert result["schema_version"] == SCHEMA_VERSION


def test_map_marker_only_noop():
    step = _make_step()
    cps = _cps(lane="map_marker")
    step._load_prev_checkpoint = lambda sid: cps.get(sid)
    p1, p2 = _patches()
    with p1, p2:
        result = step._execute()
    assert result["applicable_count"] == 0


def test_persisted_plan_revalidation_fail_closed():
    """가짜 인용이 담긴 persisted plan = AppError (소비 직전 재검증)."""
    from app.core.errors import AppError

    step = _make_step()
    cps = _cps(quote="원문에 존재하지 않는 인용")
    step._load_prev_checkpoint = lambda sid: cps.get(sid)
    p1, p2 = _patches()
    with p1, p2, pytest.raises(AppError) as ei:
        step._execute()
    assert ei.value.code == (
        "step.contract_violation.outdoor_structure_seed")


def test_spec_missing_group_failed_isolated():
    step = _make_step()
    cps = _cps()
    cps["outdoor_place_spec"] = {"data": {"groups": {}}}
    # spec 이 없으면 producer 도 그 그룹을 필수가 아니라 no_spec 으로 남긴다
    # — CP 와 현재 spec 이 어긋나면 그건 별개 결함(분할 드리프트)이고, 여기서
    # 보려는 것은 **근거 없는 그룹의 그룹 단위 격리**다.
    cps["outdoor_structure_form_reference"] = _formref_cp(
        gids=(), no_spec=("sample_site",))
    step._load_prev_checkpoint = lambda sid: cps.get(sid)
    p1, p2 = _patches()
    with p1, p2:
        result = step._execute()
    assert result["failed_count"] == 1
    g = result["data"]["groups"]["sample_site"]
    assert g["status"] == "failed"
    assert "seed 입력 SOT" in g["error"]


def test_generation_failure_isolated_and_inputs_are_sot(tmp_path):
    """생성 예외=그룹 격리 + seed 프롬프트 입력이 spec SOT 파생인지 검증."""
    step = _make_step()
    cps = _cps()
    step._load_prev_checkpoint = lambda sid: cps.get(sid)
    captured = {}

    def fake_multiroll(**kwargs):
        captured["prompt"] = kwargs["prompt"]
        raise RuntimeError("SAMPLE 생성 실패")

    p1, p2 = _patches()
    with p1, p2, patch(
        "app.modules.pipeline.multiroll_select.run_multiroll_select",
        side_effect=fake_multiroll,
    ), patch(
        "app.modules.pipeline.gpt_image_gen.make_gpt_image_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.core.config.settings.projects_dir", str(tmp_path),
    ):
        result = step._execute()
    assert result["failed_count"] == 1
    assert "SAMPLE 생성 실패" in (
        result["data"]["groups"]["sample_site"]["error"])
    # seed 입력 SOT: persistent_site 검증 서술+negative contract+규모 앵커
    p = captured["prompt"]
    assert "A SAMPLE path reaches the gate." in p
    assert "- SAMPLE main building: central mass" in p
    assert "SAMPLE movable cart" in p
    assert "거실과 방" in p
    assert "contemporary SAMPLE region" in p
    assert "EXT. SAMPLE 구조물 - 낮" in p


def test_config_hash_includes_packs_and_models():
    step = _make_step()
    h1 = step._config_hash()
    with patch(
        "app.core.steps.outdoor_structure_seed_step.SEED_PACK_VERSION", "1",
    ):
        h2 = step._config_hash()
    assert h1 != h2
    # v2 (2026-07-16 R7): siteplan 산출 제거 — siteplan 팩·모델 스탬프 소멸
    with patch(
        "app.core.steps.outdoor_structure_seed_step.SCHEMA_VERSION", 1,
    ):
        h3 = step._config_hash()
    assert h1 != h3


# ── seed 품질 2R — LLM 변형 저작 모드 (2026-07-16) ────────────────────


def _author_payload():
    # v9 (Codex BLOCKING-1·HIGH-3 + v8b ③④): 구조화 필드 전부 required —
    # 활성 팩 계약과 fixture 동기
    return {
        "location_line_en": "a SAMPLE residential alley structure",
        "conformance_brief_en": "one main SAMPLE building; openings "
                                "exactly as evidenced",
        "room_inventory_applicable": True,
        "room_inventory": [
            # provenance 검사(v9 validator): 인용=공급 씬(scene_index=3)
            # 원문의 whitespace-normalized exact substring
            {"label_en": "SAMPLE room", "scene_index": 3,
             "evidence_ko": "대문 안 구조물"},
        ],
        "lettering_text": "",
        "variants": [
            {"variant_id": f"v{i + 1}", "approach_ko": f"접근 {i + 1}",
             "evidence_ko": ["대문 안 구조물"] if i == 0 else [],
             "prompt_en": f"SAMPLE variant body {i + 1}"}
            for i in range(3)
        ],
    }


def _variants_patches():
    p1, p2 = _patches()
    p3 = patch(
        "app.core.config.settings.structure_seed_variants_enabled",
        True, create=True)
    return p1, p2, p3


def test_config_hash_variants_flag_stamped_only_when_on():
    step = _make_step()
    base = step._config_hash()
    with patch(
        "app.core.config.settings.structure_seed_variants_enabled",
        True, create=True,
    ):
        on = step._config_hash()
    with patch(
        "app.core.config.settings.structure_seed_variants_enabled",
        False, create=True,
    ):
        off = step._config_hash()
    assert on != base       # ON=팩·모델 스탬프
    assert off == base      # OFF=무스탬프 (기존 프로젝트 hash 불변)


def test_variants_mode_full_wiring(tmp_path):
    """저작→변형 롤→브리프 판정 배선 + 진단 ③(raw 헤딩 미주입) 계약.

    Codex 재리뷰 반영 잠금: ①판정 system=seed 전용 v2(film-still 계약
    부재) ②변형 수=3 고정(still_recipe_roll_count 와 독립) ⑤sidecar
    persist+CP decision 기록.
    """
    step = _make_step()
    cps = _cps()
    step._load_prev_checkpoint = lambda sid: cps.get(sid)
    step._upsert_seed_asset = MagicMock(side_effect=_asset_id_by_kind)

    author_calls = {}

    def fake_call_structured(step_tag, sys_prompt, user_content, schema,
                             **kwargs):
        author_calls["step_tag"] = step_tag
        author_calls["sys"] = sys_prompt
        author_calls["user"] = user_content
        author_calls["schema"] = schema
        author_calls["pc"] = kwargs.get("project_config")
        return _author_payload()

    mr = {}

    def fake_multiroll(**kwargs):
        mr.update(kwargs)
        kwargs["persist_record_fn"]({"selected": "B"})  # durable persist
        sel = kwargs["out_stem"].parent / (
            kwargs["out_stem"].name + "_sel.png")
        sel.parent.mkdir(parents=True, exist_ok=True)
        sel.write_bytes(b"img")
        return sel, {
            "selected": "B",
            "roll_prompts": dict(kwargs["roll_prompts"]),
            "totals": {"A": 5, "B": 9, "C": 7},
            "ranking": ["B", "C", "A"],
            "verdicts": [{"label": "B", "score": 9, "verdict_ko": "ok"}],
        }

    judge_kwargs = {}

    def fake_judge_factory(**kwargs):
        judge_kwargs.update(kwargs)
        return MagicMock()

    # ★도해 생성은 multiroll 밖에서 gen_fn 을 직접 부른다. MagicMock 을 그냥
    #  돌려주면 그 호출이 실제 이미지 API 로 나가므로, 파일을 만드는 대역을
    #  준다(2026-08-03 실측: 대역이 빗나가 유료 경로가 열렸다).
    def fake_gen_fn(tag, prompt, refs, out_path):
        Path(out_path).parent.mkdir(parents=True, exist_ok=True)
        Path(out_path).write_bytes(b"png")
        return Path(out_path)

    p1, p2, p3 = _variants_patches()
    with p1, p2, p3, patch(
        "app.modules.llm.llm_client.call_structured",
        side_effect=fake_call_structured,
    ), patch(
        "app.modules.pipeline.multiroll_select.run_multiroll_select",
        side_effect=fake_multiroll,
    ), patch(
        "app.modules.pipeline.gpt_image_gen.make_gpt_image_gen_fn",
        return_value=fake_gen_fn,
    ), patch(
        "app.modules.pipeline.multiroll_gemini.make_gemini_judge_fn",
        side_effect=fake_judge_factory,
    ), patch(
        "app.modules.pipeline.multiroll_gemini.make_gemini_critique_fn",
        return_value=MagicMock(),
    ), patch(
        "app.core.config.settings.projects_dir", str(tmp_path),
    ), patch(
        # HIGH-2: 스틸 롤 수를 바꿔도 seed 변형 수는 3 고정
        "app.core.config.settings.still_recipe_roll_count", 2,
    ):
        result = step._execute()

    assert result["completed_count"] == 1
    g = result["data"]["groups"]["sample_site"]

    # 저작 입력: 시나리오 원문 전문+SOT 입력 (④)
    assert author_calls["step_tag"] == "seed_variant_author"
    assert "대문 안 구조물 앞에 선다." in author_calls["user"]
    assert "A SAMPLE path reaches the gate." in author_calls["user"]
    assert author_calls["pc"]["seed_variant_author"] == {"model": "gpt"}

    # ★롤 = 경로 둘(2026-08-03). 저작 변형은 여러 벌 그대로 받되 두 경로가
    # **같은 한 벌**을 쓴다 — 문장이 갈리면 경로 비교가 아니라 문장 비교가
    # 된다. raw 씬 헤딩 미주입(③)은 그대로.
    rp = mr["roll_prompts"]
    assert mr["roll_count"] == 2
    assert set(rp) == {"A", "B"}
    assert "SAMPLE variant body 1" in rp["A"]
    assert "SAMPLE variant body 1" in rp["B"]
    for p in rp.values():
        assert "THE LOCATION: a SAMPLE residential alley structure" in p
        assert "EXT. SAMPLE 구조물" not in p
        assert "no people" in " ".join(p.lower().split())
        assert "SAMPLE movable cart" in p

    # ★[2026-08-01 Codex BLOCKING 2] 관할 계약이 **실제 생성 프롬프트**에
    # 실린다. multiroll 은 roll_prompts 가 있으면 base prompt 를 통째로
    # 무시하고 roll_prompts[label] 만 gen_fn 에 넘긴다 — 그래서 절을 전
    # 롤에 주입하지 않으면 웹 사진만 첨부되고 관할 계약은 어느 롤에도
    # 실리지 않는다. 카나리 A/B: 주 표시면 기관 표장 A 2/2 없음 → B 2/2 있음.
    # 경로마다 관할이 다르다 — 직접은 사진 한 장이 형태·표면을 다 맡고,
    # 스케치 경유는 형태를 도해에 넘긴다.
    assert "FORM REFERENCE" in rp["A"], "직접 경로에 관할절 없음"
    assert "LINE DRAWING" in rp["B"], "스케치 경로에 관할절 없음"
    # 참조도 경로별로 갈린다: 직접=사진만 / 스케치 경유=도해+사진.
    rr = mr["roll_refs"]
    assert [lab for lab, _ in rr["A"]] == ["FORM REFERENCE"]
    assert [lab for lab, _ in rr["B"]] == ["STRUCTURE DIAGRAM",
                                           "FORM REFERENCE"]
    # 자리 편향 — 좌우를 바꿔 두 번 묻고, 동점은 직접 경로가 가져간다.
    assert mr["judge_flip"] is True
    assert list(mr["flip_priority"]) == ["A", "B"]
    # 소비자가 보는 prompt_used 도 실제 계약이어야 한다(감사 가능성).
    assert "LINE DRAWING" in g["seed_prompt"]   # 선정=B(스케치 경로)
    # 도해가 어느 파일이고 어느 경로가 이겼는지 CP 에 남는다.
    assert g["seed_paths"]["selected"] == "B"
    assert g["seed_paths"]["sketch_png_path"]

    # 3차 BLOCKING-1(2): inner 지문에 모델·판정 계약 포함 (sidecar 경계)
    efp = mr["extra_fingerprint"]
    assert set(efp) >= {"gen_model", "judge_physical_model",
                        "seed_variants_pack", "judge_fix_pack",
                        "contract_texts_sha"}
    assert mr["force"] is False  # resume 실행

    # 판정: 브리프+seed 전용 v2 system+헤더 교체 (② / BLOCKING-1)
    assert mr["prompt"].startswith("SHARED CONFORMANCE BRIEF")
    assert "openings exactly as evidenced" in mr["prompt"]
    js = judge_kwargs["judge_sys"].lower()
    assert "different authored phrasing" in js
    assert "brief conformance" in js
    for forbidden in ("all generated from that exact prompt",
                      "shot text", "wardrobe", "carried-state",
                      "immobility"):
        assert forbidden not in js, forbidden
    assert "generated from a different authored phrasing" in (
        judge_kwargs["prompt_header"])

    # 산출: 선정 변형 전문=seed_prompt + 저작·판정 감사 기록 (NARROW-5)
    assert g["seed_prompt"] == rp["B"]
    assert g["seed_variants"]["variants"][0]["variant_id"] == "v1"
    # [2026-08-01] 필수 그룹은 검색 선택 실사를 형태 참조로 물고 내려간다.
    # 그래서 생성에 넘어가는 prompt = 저작 브리프 + 참조 관할 계약이고,
    # `seed_brief` 는 **저작본 전문 그대로**여야 한다(감사 기준선).
    assert mr["prompt"].startswith(g["seed_brief"])
    assert "FORM REFERENCE" in mr["prompt"]
    assert g["seed_decision"]["ranking"] == ["B", "C", "A"]
    assert g["seed_decision"]["verdicts"][0]["label"] == "B"

    # sidecar: 저작본+record durable persist
    import json as _json
    from pathlib import Path as _Path

    records_path = (
        _Path(str(tmp_path)) / "SAMPLE_PROJECT" / "images"
        / "background_chain" / "seed_variant_records.json"
    )
    side = _json.loads(records_path.read_text())
    assert side["sample_site"]["author"]["variants"][0][
        "variant_id"] == "v1"
    assert side["sample_site"]["record"] == {"selected": "B"}
    assert side["sample_site"]["author_fp"]

    # ★[2026-08-01 Codex BLOCKING 2 후반] 씨드 자산의 계보.
    # 검색 선택 실사를 참조로 물고 그렸는데 input_image_ids 가 비어 있으면
    # 어느 참조에서 나온 씨드인지 추적할 수 없다.
    seed_call = step._upsert_seed_asset.call_args_list[0]
    assert seed_call.kwargs["input_image_ids"] == ["asset-sample_site"]


def test_variants_author_sidecar_reuse_skips_llm(tmp_path):
    """NARROW-5: 동일 입력 재실행=sidecar 저작본 재사용(LLM 재호출 0)."""
    calls = []

    def fake_call_structured(*args, **kwargs):
        # 사전조사(typology_*)는 저작과 다른 계약이다 — 구분하지 않으면
        # 대역이 저작 응답을 돌려주고, 저작 호출 수를 세는 아래 단언이
        # 오염된다. 결손 0 = 조사 없이 빈 블록(정상 경로).
        if str(args[0] if args else kwargs.get("step")).startswith(
                "typology"):
            return {"questions": []}
        calls.append(1)
        return _author_payload()

    def fake_multiroll(**kwargs):
        sel = kwargs["out_stem"].parent / (
            kwargs["out_stem"].name + "_sel.png")
        sel.parent.mkdir(parents=True, exist_ok=True)
        sel.write_bytes(b"img")
        return sel, {"selected": "A",
                     "roll_prompts": dict(kwargs["roll_prompts"])}

    def run_once():
        step = _make_step()
        cps = _cps()
        step._load_prev_checkpoint = lambda sid: cps.get(sid)
        step._upsert_seed_asset = MagicMock(side_effect=_asset_id_by_kind)
        p1, p2, p3 = _variants_patches()
        with p1, p2, p3, patch(
            "app.modules.llm.llm_client.call_structured",
            side_effect=fake_call_structured,
        ), patch(
            "app.modules.pipeline.multiroll_select.run_multiroll_select",
            side_effect=fake_multiroll,
        ), patch(
            "app.modules.pipeline.gpt_image_gen.make_gpt_image_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.core.config.settings.projects_dir", str(tmp_path),
        ):
            return step._execute()

    r1 = run_once()
    assert r1["completed_count"] == 1 and len(calls) == 1
    r2 = run_once()
    assert r2["completed_count"] == 1
    assert len(calls) == 1  # 재실행=저작 재사용


def test_variants_force_bypasses_sidecar_cache(tmp_path):
    """3차 BLOCKING-1(1): force=저작 캐시 우회+multiroll force 전달."""
    calls = []

    def fake_call_structured(*args, **kwargs):
        # 사전조사(typology_*)는 저작과 다른 계약이다 — 구분하지 않으면
        # 대역이 저작 응답을 돌려주고, 저작 호출 수를 세는 아래 단언이
        # 오염된다. 결손 0 = 조사 없이 빈 블록(정상 경로).
        if str(args[0] if args else kwargs.get("step")).startswith(
                "typology"):
            return {"questions": []}
        calls.append(1)
        return _author_payload()

    mr_forces = []

    def fake_multiroll(**kwargs):
        mr_forces.append(kwargs["force"])
        sel = kwargs["out_stem"].parent / (
            kwargs["out_stem"].name + "_sel.png")
        sel.parent.mkdir(parents=True, exist_ok=True)
        sel.write_bytes(b"img")
        return sel, {"selected": "A",
                     "roll_prompts": dict(kwargs["roll_prompts"])}

    def run_once(mode):
        step = _make_step()
        cps = _cps()
        step._load_prev_checkpoint = lambda sid: cps.get(sid)
        step._upsert_seed_asset = MagicMock(side_effect=_asset_id_by_kind)
        p1, p2, p3 = _variants_patches()
        with p1, p2, p3, patch(
            "app.modules.llm.llm_client.call_structured",
            side_effect=fake_call_structured,
        ), patch(
            "app.modules.pipeline.multiroll_select.run_multiroll_select",
            side_effect=fake_multiroll,
        ), patch(
            "app.modules.pipeline.gpt_image_gen.make_gpt_image_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.core.config.settings.projects_dir", str(tmp_path),
        ):
            return step._execute(mode)

    run_once("resume")
    assert len(calls) == 1 and mr_forces == [False]
    run_once("force")
    assert len(calls) == 2  # force=저작 캐시 무시, 재저작
    assert mr_forces == [False, True]


def test_config_hash_variants_roll_count_independent():
    """3차 MINOR-2: ON 시 스틸 롤 수 변경이 seed hash 를 안 흔든다."""
    step = _make_step()
    with patch(
        "app.core.config.settings.structure_seed_variants_enabled",
        True, create=True,
    ):
        with patch(
            "app.core.config.settings.still_recipe_roll_count", 2,
        ):
            on_2 = step._config_hash()
        with patch(
            "app.core.config.settings.still_recipe_roll_count", 4,
        ):
            on_4 = step._config_hash()
    assert on_2 == on_4
    # OFF(레거시)는 롤 수가 실질 입력 — 기존 동작 유지
    with patch("app.core.config.settings.still_recipe_roll_count", 2):
        off_2 = step._config_hash()
    with patch("app.core.config.settings.still_recipe_roll_count", 4):
        off_4 = step._config_hash()
    assert off_2 != off_4


def test_variants_author_retry_includes_prior_violations(tmp_path):
    """NARROW-4: 재시도 입력에 이전 위반 교정 블록 병기."""
    step = _make_step()
    cps = _cps()
    step._load_prev_checkpoint = lambda sid: cps.get(sid)
    step._upsert_seed_asset = MagicMock(side_effect=_asset_id_by_kind)

    bad = {"location_line_en": "", "conformance_brief_en": "",
           "variants": []}
    users = []

    def fake_call_structured(step_tag, sys_prompt, user_content, schema,
                             **kwargs):
        users.append(user_content)
        return bad if len(users) == 1 else _author_payload()

    def fake_multiroll(**kwargs):
        sel = kwargs["out_stem"].parent / (
            kwargs["out_stem"].name + "_sel.png")
        sel.parent.mkdir(parents=True, exist_ok=True)
        sel.write_bytes(b"img")
        return sel, {"selected": "A",
                     "roll_prompts": dict(kwargs["roll_prompts"])}

    p1, p2, p3 = _variants_patches()
    with p1, p2, p3, patch(
        "app.modules.llm.llm_client.call_structured",
        side_effect=fake_call_structured,
    ), patch(
        "app.modules.pipeline.multiroll_select.run_multiroll_select",
        side_effect=fake_multiroll,
    ), patch(
        "app.modules.pipeline.gpt_image_gen.make_gpt_image_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.core.config.settings.projects_dir", str(tmp_path),
    ):
        result = step._execute()

    assert result["completed_count"] == 1
    assert len(users) == 2
    assert "PREVIOUS ATTEMPT REJECTED" not in users[0]
    assert "PREVIOUS ATTEMPT REJECTED" in users[1]
    assert "결손" in users[1]  # 이전 위반 목록 병기


def test_config_hash_author_physical_model_sensitivity():
    """HIGH-3: alias 뒤 물리 모델(openai_model) 교체=ON 시 hash 드리프트."""
    step = _make_step()
    with patch(
        "app.core.config.settings.structure_seed_variants_enabled",
        True, create=True,
    ):
        with patch(
            "app.core.config.settings.openai_model", "SAMPLE-sol-a",
            create=True,
        ):
            h_a = step._config_hash()
        with patch(
            "app.core.config.settings.openai_model", "SAMPLE-sol-b",
            create=True,
        ):
            h_b = step._config_hash()
    assert h_a != h_b
    # OFF=물리 모델 무관 (기존 프로젝트 hash 불변)
    with patch(
        "app.core.config.settings.openai_model", "SAMPLE-sol-a",
        create=True,
    ):
        off_a = step._config_hash()
    with patch(
        "app.core.config.settings.openai_model", "SAMPLE-sol-b",
        create=True,
    ):
        off_b = step._config_hash()
    assert off_a == off_b


def test_variants_author_invalid_retry_then_fail_isolated(tmp_path):
    step = _make_step()
    cps = _cps()
    step._load_prev_checkpoint = lambda sid: cps.get(sid)

    bad = {"location_line_en": "", "conformance_brief_en": "",
           "variants": []}
    calls = []

    def fake_call_structured(*args, **kwargs):
        if str(args[0] if args else kwargs.get("step")).startswith(
                "typology"):
            return {"questions": []}
        calls.append(1)
        return bad

    p1, p2, p3 = _variants_patches()
    with p1, p2, p3, patch(
        "app.modules.llm.llm_client.call_structured",
        side_effect=fake_call_structured,
    ), patch(
        "app.modules.pipeline.gpt_image_gen.make_gpt_image_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.core.config.settings.projects_dir", str(tmp_path),
    ):
        result = step._execute()

    assert len(calls) == 2  # 1회 재시도 후 소진
    assert result["failed_count"] == 1
    g = result["data"]["groups"]["sample_site"]
    assert g["status"] == "failed"
    assert "변형 저작 검증 실패" in g["error"]


def test_variants_interior_scene_join_wiring(tmp_path):
    """v9 (Codex v8b ③ 배선 결손): 같은 building 실내 멤버(loc) 씬 원문이
    저작 입력 INTERIOR SCENES 블록으로 합류 + 그 씬 인용의 provenance
    통과 — 야외 씬만으로 방 전수를 못 세던 결손 봉합."""
    step = _make_step()
    cps = _cps()
    cps["scene_save"]["data"]["segments"].append(
        {"scene_index": 5, "heading": "INT. SAMPLE 내부 - 밤",
         "text": "안방과 자기방이 따로 있다."})
    cps["scene_director"] = {"data": {"scenes": [
        {"scene_index": 3, "primary_location": "L01"},
        {"scene_index": 5, "primary_location": "L02"},
    ]}}
    step._load_prev_checkpoint = lambda sid: cps.get(sid)
    step._upsert_seed_asset = MagicMock(side_effect=_asset_id_by_kind)

    users = []

    def fake_call_structured(step_tag, sys_prompt, user_content, schema,
                             **kwargs):
        users.append(user_content)
        payload = _author_payload()
        # 실내 씬(scene 5) 원문 인용 — validator 의 씬 단위 provenance
        # 대조를 실제로 통과해야 한다
        payload["room_inventory"].append(
            {"label_en": "SAMPLE bedroom", "scene_index": 5,
             "evidence_ko": "안방과 자기방"})
        return payload

    def fake_multiroll(**kwargs):
        sel = kwargs["out_stem"].parent / (
            kwargs["out_stem"].name + "_sel.png")
        sel.parent.mkdir(parents=True, exist_ok=True)
        sel.write_bytes(b"img")
        return sel, {"selected": "A",
                     "roll_prompts": dict(kwargs["roll_prompts"])}

    p1, p2, p3 = _variants_patches()
    with p1, p2, p3, patch(
        "app.modules.llm.llm_client.call_structured",
        side_effect=fake_call_structured,
    ), patch(
        "app.modules.pipeline.multiroll_select.run_multiroll_select",
        side_effect=fake_multiroll,
    ), patch(
        "app.modules.pipeline.gpt_image_gen.make_gpt_image_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.core.config.settings.projects_dir", str(tmp_path),
    ):
        result = step._execute()

    assert result["completed_count"] == 1
    assert len(users) == 1  # provenance 통과 = 재시도 없음
    u = users[0]
    assert "INTERIOR SCENES" in u
    assert "안방과 자기방이 따로 있다." in u
    # 야외 씬은 기존 블록에만
    assert u.index("SCENARIO SCENES") < u.index("INTERIOR SCENES")


def test_collect_interior_scene_indices_pure():
    from app.modules.pipeline.outdoor_structure_seed import (
        collect_interior_scene_indices,
    )

    bg = {"members": [
        {"loc_id": "L01", "is_indoor": False},
        {"loc_id": "L02", "is_indoor": True},
        {"loc_id": "L03"},  # 미분류=default-deny
    ]}
    mapping = {3: "L01", 5: "L02", 7: "L02", 9: "L09"}
    assert collect_interior_scene_indices(
        building_group=bg, scene_primary_loc=mapping) == [5, 7]
    # exclude=이미 공급된 씬 중복 제외
    assert collect_interior_scene_indices(
        building_group=bg, scene_primary_loc=mapping,
        exclude=[5]) == [7]
    assert collect_interior_scene_indices(
        building_group={}, scene_primary_loc=mapping) == []


def test_execute_rejects_malformed_formref_checkpoint_as_app_error():
    """★계약 검증이 **실제 경로**에서 실행되는지 (Codex 재리뷰 NARROW-2).

    helper 단위 테스트만 있으면 helper 는 malformed 를 AppError 로 세우는데
    _execute 는 그 helper 를 부르기 전에 `.get` 연쇄로 AttributeError 를 낸다.
    계약 위반이 "스텝 고장"으로 둔갑하고 422 대신 500 이 뜬다.
    """
    from app.core.errors import AppError

    step = _make_step()
    cps = _cps()
    cps["outdoor_structure_form_reference"] = ["not", "a", "dict"]
    step._load_prev_checkpoint = lambda sid: cps.get(sid)
    p1, p2 = _patches()
    with p1, p2, pytest.raises(AppError) as ei:
        step._execute()
    assert ei.value.code.startswith("structure_seed.form_reference")


# ── 유형 사전지식 배선 (2026-08-03) ──────────────────────────────────
#
# ★조사는 이 스텝이 하지 않는다. 앞 스텝(form_reference)이 **이미지 검색보다
#  먼저** 조사하고, 씨드는 그 기록을 읽어 저작 입력에 싣기만 한다. 예전에는
#  씨드가 직접 조사했는데 그 자리는 검색이 다 끝난 뒤라 조사 결과가 질의로
#  이어질 수 없었다.


def _typology_run(tmp_path, *, formref_typology=None):
    """공개 진입점으로 태운다 — helper 단언만으로는 도달을 못 세운다."""
    step = _make_step()
    cps = _cps()
    cps["outdoor_structure_form_reference"] = _formref_cp(
        typology=formref_typology)
    step._load_prev_checkpoint = lambda sid: cps.get(sid)
    step._upsert_seed_asset = MagicMock(side_effect=_asset_id_by_kind)
    users = []
    gap_calls = []

    def fake_call_structured(*args, **kwargs):
        users.append(args[2] if len(args) > 2 else kwargs.get("user_prompt"))
        return _author_payload()

    def fake_multiroll(**kwargs):
        sel = kwargs["out_stem"].parent / (
            kwargs["out_stem"].name + "_sel.png")
        sel.parent.mkdir(parents=True, exist_ok=True)
        sel.write_bytes(b"img")
        return sel, {"selected": "A",
                     "roll_prompts": dict(kwargs["roll_prompts"])}

    def counting_gap(**kw):
        gap_calls.append(1)
        return {"questions": []}

    gap_patch = patch(
        "app.modules.pipeline.typology_prior.extract_gap_questions",
        side_effect=counting_gap)
    p1, p2, p3 = _variants_patches()
    with p1, p2, p3, gap_patch, patch(
        "app.modules.llm.llm_client.call_structured",
        side_effect=fake_call_structured,
    ), patch(
        "app.modules.pipeline.multiroll_select.run_multiroll_select",
        side_effect=fake_multiroll,
    ), patch(
        "app.modules.pipeline.gpt_image_gen.make_gpt_image_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.core.config.settings.projects_dir", str(tmp_path),
    ):
        result = step._execute()
    return step, result, users, gap_calls


def test_typology_facts_from_prev_step_reach_author(tmp_path):
    """앞 스텝이 조사한 사실이 저작 입력에 도달한다."""
    marker = "MULTIPLES-OF-A-PERSON-MARKER"
    record = {"block": f"TYPOLOGY FACTS:\n- {marker}",
              "items": [{"adopted": True}], "questions": [{}]}
    _, result, users, gap_calls = _typology_run(
        tmp_path, formref_typology=record)

    assert result["completed_count"] == 1
    assert any(marker in (u or "") for u in users), (
        "사전조사 블록이 저작 입력에 도달하지 않았다")
    # ★씨드는 조사하지 않는다 — 돌면 앞 스텝과 두 벌이 되고 유료가 두 번이다.
    assert gap_calls == []


def test_missing_typology_record_does_not_kill_the_group(tmp_path):
    """앞 스텝 조사 기록이 없어도 빈 블록으로 계속 간다."""
    _, result, users, gap_calls = _typology_run(
        tmp_path, formref_typology=None)
    assert result["completed_count"] == 1 and result["failed_count"] == 0
    assert users, "저작이 호출되지 않았다"
    assert gap_calls == []