"""`location_part` 참조 **묶음** — 맥락 + 상세. ★inert. 유료 0.

## 무엇인가

사용자 확정 (2026-08-31): 「1960년대 이발소 해서 하나로 검색 조사해야 해 …
아님 **둘을 찾아서 참조 두 개 이상으로** … 참조 이미지 여러 개로 해도 돼!」

`location_part` 하나가 **논리 멤버 둘**을 갖는다 —

    맥락(context)  그 장소   — 「1960년대 이발소」
    상세(detail)   그 부분   — 「이발소 회전 간판」

**같은 사진이면 한 장으로 합치고**, 다르면 둘 다 기존 `background` 자리에
붙인다. ★합쳐도 **논리 멤버 둘의 처분·신원·계보는 따로 남는다**.

## 세 가지를 고쳤다 (Codex 재현 2026-08-31)

    ①하류 SOT 가 뒤집혀 있었다 — raw `status` 를 봤다.
      정본은 **`outcome`**(`selected` | `reference_unavailable`)이고 raw
      status 는 **감사용**이다. `outcome` 이 없거나 모르는 값이면 **조용히
      건너뛰지 않고 선다** — 그러면 `selected` 의무가 사라진다.
    ②exact 열쇠에 **실제 획득 신원**이 빠졌다. `grounding:LP01:detail` 만으로는
      **옛 자산**이 신원 바뀐 새 의무를 **만족시킨다**(재현됨).
    ③「같은 bytes 면 합친다」가 **구현이 없었다.** 같은 사진을 넣으면 2장이
      나오고 역할 차례도 뒤집혔다.

## 왜 새 슬롯을 안 만드나

야외 직행 경로가 **이미** 사진+지도 두 장을 같은 `background` kind 로 넣는다.
그 모양을 그대로 쓴다 — 네 평행 목록에 **덧붙인다**.
★실측: **맥락·상세는 기존 배경판을 대신하지 못한다**(안 붙이면
`StaleUpstreamError`). 그래서 이것은 **더하는** 설계다.

## 아직 안 쓰인다

`app/` 어디서도 부르지 않는다. D 원자적 cutover 에서 general scene 경로가
**RPC sidecar 가 있을 때만** 이것을 부른다.
"""
from __future__ import annotations

import hashlib
from typing import Any, Dict, List, Optional, Sequence, Tuple

from app.modules.pipeline.grounding_host_context import (PURPOSE_CONTEXT,
                                                         PURPOSE_DETAIL,
                                                         PURPOSES)

#: ★★이 모듈이 **묶는 규칙**의 판. `SINGLE_SUBJECT_KINDS`·dedupe 열쇠·
#:  `meta_value` 모양·역할 대응이 바뀌면 **여기를 올린다**.
#:  ★재개 지문(`SceneDetailStep._config_hash`)이 이 값을 접는다 — 안 접으면
#:   묶는 규칙을 바꿔도 옛 카드가 current 로 읽힌다 (Codex NON-BLOCK 09-02).
#: ★2026-09-02: `selected` 인데 **일부러 신원이 없는** 줄이 생겼다
#:  (고증 미확인). 계약이 그것을 결손으로 안 읽는다 — 판을 올린다.
BUNDLE_CONTRACT_VERSION = "2.202609020800"

#: 붙는 자리. ★**새 슬롯을 안 만든다** — 기존 kind 를 쓴다.
BUNDLE_KIND = "background"

#: 멤버가 **제 자리를 스스로 말한다**. 없으면 배경 자리다.
#:  ★★2026-09-01 아웃룩이 들어오면서 필요해졌다 — 아웃룩은 기존
#:   `character_outlook` 자리에 붙고, 장소·장소부분은 `background` 다.
#:   한 상수로 두면 아웃룩 사진이 **배경으로 실린다** (Codex 09-01).
MEMBER_KIND_KEY = "ref_kind"
MEMBER_ROLE_KEY = "ref_role"

#: 아웃룩 고증 사진의 역할. ★**새 역할을 안 만든다** — 이미 선언된
#:  `outfit_ref_explicit` 을 쓴다. 그 지시가 「standalone outfit/costume
#:  reference · dress the character in the outfit」이라 시대 복장 조사 사진에
#:  맞고, production emitter 가 **0곳**이라 기존 쓰임과 안 부딪힌다(실측).
#:  ★`outfit_ref_inline` 은 안 쓴다 — 그 지시는 「그 사람의 신원과 옷을
#:   맞춰라」라서, 사진 속 **남의 얼굴**을 인물 신원으로 가져간다.
OUTLOOK_KIND = "character_outlook"
OUTLOOK_ROLE = "outfit_ref_explicit"

