"""Reference attachment contract validator — D5 identity-level + Area #5 sidecar edition.

spec: docs/superpowers/specs/2026-05-09-attached-reference-identity-contract-design.md §4.4
      docs/superpowers/specs/2026-05-18-area-5-reference-phrase-phantom-guard-replacement-design.md (Area #5 v1)
plan: docs/superpowers/plans/2026-05-09-attached-reference-identity-contract-implementation.md Task 3
      docs/superpowers/plans/2026-05-18-area-5-reference-phrase-phantom-guard-replacement-implementation.md

required_refs 의 (kind, id) 가 attached_meta 에 exact match 로 존재해야 통과.
substring/text 매칭 폐기. P1 (label 추론 X) + P2 (fallback meta 위조 X).

핵심 contract (D5):
- character_outlook required → attached_meta 에 ("character_outlook", id) exact 존재.
  base "character" / "character_state" 는 만족 X (close_framing 무관).
- character required (FINDING 9 W2 Cat2 — base_id_required subject) → attached_meta
  에 ("character", id) exact 존재. "character_outlook" 은 만족 X. (P8 Inc2-b:
  "character_state" 는 같은 캐릭터의 state-variant ref 이므로 sid prefix 일치 시 만족.)
- background required + NOT close → attached_meta 에 ("background", bg_id) exact
  존재 또는 ("background_prev_shot", chain_bg_lookup(bg_id)) lineage 일치.
  chain_bg_lookup(bg_id)=None 이면 lineage 확인 불가 = missing.
- readiness_policy=block_if_missing + required_refs empty → drift fail-fast.
- meta length mismatch → 즉시 fail-fast (invariant 1).
- sidecar phantom guard (Area #5 v1): reference_phrase_kinds (per-variation, producer
  emit) 의 각 kind 가 attached_meta 의 kind set 과 strict exact compare. legacy
  prompt-prose classifier + window-based heuristic 폐기. step 4 의 close-framing
  background skip 은 보존하지만 step 6 은 strict (axis 분리).

D2 strict extractor 그대로 보존 (사용자 binding G1) — `or {}` silent fallback 금지.
"""
from __future__ import annotations

import logging
from typing import Any, Callable, Dict, List, Optional, Set, Tuple

from app.core.errors import AppError


logger = logging.getLogger(__name__)


class RefContractError(AppError):
    """Required ref 미첨부 또는 from-the-reference guard 위반 — generate 차단.

    HTTP 422 (validation error). caller 가 retry 1 회 (var_t2i rebuild) 또는 propagate.
    """

    def __init__(self, detail: str):
        super().__init__(
            code="ref_contract.violation",
            message=detail,
            status_code=422,
        )


# Area #5 v1: legacy prompt-prose classifier helpers and token/window regex machinery
# were removed. Step 6 phantom guard is now producer-emitted reference_phrase_kinds
# sidecar exact compare.


def _meta_has_kind(attached_set: Set[Tuple[str, str]], kinds: Tuple[str, ...]) -> bool:
    """attached_meta set 안에 지정 kind 가 하나라도 있는지 (D5 §4.4 phantom guard)."""
    attached_kinds = {k for k, _ in attached_set}
    return any(kind in attached_kinds for kind in kinds)


