"""W20B: shot_aware_bg_render_plan_llm_provider tests.

direct litellm.completion single-call surface — monkeypatched in every
test, so NO real network call ever happens. LLM / image / VLM API call 0.

Covers:
- preflight fail-closed: malformed args, OPENAI_API_KEY empty,
  prompt pack missing/empty, capability preflight (supports_response_schema,
  response_format) all raise BEFORE completion.
- happy path: completion called exactly once, model lock, strict
  json_schema response_format pointing to sanitized schema, no
  temperature, num_retries=0, max_completion_tokens=6000.
- sanitized schema strips every OpenAI-strict-mode unsupported key
  (minLength / minItems / maxItems / minimum / maximum / uniqueItems /
  pattern / format / anyOf / oneOf / allOf / etc).
- response guards: empty choices, missing message, truthy refusal,
  empty content, finish_reason != 'stop', malformed JSON, local
  jsonschema fail, production validate_llm_output fail.
- return shape: parsed graph dict ready for build_render_plan_for_fp.
"""
from __future__ import annotations

import json as _json
from pathlib import Path
from types import SimpleNamespace
from typing import Any, Dict
from unittest.mock import MagicMock

import jsonschema
import pytest

from app.modules.pipeline.shot_aware_bg_render_plan_llm_provider import (
    MAX_COMPLETION_TOKENS_DEFAULT,
    PLANNER_MODEL_DEFAULT,
    PROMPT_VERSION,
    RESPONSE_SCHEMA_NAME,
    TIMEOUT_SECONDS_DEFAULT,
    ShotAwareBgRenderPlanProviderError,
    _OPENAI_UNSUPPORTED_SCHEMA_KEYS,
    _PROMPT_PACK_DIR,
    _load_prompt_pack,
    _sanitize_schema_for_openai_strict,
    litellm_shot_aware_bg_render_plan_provider,
)


_RENDER_GUIDANCE_FIELDS = (
    "visible_space_directive",
    "camera_framing_directive",
    "subject_position_directive",
    "state_cue_directive",
    "negative_continuity_directive",
)


def _render_guidance_fixture() -> Dict[str, str]:
    return {f: f"directive for {f}" for f in _RENDER_GUIDANCE_FIELDS}


# ─────────────────────────────── fixtures ───────────────────────────────


def _dossier() -> Dict[str, Any]:
    return {
        "fp_id": "fp_a",
        "fp_image_path": "/p/c/e/floor_plan_render/fp_a.png",
        "grid_size": [10, 10],
        "base_marker_inventory": [
            {"number": 1, "label": "u1", "category": "area",
             "position_hint": "",
             "base_layer_decision": "base_structural_unit"},
            {"number": 2, "label": "u2", "category": "area",
             "position_hint": "",
             "base_layer_decision": "base_structural_unit"},
        ],
        "per_bg_render_facts_by_bg_id": {
            "L01B01": {
                "bg_id": "L01B01", "fp_id": "fp_a",
                "target_unit_marker_numbers": [1],
                "dominant_target_unit_marker_number": 1,
                "use_numbered_elements": [1],
                "ignore_numbered_elements": [],
                "base_marker_numbers_to_reference": [1],
                "transient_marker_numbers_to_describe": [],
                "ignored_state_overlay_marker_numbers": [],
                "clean_background_expected": True,
                "applies_to_shots": ["S1_Shot1"],
                "depends_on_bg": [],
                "diagnostics": [],
            },
        },
        # W20E7-C: clean anchor candidate surface (L01B01 is the only
        # clean_background_expected=True bg in this fixture).
        "anchor_selection_metadata": {
            "candidate_bg_ids": ["L01B01"],
            "selection_diagnostics": [],
            "selected_anchor_bg_id": None,
        },
    }


def _geometry() -> Dict[str, Any]:
    return {
        "fp_id": "fp_a",
        "grid_size": [10, 10],
        "readback_status": "ok",
        "camera_cell_candidates_per_unit": {"1": [[0, 0]]},
        "look_at_cell_candidates_per_unit": {"1": [[2, 0]]},
    }


def _shot_readiness() -> Dict[str, Any]:
    return {
        "ok": True,
        "blockers": [],
        "per_bg": {
            "L01B01": {
                "applies_to_shots": ["S1_Shot1"],
                "surfaced_shots": [
                    {"shot_id": "S1_Shot1", "staging": {"shot_id": "S1_Shot1"}}
                ],
                "ok": True,
                "blockers": [],
            },
        },
    }


def _good_graph_payload() -> Dict[str, Any]:
    return {
        "graph": {
            "nodes": [
                {
                    "bg_id": "L01B01",
                    "node_index": 0,
                    "mode": "fp_seeded_anchor",
                    "is_dwelling_identity_anchor": True,
                    "rationale": "anchor",
                    "reference_decision": {
                        "selected_refs": [],
                        "rejected_refs": [],
                        "same_physical_space_dedup_decision": "single_ref",
                        "why_single_ref_or_two_refs": "anchor mode",
                        "physical_space_id_per_ref": [],
                        # W21B-w3 Commit 1: v2 same-space low-delta reuse
                        # candidate 신호 (anchor 라 후보 아님 → empty sentinel).
                        "same_physical_space_low_delta_candidate": False,
                        "low_delta_reuse_target_bg_id": "",
                        "low_delta_rationale": "",
                    },
                    "camera_decision": {
                        "camera_unit": 1,
                        "camera_cell": [0, 0],
                        "look_at_unit": 1,
                        "look_at_cell": [2, 0],
                        "lens_enum": "normal",
                        "fov_deg": 50,
                        "framing_notes": "anchor framing",
                    },
                    "render_guidance": _render_guidance_fixture(),
                }
            ]
        }
    }