#: ★★한 장이 **주인 하나**만 가져야 하는 자리들.
#:  배경은 부모 장소 사진 한 장이 형제 `LP` 여럿의 의무를 함께 덮는 것이
#:  옳다 — 지시가 「그 장소의 빛·건축·분위기」라 주인이 여럿이어도 말이 된다.
#:  아웃룩은 다르다: 지시가 「**이 사람에게** 이 옷을 입혀라」라서 주인이
#:  둘이면 모델이 어느 쪽인지 못 읽는다. 그래서 같은 사진이라도 **주인마다
#:  따로** 붙인다 (Codex BLOCK 2026-09-02).
SINGLE_SUBJECT_KINDS = (OUTLOOK_KIND,)

#: `attached_meta` 값의 앞머리. ★야외가 `outdoor_canon:<place>` 를 쓰는 것과
#:  같은 꼴 — 값 자체가 **어디서 왔는지**를 말한다.
SOURCE_PREFIX = "grounding"

#: 하류가 보는 **처분**. ★raw `status` 는 감사용이고 의무를 정하지 않는다.
OUTCOME_SELECTED = "selected"
OUTCOME_UNAVAILABLE = "reference_unavailable"
OUTCOMES = (OUTCOME_SELECTED, OUTCOME_UNAVAILABLE)

#: 쓰임 조합 → 프롬프트 역할.
#:  ★맥락 단독은 기존 `background_general` 을 **재사용**한다 — 그 지시문이
#:   「그 장소의 빛·건축·분위기를 쓰라」라서 맞는다(실측).
#:  ★상세와 결합은 **새 역할**이다 — 근접 사진에서 「분위기」를 가져오라는
#:   말이 되면 안 된다. D cutover 에서 `REF_ROLE_VALUES` 에 선언한다.
ROLE_BY_PURPOSES: Dict[Tuple[str, ...], str] = {
    (PURPOSE_CONTEXT,): "background_general",
    (PURPOSE_DETAIL,): "grounding_part_detail_ref",
    (PURPOSE_CONTEXT, PURPOSE_DETAIL): "grounding_context_detail_ref",
}


class BundleContractError(RuntimeError):
    """묶음이 계약을 어겼다. ★조용히 넘어가지 않는다."""


def _purposes_key(purposes: Sequence[str]) -> Tuple[str, ...]:
    """쓰임들의 **정본 차례**. ★집합을 결정적으로 만든다."""
    got = sorted({str(p) for p in purposes})
    for p in got:
        if p not in PURPOSES:
            raise BundleContractError(f"모르는 쓰임 {p!r} — {PURPOSES}")
    if not got:
        raise BundleContractError("쓰임이 없다")
    return tuple(got)


def _outcome_of(m: Dict[str, Any]) -> str:
    """멤버의 **처분**. ★없거나 모르는 값이면 **선다**.

    ★★★조용히 건너뛰면 `selected` 의무가 사라진다 (Codex 재현 2026-08-31).
    앞 판은 raw `status` 를 봤고, `outcome` 이 비면 「아닌가 보다」로 지나갔다.
    """
    got = str(m.get("outcome") or "").strip()
    if got not in OUTCOMES:
        raise BundleContractError(
            f"처분이 {got!r} 다 — {OUTCOMES} 중 하나여야 한다. "
            f"(raw status 는 감사용이고 의무를 정하지 않는다)")
    return got


def _fold(parts: Sequence[str]) -> str:
    h = hashlib.sha256()
    for p in parts:
        h.update(str(p).encode("utf-8"))
        h.update(b"\x00")
    return h.hexdigest()[:16]


def meta_value(*, subject_final_id, purposes: Sequence[str],
               member_identities: Sequence[str],
               covered_final_ids: Sequence[str] = ()) -> str:
    """`attached_meta` 에 실을 값. ★**출처·쓰임·신원·덮는 의무**를 다 나른다.

    ★★★신원이 빠지면 **옛 자산이 새 의무를 만족시킨다** (Codex 재현) —
    획득 계약이 바뀌어 다시 사야 하는데 옛 사진이 붙어 있으면 통과했다.

    Args:
        subject_final_id: 이 참조의 **주인**. 부모 장소 맥락이면 `L01`,
            그 장소 부분 상세면 `LP01` 이다.
        covered_final_ids: 이 한 장이 **덮는 의무들**. 부모 장소 맥락 한 장이
            형제 `LP01`·`LP02` 의 맥락 의무를 같이 덮는다 — 그때 여기에 둘 다
            적힌다. ★★앞 판은 `subject_final_id` 밖에 없어서 형제마다 **같은
            사진을 따로** 넣게 되어 있었다 (Codex BLOCK · 09-01): 모델에 같은
            사진이 중복되고 참조 장수 상한도 낭비한다.
            ★비면 주인 자신만 덮는다.
    """
    # ★★한 장이 **여러 주인**의 것일 수 있다 — 같은 사진이면 한 장으로 합치고
    #  주인들을 union 한다 (Codex BLOCK · 09-01). 그래서 목록도 받는다.
    subjects = ([str(subject_final_id)]
                if isinstance(subject_final_id, str)
                else sorted({str(x) for x in subject_final_id}))
    subjects = [x for x in subjects if x.strip()]
    if not subjects:
        raise BundleContractError("`subject_final_id` 가 비었다")
    ids = sorted({str(x) for x in member_identities if str(x or "").strip()})
    if not ids:
        raise BundleContractError(
            "획득 신원이 없다 — 무엇으로 샀는지가 값에 안 실린다")
    ps = _purposes_key(purposes)
    cov = sorted({str(x) for x in covered_final_ids if str(x or "").strip()}
                 or set(subjects))
    return (f"{SOURCE_PREFIX}:{'+'.join(subjects)}:{'+'.join(ps)}:"
            f"{_fold(ids)}:{_fold(cov)}")