def _normalize_required_refs(raw: Any) -> Dict[str, List[str]]:
    """required_refs canonical normalization — manifest list shape ↔ dict shape.

    Two accepted shapes (D2):
      - manifest canonical (list of dicts):
          [{"kind": "character_outlook", "id": "C01O02", "policy": "required"}, ...]
      - legacy dict:
          {"character_outlook": ["C01O02"], "background": ["L01"]}

    Returns dict[kind, list[id]]. 빈/None → {}.

    Iter1+iter2 BLOCKING (Codex review): malformed input 은 silent drop 금지 —
    explicit []/{}/None 은 허용하되, wrong type / list item non-dict /
    kind|id 누락 / dict value non-list / id non-string / policy drift 는
    RefContractError fail-fast (silent fallback 회로 차단, 사용자 binding G1).
    """
    if raw is None:
        return {}
    if isinstance(raw, dict):
        if not raw:
            return {}
        normalized: Dict[str, List[str]] = {}
        for k, v in raw.items():
            if not isinstance(k, str):
                raise RefContractError(
                    f"required_refs malformed: dict key must be str (kind), "
                    f"got {type(k).__name__}={k!r}"
                )
            if not isinstance(v, list):
                raise RefContractError(
                    f"required_refs malformed: dict value for kind={k!r} must "
                    f"be list, got {type(v).__name__}={v!r}"
                )
            for entry in v:
                if not isinstance(entry, str):
                    raise RefContractError(
                        f"required_refs malformed: dict value list for "
                        f"kind={k!r} must contain str ids, "
                        f"got {type(entry).__name__}={entry!r}"
                    )
            normalized[k] = list(v)
        return normalized
    if isinstance(raw, list):
        if not raw:
            return {}
        normalized = {}
        for idx, item in enumerate(raw):
            if not isinstance(item, dict):
                raise RefContractError(
                    f"required_refs malformed: list item[{idx}] must be dict, "
                    f"got {type(item).__name__}={item!r}"
                )
            kind = item.get("kind")
            rid = item.get("id")
            if not isinstance(kind, str) or not kind:
                raise RefContractError(
                    f"required_refs malformed: list item[{idx}] 'kind' must "
                    f"be non-empty str, got {type(kind).__name__}={kind!r}"
                )
            if not isinstance(rid, str) or not rid:
                raise RefContractError(
                    f"required_refs malformed: list item[{idx}] 'id' must "
                    f"be non-empty str, got {type(rid).__name__}={rid!r}"
                )
            policy = item.get("policy")
            if not isinstance(policy, str):
                raise RefContractError(
                    f"required_refs malformed: list item[{idx}] 'policy' must "
                    f"be str (only 'required' is allowed), "
                    f"got {type(policy).__name__}={policy!r}"
                )
            if policy != "required":
                raise RefContractError(
                    f"required_refs malformed: list item[{idx}] 'policy' must "
                    f"be exactly 'required' (producer drift), got {policy!r}"
                )
            normalized.setdefault(kind, []).append(rid)
        return normalized
    raise RefContractError(
        f"required_refs malformed: must be list or dict, "
        f"got {type(raw).__name__}={raw!r}"
    )


