"""D6 T8 Phase A — dispatcher_preflight + StaleUpstreamError unit tests.

unit-style pivot (T6/T7 와 같은 결정) — d6_episode* fixture 미존재.
``tmp_path`` 에 minimal manifest 직접 기록 + ``settings.projects_dir`` monkeypatch.

검증 범위:
  - 4 consumer (T5-fix iter7) 모두 catalog + binding 양쪽 비교.
  - corrupt manifest → StaleUpstreamError fail-fast (silent skip 폐기).
  - master_plan 에 D6 hash 있는데 consumer manifest missing → StaleUpstreamError.
  - master_plan 자체 missing 또는 D6 hash 부재 → silent return (D5 fallback).
  - custom_prompt path (enforce_shot_binding=False) — catalog 만 검증, binding skip.
  - StaleUpstreamError details + handler serialization (404 회귀 방지).
  - 기존 AppError 호출자 backward compat — handler 가 details 없는 응답 그대로.

spec: docs/superpowers/specs/2026-05-09-deterministic-bg-id-and-catalog-lineage.md §4.8
plan: T8 Phase A (unit-style pivot)
"""
from __future__ import annotations

import asyncio
import json
from pathlib import Path

import pytest


_CATALOG_HASH = "cat_001"
_BINDING_HASH = "bind_001"
_CONSUMERS = (
    "floor_plan_prompt",
    "background_prompt",
    "background_render",
    "scene_detail",
)


def _write_manifest(projects_dir, pid, eid, step, data, schema_version=1):
    p = (Path(projects_dir) / pid / "checkpoints"
         / "episodes" / eid / step / "manifest.json")
    p.parent.mkdir(parents=True, exist_ok=True)
    payload = {"schema_version": schema_version, "data": data}
    p.write_text(json.dumps(payload), encoding="utf-8")
    return p


def _setup_consistent_d6_episode(
    tmp_path, monkeypatch, *,
    pid="p1", eid="e1",
    catalog_hash=_CATALOG_HASH, binding_hash=_BINDING_HASH,
):
    """master_plan + 4 consumer 모두 hash 일치 — 정상 D6 episode."""
    monkeypatch.setattr("app.core.config.settings.projects_dir", str(tmp_path))
    _write_manifest(tmp_path, pid, eid, "background_master_plan", {
        "bg_catalog_hash": catalog_hash,
        "shot_binding_hash": binding_hash,
        "background_catalog": {},
        "shot_background_map": {},
    })
    for step in _CONSUMERS:
        _write_manifest(tmp_path, pid, eid, step, {
            "consumed_bg_catalog_hash": catalog_hash,
            "consumed_shot_binding_hash": binding_hash,
        })
    return pid, eid


# ──────────────────────────────────────────────────────────────────────
# happy path + D5 fallback
# ──────────────────────────────────────────────────────────────────────


def test_preflight_passes_when_all_4_consumers_match(tmp_path, monkeypatch):
    pid, eid = _setup_consistent_d6_episode(tmp_path, monkeypatch)
    from app.services.dispatcher_preflight import check_bg_catalog_freshness

    check_bg_catalog_freshness(pid, eid, enforce_shot_binding=True)


def test_preflight_passes_with_enforce_binding_false(tmp_path, monkeypatch):
    pid, eid = _setup_consistent_d6_episode(tmp_path, monkeypatch)
    from app.services.dispatcher_preflight import check_bg_catalog_freshness

    check_bg_catalog_freshness(pid, eid, enforce_shot_binding=False)


def test_preflight_returns_silently_when_master_plan_missing(
    tmp_path, monkeypatch,
):
    """D5 fallback — master_plan cp 부재 = D6 미적용."""
    monkeypatch.setattr("app.core.config.settings.projects_dir", str(tmp_path))
    from app.services.dispatcher_preflight import check_bg_catalog_freshness

    # 어떤 consumer manifest 도 없어도 silent return
    check_bg_catalog_freshness("p-no-mp", "e1", enforce_shot_binding=True)


