"""중앙 참조 조사기의 **입구 계약** — 갈래 중립. ★조사는 여기서 안 한다.

## 왜 갈래 중립인가

Codex BLOCK (2026-09-01) — 앞 판은 중앙 입구가 `grounding_outlook_binding` 을
직접 읽어서 **outlook 전용**이었다. `bind` 는 `outlook_phase3` 를 받고
`character_id`/`outlook_id` 로 잇는다. 그래서 location·location_part 가
**정상 경로로 중앙에 들어올 길이 없었다** — 손으로 줄을 지어 시험을 쓰면
「받는 쪽만 있고 내는 쪽이 없는」 결함이 또 난다.

갈래마다 **잇는 법은 다르다.** 그것은 갈래별 resolver 로 남긴다 —

    character · location · prop   등록된 엔티티 줄로 **바로** 해소된다
    location_part                 `grounding_facet_binding`/host_context
    outlook                       phase3 뒤 `grounding_outlook_binding`

**해소된 뒤의 모양은 하나**다. 이 모듈이 그 하나를 정하고, 회계·자리 가름을
한 벌만 갖는다. ★새 조사기를 만드는 것이 아니라, 갈래별 결과를 중앙 입구
모양으로 **정규화하는 얇은 층**이다.

## 최소 공통 줄

    owner_type · research_subject_id · screen · status
    final_id                 해소됐으면 그 정본 ID
    parent_final_id          있을 때만 (location_part 의 장소, outlook 의 인물)
    source_evidence          원문 증거
    grounding_producer_payload  판별이 낸 것 (있을 때)
"""
from __future__ import annotations

import copy
from typing import Any, Dict, List, Optional, Sequence

LEDGER_CONTRACT_VERSION = "1.202609012400"


class LedgerContractError(RuntimeError):
    """장부가 계약을 어긴다. ★모르는 값을 정상으로 접지 않는다."""


#: 결속·해소가 **끝났다**. ★값은 옛 장부와 같다(`"bound"`) — 이름만 갈래
#:  중립으로 바꿨다. 값을 바꾸면 이미 적힌 장부를 못 읽는다.
RESOLVED = "bound"
#: 후보가 여럿이거나 소유가 어긋나 **못 정했다**.
UNRESOLVED = "unresolved"
#: 이 장부가 낼 수 있는 상태. ★그 밖은 **깨진 줄**이다.
STATUSES = (RESOLVED, UNRESOLVED)

#: 장부 한 줄이 갈 수 있는 **세 자리**. ★네 번째는 없다.
LANE_BUY = "buy"                         # 조사해야 하고 붙일 데도 안다
LANE_AUTO_DONE = "auto_completed"        # 조사해야 하는데 못 붙였다 — 그냥 간다
LANE_NOT_APPLICABLE = "not_applicable"   # 애초에 조사할 것이 아니었다

#: 최소 공통 줄의 칸. ★부르는 쪽이 손으로 적지 않게 한다.
ROW_FIELDS = ("owner_type", "research_subject_id", "screen", "status",
              "final_id", "parent_final_id", "source_evidence")


def known_screens() -> frozenset:
    """판별이 낼 수 있는 값. ★**여기서 다시 적지 않는다** — 그 모듈에서 온다."""
    from app.modules.pipeline import grounding_screen as gs

    return frozenset(gs.SCREENS)


def row(*, owner_type: str, research_subject_id: str, screen: str,
        status: str, final_id: Optional[str] = None,
        parent_final_id: Optional[str] = None,
        source_evidence: Optional[Dict[str, Any]] = None,
        producer_payload: Optional[Dict[str, Any]] = None,
        **extra: Any) -> Dict[str, Any]:
    """갈래별 resolver 의 산출 → **중앙 입구 한 모양**.

    ★해소됐다면서 `final_id` 가 없으면 **선다** — 「무엇에 붙일지 안다」가
    결속의 뜻인데 그 값이 없으면 아는 것이 아니다.
    """
    from app.modules.pipeline.grounding_entity_contract import (
        MATERIALIZABLE_OWNER_TYPES)

    if owner_type not in MATERIALIZABLE_OWNER_TYPES:
        raise LedgerContractError(
            f"모르는 갈래 {owner_type!r} — {MATERIALIZABLE_OWNER_TYPES}")
    if status not in STATUSES:
        raise LedgerContractError(f"모르는 상태 {status!r} — {STATUSES}")
    if screen not in known_screens():
        raise LedgerContractError(
            f"모르는 판별 {screen!r} — {sorted(known_screens())}")
    if status == RESOLVED and not str(final_id or "").strip():
        raise LedgerContractError(
            f"{research_subject_id!r} 가 해소됐다는데 `final_id` 가 없다")
    out: Dict[str, Any] = {
        "owner_type": str(owner_type),
        "research_subject_id": str(research_subject_id or ""),
        "screen": str(screen), "status": str(status),
        "final_id": (str(final_id) if final_id else None),
        "parent_final_id": (str(parent_final_id) if parent_final_id else None),
        "source_evidence": copy.deepcopy(source_evidence or {}),
    }
    if producer_payload:
        out["grounding_producer_payload"] = copy.deepcopy(producer_payload)
    out.update(extra)                    # ★갈래별 감사 칸은 그대로 실린다
    return out