def parse_meta_value(value: str) -> Optional[Dict[str, Any]]:
    """`attached_meta` 값을 되읽는다. 조사 것이 아니면 `None`.

    ★이름을 해석하지 않는다 — **우리가 만든 값**의 모양만 읽는다.
    """
    parts = str(value or "").split(":")
    if len(parts) != 5 or parts[0] != SOURCE_PREFIX:
        return None
    try:
        ps = _purposes_key(parts[2].split("+"))
    except BundleContractError:
        return None
    return {"subject_final_ids": parts[1].split("+"), "purposes": list(ps),
            "identity_fold": parts[3], "covered_fold": parts[4]}


#: 좌표 갈래. ★**모호하게 두지 않는다** (Codex 2026-08-31) —
#:  `asset_id or path` 로 두면 **읽지도 않은 asset UUID 가 계보로 남는다**.
SOURCE_ASSET = "asset"
SOURCE_FILE = "file"
SOURCES = (SOURCE_ASSET, SOURCE_FILE)


def coordinate_of(m: Dict[str, Any]) -> Dict[str, str]:
    """멤버 하나의 **좌표**. ★갈래를 명시로 받는다.

        {"source": "asset", "asset_id": "<UUID>"}   ← DB/service 가 읽는다
        {"source": "file",  "path": "<정본 경로>"}   ← 파일에서 읽는다.
                                                      ★asset_id 를 **주장 안 한다**

    ★`ref_image_map` 의 열쇠는 **참조 map key** 지 ImageAsset UUID 가 아니다
    (실측). 그것을 쓰려면 `ref_map_key` 로 **따로** 부른다 — `asset_id` 라고
    부르면 읽지 않은 UUID 가 계보로 남는다.
    """
    src = str(m.get("source") or "").strip()
    if src not in SOURCES:
        raise BundleContractError(
            f"좌표 갈래가 {src!r} 다 — {SOURCES} 중 하나여야 한다. "
            "`asset_id or path` 로 모호하게 두지 않는다")
    if src == SOURCE_ASSET:
        v = str(m.get("asset_id") or "").strip()
        if not v:
            raise BundleContractError("`source=asset` 인데 `asset_id` 가 없다")
        return {"source": src, "asset_id": v}
    v = str(m.get("path") or "").strip()
    if not v:
        raise BundleContractError("`source=file` 인데 `path` 가 없다")
    return {"source": src, "path": v}


def _coord_key(c: Dict[str, str]) -> Tuple[str, str]:
    return (str(c.get("source") or ""),
            str(c.get("asset_id") or c.get("path") or ""))