def test_preflight_returns_silently_when_master_plan_lacks_d6_hash(
    tmp_path, monkeypatch,
):
    """legacy / pre-D6 master_plan — bg_catalog_hash 부재."""
    monkeypatch.setattr("app.core.config.settings.projects_dir", str(tmp_path))
    _write_manifest(tmp_path, "p1", "e1", "background_master_plan", {
        "plans": {},
        # 의도적으로 bg_catalog_hash 없음
    })
    from app.services.dispatcher_preflight import check_bg_catalog_freshness

    check_bg_catalog_freshness("p1", "e1", enforce_shot_binding=True)


# ──────────────────────────────────────────────────────────────────────
# catalog hash mismatch
# ──────────────────────────────────────────────────────────────────────


@pytest.mark.parametrize("stale_step", _CONSUMERS)
def test_preflight_raises_on_catalog_hash_mismatch_for_any_consumer(
    tmp_path, monkeypatch, stale_step,
):
    """4 consumer 중 어느 하나의 catalog hash 가 stale 이면 첫 발견 시점에 raise."""
    pid, eid = _setup_consistent_d6_episode(tmp_path, monkeypatch)
    _write_manifest(tmp_path, pid, eid, stale_step, {
        "consumed_bg_catalog_hash": "STALE_CATALOG",
        "consumed_shot_binding_hash": _BINDING_HASH,
    })
    from app.services.dispatcher_preflight import check_bg_catalog_freshness
    from app.core.errors import StaleUpstreamError

    with pytest.raises(StaleUpstreamError) as exc_info:
        check_bg_catalog_freshness(pid, eid, enforce_shot_binding=True)
    err = exc_info.value
    assert err.upstream == stale_step
    assert err.expected_bg_catalog_hash == _CATALOG_HASH
    assert err.observed_bg_catalog_hash == "STALE_CATALOG"
    assert err.status_code == 422
    assert err.code == "STALE_UPSTREAM"


# ──────────────────────────────────────────────────────────────────────
# binding hash mismatch — 4 consumer 모두 (T5-fix iter7 정합)
# ──────────────────────────────────────────────────────────────────────


@pytest.mark.parametrize("stale_step", _CONSUMERS)
def test_preflight_raises_on_binding_hash_mismatch_when_enforced(
    tmp_path, monkeypatch, stale_step,
):
    """T5-fix iter7: 4 consumer 모두 binding stamp — 각각 mismatch detection."""
    pid, eid = _setup_consistent_d6_episode(tmp_path, monkeypatch)
    _write_manifest(tmp_path, pid, eid, stale_step, {
        "consumed_bg_catalog_hash": _CATALOG_HASH,  # catalog OK
        "consumed_shot_binding_hash": "STALE_BINDING",
    })
    from app.services.dispatcher_preflight import check_bg_catalog_freshness
    from app.core.errors import StaleUpstreamError

    with pytest.raises(StaleUpstreamError) as exc_info:
        check_bg_catalog_freshness(pid, eid, enforce_shot_binding=True)
    err = exc_info.value
    assert err.upstream == stale_step
    assert err.expected_shot_binding_hash == _BINDING_HASH
    assert err.observed_shot_binding_hash == "STALE_BINDING"


def test_preflight_skips_binding_check_when_custom_prompt(tmp_path, monkeypatch):
    """custom_prompt path (enforce_shot_binding=False) — binding stale 무시, catalog 만 검증."""
    pid, eid = _setup_consistent_d6_episode(tmp_path, monkeypatch)
    _write_manifest(tmp_path, pid, eid, "scene_detail", {
        "consumed_bg_catalog_hash": _CATALOG_HASH,
        "consumed_shot_binding_hash": "STALE_BINDING",
    })
    from app.services.dispatcher_preflight import check_bg_catalog_freshness

    check_bg_catalog_freshness(pid, eid, enforce_shot_binding=False)


