"""D6 T9 — validator fallback STALE_UPSTREAM 분리 (preflight safety net).

`ref_contract_validator.validate_attached_refs` 에서 background ref missing 시:
  - bg_id 가 ``BG_ID_RE`` (`^L\\d{2,3}B\\d{2,3}$`) match → ``StaleUpstreamError``
    (D6 path — render manifest stale 의 표현, T8 preflight 가 잡았어야 했지만
    bypass 된 fallback safety net).
  - legacy `bg_*` 형식 → 기존 ``RefContractError`` (D5 정합 보존, transitional
    cp 호환).

unit-style — validate_attached_refs 직접 호출, chain_bg_lookup lambda mock.

spec: docs/superpowers/specs/2026-05-09-deterministic-bg-id-and-catalog-lineage.md §4.8 (preflight) + R2 I3 (validator fallback split)
plan: T9
"""
from __future__ import annotations

import pytest


# ──────────────────────────────────────────────────────────────────────
# D6 path — bg_id 가 BG_ID_RE match 시 StaleUpstreamError
# ──────────────────────────────────────────────────────────────────────


def test_validator_emits_stale_upstream_when_d6_bg_id_missing_no_lookup():
    """D6 bg_id missing + chain_bg_lookup=None → StaleUpstreamError."""
    from app.core.ref_contract_validator import validate_attached_refs
    from app.core.errors import StaleUpstreamError

    rpc = {"asset_requirements": {
        "required_refs": [
            {"kind": "background", "id": "L09B01", "policy": "required"},
        ],
    }}
    with pytest.raises(StaleUpstreamError) as exc_info:
        validate_attached_refs(
            rpc, [], [], prompt="ref-free prompt", is_close_framing=False,
            reference_phrase_kinds=[],
        )
    err = exc_info.value
    assert err.upstream == "background_render"
    assert err.missing_in_render == ["L09B01"]
    assert err.code == "STALE_UPSTREAM"
    assert err.status_code == 422


def test_validator_emits_stale_upstream_with_chain_lookup_returning_none():
    """D6 bg_id + chain_bg_lookup=lambda x: None → StaleUpstreamError (D5 G3 lineage 확인 불가)."""
    from app.core.ref_contract_validator import validate_attached_refs
    from app.core.errors import StaleUpstreamError

    rpc = {"asset_requirements": {
        "required_refs": [
            {"kind": "background", "id": "L09B01", "policy": "required"},
        ],
    }}
    with pytest.raises(StaleUpstreamError) as exc_info:
        validate_attached_refs(
            rpc, [], [], prompt="ref-free prompt", is_close_framing=False,
            chain_bg_lookup=lambda x: None,
            reference_phrase_kinds=[],
        )
    assert exc_info.value.upstream == "background_render"
    assert exc_info.value.missing_in_render == ["L09B01"]


def test_validator_emits_stale_upstream_when_lookup_returns_loc_but_lineage_not_attached():
    """D6 bg_id + chain_bg_lookup=loc 매핑 있으나 ('background_prev_shot', loc) attach 안 됨 → STALE."""
    from app.core.ref_contract_validator import validate_attached_refs
    from app.core.errors import StaleUpstreamError

    rpc = {"asset_requirements": {
        "required_refs": [
            {"kind": "background", "id": "L09B01", "policy": "required"},
        ],
    }}
    # chain_bg_lookup 은 loc_id 반환하지만 attached_meta 에 lineage 없음
    with pytest.raises(StaleUpstreamError):
        validate_attached_refs(
            rpc, [], [], prompt="ref-free prompt", is_close_framing=False,
            chain_bg_lookup=lambda bid: "L09" if bid == "L09B01" else None,
            reference_phrase_kinds=[],
        )


def test_validator_stale_upstream_details_include_remediation_and_missing():
    """details payload — handler 가 응답에 surface 할 metadata 검증."""
    from app.core.ref_contract_validator import validate_attached_refs
    from app.core.errors import StaleUpstreamError

    rpc = {"asset_requirements": {
        "required_refs": [
            {"kind": "background", "id": "L05B02", "policy": "required"},
        ],
    }}
    with pytest.raises(StaleUpstreamError) as exc_info:
        validate_attached_refs(
            rpc, [], [], prompt="ref-free prompt", is_close_framing=False,
            reference_phrase_kinds=[],
        )
    err = exc_info.value
    assert err.details["upstream"] == "background_render"
    assert err.details["missing_in_render"] == ["L05B02"]
    assert "remediation" in err.details
    assert "background_render" in err.details["remediation"].lower() or \
           "force" in err.details["remediation"].lower()