def merge(*ledgers: Dict[str, Any]) -> Dict[str, Any]:
    """갈래별 장부들 → **하나**. ★같은 대상이 두 번 들어오면 선다.

    ★★두 resolver 가 같은 `research_subject_id` 를 내면 그것은 소유가
    겹친다는 뜻이고, 그대로 두면 **같은 대상을 두 번 조사한다**.
    """
    rows: List[Dict[str, Any]] = []
    seen: Dict[str, str] = {}
    for led in ledgers:
        for r in (led or {}).get("rows") or ():
            rsid = str(r.get("research_subject_id") or "")
            if rsid in seen:
                raise LedgerContractError(
                    f"{rsid!r} 를 {seen[rsid]!r} 와 "
                    f"{r.get('owner_type')!r} 가 **둘 다** 냈다 — 같은 대상을 "
                    "두 번 조사하게 된다")
            seen[rsid] = str(r.get("owner_type") or "")
            rows.append(copy.deepcopy(r))
    return {"contract_version": LEDGER_CONTRACT_VERSION, "rows": rows,
            "counts": {s: sum(1 for x in rows if x.get("status") == s)
                       for s in STATUSES}}


def acquisition_targets(ledger: Dict[str, Any]) -> List[Dict[str, Any]]:
    """중앙 조사기가 **조사할 대상**. ★두 문을 **다** 지나야 한다.

        ①해소됐다(`RESOLVED`) — 무엇에 붙일지 안다
        ②판별이 **의무**라고 했다 — 조사할 까닭이 있다

    ★★앞 판은 ①만 봤다. 그래서 판별이 「참조 불필요」(`not_target`)라고 한
    줄도 결속만 되면 **그대로 조사했다** (Codex 재현 · 09-01). 판별을 버리고
    결속만 보면, 우리가 돈을 쓰는 까닭이 사라진다.
    """
    from app.modules.pipeline.grounding_screen import SCREEN_OBLIGATION

    return [r for r in (ledger or {}).get("rows") or ()
            if r.get("status") == RESOLVED
            and str(r.get("screen") or "") == SCREEN_OBLIGATION]


def auto_completed(ledger: Dict[str, Any]) -> List[Dict[str, Any]]:
    """**사람 없이 그냥 닫는** 줄들 — 조사해야 하는데 붙일 데를 못 찾은 것.

    ★판별이 의무라고 했는데 해소가 안 된 것만이다. 비대상은 애초에 조사할
    것이 아니었으므로 「참조를 못 구했다」가 아니다.
    """
    from app.modules.pipeline.grounding_screen import SCREEN_OBLIGATION

    return [r for r in (ledger or {}).get("rows") or ()
            if r.get("status") == UNRESOLVED
            and str(r.get("screen") or "") == SCREEN_OBLIGATION]


def unresolved_outcome(row_: Dict[str, Any]) -> str:
    """못 이은 줄의 **처지**. ★사람 대기를 만들지 않는다."""
    from app.modules.pipeline.reference_acquisition import STATUS_UNAVAILABLE

    del row_
    return STATUS_UNAVAILABLE