def test_preflight_still_raises_on_catalog_mismatch_with_enforce_false(
    tmp_path, monkeypatch,
):
    """custom_prompt path 도 catalog freshness 는 강제."""
    pid, eid = _setup_consistent_d6_episode(tmp_path, monkeypatch)
    _write_manifest(tmp_path, pid, eid, "background_render", {
        "consumed_bg_catalog_hash": "STALE_CATALOG",
        "consumed_shot_binding_hash": _BINDING_HASH,
    })
    from app.services.dispatcher_preflight import check_bg_catalog_freshness
    from app.core.errors import StaleUpstreamError

    with pytest.raises(StaleUpstreamError):
        check_bg_catalog_freshness(pid, eid, enforce_shot_binding=False)


# ──────────────────────────────────────────────────────────────────────
# fail-fast: missing consumer manifest under D6
# ──────────────────────────────────────────────────────────────────────


@pytest.mark.parametrize("absent_step", _CONSUMERS)
def test_preflight_raises_when_any_consumer_manifest_absent(
    tmp_path, monkeypatch, absent_step,
):
    """master_plan 에 D6 hash 있으면 4 consumer 모두 manifest 의무 (silent skip 폐기)."""
    monkeypatch.setattr("app.core.config.settings.projects_dir", str(tmp_path))
    _write_manifest(tmp_path, "p1", "e1", "background_master_plan", {
        "bg_catalog_hash": _CATALOG_HASH,
        "shot_binding_hash": _BINDING_HASH,
    })
    for step in _CONSUMERS:
        if step == absent_step:
            continue  # absent_step manifest 작성 X
        _write_manifest(tmp_path, "p1", "e1", step, {
            "consumed_bg_catalog_hash": _CATALOG_HASH,
            "consumed_shot_binding_hash": _BINDING_HASH,
        })
    from app.services.dispatcher_preflight import check_bg_catalog_freshness
    from app.core.errors import StaleUpstreamError

    with pytest.raises(StaleUpstreamError) as exc_info:
        check_bg_catalog_freshness("p1", "e1", enforce_shot_binding=True)
    err = exc_info.value
    assert err.upstream == absent_step
    assert err.observed_bg_catalog_hash == "(manifest missing)"


# ──────────────────────────────────────────────────────────────────────
# fail-fast: corrupt manifest
# ──────────────────────────────────────────────────────────────────────


def test_preflight_raises_on_corrupt_consumer_manifest(tmp_path, monkeypatch):
    """parse 실패 → StaleUpstreamError (silent return None 폐기)."""
    monkeypatch.setattr("app.core.config.settings.projects_dir", str(tmp_path))
    _write_manifest(tmp_path, "p1", "e1", "background_master_plan", {
        "bg_catalog_hash": _CATALOG_HASH,
        "shot_binding_hash": _BINDING_HASH,
    })
    # background_prompt 만 corrupt — 나머지는 정상
    bp = (Path(tmp_path) / "p1" / "checkpoints" / "episodes" / "e1"
          / "background_prompt" / "manifest.json")
    bp.parent.mkdir(parents=True, exist_ok=True)
    bp.write_text("{not json", encoding="utf-8")
    for step in ("floor_plan_prompt", "background_render", "scene_detail"):
        _write_manifest(tmp_path, "p1", "e1", step, {
            "consumed_bg_catalog_hash": _CATALOG_HASH,
            "consumed_shot_binding_hash": _BINDING_HASH,
        })
    from app.services.dispatcher_preflight import check_bg_catalog_freshness
    from app.core.errors import StaleUpstreamError

    with pytest.raises(StaleUpstreamError) as exc_info:
        check_bg_catalog_freshness("p1", "e1", enforce_shot_binding=True)
    err = exc_info.value
    assert err.upstream == "background_prompt"
    assert "corrupt" in err.remediation.lower()


