"""후보 ↔ 엔티티 **결속**을 구조화 ID 로 — 이름·부분문자열이 아니라.

## 왜

지금 결속은 `grounding_carry._hits` 의 **양방향 부분문자열**이다. 그것으로
「같은 대상인가」를 정하면 —

    「가방」과 「손가방」이 같은 것이 된다     ← 잘못 합친다
    추출이 이름을 크게 바꾸면 못 붙는다        ← 놓친다

둘 다 **뜻을 글자로 판단**한 것이고, 사용자 계약 3·14 가 금하는 자리다.
문자열 포함은 **완전성 경고**까지다.

## 어디서 나오나

가장 이른 올바른 자리는 `entity_all` 이다 — A0 후보 목록과 추출 대상을
**동시에 보는** 유일한 호출이기 때문이다. 그 모델이 「이 행은 저 후보다」를
직접 적으면, 그 뒤로는 코드가 기계적으로 나른다.

## 계약 여섯 (Codex 2026-08-31)

1. 칸은 **배열**이다 — `grounding_candidate_ids: array[string]`, required,
   uniqueItems, 빈 배열 허용. 한 후보가 여러 행으로 갈리고 같은 이름 행이
   합쳐지므로 단일값은 정보를 버린다.
2. 허용 ID 는 팩에 안 박는다. **그 호출에 실린 후보 ID 만 runtime enum** 이다.
3. 한 후보 ID 가 **두 행 이상**에 붙으면 임의로 고르지 않는다 — `contested`.
   여러 후보 ID 가 한 행을 가리키는 것은 **배열로 보존**한다.
4. 상세 추출은 모델에게 후보 ID 를 **다시 고르게 하지 않는다.** `short_id` 를
   되받고 코드가 그것으로 재결속한다.
5. scene-chain fallback 도 같은 catalog 를 받고, 중복이면 ID 집합을 **union**.
6. `entity_merge` 가 행을 지울 때 그 행의 ID 를 **keep 행에 union** 한다.
   안 그러면 병합에서 provenance 가 사라진다.
"""
from __future__ import annotations

import copy
import logging
from typing import Any, Dict, Iterable, List, Optional, Sequence, Tuple

logger = logging.getLogger(__name__)

#: 결속 계약이 바뀌면 올린다 — 지문에 접혀 resume 이 옛 CP 를 안 건너뛴다.
BINDING_CONTRACT_VERSION = 1

#: 엔티티 행에 실리는 칸 이름. **한 곳에서만** 적는다.
FIELD = "grounding_candidate_ids"

#: ★★**후보별 장부**가 앉는 칸. 계산해 놓고 안 남기면, 체크포인트에는 빈
#:  배열만 남아 「다퉈서 못 붙였다」와 「애초에 아무 후보도 아니다」가
#:  구별이 안 된다 — 그러면 승격 금지해야 할 것이 승격된다 (Codex BLOCK-3).
LEDGER_KEY = "binding_ledger"

#: 후보 하나의 행선지. ★`unbound` 만 「새로 만들어도 되는 것」이다.
BIND_BOUND = "bound"            # 한 행에 붙었다
BIND_CONTESTED = "contested"    # 두 행 이상이 가져갔다 — 못 정했다
BIND_LOST = "lost"              # 붙었다가 뒤 단계에서 끊겼다
BIND_UNBOUND = "unbound"        # 아무 행도 안 불렀다
BIND_DISPOSITIONS = (BIND_BOUND, BIND_CONTESTED, BIND_LOST, BIND_UNBOUND)

#: 승격·보호를 **하면 안 되는** 행선지. 「못 정했다」를 「없다」로 읽으면
#: 같은 대상이 둘이 된다.
BIND_NOT_PROMOTABLE = frozenset({BIND_CONTESTED, BIND_LOST})