def plan_bundle(members: Sequence[Dict[str, Any]]
                ) -> Tuple[List[Dict[str, Any]], List[Tuple[str, str]]]:
    """논리 멤버들 → **붙일 자리들** + **정확히 붙어야 하는 짝**.

    ★★★**bytes 를 안 받는다** — CP 는 JSON 이라 bytes 를 못 담는다.
    durable sidecar 는 **좌표**(`source`+`asset_id|path`)와
    `content_sha256` 을 적고, 소비하는 쪽이 읽어 **해시를 확인**한다.

    ★★★**같은 사진이면 한 장**으로 합치되 — 합쳐도 **논리 멤버 둘의
    좌표·신원·계보를 전부 보존**한다 (Codex 재현 2026-08-31: 앞 판은
    `setdefault` 가 **첫 좌표만** 잡아 둘째 자산 계보를 잃었고,
    `plan_bundle(AB) != plan_bundle(BA)` 였다).

    ★서로 다른 좌표가 **같은 sha** 를 말하면 둘 다 남기고, **어느 것으로
    읽을지는 결정적으로** 고른다(`coordinates[0]`). 읽은 뒤 해시를 확인하므로
    거짓 주장은 그때 걸린다.
    """
    by_hash: Dict[str, Dict[str, Any]] = {}
    for m in members:
        outcome = _outcome_of(m)
        lp = str(m.get("subject_final_id") or "")
        if not lp:
            raise BundleContractError("`subject_final_id` 가 비었다")
        purpose = str(m.get("purpose") or "")
        _purposes_key([purpose])                    # ★모르는 쓰임이면 선다
        # ★★★「아직 아무도 안 봤다」와 「있어야 하는데 없다」를 **가른다**
        #  (2026-09-02). 앞 판은 신원이 없으면 무조건 지나가게 넓혔는데,
        #  그러면 **진짜 결손**(붙어야 하는데 신원이 빠진 줄)까지 조용히
        #  통과한다 — 잠가 둔 시험이 잡았다. 투영이 **일부러 안 만든 것**만
        #  지나간다. 그 표시는 투영이 적는다.
        deliberately_none = m.get("not_attached") is not None
        if outcome != OUTCOME_SELECTED or deliberately_none:
            # ★못 구한 것은 붙일 자리도 없고 **required 도 아니다**.
            #  ★★신원도 해시도 **없는 것이 맞다** — 그것을 요구하면 투영이
            #  없는 값을 지어내게 된다 (Codex · 09-01). 논리 의무는
            #  `grounding_bundle_projection.unavailable_members` 가 남긴다.
            #  ★★★2026-09-02: **골랐지만 아직 아무도 안 본 것**도 여기로
            #   온다. `outcome` 은 `selected` 인데 붙일 신원이 없다 —
            #   고증이 확인되기 전에는 투영이 신원·해시를 **안 만든다**
            #   (Codex BLOCK). 그것을 「계약 위반」으로 읽으면 안 된다.
            continue
        ident = str(m.get("member_identity") or "")
        if not ident:
            raise BundleContractError(
                f"{lp}/{purpose} 에 `member_identity` 가 없다")
        sha = str(m.get("content_sha256") or "").strip()
        if not sha:
            raise BundleContractError(
                f"{lp}/{purpose} 가 `selected` 인데 `content_sha256` 이 없다 "
                "— 사진이 맞는지 확인할 길이 없다")
        coord = coordinate_of(m)
        # ★★★물리 dedupe 는 **내용 해시**로만 한다 (Codex BLOCK · 09-01).
        #  앞 판은 `주인:sha` 라서 `L01` 맥락과 `LP01` 상세가 **같은 사진이어도
        #  두 장**이 나갔다 — 「같은 bytes 면 한 장」이 같은 주인 안에서만
        #  성립했다. 내용이 다르면 여전히 두 장이다.
        # ★★붙는 **자리가 다르면 합치지 않는다** — 같은 bytes 라도 배경
        #  자리와 인물 아웃룩 자리는 서로 다른 요구다 (Codex 09-01).
        kind = str(m.get(MEMBER_KIND_KEY) or BUNDLE_KIND)
        role = str(m.get(MEMBER_ROLE_KEY) or "") or None
        # ★주인을 하나만 갖는 자리는 **주인까지** 열쇠에 넣는다 — 그래야
        #  같은 사진이 두 복장을 덮어도 지시가 누구 것인지 말할 수 있다.
        key = (f"{kind}\u0000{lp}\u0000{sha}"
               if kind in SINGLE_SUBJECT_KINDS else f"{kind}\u0000{sha}")
        slot = by_hash.setdefault(key, {
            "subject_final_ids": [], "content_sha256": sha,
            "coordinates": [], "purposes": [], "member_identities": [],
            "labels": [], "parent_final_ids": [], "covered_final_ids": [],
            "ref_kind": kind, "ref_role": role,
        })
        if slot["ref_role"] != role:
            raise BundleContractError(
                f"{lp} 의 역할이 같은 자리 안에서 갈린다 — "
                f"{slot['ref_role']!r} vs {role!r}")
        slot["subject_final_ids"].append(lp)
        slot["coordinates"].append(coord)
        slot["purposes"].append(purpose)
        slot["member_identities"].append(ident)
        # ★★이 한 장이 **어느 의무들을 덮나**. 부모 장소 맥락 한 장이 형제
        #  LP 여럿의 맥락 의무를 같이 덮는다 (Codex BLOCK · 09-01) — 안 적으면
        #  형제마다 같은 사진을 따로 넣게 된다.
        slot["covered_final_ids"].extend(
            str(x) for x in (m.get("covers") or (lp,)) if str(x or "").strip())
        if m.get("label"):
            slot["labels"].append(str(m["label"]))
        if m.get("parent_final_id"):
            slot["parent_final_ids"].append(str(m["parent_final_id"]))

    plan: List[Dict[str, Any]] = []
    required: List[Tuple[str, str]] = []
    # ★차례를 **정본**으로 — 입력 차례가 결과를 안 바꾸게
    for key in sorted(by_hash):
        s = by_hash[key]
        ps = _purposes_key(s["purposes"])
        subjects = sorted(set(s["subject_final_ids"]))
        val = meta_value(subject_final_id=subjects, purposes=ps,
                         member_identities=s["member_identities"],
                         covered_final_ids=s["covered_final_ids"])
        # ★멤버가 역할을 말했으면 그것을 쓴다 — 아웃룩은 쓰임이 아니라
        #  **자리**가 역할을 정한다. 안 말했으면 쓰임 조합에서 찾는다.
        role = s["ref_role"] or ROLE_BY_PURPOSES.get(ps)
        if role is None:
            raise BundleContractError(
                f"쓰임 조합 {ps} 에 맞는 역할이 없다 — "
                f"{sorted(ROLE_BY_PURPOSES)}")
        # ★★좌표를 **전부** 남기고 **결정적으로** 정렬한다 — 둘째 계보를
        #  잃지 않고, 입력 차례가 바뀌어도 같은 bytes 가 나온다.
        coords = sorted({_coord_key(c): c
                         for c in s["coordinates"]}.items())
        plan.append({
            "subject_final_ids": subjects, "purposes": list(ps),
            "role": role, "meta_value": val,
            "content_sha256": s["content_sha256"],
            "coordinates": [c for _k, c in coords],
            "label": " · ".join(sorted(set(s["labels"]))) or "",
            "member_identities": sorted(set(s["member_identities"])),
            "parent_final_ids": sorted(set(s["parent_final_ids"])),
            # ★한 장이 덮는 의무들 — 형제 LP 가 같은 부모 사진을 **함께** 쓴다
            "covered_final_ids": sorted(set(s["covered_final_ids"])),
            # ★어느 자리에 붙나 — 멤버가 말한 것 (기본은 배경)
            "ref_kind": s["ref_kind"],
        })
        required.append((s["ref_kind"], val))
    return plan, required