# ──────────────────────────────────────────────────────────────────────
# legacy path — bg_* 형식은 기존 RefContractError 보존
# ──────────────────────────────────────────────────────────────────────


@pytest.mark.parametrize("legacy_bg_id", [
    "bg_supermarket_dusk",
    "bg_kitchen",
    "bg_office_normal",
    "",  # empty string — BG_ID_RE 도 안 맞음, legacy path 로 fall through
])
def test_validator_keeps_ref_contract_error_for_legacy_bg_id(legacy_bg_id):
    """legacy `bg_*` (D5 transitional cp) 는 기존 RefContractError 유지 — D6 분리 누락 차단."""
    from app.core.ref_contract_validator import (
        validate_attached_refs, RefContractError,
    )
    from app.core.errors import StaleUpstreamError

    rpc = {"asset_requirements": {
        "required_refs": [
            {"kind": "background", "id": legacy_bg_id, "policy": "required"},
        ],
    }}
    # raise 가 발생하되 종류는 StaleUpstreamError 아님 (D5 정합)
    with pytest.raises(RefContractError) as exc_info:
        validate_attached_refs(
            rpc, [], [], prompt="ref-free prompt", is_close_framing=False,
            reference_phrase_kinds=[],
        )
    # subclass check 도 — legacy path 는 StaleUpstreamError 가 아니어야 함
    assert not isinstance(exc_info.value, StaleUpstreamError), (
        f"legacy bg_id {legacy_bg_id!r} 가 StaleUpstreamError 로 잘못 분류됨"
    )


# ──────────────────────────────────────────────────────────────────────
# 변경 없음 검증 (regression guard)
# ──────────────────────────────────────────────────────────────────────


def test_validator_close_framing_skips_background_check_d6_bg_id():
    """close framing 시 D6 bg_id 도 background 검사 자체 skip — 분기 변경 후에도 보존."""
    from app.core.ref_contract_validator import validate_attached_refs

    rpc = {"asset_requirements": {
        "required_refs": [
            {"kind": "background", "id": "L09B01", "policy": "required"},
        ],
    }}
    # close framing → background 검사 자체 skip (StaleUpstreamError 도 raise 안 됨)
    validate_attached_refs(
        rpc, [], [], prompt="extreme close-up.", is_close_framing=True,
        reference_phrase_kinds=[],
    )


def test_validator_satisfied_d6_bg_id_attached_no_raise():
    """D6 bg_id 가 ('background', bg_id) 로 attach 되면 통과 (분기 진입 안 함)."""
    from app.core.ref_contract_validator import validate_attached_refs

    rpc = {"asset_requirements": {
        "required_refs": [
            {"kind": "background", "id": "L09B01", "policy": "required"},
        ],
    }}
    labeled_refs = [("background L09B01", b"x")]
    attached_meta = [("background", "L09B01")]
    validate_attached_refs(
        rpc, labeled_refs, attached_meta,
        prompt="wide shot.", is_close_framing=False,
        reference_phrase_kinds=[],
    )


# ──────────────────────────────────────────────────────────────────────
# T9-fix BLOCKING — batch propagation
# ──────────────────────────────────────────────────────────────────────


def _stale_for_test():
    """Helper: raise a fresh StaleUpstreamError instance."""
    from app.core.errors import StaleUpstreamError

    raise StaleUpstreamError(
        upstream="background_render",
        expected_bg_catalog_hash="EXPECTED",
        observed_bg_catalog_hash="STALE",
        remediation="POST /steps/background_render?mode=force",
    )