def build_candidate_catalog(
    candidates: Sequence[Dict[str, Any]], entity_type: str,
) -> Tuple[List[str], List[str]]:
    """그 갈래가 볼 후보의 **동적 슬롯**과 **허용 ID 목록**.

    ★슬롯만 싣는다 — `{id, owner_type, source_anchor, source_quote,
    surface_form}`. 구체 대상 예시는 **0** 이다(사용자 계약 1·2).

    Returns:
        (프롬프트 줄들, 허용 ID 목록). 후보가 없으면 **둘 다 비었다** —
        그러면 legacy 는 한 바이트도 안 달라진다.
    """
    from app.modules.pipeline.grounding_overlay import candidates_for

    mine = candidates_for(candidates, entity_type)
    # ★★**조용히 버리지 않는다.** 앞 판은 빈 id 와 겹친 id 를 `continue` 로
    #  넘겨서, 그 후보는 목록에도 허용 enum 에도 안 들어가 **모델이 결속할
    #  길이 아예 없었다** — 그런데 아무도 그 사실을 몰랐다.
    blank = [i for i, c in enumerate(mine)
             if not str(c.get("research_subject_id") or "").strip()]
    if blank:
        raise AssertionError(
            f"{entity_type} 후보에 `research_subject_id` 가 빈 것이 "
            f"{len(blank)}개 있다 (자리 {blank[:5]}) — 결속할 열쇠가 없다")
    seen_ids = [str(c["research_subject_id"]).strip() for c in mine]
    dup = sorted({x for x in seen_ids if seen_ids.count(x) > 1})
    if dup:
        raise AssertionError(
            f"{entity_type} 후보에 같은 `research_subject_id` 가 "
            f"{len(dup)}개 겹친다 {dup[:5]} — 어느 쪽이 진짜인지 못 정한다")

    ids: List[str] = []
    lines: List[str] = []
    for c in mine:
        rid = str(c["research_subject_id"]).strip()
        ids.append(rid)
        lines.append(
            f"- id={rid} | owner={c.get('owner_type', '')} "
            f"| anchor={c.get('source_anchor', '')} "
            f"| surface={c.get('surface_form', '')} "
            f"| quote={c.get('source_quote', '')}"
        )
    return lines, ids


#: ★★규칙문은 **팩에 있다** (Codex BLOCK-1). 소스 문자열로 박으면 팩 버전도
#:  원문 hash 도 audit 도 없다. 동적 slot 과 runtime enum 만 코드에 둔다.
_MODULE = "grounding_binding"
PROMPT_PACK_VERSION = "1.202608310100"
STEM_CATALOG_HEAD = "candidate_catalog_head"
STEM_BINDING = "candidate_binding"
STEM_SHORT_ID = "short_id_echo"
STEM_MERGE = "merge_mapping"
STEMS = (STEM_CATALOG_HEAD, STEM_BINDING, STEM_SHORT_ID, STEM_MERGE)


def load_pack(*, db=None, version: Optional[str] = None) -> Dict[str, Any]:
    """결속 지문 팩.

    `raw_content_hash` 를 소비 지문에 접어야 「지문을 고쳤는데 resume 이 옛
    체크포인트를 건너뛴다」가 안 난다.

    ★**지금 호출부는 `db` 를 안 넘긴다 — file pack 만 본다** (Codex 정정).
    `resolve_effective` 는 DB override 를 지원하지만, `_config_hash` ·
    `build_binding_block` · 두 지시문 helper 중 어느 것도 `db` 를 전파하지
    않는다. DB override 까지 된다고 쓰려면 한 경로로 `db` 를 흘리고 **지문과
    실제 요청이 같은 실효 팩을 보는 끝점**이 있어야 한다. 그 전에는
    file-only 다.
    """
    from app.modules.prompt_loader import resolve_effective

    ver = version or PROMPT_PACK_VERSION
    resolved = {st: resolve_effective(_MODULE, st, kind="prompt",
                                      version=ver, db=db) for st in STEMS}
    import hashlib

    manifest = "|".join(
        f"{st}:{r['source']}:{r['version']}:{r['raw_content_hash']}"
        for st, r in sorted(resolved.items()))
    return {"module": _MODULE, "version": ver, "stems": resolved,
            "pack_manifest_hash": hashlib.sha256(
                manifest.encode("utf-8")).hexdigest()[:16]}