def insert_bundle_refs(labeled_refs: List, attached_meta: List,
                       ref_roles: List, ref_role_metadata: List,
                       plan: Sequence[Dict[str, Any]], *,
                       load_bytes: Any, where: str = "") -> int:
    """`plan_bundle` 이 낸 **좌표**로 사진을 읽어 네 목록에 **덧붙인다**.

    Args:
        load_bytes: `(coordinate) -> (bytes, resolved)`.
            ★반환이 **bytes 만이면 안 된다** — 그러면 읽지도 않은 자산 UUID 가
            계보로 남는다.

    ★★★**좌표를 전부 읽어 전부 확인한다** (Codex 재현 2026-08-31).
    앞 판은 `coordinates[0]` 만 확인하면서 metadata 에는 **좌표 전부**를
    계보로 적었다 — 둘째 좌표가 **다른 사진인데 같은 해시를 주장**해도
    통과하고 그 경로가 계보로 남았다. 즉 **확인하지 않은 자산이 계보**가 됐다.
    읽는 것은 **무료**이므로 전부 읽고, 하나라도 없거나 해시가 다르면
    **provider 앞에서 선다**. 쓸 것은 **첫 좌표**로 결정적으로 고른다.

    ★기존 것을 안 지운다. 차례는 `plan_bundle` 이 정한 정본을 지킨다.
    """
    got = list(plan)
    n = 0
    for m in reversed(got):                         # ★정본 차례를 지킨다
        coords = list(m.get("coordinates") or ())
        if not coords:
            raise BundleContractError(
                f"{'+'.join(m['subject_final_ids'])}/{m['purposes']} 에 좌표가 없다")
        want = str(m.get("content_sha256") or "")
        first_raw: Optional[bytes] = None
        first_resolved: Dict[str, Any] = {}
        checked: List[Dict[str, Any]] = []
        for i, c in enumerate(coords):
            raw, resolved = load_bytes(c)
            if not raw:
                raise BundleContractError(
                    f"{'+'.join(m['subject_final_ids'])}/{m['purposes']} 의 사진을 못 읽었다 "
                    f"({c}) — 계보에 적을 좌표는 **전부** 읽혀야 한다")
            sha = hashlib.sha256(raw).hexdigest()
            if sha != want:
                raise BundleContractError(
                    f"{'+'.join(m['subject_final_ids'])}/{m['purposes']} 의 좌표 {c} 가 "
                    f"**다른 사진**을 가리킨다 — 적힌 해시 {want!r} · 읽은 것 "
                    f"{sha!r}. 확인 안 한 자산을 계보로 남기지 않는다")
            checked.append(dict(resolved or {}))
            if i == 0:
                first_raw, first_resolved = raw, dict(resolved or {})
        assert first_raw is not None
        labeled_refs.insert(0, (str(m.get("label") or ""), first_raw))
        attached_meta.insert(0, (str(m.get("ref_kind") or BUNDLE_KIND),
                                 m["meta_value"]))
        ref_roles.insert(0, m["role"])
        meta: Dict[str, Any] = {
            "pipeline_role": m["role"],
            "purposes": list(m["purposes"]),
            # ★한 장이 **여러 주인**의 것일 수 있다 — 같은 사진이면 합쳤다
            "subject_final_ids": list(m["subject_final_ids"]),
            "grounding_subject_ids": list(m["subject_final_ids"]),
            "covered_final_ids": list(m.get("covered_final_ids") or ()),
            "member_identities": list(m["member_identities"]),
            "content_sha256": m["content_sha256"],
            # ★★계보는 **전부 읽어 확인한 것**만 적는다
            "coordinates": list(coords),
            "verified_sources": checked,
            "resolved_source": first_resolved,
        }
        if m.get("parent_final_ids"):
            meta["parent_final_ids"] = list(m["parent_final_ids"])
        if where:
            meta["where"] = where
        ref_role_metadata.insert(0, meta)
        n += 1
    return n