def test_await_variation_results_propagates_stale_upstream():
    """variation future loop (`_await_variation_results`) 가 StaleUpstreamError
    를 별도 catch 후 re-raise (silent skip 차단).

    review iter1 BLOCKING — 이전 ``except Exception`` 만으로는 generic logging 으로
    swallow 되어 var_results 빈 list 반환 → "all variations failed" 로 무력화.
    """
    from concurrent.futures import ThreadPoolExecutor
    from unittest.mock import MagicMock

    from app.core.errors import StaleUpstreamError
    from app.services.scene_generation_coordinator import SceneGenerationCoordinator

    coord = SceneGenerationCoordinator.__new__(SceneGenerationCoordinator)
    coord._db = MagicMock()
    coord._project_id = "p1"
    coord._actor_id = "u1"
    coord._logger = MagicMock()

    def _ok_worker():
        return {"file_path": "ok.png"}

    with ThreadPoolExecutor(max_workers=2) as ex:
        var_futures = {
            ex.submit(_stale_for_test): 0,
            ex.submit(_ok_worker): 1,  # 다른 variation 은 정상 — stale 가 우선
        }
        with pytest.raises(StaleUpstreamError) as exc_info:
            coord._await_variation_results(var_futures)
        assert exc_info.value.upstream == "background_render"
        assert exc_info.value.observed_bg_catalog_hash == "STALE"


def test_await_variation_results_captures_generic_exception_as_typed_failure():
    """Fix C (2026-05-10): generic Exception 은 silent swallow 하지 않고 typed
    failure dict (`_failure_reason="GENERIC_EXCEPTION"`) 로 var_results 에
    누적 — scene_image_service 가 cp.failed 메시지 직렬화 시 이 reason 사용해
    24 shot deterministic 실패 root cause 가시화. 옛 silent skip 정책 폐기.
    """
    from concurrent.futures import ThreadPoolExecutor
    from unittest.mock import MagicMock

    from app.services.scene_generation_coordinator import SceneGenerationCoordinator

    coord = SceneGenerationCoordinator.__new__(SceneGenerationCoordinator)
    coord._db = MagicMock()
    coord._project_id = "p1"
    coord._actor_id = "u1"
    coord._logger = MagicMock()

    def _flaky_worker():
        raise ValueError("transient gemini timeout")

    def _ok_worker():
        return {"file_path": "ok.png"}

    with ThreadPoolExecutor(max_workers=2) as ex:
        var_futures = {
            ex.submit(_flaky_worker): 0,
            ex.submit(_ok_worker): 1,
        }
        results = coord._await_variation_results(var_futures)

    # 성공 + typed failure 둘 다 반환 (scene_image_service 가 분리)
    assert len(results) == 2
    successes = [r for r in results if not r.get("_failure_reason")]
    failures = [r for r in results if r.get("_failure_reason")]
    assert len(successes) == 1
    assert successes[0]["file_path"] == "ok.png"
    assert len(failures) == 1
    assert failures[0]["_failure_reason"] == "GENERIC_EXCEPTION"
    assert "transient gemini timeout" in failures[0]["_failure_detail"]
    assert "_failure_traceback" in failures[0]


def test_validator_stale_upstream_propagates_through_threadpool_future():
    """validator 의 StaleUpstreamError 가 future.result() 를 통해 그대로 propagate
    (RefContractError 만 catch 하는 variation handler 가 swallow 안 함).

    이는 _generate_variation_in_loop 의 try/except RefContractError (line 1312)
    가 StaleUpstreamError 를 우회한다는 invariant 검증 — AppError 분기 (line 1398)
    는 sanitize 루프 안이라 validate_attached_refs raise 시점에는 미도달.
    """
    from concurrent.futures import ThreadPoolExecutor
    from app.core.errors import StaleUpstreamError
    from app.core.ref_contract_validator import (
        validate_attached_refs, RefContractError,
    )

    rpc = {"asset_requirements": {
        "required_refs": [
            {"kind": "background", "id": "L09B01", "policy": "required"},
        ],
    }}

    def _worker_calls_validator():
        try:
            validate_attached_refs(
                rpc, [], [], prompt="ref-free", is_close_framing=False,
                reference_phrase_kinds=[],
            )
        except RefContractError:
            # variation handler 가 catch — variation skip path
            return None
        # 정상 흐름

    with ThreadPoolExecutor(max_workers=1) as ex:
        future = ex.submit(_worker_calls_validator)
        with pytest.raises(StaleUpstreamError):
            future.result()