def lane_of(row_: Dict[str, Any]) -> str:
    """이 줄이 갈 자리. ★**셋 중 하나**이되, **아는 값일 때만** 고른다.

    ★★★앞 판은 「obligation 이 아니면 무엇이든 `not_applicable`, obligation
    인데 resolved 가 아니면 무엇이든 `auto_completed`」로 접었다 (Codex ·
    09-01). 그러면 **계약 drift 나 깨진 장부가 「비대상」이나 「사람 없이 자동
    완료」로 정상 처리된다.** 그리고 합은 늘 맞으므로 회계가 **항진식**이 된다.

    Raises:
        LedgerContractError: 모르는 판별 값이거나 모르는 상태다.
    """
    from app.modules.pipeline.grounding_screen import SCREEN_OBLIGATION

    screen = str(row_.get("screen") or "")
    if screen not in known_screens():
        raise LedgerContractError(
            f"{row_.get('research_subject_id')!r} 의 판별이 {screen!r} 다 — "
            f"아는 값은 {sorted(known_screens())} 뿐이다. 모르는 것을 "
            "「비대상」으로 접으면 계약이 어긋난 것이 정상으로 보인다")
    status = row_.get("status")
    if status not in STATUSES:
        raise LedgerContractError(
            f"{row_.get('research_subject_id')!r} 의 결속이 {status!r} 다 — "
            f"아는 값은 {sorted(STATUSES)} 뿐이다")
    if screen != SCREEN_OBLIGATION:
        return LANE_NOT_APPLICABLE
    return LANE_BUY if status == RESOLVED else LANE_AUTO_DONE


def accounting(ledger: Dict[str, Any]) -> Dict[str, Any]:
    """장부의 **회계**. ★들어온 줄 수 = 세 자리의 합이어야 한다.

    ★안 맞으면 **선다**. 조용히 사라지는 줄이 있으면 그 줄만큼 우리가 모르는
    일이 일어난 것이다 — 조사하거나, 안 하거나, 둘 다 아닌 채로.
    """
    rows = list((ledger or {}).get("rows") or ())
    lanes: Dict[str, List[Dict[str, Any]]] = {
        LANE_BUY: [], LANE_AUTO_DONE: [], LANE_NOT_APPLICABLE: []}
    for r in rows:
        lanes[lane_of(r)].append(r)
    total = sum(len(v) for v in lanes.values())
    if total != len(rows):
        # ★이 합은 `lane_of` 가 **설 수 있어야** 뜻이 있다 — 늘 무언가를
        #  돌려주면 이 검사는 항진식이다.
        raise LedgerContractError(
            f"줄 {len(rows)} 인데 자리 합이 {total} 이다 — 조용히 사라진 줄이 있다")
    return {"counts": {k: len(v) for k, v in lanes.items()},
            "rows": len(rows), "lanes": lanes}


def owners_present(ledger: Dict[str, Any]) -> Sequence[str]:
    """이 장부에 **어느 갈래가 들어와 있나**. ★공개 끝점.

    다섯 갈래 완주를 잴 때 이것으로 본다 — 「받는 쪽만 있고 내는 쪽이 없는」
    것을 여기서 잡는다.
    """
    return tuple(sorted({str(r.get("owner_type") or "")
                         for r in (ledger or {}).get("rows") or ()}))


# ─────────────────────────────────────────────────────────────────────
# 갈래별 resolver — **잇는 법은 다르고, 나온 모양은 하나**
# ─────────────────────────────────────────────────────────────────────

#: 못 해소한 까닭. ★「없음」과 「틀림」을 가른다.
WHY_NO_ENTITY_ROW = "no_registered_entity_row"
WHY_NO_FACET_COORDINATE = "no_facet_coordinate"
WHY_FACET_NOT_BOUND = "facet_not_bound_to_parent"
WHY_OWNER_MISMATCH = "owner_type_disagrees_with_final_id"


def _screen_of(r: Dict[str, Any]) -> str:
    return str((r or {}).get("screen") or "")


def resolve_entity_backed(rows: Sequence[Dict[str, Any]], *,
                          final_id_by_subject: Dict[str, str]
                          ) -> Dict[str, Any]:
    """엔티티 줄로 **바로** 해소되는 갈래 — character · location · prop.

    Args:
        rows: `grounding_screen` 이 낸 정본 행 중 이 갈래들.
        final_id_by_subject: `research_subject_id` → 등록된 `short_id`.
            ★`grounding_plan` 이 남긴 `decided[]._short_id` 다 — 이름으로
            다시 짝짓지 않는다.

    ★facet 갈래(`location_part`·`outlook`)는 여기 오면 **선다** — 그것들은
    뒤에 producer 가 있어서 잇는 법이 다르다.
    """
    from app.modules.pipeline.grounding_entity_contract import (
        GENERIC_PROMOTION_OWNERS, owner_of_final_id)

    out: List[Dict[str, Any]] = []
    for r in rows or ():
        owner = str((r or {}).get("owner_type") or "")
        if owner not in GENERIC_PROMOTION_OWNERS:
            raise LedgerContractError(
                f"{owner!r} 는 엔티티 줄로 바로 해소되는 갈래가 아니다 — "
                f"{GENERIC_PROMOTION_OWNERS} 만 여기로 온다")
        rsid = str((r or {}).get("research_subject_id") or "")
        fid = str(final_id_by_subject.get(rsid) or "").strip()
        common = {
            "owner_type": owner, "research_subject_id": rsid,
            "screen": _screen_of(r),
            "source_evidence": (r or {}).get("source_evidence"),
            "producer_payload": (r or {}).get("grounding_producer_payload"),
        }
        if not fid:
            out.append(row(**common, status=UNRESOLVED,
                           why=WHY_NO_ENTITY_ROW))
            continue
        # ★신원이 정말 그 갈래인지 **계약에 묻는다** — 접두를 안 견준다
        if owner_of_final_id(fid) != owner:
            out.append(row(**common, status=UNRESOLVED,
                           why=WHY_OWNER_MISMATCH, seen_final_id=fid))
            continue
        out.append(row(**common, status=RESOLVED, final_id=fid))
    return {"contract_version": LEDGER_CONTRACT_VERSION, "rows": out}