def _good_response_content() -> str:
    return _json.dumps(_good_graph_payload())


def _make_fake_response(
    *,
    content: str | None = None,
    finish_reason: str = "stop",
    refusal: str | None = None,
    choices: list | None = None,
    message: object | None = None,
):
    if choices is None:
        if message is None:
            msg = SimpleNamespace(content=content, refusal=refusal)
        else:
            msg = message
        ch = SimpleNamespace(message=msg, finish_reason=finish_reason)
        choices = [ch]
    return SimpleNamespace(choices=choices)


def _stub_preflight(
    monkeypatch,
    *,
    supports_schema: bool = True,
    supported_params: list[str] | None = None,
    openai_key: str | None = "sk-test-do-not-call-real-api",
) -> None:
    import litellm
    if openai_key is None:
        # [2026-08-01] 키 유무의 권위가 환경변수에서 슬롯 브로커로 옮겼다 —
        # 보조 슬롯만 있어도 '있음'이므로 env 만 지우면 fail-closed 가 아니다.
        monkeypatch.delenv("OPENAI_API_KEY", raising=False)
        monkeypatch.setattr(
            "app.core.config.settings.openai_api_key", "", raising=False)
        monkeypatch.setattr(
            "app.core.config.settings.openai_api_key_secondary", "",
            raising=False)
    else:
        # 슬롯도 같은 값으로 세운다 — 공백 키가 "있음"으로 통과하면 이 테스트가
        # 검사하려던 fail-closed 자체가 성립하지 않는다.
        monkeypatch.setenv("OPENAI_API_KEY", openai_key)
        monkeypatch.setattr(
            "app.core.config.settings.openai_api_key", openai_key,
            raising=False)
        monkeypatch.setattr(
            "app.core.config.settings.openai_api_key_secondary", "",
            raising=False)
    monkeypatch.setattr(
        litellm,
        "supports_response_schema",
        lambda **kw: supports_schema,
        raising=False,
    )
    monkeypatch.setattr(
        litellm,
        "get_supported_openai_params",
        lambda **kw: list(
            supported_params
            if supported_params is not None
            else ["response_format", "max_completion_tokens", "timeout"]
        ),
        raising=False,
    )


def _stub_completion(monkeypatch, *, response) -> MagicMock:
    import litellm
    mock = MagicMock(return_value=response)
    monkeypatch.setattr(litellm, "completion", mock, raising=False)
    return mock


def _call(**overrides) -> Dict[str, Any]:
    kwargs = dict(
        fp_id="fp_a",
        dossier=_dossier(),
        geometry=_geometry(),
        shot_readiness=_shot_readiness(),
        candidate_catalog=[],
    )
    kwargs.update(overrides)
    return litellm_shot_aware_bg_render_plan_provider(**kwargs)


# ─────────────────────────── arg-shape preflight ───────────────────────────


@pytest.mark.parametrize(
    "field,bad",
    [
        ("fp_id", ""),
        ("fp_id", None),
        ("fp_id", 123),
        ("dossier", None),
        ("dossier", "string"),
        ("geometry", None),
        ("geometry", "string"),
        ("shot_readiness", None),
        ("shot_readiness", "string"),
        ("candidate_catalog", "string"),
        ("candidate_catalog", {"k": "v"}),
    ],
)
def test_provider_rejects_malformed_args_before_network(
    monkeypatch, field, bad
):
    _stub_preflight(monkeypatch)
    mock = _stub_completion(
        monkeypatch,
        response=_make_fake_response(content=_good_response_content()),
    )
    with pytest.raises(ShotAwareBgRenderPlanProviderError):
        _call(**{field: bad})
    assert mock.call_count == 0


# ─────────────────────────── env preflight ───────────────────────────


def test_provider_fails_closed_when_openai_api_key_missing(monkeypatch):
    _stub_preflight(monkeypatch, openai_key=None)
    mock = _stub_completion(
        monkeypatch,
        response=_make_fake_response(content=_good_response_content()),
    )
    with pytest.raises(ShotAwareBgRenderPlanProviderError) as exc:
        _call()
    assert "OPENAI_API_KEY" in str(exc.value)
    assert mock.call_count == 0


def test_provider_fails_closed_when_openai_api_key_blank(monkeypatch):
    _stub_preflight(monkeypatch, openai_key="   ")
    mock = _stub_completion(
        monkeypatch,
        response=_make_fake_response(content=_good_response_content()),
    )
    with pytest.raises(ShotAwareBgRenderPlanProviderError):
        _call()
    assert mock.call_count == 0


# ─────────────────────────── prompt pack preflight ───────────────────────────


def test_provider_fails_closed_when_prompt_pack_dir_missing(monkeypatch):
    import app.modules.pipeline.shot_aware_bg_render_plan_llm_provider as prov
    _stub_preflight(monkeypatch)
    mock = _stub_completion(
        monkeypatch,
        response=_make_fake_response(content=_good_response_content()),
    )
    monkeypatch.setattr(
        prov, "_PROMPT_PACK_DIR", Path("/tmp/__nonexistent_pack_dir__")
    )
    with pytest.raises(ShotAwareBgRenderPlanProviderError) as exc:
        _call()
    assert "prompt pack" in str(exc.value)
    assert mock.call_count == 0