def pack_fingerprint(*, db=None, version: Optional[str] = None) -> Dict[str, Any]:
    """소비 지문에 접을 좌표. ★상수만 올리고 **bytes 를 안 접으면** 안 움직인다."""
    pack = load_pack(db=db, version=version)
    return {"binding_contract": BINDING_CONTRACT_VERSION,
            "binding_pack": pack["version"],
            "binding_pack_hash": pack["pack_manifest_hash"]}


def _text(stem: str, *, db=None, version: Optional[str] = None) -> str:
    return load_pack(db=db, version=version)["stems"][stem]["content"]


def short_id_instruction(*, db=None, version: Optional[str] = None) -> str:
    """상세 추출에 붙일 `short_id` 반환 지시. ★팩에서 읽는다."""
    return _text(STEM_SHORT_ID, db=db, version=version)


def merge_instruction(*, db=None, version: Optional[str] = None) -> str:
    """`entity_merge` 에 붙일 keep/remove 대응 지시. ★팩에서 읽는다."""
    return _text(STEM_MERGE, db=db, version=version)


def build_binding_block(candidates: Sequence[Dict[str, Any]],
                        entity_type: str, *, db=None,
                        version: Optional[str] = None) -> Tuple[str, List[str]]:
    """프롬프트에 붙일 **한 덩어리**와 허용 ID.

    ★목록과 문안을 **함께** 낸다. 따로 두면 한쪽만 붙는 판이 생기고,
    그러면 모델이 채울 수 없는 칸을 required 로 요구하게 된다.

    Returns:
        (붙일 텍스트, 허용 ID). 후보가 없으면 `("", [])`.
    """
    lines, ids = build_candidate_catalog(candidates, entity_type)
    if not ids:
        return "", []
    pack = load_pack(db=db, version=version)
    body = [
        "",
        pack["stems"][STEM_CATALOG_HEAD]["content"].strip(),
        *lines,
        "",
        pack["stems"][STEM_BINDING]["content"].format(field=FIELD).strip(),
    ]
    return "\n".join(body), ids


def patch_schema_with_candidate_ids(schema: Dict[str, Any],
                                    allowed_ids: Sequence[str]) -> Dict[str, Any]:
    """산출 schema 에 **runtime enum** 칸을 더한다.

    ★`allowed_ids` 가 비면 **schema 를 안 건드린다** — 빈 enum 은 어떤 값도
    못 받아 모델이 설 수 있고, 후보가 없는 판은 legacy 와 같아야 한다.

    ★`additionalProperties: false` 라 이 패치 없이는 모델이 칸을 못 낸다.
    `_patch_schema_shot_count` 와 같은 전례다.
    """
    if not allowed_ids:
        return schema
    out = copy.deepcopy(schema)
    for key in list(out.get("properties", {}).keys()):
        arr = out["properties"][key]
        if arr.get("type") != "array" or "items" not in arr:
            continue
        props = arr["items"].setdefault("properties", {})
        req = arr["items"].setdefault("required", [])
        props[FIELD] = {
            "type": "array",
            "items": {"type": "string", "enum": list(allowed_ids)},
            "uniqueItems": True,
        }
        # ★**required** 다. optional 이면 모델이 그냥 빼고, 그러면 「안 붙었다」와
        #  「안 물어봤다」가 구별이 안 된다.
        if FIELD not in req:
            req.append(FIELD)
    return out


def patch_schema_with_short_ids(schema: Dict[str, Any],
                              allowed: Sequence[str]) -> Dict[str, Any]:
    """상세 추출 산출에 **`short_id` runtime enum** 을 더한다.

    ★계약 4 — 여기서 모델에게 후보 ID 를 **다시 고르게 하지 않는다.** 앞
    단계가 정한 `short_id` 만 되받고, 후보 ID 는 그것으로 코드가 잇는다.
    이름은 표시·audit 일 뿐 SOT 가 아니다.

    ★`allowed` 가 비면 schema 를 안 건드린다 — legacy 불변.
    """
    if not allowed:
        return schema
    out = copy.deepcopy(schema)
    for key in list(out.get("properties", {}).keys()):
        arr = out["properties"][key]
        if arr.get("type") != "array" or "items" not in arr:
            continue
        props = arr["items"].setdefault("properties", {})
        req = arr["items"].setdefault("required", [])
        props["short_id"] = {"type": "string", "enum": list(allowed)}
        if "short_id" not in req:
            req.append("short_id")
    return out