def test_preflight_raises_on_corrupt_master_plan_manifest(
    tmp_path, monkeypatch,
):
    """master_plan corrupt → fail-fast (D5 fallback 으로 잘못 진행 차단)."""
    monkeypatch.setattr("app.core.config.settings.projects_dir", str(tmp_path))
    mp = (Path(tmp_path) / "p1" / "checkpoints" / "episodes" / "e1"
          / "background_master_plan" / "manifest.json")
    mp.parent.mkdir(parents=True, exist_ok=True)
    mp.write_text("{not json", encoding="utf-8")
    from app.services.dispatcher_preflight import check_bg_catalog_freshness
    from app.core.errors import StaleUpstreamError

    with pytest.raises(StaleUpstreamError) as exc_info:
        check_bg_catalog_freshness("p1", "e1", enforce_shot_binding=True)
    assert exc_info.value.upstream == "background_master_plan"


# ──────────────────────────────────────────────────────────────────────
# StaleUpstreamError details + handler serialization
# ──────────────────────────────────────────────────────────────────────


def test_stale_upstream_error_details_include_remediation(tmp_path, monkeypatch):
    """response payload 에 들어갈 details — code/upstream/hashes/remediation."""
    pid, eid = _setup_consistent_d6_episode(tmp_path, monkeypatch)
    _write_manifest(tmp_path, pid, eid, "background_render", {
        "consumed_bg_catalog_hash": "STALE_CATALOG",
        "consumed_shot_binding_hash": _BINDING_HASH,
    })
    from app.services.dispatcher_preflight import check_bg_catalog_freshness
    from app.core.errors import StaleUpstreamError

    with pytest.raises(StaleUpstreamError) as exc_info:
        check_bg_catalog_freshness(pid, eid, enforce_shot_binding=True)
    err = exc_info.value
    assert err.details["upstream"] == "background_render"
    assert err.details["expected_bg_catalog_hash"] == _CATALOG_HASH
    assert err.details["observed_bg_catalog_hash"] == "STALE_CATALOG"
    assert "remediation" in err.details
    assert "background_render" in err.details["remediation"]
    # binding hash details 는 catalog mismatch 인 경우 미포함
    assert "expected_shot_binding_hash" not in err.details


def test_stale_upstream_binding_mismatch_includes_binding_details(
    tmp_path, monkeypatch,
):
    pid, eid = _setup_consistent_d6_episode(tmp_path, monkeypatch)
    _write_manifest(tmp_path, pid, eid, "scene_detail", {
        "consumed_bg_catalog_hash": _CATALOG_HASH,
        "consumed_shot_binding_hash": "STALE_BINDING",
    })
    from app.services.dispatcher_preflight import check_bg_catalog_freshness
    from app.core.errors import StaleUpstreamError

    with pytest.raises(StaleUpstreamError) as exc_info:
        check_bg_catalog_freshness(pid, eid, enforce_shot_binding=True)
    err = exc_info.value
    assert err.details["expected_shot_binding_hash"] == _BINDING_HASH
    assert err.details["observed_shot_binding_hash"] == "STALE_BINDING"


def test_app_error_handler_serializes_stale_upstream_details():
    """handler 가 details 를 응답 ``error`` dict 에 merge."""
    from app.core.errors import StaleUpstreamError, app_error_handler

    err = StaleUpstreamError(
        upstream="scene_detail",
        expected_bg_catalog_hash="A",
        observed_bg_catalog_hash="B",
        remediation="POST /steps/scene_detail?mode=force",
    )
    loop = asyncio.new_event_loop()
    try:
        response = loop.run_until_complete(app_error_handler(None, err))
    finally:
        loop.close()
    body = json.loads(response.body)
    assert response.status_code == 422
    assert body["error"]["code"] == "STALE_UPSTREAM"
    assert body["error"]["upstream"] == "scene_detail"
    assert body["error"]["expected_bg_catalog_hash"] == "A"
    assert body["error"]["observed_bg_catalog_hash"] == "B"
    assert body["error"]["remediation"] == "POST /steps/scene_detail?mode=force"
    assert "message" in body["error"]