def test_provider_fails_closed_when_prompt_pack_file_missing(
    monkeypatch, tmp_path
):
    import app.modules.pipeline.shot_aware_bg_render_plan_llm_provider as prov
    _stub_preflight(monkeypatch)
    mock = _stub_completion(
        monkeypatch,
        response=_make_fake_response(content=_good_response_content()),
    )
    # Create pack dir but omit schema.json.
    (tmp_path / "system.md").write_text("system", encoding="utf-8")
    (tmp_path / "user_template.md").write_text(
        "u {fp_id} {readback_status} {dossier_block} {geometry_block} "
        "{per_bg_facts_block} {shot_staging_block} "
        "{candidate_catalog_block}",
        encoding="utf-8",
    )
    monkeypatch.setattr(prov, "_PROMPT_PACK_DIR", tmp_path)
    with pytest.raises(ShotAwareBgRenderPlanProviderError):
        _call()
    assert mock.call_count == 0


# ─────────────────────────── litellm capability preflight ───────────────


def test_provider_fails_closed_when_supports_response_schema_false(monkeypatch):
    _stub_preflight(monkeypatch, supports_schema=False)
    mock = _stub_completion(
        monkeypatch,
        response=_make_fake_response(content=_good_response_content()),
    )
    with pytest.raises(ShotAwareBgRenderPlanProviderError) as exc:
        _call()
    assert "response_schema" in str(exc.value)
    assert mock.call_count == 0


def test_provider_fails_closed_when_response_format_not_supported(monkeypatch):
    _stub_preflight(monkeypatch, supported_params=["timeout"])  # no response_format
    mock = _stub_completion(
        monkeypatch,
        response=_make_fake_response(content=_good_response_content()),
    )
    with pytest.raises(ShotAwareBgRenderPlanProviderError) as exc:
        _call()
    assert "response_format" in str(exc.value)
    assert mock.call_count == 0


# ─────────────────────────── happy path completion ───────────────────────


def test_provider_invokes_completion_exactly_once(monkeypatch):
    _stub_preflight(monkeypatch)
    fake = _make_fake_response(content=_good_response_content())
    mock = _stub_completion(monkeypatch, response=fake)
    result = _call()
    assert mock.call_count == 1
    assert "graph" in result


def test_provider_request_uses_locked_model_and_kwargs(monkeypatch):
    _stub_preflight(monkeypatch)
    fake = _make_fake_response(content=_good_response_content())
    mock = _stub_completion(monkeypatch, response=fake)
    _call()
    kwargs = mock.call_args.kwargs
    assert kwargs["model"] == PLANNER_MODEL_DEFAULT
    assert kwargs["max_completion_tokens"] == MAX_COMPLETION_TOKENS_DEFAULT
    assert kwargs["timeout"] == TIMEOUT_SECONDS_DEFAULT
    assert kwargs["num_retries"] == 0
    # Temperature MUST NOT be passed (router temperature/strict-schema bug).
    assert "temperature" not in kwargs
    rf = kwargs["response_format"]
    assert rf["type"] == "json_schema"
    inner = rf["json_schema"]
    assert inner["name"] == RESPONSE_SCHEMA_NAME
    assert inner["strict"] is True
    assert "schema" in inner


def test_provider_messages_include_system_and_user_sections(monkeypatch):
    _stub_preflight(monkeypatch)
    fake = _make_fake_response(content=_good_response_content())
    mock = _stub_completion(monkeypatch, response=fake)
    _call()
    messages = mock.call_args.kwargs["messages"]
    roles = [m["role"] for m in messages]
    assert "system" in roles
    assert "user" in roles
    system_text = next(m["content"] for m in messages if m["role"] == "system")
    user_text = next(m["content"] for m in messages if m["role"] == "user")
    assert isinstance(system_text, str) and system_text.strip()
    assert isinstance(user_text, str) and user_text.strip()
    # user prompt carries the structured fp_id verbatim.
    assert "fp_a" in user_text


def test_provider_returns_parsed_graph_dict_for_build_render_plan(monkeypatch):
    _stub_preflight(monkeypatch)
    fake = _make_fake_response(content=_good_response_content())
    _stub_completion(monkeypatch, response=fake)
    result = _call()
    nodes = (result.get("graph") or {}).get("nodes") or []
    assert isinstance(nodes, list) and len(nodes) == 1
    assert nodes[0]["bg_id"] == "L01B01"
    assert nodes[0]["is_dwelling_identity_anchor"] is True


# ─────────────────────────── sanitized response schema ──────────────────


def _walk_schema_for_keys(node, banned) -> list[str]:
    leaks: list[str] = []
    if isinstance(node, dict):
        for k, v in node.items():
            if k in banned:
                leaks.append(k)
            leaks.extend(_walk_schema_for_keys(v, banned))
    elif isinstance(node, list):
        for item in node:
            leaks.extend(_walk_schema_for_keys(item, banned))
    return leaks


def test_sanitized_schema_has_no_openai_strict_unsupported_keys(monkeypatch):
    _stub_preflight(monkeypatch)
    fake = _make_fake_response(content=_good_response_content())
    mock = _stub_completion(monkeypatch, response=fake)
    _call()
    sent_schema = mock.call_args.kwargs["response_format"]["json_schema"]["schema"]
    leaks = _walk_schema_for_keys(sent_schema, _OPENAI_UNSUPPORTED_SCHEMA_KEYS)
    assert not leaks, (
        f"sanitized response schema still contains OpenAI-strict-mode "
        f"unsupported keys: {sorted(set(leaks))}"
    )