def assert_grounding_attached(required: Sequence[Tuple[str, str]],
                              attached_meta: Sequence[Tuple[str, str]]
                              ) -> None:
    """**durable 로 적힌** 요구가 실제로 붙었나. ★안 붙었으면 선다.

    ★★`required` 는 **RPC 에 적힌 것**을 넘긴다 — 붙이는 쪽이 쓴 목록을
    그대로 다시 넘기면 **둘 다 빠뜨려도 통과한다**(Codex).

    ★기존 plate/space/outdoor 면제는 이것을 **대신 만족시키지 못한다** —
    값이 `grounding:<lp>:<purposes>:<신원>` 이라 다른 것과 안 겹친다.
    """
    have = {(str(k), str(v)) for k, v in attached_meta}
    missing = [(str(k), str(v)) for k, v in required
               if (str(k), str(v)) not in have]
    if missing:
        raise BundleContractError(
            f"조사가 고른 참조 {len(missing)}개가 안 붙었다: {missing} — "
            "기존 배경판이나 클로즈업 면제가 이것을 대신하지 못한다")


def undeclared_roles(known: Sequence[str]) -> List[str]:
    """이 묶음이 쓰는 역할 중 **아직 선언 안 된 것**.

    ★`REF_ROLE_VALUES` 는 **닫힌 목록**이라 모르는 역할이면 즉시 선다.
    D cutover 에서 이 목록이 **비어야** 켤 수 있다.
    """
    have = set(known)
    used = set(ROLE_BY_PURPOSES.values()) | {OUTLOOK_ROLE}
    return sorted({r for r in used if r not in have})


# ─────────────────────────────────────────────────────────────────────
# RPC sidecar — **있을 때만** 새 길로 간다
#
# ★★Codex 2026-08-31: validator 의 required SOT 는 **붙이는 쪽이 넘긴 목록**이
#  아니라 **RPC 에 durable 하게 적힌 것**이어야 한다. 같은 목록을 두 번 쓰면
#  builder 가 빠뜨린 것을 validator 도 같이 빠뜨려 **거짓 통과**한다.
#
# ★그리고 sidecar 가 **없으면** 활성 경로가 한 글자도 안 바뀐다 — 그것이
#  이 문이 있는 이유다.
# ─────────────────────────────────────────────────────────────────────

#: RPC 에 적히는 자리 이름.
RPC_MEMBERS_KEY = "grounding_bundle_members"
RPC_REQUIRED_KEY = "grounding_required_attachments"


def members_from_rpc(rpc: Any) -> Optional[List[Dict[str, Any]]]:
    """RPC 에서 묶음 멤버를 꺼낸다. ★**키가 없을 때만** `None`(옛 길).

    ★★★앞 판은 `rpc.get(...)` 이 `None` 을 주면 **키가 없는 것과 똑같이**
    옛 길로 접었다 (Codex 재현 2026-08-31): `{members: None}` 이 조용히
    통과했다. schema marker 에 쓴 것과 **같은 원칙**을 여기도 쓴다 —

        키 없음              → `None`. 옛 CP 라 옛 길이 맞다
        키 있고 목록         → 그 목록 (빈 목록은 **명시 「없음」**)
        키 있는데 목록 아님   → **선다** (`None` 포함)
    """
    if not isinstance(rpc, dict):
        return None
    if RPC_MEMBERS_KEY not in rpc:
        return None                                 # ★키 없음 = 옛 CP
    got = rpc[RPC_MEMBERS_KEY]
    if not isinstance(got, list):
        raise BundleContractError(
            f"`{RPC_MEMBERS_KEY}` 가 있는데 목록이 아니다: "
            f"{type(got).__name__} — 모르는 것을 옛 것으로 읽지 않는다")
    for x in got:
        if not isinstance(x, dict):
            raise BundleContractError(
                f"멤버가 dict 가 아니다: {type(x).__name__}")
    return [dict(x) for x in got]