def test_app_error_handler_backward_compat_for_plain_app_error():
    """기존 AppError(code, message, status_code) 호출자 — details 없으면 응답 shape 변경 0."""
    from app.core.errors import AppError, app_error_handler

    err = AppError(code="x.y", message="hi", status_code=400)
    loop = asyncio.new_event_loop()
    try:
        response = loop.run_until_complete(app_error_handler(None, err))
    finally:
        loop.close()
    body = json.loads(response.body)
    assert response.status_code == 400
    assert body == {"error": {"code": "x.y", "message": "hi"}}


# ──────────────────────────────────────────────────────────────────────
# review iter1 IMPORTANT 1 — AppError.details merge order 안전
# ──────────────────────────────────────────────────────────────────────


def test_app_error_rejects_reserved_details_keys():
    """details 에 code/message 들어있으면 __init__ 시점 fail-fast (defense in depth)."""
    from app.core.errors import AppError

    with pytest.raises(ValueError, match="reserved keys"):
        AppError(code="x.y", message="hi", details={"code": "evil"})
    with pytest.raises(ValueError, match="reserved keys"):
        AppError(code="x.y", message="hi", details={"message": "evil"})
    with pytest.raises(ValueError, match="reserved keys"):
        AppError(code="x.y", message="hi", details={"code": "a", "extra": "b"})


def test_app_error_handler_base_fields_win_over_details():
    """handler 가 details 의 reserved key 를 제거 후 base 우선 적용 — fail-fast 우회 시에도 안전."""
    from app.core.errors import AppError, app_error_handler

    # Bypass __init__ guard by direct attribute mutation (subclass mistake 시뮬).
    err = AppError(code="real", message="real-msg", status_code=400)
    err.details = {"code": "fake", "message": "fake-msg", "extra": "OK"}

    loop = asyncio.new_event_loop()
    try:
        response = loop.run_until_complete(app_error_handler(None, err))
    finally:
        loop.close()
    body = json.loads(response.body)
    assert body["error"]["code"] == "real", "handler 가 details 의 fake code 에 덮였음"
    assert body["error"]["message"] == "real-msg"
    assert body["error"]["extra"] == "OK"


# ──────────────────────────────────────────────────────────────────────
# review iter1 BLOCKING — empty-input D6 episode invariant
# ──────────────────────────────────────────────────────────────────────


def test_preflight_passes_for_empty_d6_episode_with_zero_chain_bg(
    tmp_path, monkeypatch,
):
    """D6 mode 에 chain bg job 0 인 episode — 4 consumer 가 empty-path manifest +
    valid stamp 를 남기면 preflight 통과.

    각 step 의 empty-path 분기 (`floor_plan_prompt_step.py:131-147`,
    `background_prompt_step.py:184-198`, `background_render_step.py:238-244`)
    가 모두 hash stamp 하므로 catalog 가 비어있어도 (hash 는 빈 catalog 의
    deterministic SHA) 정합성 유지.
    """
    from app.core.bg_catalog import compute_bg_catalog_hash, compute_shot_binding_hash

    # 빈 catalog / 빈 shot_background_map 의 deterministic hash
    empty_catalog_hash = compute_bg_catalog_hash({})
    empty_binding_hash = compute_shot_binding_hash({})

    monkeypatch.setattr("app.core.config.settings.projects_dir", str(tmp_path))
    _write_manifest(tmp_path, "p1", "e1", "background_master_plan", {
        "bg_catalog_hash": empty_catalog_hash,
        "shot_binding_hash": empty_binding_hash,
        "background_catalog": {},
        "shot_background_map": {},
        "plans": {},
    })
    # 4 consumer 모두 empty-path 분기 시뮬 — 빈 data + 같은 hash stamp.
    for step in _CONSUMERS:
        _write_manifest(tmp_path, "p1", "e1", step, {
            "consumed_bg_catalog_hash": empty_catalog_hash,
            "consumed_shot_binding_hash": empty_binding_hash,
        })

    from app.services.dispatcher_preflight import check_bg_catalog_freshness
    check_bg_catalog_freshness("p1", "e1", enforce_shot_binding=True)