def test_original_pack_schema_keeps_unsupported_keys_for_local_validation(
    monkeypatch,
):
    """The original prompt-pack schema is the local jsonschema SOT and
    MUST retain its minLength/minItems/maxItems/enum constraints; the
    sanitized form is a separate copy used only for the LLM request."""
    pack = _load_prompt_pack()
    raw = pack["schema"]
    leaks = _walk_schema_for_keys(raw, _OPENAI_UNSUPPORTED_SCHEMA_KEYS)
    # We expect at least minLength / minItems / maxItems somewhere in the
    # original schema — they are present in the v1 pack.
    assert leaks, (
        "original prompt-pack schema is missing the constraint keys "
        "that the sanitized copy strips — local jsonschema validation "
        "would lose its bite"
    )


def test_sanitize_does_not_mutate_input():
    raw = {
        "type": "object",
        "properties": {
            "x": {"type": "string", "minLength": 1},
            "ys": {"type": "array", "maxItems": 2,
                   "items": {"type": "integer", "minimum": 0}},
        },
        "anyOf": [{"required": ["x"]}],
    }
    original_copy = _json.loads(_json.dumps(raw))
    sanitized = _sanitize_schema_for_openai_strict(raw)
    assert raw == original_copy  # input untouched
    leaks = _walk_schema_for_keys(sanitized, _OPENAI_UNSUPPORTED_SCHEMA_KEYS)
    assert not leaks


# ─────────────── OpenAI strict required-all (Codex patch A) ───────────────


def _walk_object_required_pairs(node, path: str = "$"):
    """Yield ``(path, props_set, required_set)`` for every dict in ``node``
    that looks like an object schema (``type == 'object'`` AND a dict
    ``properties`` map). Recurses through nested dicts and lists."""
    if isinstance(node, dict):
        if node.get("type") == "object" and isinstance(
            node.get("properties"), dict
        ):
            props = set(node["properties"].keys())
            required = set(node.get("required") or [])
            yield path, props, required
        for k, v in node.items():
            yield from _walk_object_required_pairs(v, f"{path}.{k}")
    elif isinstance(node, list):
        for i, item in enumerate(node):
            yield from _walk_object_required_pairs(item, f"{path}[{i}]")


def test_sanitized_schema_requires_every_property_for_strict_mode():
    """OpenAI Structured Outputs strict 모드는 모든 object property를
    ``required`` 에 포함해야 함. sanitized request copy 는 prompt pack
    원본의 optional field 를 모두 채워서 strict request 가 모델 호출
    전에 schema validation 으로 reject 되지 않도록 보강."""
    pack = _load_prompt_pack()
    sanitized = _sanitize_schema_for_openai_strict(pack["schema"])
    mismatches = []
    for path, props, required in _walk_object_required_pairs(sanitized):
        if props != required:
            mismatches.append(
                f"{path}: properties={sorted(props)} required={sorted(required)}"
            )
    assert not mismatches, (
        "sanitized schema has objects whose required != properties: "
        + "; ".join(mismatches)
    )


def test_sanitized_schema_selected_refs_item_requires_space_description():
    """Specific regression — original pack schema marks
    ``space_description`` optional; sanitized request copy must add it
    to required so strict mode does not reject the response_format."""
    pack = _load_prompt_pack()
    sanitized = _sanitize_schema_for_openai_strict(pack["schema"])
    item_schema = (
        sanitized["properties"]["graph"]
        ["properties"]["nodes"]
        ["items"]
        ["properties"]["reference_decision"]
        ["properties"]["selected_refs"]
        ["items"]
    )
    assert "space_description" in item_schema.get("required", [])


def test_sanitized_schema_camera_decision_requires_framing_notes():
    """Specific regression — original pack schema marks
    ``framing_notes`` optional; sanitized request copy must add it to
    required."""
    pack = _load_prompt_pack()
    sanitized = _sanitize_schema_for_openai_strict(pack["schema"])
    cam = (
        sanitized["properties"]["graph"]
        ["properties"]["nodes"]
        ["items"]
        ["properties"]["camera_decision"]
    )
    assert "framing_notes" in cam.get("required", [])


def test_original_pack_schema_keeps_space_description_optional():
    """Local jsonschema SOT must keep optional fields optional. The
    sanitized copy is a separate tree — verifying the original is
    untouched is what lets ``space_description: null`` and
    ``framing_notes`` omissions pass local validation when an upstream
    consumer chooses to skip them."""
    pack = _load_prompt_pack()
    item_schema = (
        pack["schema"]["properties"]["graph"]
        ["properties"]["nodes"]
        ["items"]
        ["properties"]["reference_decision"]
        ["properties"]["selected_refs"]
        ["items"]
    )
    assert "space_description" not in (item_schema.get("required") or [])


def test_original_pack_schema_keeps_framing_notes_optional():
    pack = _load_prompt_pack()
    cam = (
        pack["schema"]["properties"]["graph"]
        ["properties"]["nodes"]
        ["items"]
        ["properties"]["camera_decision"]
    )
    assert "framing_notes" not in (cam.get("required") or [])


# ─────────────────────────── response guards ───────────────────────────