def rebind_by_short_id(
    detailed: Sequence[Dict[str, Any]], listed: Sequence[Dict[str, Any]],
) -> Dict[str, Any]:
    """상세 행에 **앞 단계의 후보 ID 를 기계적으로 옮긴다.**

    ★계약 4 — 잇는 열쇠는 `short_id` 다. 이름으로 이으면 모델이 이름을
    바꾼 순간 결속이 통째로 끊기고, 비슷한 이름끼리 잘못 붙는다.

    ★같은 `short_id` 를 두 상세 행이 주장하면 **양쪽 다 안 잇는다** —
    어느 쪽이 그 행인지 못 정한다.

    Returns:
        `carried` · `unmatched`(앞 단계에 없던 short_id) ·
        `contested`(두 행이 주장한 short_id) · `lost`(안 이어진 후보 ID).
    """
    by_sid = {str(e.get("short_id") or "").strip(): e for e in listed or ()
              if str(e.get("short_id") or "").strip()}
    claims: Dict[str, List[int]] = {}
    for i, d in enumerate(detailed or ()):
        sid = str((d or {}).get("short_id") or "").strip()
        if sid:
            claims.setdefault(sid, []).append(i)

    contested = sorted(s for s, idx in claims.items() if len(idx) > 1)
    unmatched = sorted(s for s in claims if s not in by_sid)
    carried = 0
    for sid, idx in claims.items():
        if len(idx) > 1 or sid not in by_sid:
            continue
        src = by_sid[sid]
        if src.get(FIELD):
            carry_into(detailed[idx[0]], src)
            carried += 1

    got = set()
    for d in detailed or ():
        got.update(d.get(FIELD) or [])
    want = set()
    for e in listed or ():
        want.update(e.get(FIELD) or [])
    lost = sorted(want - got)
    if lost:
        logger.warning("grounding binding: 상세 단계에서 후보 %d개가 끊겼다",
                       len(lost))
    return {"carried": carried, "unmatched": unmatched,
            "contested": contested, "lost": lost}


def normalize_binding(
    entities: Sequence[Dict[str, Any]], allowed_ids: Sequence[str],
) -> Dict[str, Any]:
    """모델이 낸 결속을 **다듬고 갈린 것을 가른다.**

    하는 일:

    - 허용 목록 밖의 ID 는 **버린다**(모델이 지어낸 것이다).
    - 한 행 안에서 겹친 ID 는 접는다.
    - 한 ID 가 **두 행 이상**에 붙었으면 그 ID 를 `contested` 로 빼고
      **양쪽 행에서 지운다** — 임의로 고르면 다른 대상의 근거를 물려받는다.

    ★행 자체는 안 지운다. 결속이 안 된 행도 엔티티로는 정상이다.

    Returns:
        `bound`(id → short_id 또는 이름) · `contested` · `unknown` · `counts`.
    """
    ok = set(allowed_ids)
    unknown: set = set()
    claims: Dict[str, List[int]] = {}
    for i, e in enumerate(entities):
        got: List[str] = []
        for rid in (e.get(FIELD) or []):
            rid = str(rid or "").strip()
            if not rid:
                continue
            if rid not in ok:
                unknown.add(rid)
                continue
            if rid not in got:
                got.append(rid)
        e[FIELD] = got
        for rid in got:
            claims.setdefault(rid, []).append(i)

    contested = sorted(r for r, idx in claims.items() if len(idx) > 1)
    for rid in contested:
        for i in claims[rid]:
            entities[i][FIELD] = [x for x in entities[i][FIELD] if x != rid]

    bound = {rid: idx[0] for rid, idx in claims.items() if len(idx) == 1}
    ledger = {
        rid: (BIND_CONTESTED if rid in contested
              else BIND_BOUND if rid in bound else BIND_UNBOUND)
        for rid in allowed_ids
    }
    if unknown:
        logger.warning("grounding binding: 목록에 없는 후보 id %d개 버림",
                       len(unknown))
    if contested:
        logger.warning("grounding binding: 두 행이 가져간 후보 %d개 — 안 붙인다",
                       len(contested))
    return {
        "contract_version": BINDING_CONTRACT_VERSION,
        "bound": bound,
        "contested": contested,
        "unknown": sorted(unknown),
        # ★★**후보마다 한 줄.** 이것이 체크포인트에 남아야 다음 소비자가
        #  「다퉈서 못 붙었다」와 「아무것도 아니다」를 가른다.
        "ledger": ledger,
        "counts": {"allowed": len(ok), "bound": len(bound),
                   "contested": len(contested), "unknown": len(unknown),
                   "unbound": sum(1 for v in ledger.values()
                                  if v == BIND_UNBOUND)},
    }