def required_from_rpc(rpc: Any) -> Optional[List[Tuple[str, str]]]:
    """RPC 에 **적힌** 요구를 읽는다. ★**키가 없으면 `None`** — 「없음」과 다르다.

    ★★★앞 판은 `rpc.get(...) or []` 라 **키 없음**과 **명시 빈 목록**을
    못 갈랐고, 잘못된 타입(`{}` 등)도 조용히 삼켰다 (Codex 재현).
    그러면 만드는 쪽이 요구를 안 적었는데 「없다고 적었다」로 읽힌다.

        키 없음              → `None`  (만드는 쪽이 안 적었다)
        키 있고 목록         → 그 목록 (빈 목록은 **명시 「요구 없음」**)
        키 있는데 목록 아님   → **선다**
    """
    if not isinstance(rpc, dict):
        return None
    if RPC_REQUIRED_KEY not in rpc:
        return None
    got = rpc[RPC_REQUIRED_KEY]
    if not isinstance(got, list):
        raise BundleContractError(
            f"`{RPC_REQUIRED_KEY}` 가 있는데 목록이 아니다: "
            f"{type(got).__name__}")
    out: List[Tuple[str, str]] = []
    for x in got:
        if isinstance(x, dict):
            out.append((str(x.get("kind") or ""), str(x.get("value") or "")))
        elif isinstance(x, (list, tuple)) and len(x) == 2:
            out.append((str(x[0]), str(x[1])))
        else:
            raise BundleContractError(f"요구 모양이 이상하다: {x!r}")
    return out


def write_sidecar(rpc: Dict[str, Any],
                  members: Sequence[Dict[str, Any]]) -> List[Tuple[str, str]]:
    """**만드는 쪽**이 members 와 required 를 **한 번에** 적는다. ★원자적.

    ★★★canary 나 시험이 sidecar 를 **손으로 두 벌** 만들면, 그 둘이 어긋난
    채로 굳는다 (Codex 2026-08-31). 그래서 **이 함수 하나**가 —

        ①JSON 으로 담을 수 있는 모양인지 본다 (bytes 가 섞이면 선다)
        ②`plan_bundle` 로 요구를 낸다
        ③members 와 required 를 **같은 RPC 에 같이** 적는다

    ★소비하는 쪽은 **이 함수를 안 부른다** — 읽고 대조만 한다.
    """
    import json

    got = [dict(x) for x in members]
    try:
        json.dumps(got, ensure_ascii=False)
    except TypeError as exc:
        raise BundleContractError(
            f"멤버를 CP 에 못 담는다 ({exc}) — durable sidecar 는 bytes 가 "
            "아니라 **좌표**를 적는다") from exc
    _plan, required = plan_bundle(got)
    rpc[RPC_MEMBERS_KEY] = got
    rpc[RPC_REQUIRED_KEY] = [{"kind": k, "value": v} for k, v in required]
    return required


def attach_from_rpc(labeled_refs: List, attached_meta: List,
                    ref_roles: List, ref_role_metadata: List,
                    rpc: Any, *, load_bytes: Any, where: str = "") -> int:
    """RPC 에 sidecar 가 **있을 때만** 붙인다. ★없으면 **아무것도 안 한다**.

    ★★★**요구를 쓰지 않는다** — 읽고 **대조만** 한다. 쓰면 SOT 가 아니게
    되고, 조립이 이 호출을 빠뜨리면 쓰기도 같이 빠져 거짓 통과한다.

        멤버 키 없음                  → 0. 목록 한 글자도 안 바뀐다
        멤버 키 있는데 목록 아님       → **선다**
        멤버는 있는데 요구 키가 없음   → **선다**
        요구가 다시 낸 것과 **다름**   → **선다**
        맞으면                        → 좌표로 읽어 붙인다(해시 확인)
    """
    members = members_from_rpc(rpc)
    if members is None:
        return 0                                    # ★옛 길 — 무변
    plan, expected = plan_bundle(members)
    persisted = required_from_rpc(rpc)
    if persisted is None:
        raise BundleContractError(
            f"sidecar 는 있는데 `{RPC_REQUIRED_KEY}` 키가 **없다** — "
            "만드는 쪽이 둘을 같이 적어야 한다(`write_sidecar`). "
            "소비하는 쪽은 안 만든다")
    if sorted(persisted) != sorted(expected):
        raise BundleContractError(
            f"적힌 요구와 멤버에서 다시 낸 것이 **다르다**: "
            f"적힌 것 {sorted(persisted)} · 다시 낸 것 {sorted(expected)}")
    return insert_bundle_refs(labeled_refs, attached_meta, ref_roles,
                              ref_role_metadata, plan,
                              load_bytes=load_bytes, where=where)