@pytest.mark.parametrize(
    "case",
    [
        "malformed_json",
        "refusal",
        "empty_content",
        "finish_reason_length",
        "finish_reason_content_filter",
        "empty_choices",
        "missing_message",
        "non_dict_json",
        "local_jsonschema_fail",
        "validator_fail",
    ],
)
def test_provider_raises_on_bad_response(monkeypatch, case):
    _stub_preflight(monkeypatch)
    good = _good_response_content()
    if case == "malformed_json":
        resp = _make_fake_response(content="this is not json")
    elif case == "refusal":
        resp = _make_fake_response(content=good, refusal="cannot help")
    elif case == "empty_content":
        resp = _make_fake_response(content="")
    elif case == "finish_reason_length":
        resp = _make_fake_response(content=good, finish_reason="length")
    elif case == "finish_reason_content_filter":
        resp = _make_fake_response(
            content=good, finish_reason="content_filter"
        )
    elif case == "empty_choices":
        resp = _make_fake_response(choices=[])
    elif case == "missing_message":
        ch = SimpleNamespace(finish_reason="stop")  # no .message attr → None
        resp = _make_fake_response(choices=[ch])
    elif case == "non_dict_json":
        resp = _make_fake_response(content=_json.dumps([1, 2, 3]))
    elif case == "local_jsonschema_fail":
        # Valid JSON but graph nodes are missing required fields per the
        # original prompt-pack schema (minLength on rationale enforces
        # non-empty string).
        broken = _good_graph_payload()
        broken["graph"]["nodes"][0]["rationale"] = ""
        resp = _make_fake_response(content=_json.dumps(broken))
    elif case == "validator_fail":
        # Passes local jsonschema but the camera_unit references a unit
        # not in the geometry candidates → production validator fails.
        bad = _good_graph_payload()
        bad["graph"]["nodes"][0]["camera_decision"]["camera_unit"] = 999
        resp = _make_fake_response(content=_json.dumps(bad))
    else:
        raise AssertionError(f"unknown case: {case}")

    _stub_completion(monkeypatch, response=resp)
    with pytest.raises(ShotAwareBgRenderPlanProviderError):
        _call()


# ─────────────────────────── boundary surface ───────────────────────────


def test_module_constants_locked():
    assert PLANNER_MODEL_DEFAULT == "openai/gpt-6-astra"
    assert MAX_COMPLETION_TOKENS_DEFAULT == 16000  # W20F7-A: 6000→16000 (gpt-5.5 reasoning budget)
    assert TIMEOUT_SECONDS_DEFAULT == 180
    assert PROMPT_VERSION == "3.202607231435"  # W20F10 graph/anchor 교정 retry 스템
    assert RESPONSE_SCHEMA_NAME == "shot_aware_bg_render_plan_v2"


def test_prompt_pack_dir_points_to_new_version():
    """W21B-w3 v3 hotfix: prompt pack discovery resolves to the
    same-major/new-timestamp low-delta calibration directory
    (2.202605291800). The schema is identical to the 2.202605291200
    pack (wording-only hotfix); the earlier 1.x packs and the
    pre-hotfix 2.202605291200 pack stay on disk untouched for
    traceability — no in-place overwrite."""
    assert _PROMPT_PACK_DIR.name == "3.202607231435"
    assert _PROMPT_PACK_DIR.is_dir()
    # W20F10: 교정 스템 실재 (팩 로더 fail-closed 대상)
    assert (_PROMPT_PACK_DIR / "graph_anchor_repair.md").is_file()
    legacy_w20f9 = _PROMPT_PACK_DIR.parent / "2.202605291800"
    assert legacy_w20f9.is_dir()
    # Older packs must NOT be deleted — keep on disk.
    legacy_v1 = _PROMPT_PACK_DIR.parent / "1.202605271200"
    legacy_v2 = _PROMPT_PACK_DIR.parent / "1.202605271800"
    legacy_v8 = _PROMPT_PACK_DIR.parent / "1.202605281200"
    legacy_v2_prehotfix = _PROMPT_PACK_DIR.parent / "2.202605291200"
    assert legacy_v1.is_dir()
    assert legacy_v2.is_dir()
    assert legacy_v8.is_dir()
    assert legacy_v2_prehotfix.is_dir()


# ────────── W20D render_guidance schema lock-ins (new prompt version) ──────────


def _nodes_item_schema(schema: Dict[str, Any]) -> Dict[str, Any]:
    return (
        schema["properties"]["graph"]
        ["properties"]["nodes"]
        ["items"]
    )


def test_original_pack_schema_requires_render_guidance_on_each_node():
    pack = _load_prompt_pack()
    item = _nodes_item_schema(pack["schema"])
    assert "render_guidance" in item["properties"]
    assert "render_guidance" in item["required"]
    rg = item["properties"]["render_guidance"]
    assert rg["type"] == "object"
    assert rg["additionalProperties"] is False
    for field in _RENDER_GUIDANCE_FIELDS:
        assert field in rg["properties"]
        assert rg["properties"][field]["type"] == "string"
        assert rg["properties"][field]["minLength"] == 1
        assert field in rg["required"]


def test_sanitized_schema_keeps_render_guidance_required_for_strict_mode():
    pack = _load_prompt_pack()
    sanitized = _sanitize_schema_for_openai_strict(pack["schema"])
    item = _nodes_item_schema(sanitized)
    assert "render_guidance" in item["properties"]
    assert "render_guidance" in item["required"]
    rg = item["properties"]["render_guidance"]
    # strict mode: every property must be required
    assert set(rg["properties"].keys()) == set(rg["required"])
    assert set(rg["required"]) == set(_RENDER_GUIDANCE_FIELDS)


def test_local_validation_rejects_node_missing_render_guidance(monkeypatch):
    """When the LLM emits a payload that is valid JSON but omits
    ``render_guidance``, the provider's local jsonschema validation
    (against the ORIGINAL pack schema) must fail before the production
    validators run."""
    _stub_preflight(monkeypatch)
    bad = _good_graph_payload()
    bad["graph"]["nodes"][0].pop("render_guidance")
    _stub_completion(
        monkeypatch,
        response=_make_fake_response(content=_json.dumps(bad)),
    )
    with pytest.raises(ShotAwareBgRenderPlanProviderError):
        _call()