def test_validator_satisfied_d6_bg_id_via_prev_shot_lineage_no_raise():
    """D6 bg_id + chain_bg_lookup=loc 매핑 + ('background_prev_shot', loc) attach → 통과."""
    from app.core.ref_contract_validator import validate_attached_refs

    rpc = {"asset_requirements": {
        "required_refs": [
            {"kind": "background", "id": "L09B01", "policy": "required"},
        ],
    }}
    labeled_refs = [("background_prev_shot L09", b"x")]
    attached_meta = [("background_prev_shot", "L09")]
    validate_attached_refs(
        rpc, labeled_refs, attached_meta,
        prompt="wide shot.", is_close_framing=False,
        chain_bg_lookup=lambda bid: "L09" if bid == "L09B01" else None,
        reference_phrase_kinds=[],
    )


# ──────────────────────────────────────────────────────────────────────
# W21B space_set_bg Phase 2 — space plate overlay 는 required background 를
# 충족하는 1급 background source (silent fallback 아님). loader 가 key 충돌 시
# space plate 를 우선 주입하면 attached bg_id 가 synthetic(space_set_bg:...)이
# 되어 exact L##B## match 가 깨지는 계약 충돌의 명시 해소 (E2E 실측).
# ──────────────────────────────────────────────────────────────────────


def test_validator_space_set_bg_background_satisfies_required():
    """required L##B## + attached ('background', 'space_set_bg:...') → 통과 (no raise)."""
    from app.core.ref_contract_validator import validate_attached_refs

    rpc = {"asset_requirements": {
        "required_refs": [
            {"kind": "background", "id": "L04B04", "policy": "required"},
        ],
    }}
    labeled_refs = [("space set background ref (hub room, G1)", b"x")]
    attached_meta = [("background", "space_set_bg:G1:hub_room")]
    validate_attached_refs(
        rpc, labeled_refs, attached_meta,
        prompt="wide shot.", is_close_framing=False,
        reference_phrase_kinds=[],
    )


def test_validator_space_set_bg_satisfies_multiple_required_backgrounds():
    """required background 복수 + space ref 1건 → 전부 충족 (shot 의 bg ref slot 은 1개 — contract 고정)."""
    from app.core.ref_contract_validator import validate_attached_refs

    rpc = {"asset_requirements": {
        "required_refs": [
            {"kind": "background", "id": "L04B04", "policy": "required"},
            {"kind": "background", "id": "L04B07", "policy": "required"},
        ],
    }}
    labeled_refs = [("space set background ref (hub room, G1)", b"x")]
    attached_meta = [("background", "space_set_bg:G1:hub_room")]
    validate_attached_refs(
        rpc, labeled_refs, attached_meta,
        prompt="wide shot.", is_close_framing=False,
        reference_phrase_kinds=[],
    )


def test_validator_space_set_bg_does_not_mask_missing_background():
    """space ref 부재 시 기존 StaleUpstreamError 그대로 (약화 0)."""
    from app.core.ref_contract_validator import validate_attached_refs
    from app.core.errors import StaleUpstreamError

    rpc = {"asset_requirements": {
        "required_refs": [
            {"kind": "background", "id": "L04B04", "policy": "required"},
        ],
    }}
    with pytest.raises(StaleUpstreamError):
        validate_attached_refs(
            rpc, [], [], prompt="wide shot.", is_close_framing=False,
            reference_phrase_kinds=[],
        )


def test_validator_space_set_bg_does_not_satisfy_required_prop_or_outlook():
    """space ref 는 background 만 충족 — prop/character_outlook strict 는 영향 0."""
    from app.core.ref_contract_validator import validate_attached_refs
    from app.core.ref_contract_validator import RefContractError

    rpc = {"asset_requirements": {
        "required_refs": [
            {"kind": "prop", "id": "P01", "policy": "required"},
        ],
    }}
    labeled_refs = [("space set background ref (hub room, G1)", b"x")]
    attached_meta = [("background", "space_set_bg:G1:hub_room")]
    with pytest.raises(RefContractError):
        validate_attached_refs(
            rpc, labeled_refs, attached_meta,
            prompt="wide shot.", is_close_framing=False,
            reference_phrase_kinds=[],
        )