def merge_ledger(base: Optional[Dict[str, Any]],
                 *, lost: Sequence[str] = (),
                 contested: Sequence[str] = ()) -> Dict[str, str]:
    """뒤 단계에서 끊긴 것을 장부에 **되쓴다**. ★내려가기만 한다.

    `bound` 였던 것이 상세 단계에서 사라지면 `lost` 다 — 그것을 그냥
    `unbound` 로 두면 「아무도 안 불렀다」와 같아져 승격된다.
    """
    out = dict((base or {}).get("ledger") or {})
    for rid in contested:
        out[str(rid)] = BIND_CONTESTED
    for rid in lost:
        if out.get(str(rid)) != BIND_CONTESTED:
            out[str(rid)] = BIND_LOST
    return out


def promotable(ledger: Optional[Dict[str, str]]) -> set:
    """장부에서 **새로 만들어도 되는** 후보. ★못 정한 것은 안 준다."""
    return {rid for rid, d in (ledger or {}).items()
            if d == BIND_UNBOUND}


def blocked(ledger: Optional[Dict[str, str]]) -> set:
    """★**승격 금지.** 다퉜거나 끊긴 것 — 「없다」로 읽으면 둘이 된다."""
    return {rid for rid, d in (ledger or {}).items()
            if d in BIND_NOT_PROMOTABLE}


def union_ids(*rows: Optional[Dict[str, Any]]) -> List[str]:
    """행들의 후보 ID 를 **합친다.** ★순서를 지킨다 — 첫 등장 순.

    중복 제거·병합·삭제에서 쓴다. 「합치는 자리마다 각자 합치면」 한 곳만
    고쳐진다.
    """
    out: List[str] = []
    for row in rows:
        for rid in ((row or {}).get(FIELD) or []):
            rid = str(rid or "").strip()
            if rid and rid not in out:
                out.append(rid)
    return out


def carry_into(target: Dict[str, Any], *sources: Optional[Dict[str, Any]]) -> Dict[str, Any]:
    """`target` 에 `sources` 의 ID 를 union 해 넣는다. ★제자리에서 고친다."""
    target[FIELD] = union_ids(target, *sources)
    return target


def bound_short_ids(entities: Iterable[Dict[str, Any]]) -> Dict[str, str]:
    """후보 ID → **`short_id`**. ★보호·승격이 읽는 유일한 사상이다.

    이름은 여기 안 쓴다. 값이 없는 행은 건너뛴다 — 「못 붙었다」이지
    「이름이 비슷하다」가 아니다.
    """
    # ★★**첫 값만 취하지 않는다.** 같은 후보를 두 행이 들고 오면 행 순서가
    #  보호 대상을 바꾼다 — AB 면 P01, BA 면 P02 였다. `normalize_binding`
    #  이 그런 것을 `contested` 로 지우지만, 이 함수는 그것을 안 거친 행에도
    #  불릴 수 있다. **여기서도 막는다.**
    seen: Dict[str, str] = {}
    for e in entities or ():
        sid = str((e or {}).get("short_id") or "").strip()
        if not sid:
            continue
        for rid in ((e or {}).get(FIELD) or []):
            rid = str(rid or "").strip()
            if not rid:
                continue
            prev = seen.get(rid)
            if prev is not None and prev != sid:
                raise AssertionError(
                    f"후보 {rid} 를 두 행이 들고 있다 ({prev} · {sid}) — "
                    "행 순서가 보호 대상을 정하게 둘 수 없다")
            seen[rid] = sid
    return seen