def assert_from_rpc(rpc: Any,
                    attached_meta: Sequence[Tuple[str, str]]) -> None:
    """RPC 에 **적힌** 요구가 붙었나. ★조립·검사 **뒤에** 부른다.

    ★기존 `validate_attached_refs` 의 클로즈업·plate 면제와 **별개**다 —
    조사가 고른 것은 그 면제로 대신 채워지지 않는다.
    """
    required = required_from_rpc(rpc)
    if not required:                                # ★None 이거나 명시 빈 목록
        return
    assert_grounding_attached(required, attached_meta)


# ─────────────────────────────────────────────────────────────────────
# 참조 **개수** — provider 능력 계약 한 벌을 소비한다
#
# ★★Codex 2026-08-31: 「임의 숫자를 안 박고 **고른 provider 의 capability
#  계약**을 SOT 로 삼아라. preflight 와 보내기 직전이 **같은 method** 를
#  소비하고, provider class/model + capability 를 **lock 에 접어라**」.
#
# ★이미지 **생성 호출** 상한(canary 1회)과 **참조 개수**는 **별개**다.
# ─────────────────────────────────────────────────────────────────────


class ReferenceCountRefused(RuntimeError):
    """참조 장수가 provider 가 받는 범위 밖이다. ★보내기 **전에** 선다."""


def capability_of(client: Any) -> Dict[str, Any]:
    """client 에게 **직접 묻는다**. ★registry 를 베끼지 않는다.

    ★없으면 **선다** — 「아마 다 받겠지」로 지나가면 잘린 참조가 **그림에서만**
    드러난다.
    """
    fn = getattr(client, "reference_input_capability", None)
    if not callable(fn):
        raise ReferenceCountRefused(
            f"{type(client).__name__} 에 `reference_input_capability()` 가 "
            "없다 — 몇 장을 받는지 모르는 채로 안 보낸다")
    got = fn()
    if not isinstance(got, dict):
        raise ReferenceCountRefused(
            f"능력 계약이 dict 가 아니다: {type(got).__name__}")
    return dict(got)


def assert_reference_count(capability: Dict[str, Any], n: int) -> None:
    """참조 `n` 장이 이 provider 에 맞나. ★**보내기 전에** 본다.

        `max_images` 가 수    → `n <= max` 를 **본다**
        `max_images` 가 None  → 「**모른다**」다. ★상한을 **발명하지 않는다** —
                                지나가되 부르는 쪽이 observed n 을 **기록**한다
        `min_images`          → 모자라면 선다

    ★넘으면 **조용히 자르지 않는다** — 잘린 참조는 그림에서만 드러난다.
    """
    lo = capability.get("min_images")
    hi = capability.get("max_images")
    if not capability.get("supports_labeled_refs", True) and n:
        raise ReferenceCountRefused(
            f"{capability.get('provider')} 는 이름표 붙은 참조를 안 받는다")
    if lo is not None and n < int(lo):
        raise ReferenceCountRefused(
            f"{capability.get('provider')} 는 참조 {lo}장 이상이 필요한데 "
            f"{n}장이다")
    if hi is not None and n > int(hi):
        raise ReferenceCountRefused(
            f"{capability.get('provider')} 는 참조 {hi}장까지인데 {n}장이 "
            "왔다 — 조용히 자르지 않는다")


def capability_lock(capability: Dict[str, Any]) -> Dict[str, Any]:
    """지문에 접을 것 — **provider class/model + 능력**.

    ★provider 를 바꾸면 같은 문안·같은 참조라도 **다른 산출**이다. 능력이
    바뀌면 보낼 수 있는 장수가 달라진다. 둘 다 지문에 들어가야 옛 산출이
    「지금 계약의 것」으로 통하지 않는다.
    """
    return {
        "provider": str(capability.get("provider") or ""),
        "model": str(capability.get("model") or ""),
        "supports_labeled_refs": bool(
            capability.get("supports_labeled_refs", True)),
        "min_images": capability.get("min_images"),
        "max_images": capability.get("max_images"),
    }