def validate_attached_refs(
    rpc: Optional[Dict[str, Any]],
    labeled_refs: List[Tuple[str, bytes]],
    attached_meta: List[Tuple[str, str]],
    prompt: str,
    is_close_framing: bool,
    *,
    chain_bg_lookup: Optional[Callable[[str], Optional[str]]] = None,
    reference_phrase_kinds: List[str],
) -> None:
    """spec D5 §4.4 + Area #5 v1 — identity-level required_refs vs attached_meta +
    prev_shot lineage + readiness_policy + meta length invariant + sidecar phantom guard.

    위반 시 RefContractError raise (fail-fast). HTTP 422.

    검사 순서:
      1. invariant 1 — len(labeled_refs) == len(attached_meta).
      2. rpc shape strict (D2 보존, 사용자 binding G1 — `or {}` 금지).
      3. character_outlook strict — ("character_outlook", cid) exact match.
         close_framing 분기 없음 (사용자 binding G2).
      3a/3b. prop / character (base_id_required) strict subset — ("prop", id) /
         ("character", id) exact match. close_framing 무관.
      4. background — NOT close framing 시 ("background", bg_id) exact 또는
         chain_bg_lookup(bg_id) 가 loc_id 반환 시 ("background_prev_shot", loc_id)
         lineage 일치. chain_bg_lookup=None / lookup 결과 None 이면 lineage
         확인 불가 = missing (사용자 binding G3).
      5. readiness_policy=block_if_missing consistency (required_refs empty 시 drift).
      6. Sidecar phantom guard (Area #5 v1) — `reference_phrase_kinds` (producer
         emit) 의 각 kind 가 `attached_meta` 에 해당 kind 1+ 있는지 strict exact
         compare. close-framing 무관 (legacy 30/60자 window classifier 폐기,
         strict exact compare). `prompt` arg 보존 — error message context 용,
         step 6 안에서 prompt body read 0.

    `reference_phrase_kinds` keyword-only required (no default). 누락 시 TypeError.
    None / non-list / invalid enum value 는 step 6 안에서 RefContractError fail-fast.
    """
    # 1. invariant 1 — meta length match
    if len(labeled_refs) != len(attached_meta):
        raise RefContractError(
            f"meta length mismatch: labeled_refs={len(labeled_refs)} "
            f"attached_meta={len(attached_meta)} (D5 §2.3 invariant 1)"
        )

    # 2. rpc shape strict (D2 보존)
    if rpc is None:
        required: Dict[str, List[str]] = {}
        readiness: Optional[str] = None
    else:
        if not isinstance(rpc, dict):
            raise RefContractError(
                f"rpc malformed: must be dict, got {type(rpc).__name__}={rpc!r}"
            )
        if "asset_requirements" not in rpc:
            raise RefContractError(
                "rpc malformed: 'asset_requirements' field missing — "
                "RenderPromptCard contract requires this field"
            )
        asset_req = rpc.get("asset_requirements")
        if asset_req is None:
            raise RefContractError(
                "rpc malformed: 'asset_requirements' is None — "
                "asset_requirements must include explicit required_refs: [] or {}"
            )
        if not isinstance(asset_req, dict):
            raise RefContractError(
                f"rpc malformed: 'asset_requirements' must be dict, "
                f"got {type(asset_req).__name__}={asset_req!r}"
            )
        if "required_refs" not in asset_req:
            raise RefContractError(
                "rpc malformed: 'asset_requirements.required_refs' field "
                "missing — use explicit [] or {} for no-required-refs"
            )
        raw_required_refs = asset_req.get("required_refs")
        if raw_required_refs is None:
            raise RefContractError(
                "rpc malformed: 'asset_requirements.required_refs' is None — "
                "use explicit [] or {} for no-required-refs"
            )
        required = _normalize_required_refs(raw_required_refs)
        readiness = asset_req.get("readiness_policy")  # optional — None 허용

    attached_set: Set[Tuple[str, str]] = set(attached_meta)

    # 3. character_outlook strict (close_framing 무관 — G2)
    # P8 Inc2-b 연장 (2026-07-02): immobilized subject 는 resolver §2.5 가 outfit
    # composite 대신 state-variant ref ("character_state", "C##:state") 를 attach
    # 한다 — 살아있는 outfit composite 는 시신/부상 외형과 모순이라 state ref 가
    # 그 shot 의 더 구체적 외형 SOT. 같은 base 캐릭터(C##O## 의 C## 구조 파싱)의
    # character_state attach 는 required character_outlook 을 충족으로 인정한다.
    # (이전 "outlook 은 state 로 충족 X" 계약은 resolver 의 composite→state 대체와
    # 데드락 — S12 sh10 production 실측으로 정정.)
    for cid in required.get("character_outlook", []):
        if ("character_outlook", cid) in attached_set:
            continue
        base_cid = cid.split("O", 1)[0]
        if any(
            k == "character_state"
            and isinstance(v, str)
            and v.split(":", 1)[0] == base_cid
            for k, v in attached_set
        ):
            continue
        raise RefContractError(
            f"required character_outlook {cid!r} missing — "
            f"attached={sorted(attached_set)} "
            f"(base 'character' ref 는 충족 X; 같은 캐릭터의 'character_state' 만 "
            f"immobilized 대체로 충족)"
        )

    # 3a. Patch A — prop strict subset (Tier 3).
    # required_refs(kind='prop') 의 (kind, id) 가 attached_meta 에 exact match 로
    # 존재해야 한다. character_outlook 패턴 일관. close_framing 무관 — prop 자체는
    # framing 과 독립 (사진/액자/문서 prop 은 close-up 에서도 ref 필수).
    for prop_id in required.get("prop", []):
        if ("prop", prop_id) not in attached_set:
            raise RefContractError(
                f"required prop {prop_id!r} missing — "
                f"attached={sorted(attached_set)} "
                f"(Tier 3 strict subset violation, Patch A)"
            )

    # 3b. FINDING 9 W2 (Cat2) — required character base strict subset.
    # required_refs(kind='character') 의 (kind, id) 가 attached_meta 에 exact
    # match 로 존재해야 한다. base_id_required subject 의 base ref attach 강제 —
    # step 3 character_outlook / step 3a prop 패턴 일관. close_framing 무관.
    #
    # P8 Inc2-b (2026-06-22): state-variant ref (dead/injured) 도 required character
    # 를 충족한다. bare C## immobilized subject (시신 등) 는 resolver section 2.5 가
    # base passport bytes 대신 dead/injured variant bytes 를 attach 하며 meta 를
    # ("character_state", "C##:state") 로 기록한다. 이는 같은 캐릭터의 더 구체적인
    # reference 이므로 required character ("character", "C##") 를 만족시킨다 — sid
    # prefix 매칭. (2026-07-02: step 3 character_outlook 도 같은 논리로 같은 캐릭터
    # character_state 를 충족으로 인정 — 위 step 3 주석 참조.)
    for char_id in required.get("character", []):
        if ("character", char_id) in attached_set:
            continue
        if any(
            k == "character_state"
            and isinstance(v, str)
            and v.split(":", 1)[0] == char_id
            for k, v in attached_set
        ):
            continue
        raise RefContractError(
            f"required character {char_id!r} missing — "
            f"attached={sorted(attached_set)} "
            f"(base_id_required subject base ref 미첨부, FINDING 9 W2)"
        )

    # 4. background — exact bg_id 또는 prev_shot lineage 일치 (NOT close 시)
    if not is_close_framing:
        # D6 T9 (R2 I3) — bg_id 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 호환).
        from app.core.bg_state_vocab import BG_ID_RE
        from app.core.errors import StaleUpstreamError

        # W21B space_set_bg Phase 2 — synthetic `space_set_bg:<gid>:<plate_key>`
        # background 는 opt-in overlay 가 만든 ★명시적 1급 background source★ 이지
        # stale background_render fallback 이 아니다. loader 가 key 충돌 시 space
        # plate 를 우선 주입하면 attached bg_id 가 synthetic 으로 바뀌어 required
        # L##B## exact match 가 깨진다 — required 의 의도("이 shot 은 background
        # ref 가 있어야 한다")는 충족된 상태이므로 인정한다. overlay 는
        # space_set_bg_enabled ON + shot 배정 존재 시에만 발생 — attached 의
        # space ref 존재 자체가 source 증거라 settings gate 불필요. shot 의
        # background ref slot 은 1개이므로 required 복수여도 전체 충족 처리.
        _space_set_bg_attached = any(
            k == "background" and isinstance(v, str) and v.startswith("space_set_bg:")
            for k, v in attached_set
        )
        # W22 직행 (2026-07-10) — 야외 직행 캐논(`outdoor_canon:<place>`)도
        # 동일 논리의 명시적 1급 background source: opt-in flag ON + 캐논/
        # grounding 산출 존재 시에만 attach 되므로 attach 자체가 source 증거.
        # required 의 의도("background ref 필요")는 캐논 실사+맵으로 충족.
        _outdoor_canon_attached = any(
            k == "background" and isinstance(v, str) and v.startswith("outdoor_canon:")
            for k, v in attached_set
        )

        for bg_id in required.get("background", []):
            if ("background", bg_id) in attached_set:
                continue
            if _space_set_bg_attached:
                continue
            if _outdoor_canon_attached:
                continue
            # G3: chain_bg_lookup(bg_id)=None 이면 lineage 확인 불가 → missing.
            #     "required_loc and ..." 에서 falsy 면 lineage 분기 진입 X →
            #     아래 raise 로 즉시 fail-fast.
            required_loc = chain_bg_lookup(bg_id) if chain_bg_lookup else None
            if required_loc and ("background_prev_shot", required_loc) in attached_set:
                continue
            if BG_ID_RE.match(bg_id or ""):
                # D6 path — preflight 우회된 stale render. force re-run 권장.
                raise StaleUpstreamError(
                    upstream="background_render",
                    expected_bg_catalog_hash="(unknown — validator fallback path)",
                    observed_bg_catalog_hash="",
                    missing_in_render=[bg_id],
                    remediation=(
                        f"validator detected D6 bg_id {bg_id!r} missing from "
                        f"attached_meta (preflight bypass) — POST /steps/"
                        f"background_render?mode=force"
                    ),
                )
            raise RefContractError(
                f"required background {bg_id!r} missing "
                f"(expected exact ('background', {bg_id!r}) or "
                f"('background_prev_shot', {required_loc!r})) — "
                f"attached={sorted(attached_set)}"
            )

    # 5. readiness_policy consistency
    if readiness == "block_if_missing" and not any(required.values()):
        raise RefContractError(
            "rpc drift: readiness_policy=block_if_missing but required_refs is empty"
        )

    # 6. Sidecar phantom guard (Area #5 v1) — reference_phrase_kinds vs attached_meta
    #    kind set exact 비교. legacy prompt-prose classifier + window-based heuristic
    #    폐기, strict exact compare. close-framing 무관 (step 6 = strict, legacy skip
    #    보존 X; step 4 의 close-framing background skip 은 별도 axis).
    if reference_phrase_kinds is None:
        raise RefContractError(
            "reference_phrase_kinds missing — producer (scene_detail v26) emit "
            "required, no silent fallback (Area #5 v1)"
        )
    if not isinstance(reference_phrase_kinds, list):
        raise RefContractError(
            f"reference_phrase_kinds malformed: must be list, "
            f"got {type(reference_phrase_kinds).__name__}={reference_phrase_kinds!r}"
        )
    _ALLOWED_KINDS = ("character", "background", "prop")
    for kind in reference_phrase_kinds:
        if kind not in _ALLOWED_KINDS:
            raise RefContractError(
                f"reference_phrase_kinds invalid enum value: {kind!r} "
                f"(allowed: {_ALLOWED_KINDS})"
            )
        if kind == "character" and not _meta_has_kind(
            attached_set, ("character", "character_outlook", "character_state")
        ):
            raise RefContractError(
                f"reference_phrase_kinds declares kind={kind!r} but no character "
                f"meta attached — attached={sorted(attached_set)}"
            )
        if kind == "background" and not _meta_has_kind(
            attached_set, ("background", "background_prev_shot")
        ):
            raise RefContractError(
                f"reference_phrase_kinds declares kind={kind!r} but no background "
                f"meta — attached={sorted(attached_set)}"
            )
        if kind == "prop" and not _meta_has_kind(attached_set, ("prop",)):
            raise RefContractError(
                f"reference_phrase_kinds declares kind={kind!r} but no prop meta — "
                f"attached={sorted(attached_set)}"
            )