def test_local_validation_rejects_render_guidance_field_empty(monkeypatch):
    _stub_preflight(monkeypatch)
    bad = _good_graph_payload()
    bad["graph"]["nodes"][0]["render_guidance"][
        "visible_space_directive"
    ] = ""
    _stub_completion(
        monkeypatch,
        response=_make_fake_response(content=_json.dumps(bad)),
    )
    with pytest.raises(ShotAwareBgRenderPlanProviderError):
        _call()


def test_unsupported_schema_key_set_contains_router_problem_keys():
    must_have = {
        "minLength", "minItems", "maxItems",
        "minimum", "maximum", "uniqueItems",
        "pattern", "format", "anyOf", "oneOf",
    }
    leaks = must_have - _OPENAI_UNSUPPORTED_SCHEMA_KEYS
    assert not leaks, (
        f"_OPENAI_UNSUPPORTED_SCHEMA_KEYS is missing entries that the "
        f"sanitizer must strip: {sorted(leaks)}"
    )


def test_prompt_pack_loads_three_artifacts():
    pack = _load_prompt_pack()
    assert "system_prompt" in pack
    assert "user_template" in pack
    assert "schema" in pack
    assert isinstance(pack["system_prompt"], str) and pack["system_prompt"].strip()
    assert isinstance(pack["user_template"], str) and pack["user_template"].strip()
    assert isinstance(pack["schema"], dict)
    # schema must be schema-shaped (no jsonschema parsing here — just
    # the type lock).
    assert pack["schema"].get("type") == "object"
    # The local-validation SOT must work as a Draft202012Validator.
    jsonschema.Draft202012Validator(pack["schema"])


# ─────────────── safety: no banned imports in provider module ───────────────


_ALLOWED_LLM_IMPORTS = {"litellm"}  # provider must lazy-import litellm
_BANNED_IMPORT_TOKENS = {
    "openai",
    "anthropic",
    "google.generativeai",
    "fal",
    "call_structured",
    "call_multiturn",
    "images.edit",
    "ImageAsset",
}


def test_provider_module_does_not_import_call_structured_or_router():
    import ast as _ast
    mod_path = (
        Path(__file__).resolve().parent.parent.parent
        / "app" / "modules" / "pipeline"
        / "shot_aware_bg_render_plan_llm_provider.py"
    )
    tokens: set[str] = set()
    tree = _ast.parse(mod_path.read_text(encoding="utf-8"))
    for node in _ast.walk(tree):
        if isinstance(node, _ast.Import):
            for alias in node.names:
                tokens.add(alias.name)
        elif isinstance(node, _ast.ImportFrom):
            if node.module:
                tokens.add(node.module)
            for alias in node.names:
                tokens.add(alias.name)
    leaks = tokens & _BANNED_IMPORT_TOKENS
    assert not leaks, (
        f"provider module must not import call_structured / router / "
        f"non-litellm provider SDKs; found: {sorted(leaks)}"
    )


# ───────── W21B: camera candidate deterministic snap (final fallback) ─────────


def _graph_payload_with_invalid_look_at() -> Dict[str, Any]:
    payload = _good_graph_payload()
    # [9,9] is outside look_at_cell_candidates_per_unit["1"] == [[2,0]].
    payload["graph"]["nodes"][0]["camera_decision"]["look_at_cell"] = [9, 9]
    return payload


def test_provider_snaps_camera_cell_after_w20f9_retry_still_invalid(monkeypatch):
    """When the first pass AND the W20F9 re-prompt both emit an
    out-of-candidate look_at_cell (camera-only failure each time), the
    provider deterministically snaps to the nearest candidate instead of
    hard-failing the fp, and surfaces the repair in retry metadata."""
    _stub_preflight(monkeypatch)
    invalid = _make_fake_response(
        content=_json.dumps(_graph_payload_with_invalid_look_at())
    )
    mock = _stub_completion(monkeypatch, response=invalid)
    # first pass + W20F9 retry both return the same out-of-candidate cell.
    mock.side_effect = [invalid, invalid]

    out = _call()  # _geometry(): readback ok + look_at cand [[2,0]]

    node = out["graph"]["nodes"][0]
    assert node["camera_decision"]["look_at_cell"] == [2, 0]  # snapped
    meta = out["_w20f9_retry_metadata"]
    assert meta["camera_validator_retry_attempted"] == 1
    assert meta["camera_validator_retry_succeeded"] == 0
    repairs = meta["camera_candidate_snap_repairs"]
    assert any(
        r["field"] == "look_at_cell" and r["from"] == [9, 9] and r["to"] == [2, 0]
        for r in repairs
    )
    # truncation accounting (silent-repair transparency).
    assert meta["camera_candidate_snap_repairs_total"] == 1
    assert meta["camera_candidate_snap_repairs_truncated"] is False
    # exactly two LLM calls (first + W20F9); the snap adds no extra call.
    assert mock.call_count == 2


def test_provider_does_not_snap_when_synthetic_readback_also_fails(monkeypatch):
    """Snap must NEVER mask a non-camera defect. With a synthetic readback,
    synthetic_readback_production_clear fails alongside camera, so it is
    not a camera-only failure: W20F9 does not fire, snap does not apply,
    and the provider hard-fails (fail-closed)."""
    _stub_preflight(monkeypatch)
    invalid = _make_fake_response(
        content=_json.dumps(_graph_payload_with_invalid_look_at())
    )
    mock = _stub_completion(monkeypatch, response=invalid)
    geom = _geometry()
    geom["readback_status"] = "synthetic_fixture"

    with pytest.raises(ShotAwareBgRenderPlanProviderError):
        _call(geometry=geom)

    # no W20F9 retry (not camera-only) and no snap masking.
    assert mock.call_count == 1