def resolve_facet_rows(rows: Sequence[Dict[str, Any]], *,
                       final_id_by_subject: Optional[Dict[str, str]] = None
                       ) -> Dict[str, Any]:
    """facet 갈래 — `location_part`. ★앞단이 실은 **좌표**로만 잇는다.

    ★이름도 부분문자열도 안 쓴다. `grounding_facet_binding` 이 `part_of` 와
    owner 조합으로 이미 이어 둔 것을 읽을 뿐이다.

    ★★**제 신원과 부모는 다른 자리에서 온다.** 부모 장소가 원문에 독립
    대상으로 없으면 `part_of` 가 없어 facet 결속도 없다 — 그래도 그 부분
    자체는 등록돼 있고 정본 ID 를 갖는다. 앞 판은 결속에서만 ID 를 읽어
    **부모 없는 LP 가 통째로 unresolved 로 빠졌다**(실측).
    """
    from app.modules.pipeline.grounding_entity_contract import (
        FACET_BINDING, HOST_CONTEXT, owner_of_final_id)

    out: List[Dict[str, Any]] = []
    for r in rows or ():
        owner = str((r or {}).get("owner_type") or "")
        if owner != "location_part":
            raise LedgerContractError(
                f"{owner!r} 는 이 resolver 것이 아니다 — location_part 만 온다")
        rsid = str((r or {}).get("research_subject_id") or "")
        fb = (r or {}).get(FACET_BINDING) or {}
        common = {
            "owner_type": owner, "research_subject_id": rsid,
            "screen": _screen_of(r),
            "source_evidence": (r or {}).get("source_evidence"),
            "producer_payload": (r or {}).get("grounding_producer_payload"),
            # ★맥락 칸 — 의무 계획이 이것으로 맥락 조사 재료를 얻는다
            **({HOST_CONTEXT: copy.deepcopy(r[HOST_CONTEXT])}
               if (r or {}).get(HOST_CONTEXT) else {}),
        }
        # ★제 신원 — 결속이 있으면 그것, 없으면 **등록된 정본 ID**
        fid = (str(fb.get("final_id") or "").strip()
               or str((final_id_by_subject or {}).get(rsid) or "").strip())
        if not fid:
            out.append(row(**common, status=UNRESOLVED,
                           why=WHY_NO_FACET_COORDINATE))
            continue
        if owner_of_final_id(fid) != owner:
            out.append(row(**common, status=UNRESOLVED,
                           why=WHY_OWNER_MISMATCH, seen_final_id=fid))
            continue
        out.append(row(**common, status=RESOLVED, final_id=fid,
                       parent_final_id=(str(fb.get("parent_final_id") or "")
                                        or None)))
    return {"contract_version": LEDGER_CONTRACT_VERSION, "rows": out}


def split_by_owner(rows: Sequence[Dict[str, Any]]
                   ) -> Dict[str, List[Dict[str, Any]]]:
    """정본 행을 **갈래별로** 나눈다. ★모르는 갈래면 선다.

    ★부르는 쪽이 `owner_type == "outlook"` 같은 글자를 손으로 적지 않게 한다.
    """
    from app.modules.pipeline.grounding_entity_contract import (
        MATERIALIZABLE_OWNER_TYPES)

    out: Dict[str, List[Dict[str, Any]]] = {o: []
                                            for o in MATERIALIZABLE_OWNER_TYPES}
    for r in rows or ():
        owner = str((r or {}).get("owner_type") or "")
        if owner not in out:
            raise LedgerContractError(
                f"모르는 갈래 {owner!r} — {MATERIALIZABLE_OWNER_TYPES}")
        out[owner].append(r)
    return out