# ─────────────── W20F10 — graph/anchor targeted retry (2026-07-23) ───────────────
# Codex 합의 조건: 분류=validator boolean+노드 구조(문자열 검색 금지),
# 총 completion 상한=2(W20F9 와 상호 배타·연쇄 금지), 교정 입력=팩 스템
# (진단·기대 bg·anchor 후보·이전 응답 전문), 소진=fail-closed(결정론
# fallback 금지), 실패 예외에 completion_call_count 보존.


def _empty_graph_content() -> str:
    return _json.dumps({"graph": {"nodes": []}})


def _anchorless_content() -> str:
    payload = _good_graph_payload()
    payload["graph"]["nodes"][0]["is_dwelling_identity_anchor"] = False
    payload["graph"]["nodes"][0]["mode"] = "same_physical_space_view"
    return _json.dumps(payload)


def _stub_completion_seq(monkeypatch, responses) -> MagicMock:
    import litellm

    mock = MagicMock(side_effect=list(responses))
    monkeypatch.setattr(litellm, "completion", mock, raising=False)
    return mock


def test_w20f10_empty_graph_retry_succeeds_total_two_calls(monkeypatch):
    """실측 재현: 빈 그래프 → 교정 1회 → 성공. 교정 프롬프트에 진단·기대
    bg·anchor 후보·이전 응답이 구조적으로 포함."""
    _stub_preflight(monkeypatch)
    mock = _stub_completion_seq(monkeypatch, [
        _make_fake_response(content=_empty_graph_content()),
        _make_fake_response(content=_good_response_content()),
    ])
    out = _call()
    assert mock.call_count == 2
    meta = out["_graph_anchor_retry_metadata"]
    assert meta["graph_anchor_validator_retry_attempted"] == 1
    assert meta["graph_anchor_validator_retry_succeeded"] == 1
    assert meta["first_pass_diagnostics"]
    repair_prompt = mock.call_args_list[1].kwargs["messages"][1]["content"]
    assert "W20F10" in repair_prompt
    assert '"L01B01"' in repair_prompt            # 기대 renderable bg 목록
    assert "is_dwelling_identity_anchor" in repair_prompt
    assert "graph.nodes must be a non-empty list" in repair_prompt  # 진단 전문
    assert _empty_graph_content() in repair_prompt  # 이전 응답 전문
    # W20F9 카메라 메타 키 재사용 금지
    assert "_w20f9_retry_metadata" not in out


def test_w20f10_anchorless_nonempty_graph_retry_succeeds(monkeypatch):
    """non-empty graph + anchor-only 실패(anchor 0)도 교정 대상."""
    _stub_preflight(monkeypatch)
    mock = _stub_completion_seq(monkeypatch, [
        _make_fake_response(content=_anchorless_content()),
        _make_fake_response(content=_good_response_content()),
    ])
    out = _call()
    assert mock.call_count == 2
    assert out["_graph_anchor_retry_metadata"][
        "graph_anchor_validator_retry_succeeded"] == 1


def test_w20f10_no_clean_anchor_candidates_fails_immediately(monkeypatch):
    """상류 clean-anchor 후보 0 = LLM 복구 불가 — 1콜 후 즉시 fail-closed."""
    _stub_preflight(monkeypatch)
    mock = _stub_completion_seq(monkeypatch, [
        _make_fake_response(content=_empty_graph_content()),
    ])
    dossier = _dossier()
    dossier["anchor_selection_metadata"]["candidate_bg_ids"] = []
    with pytest.raises(ShotAwareBgRenderPlanProviderError):
        _call(dossier=dossier)
    assert mock.call_count == 1


def test_w20f10_mixed_camera_and_anchor_failure_no_retry(monkeypatch):
    """camera+anchor 혼합 실패 = W20F9/W20F10 어느 쪽도 아님 — 1콜 실패."""
    _stub_preflight(monkeypatch)
    payload = _good_graph_payload()
    payload["graph"]["nodes"][0]["is_dwelling_identity_anchor"] = False
    payload["graph"]["nodes"][0]["mode"] = "same_physical_space_view"
    payload["graph"]["nodes"][0]["camera_decision"]["camera_cell"] = [9, 9]
    mock = _stub_completion_seq(monkeypatch, [
        _make_fake_response(content=_json.dumps(payload)),
    ])
    with pytest.raises(ShotAwareBgRenderPlanProviderError):
        _call()
    assert mock.call_count == 1


def test_w20f10_retry_still_failing_fail_closed_two_calls(monkeypatch):
    """교정본 재실패 = 연쇄 없이 fail-closed, 총 2콜 + 실비용 보존."""
    _stub_preflight(monkeypatch)
    mock = _stub_completion_seq(monkeypatch, [
        _make_fake_response(content=_empty_graph_content()),
        _make_fake_response(content=_empty_graph_content()),
    ])
    with pytest.raises(ShotAwareBgRenderPlanProviderError) as exc:
        _call()
    assert mock.call_count == 2
    assert exc.value.completion_call_count == 2
    assert "W20F10 retry" in str(exc.value)


def test_w20f10_retry_flipping_to_camera_failure_no_third_call(monkeypatch):
    """교정본이 camera 실패로 전환돼도 3콜/snap 연쇄 금지 — 상한 2 잠금."""
    _stub_preflight(monkeypatch)
    cam_bad = _good_graph_payload()
    cam_bad["graph"]["nodes"][0]["camera_decision"]["camera_cell"] = [9, 9]
    mock = _stub_completion_seq(monkeypatch, [
        _make_fake_response(content=_empty_graph_content()),
        _make_fake_response(content=_json.dumps(cam_bad)),
    ])
    with pytest.raises(ShotAwareBgRenderPlanProviderError) as exc:
        _call()
    assert mock.call_count == 2
    assert exc.value.completion_call_count == 2


def test_w20f10_first_pass_success_single_call_no_sentinel(monkeypatch):
    """정상 첫 응답 = 1콜·산출 byte-equivalent·sentinel 없음."""
    _stub_preflight(monkeypatch)
    mock = _stub_completion_seq(monkeypatch, [
        _make_fake_response(content=_good_response_content()),
    ])
    out = _call()
    assert mock.call_count == 1
    assert out == _good_graph_payload()
    assert "_graph_anchor_retry_metadata" not in out
    assert "_w20f9_retry_metadata" not in out


def _two_bg_fixtures():
    """renderable={L01B01, L01B02} 2-bg fixture (TEST-GAP-3)."""
    dossier = _dossier()
    facts = dossier["per_bg_render_facts_by_bg_id"]
    b2 = _json.loads(_json.dumps(facts["L01B01"]))
    b2["bg_id"] = "L01B02"
    b2["applies_to_shots"] = ["S1_Shot2"]
    facts["L01B02"] = b2
    readiness = _shot_readiness()
    readiness["per_bg"]["L01B02"] = {
        "applies_to_shots": ["S1_Shot2"],
        "surfaced_shots": [
            {"shot_id": "S1_Shot2", "staging": {"shot_id": "S1_Shot2"}}
        ],
        "ok": True,
        "blockers": [],
    }
    payload = _good_graph_payload()
    node2 = _json.loads(_json.dumps(payload["graph"]["nodes"][0]))
    node2["bg_id"] = "L01B02"
    node2["node_index"] = 1
    node2["is_dwelling_identity_anchor"] = False
    node2["mode"] = "same_physical_space_view"
    payload["graph"]["nodes"].append(node2)
    return dossier, readiness, payload


def test_w20f10_nonempty_missing_bg_retry_recovers_full_graph(monkeypatch):
    """TEST-GAP-3: non-empty(유효 anchor A)지만 B 누락 → 교정 프롬프트에
    B 누락 진단 포함 → 2콜째 완전 그래프로 복구."""
    _stub_preflight(monkeypatch)
    dossier, readiness, full_payload = _two_bg_fixtures()
    partial_payload = _json.loads(_json.dumps(full_payload))
    partial_payload["graph"]["nodes"] = [
        n for n in partial_payload["graph"]["nodes"]
        if n["bg_id"] == "L01B01"
    ]
    mock = _stub_completion_seq(monkeypatch, [
        _make_fake_response(content=_json.dumps(partial_payload)),
        _make_fake_response(content=_json.dumps(full_payload)),
    ])
    out = _call(dossier=dossier, shot_readiness=readiness)
    assert mock.call_count == 2
    assert out["_graph_anchor_retry_metadata"][
        "graph_anchor_validator_retry_succeeded"] == 1
    assert {n["bg_id"] for n in out["graph"]["nodes"]} == {
        "L01B01", "L01B02"}
    repair_prompt = mock.call_args_list[1].kwargs["messages"][1]["content"]
    assert "L01B02" in repair_prompt          # 누락 bg 진단+기대 목록
    assert "missing renderable bg_ids" in repair_prompt


def test_w20f10_anchor_candidates_disjoint_from_renderable_no_retry(
    monkeypatch,
):
    """clean-anchor set 이 nonempty 여도 renderable 과 교집합 0 = 복구
    불가 — 1콜 즉시 fail-closed."""
    _stub_preflight(monkeypatch)
    dossier = _dossier()
    dossier["anchor_selection_metadata"]["candidate_bg_ids"] = ["L99B99"]
    mock = _stub_completion_seq(monkeypatch, [
        _make_fake_response(content=_empty_graph_content()),
    ])
    with pytest.raises(ShotAwareBgRenderPlanProviderError):
        _call(dossier=dossier)
    assert mock.call_count == 1


def test_w20f10_repair_template_single_pass_byte_preserved(monkeypatch):
    """NARROW-2: 삽입 데이터(이전 응답)에 placeholder 토큰이 있어도
    재치환되지 않고 byte-preserved."""
    from app.modules.pipeline.shot_aware_bg_render_plan_llm_provider import (
        _build_graph_anchor_repair_user_prompt,
    )

    poisoned_prev = (
        '{"graph": {"nodes": []}, "rationale": '
        '"{{original_user_prompt}} {{diagnostics_block}} '
        '{{renderable_bg_ids_block}} {{clean_anchor_candidates_block}} '
        '{{previous_response_block}}"}'
    )
    out = _build_graph_anchor_repair_user_prompt(
        repair_template=(
            "HEAD\n{{diagnostics_block}}\n{{previous_response_block}}\n"
            "{{original_user_prompt}}\nTAIL"
        ),
        original_user_prompt="ORIGINAL_PROMPT_SENTINEL",
        previous_content=poisoned_prev,
        diagnostics=["diag one"],
        renderable_bg_ids=frozenset({"L01B01"}),
        clean_anchor_candidate_bg_ids=frozenset({"L01B01"}),
    )
    # 삽입된 이전 응답 전문이 그대로 보존 (내부 토큰 미확장)
    assert poisoned_prev in out
    # 템플릿 자신의 토큰만 정확히 1회 치환
    assert out.count("ORIGINAL_PROMPT_SENTINEL") == 1
    assert "- diag one" in out