"""GROUNDING-V2 §4a — 출처 붙은 claims 와 gaps.

★이 모듈이 route 의 **A(시대 차이가 실제로 있나)를 확정**한다.

왜 여기로 왔나 — 그 전에는 `grounding_classify` 가 답했는데, 그 단계는 문안
첫 줄부터 「검색을 하지 않는다」이면서 **세상 사실**을 답하고 있었다.
실측에서 그 짐작 축(`difficulty` = 계약 §2 세로축)만 흔들렸고, VLM 으로 재려던
안(§C)은 **기각**됐다 — VLM 도 그 시대 물건을 본 적이 없고, 정답 사진을 얻으면
이미 조사를 산 것이라 아낀 게 없다.

★**계약 §3 의 비대칭**: 검색 실패는 `no` 가 아니라 `unresolved` 다.
`A=no` 는 「차이가 없다」는 **긍정적 출처**가 있을 때만 낸다. 못 찾은 것은
`claims[]` 가 아니라 `gaps[]` 로 간다.

★이 모듈은 **바깥 호출을 하지 않는다.** 계약·모양·접기만 있다.
검색 전송은 별도이고, 이 모듈이 그 산출을 받아 판정한다.
"""
from __future__ import annotations

import hashlib
import json
import re
import unicodedata
from typing import Any, Dict, Iterable, List, Optional, Sequence, Set, Tuple

#: 계약 버전. claims/gaps 모양이 바뀌면 올린다.
CLAIMS_CONTRACT_VERSION = 1

#: claim 종류. ★**금지 사실을 따로 둔다** — 「이건 이 시대에 없었다」가
#: 그림을 더 많이 고친다(계약 §6 ①).
CLAIM_FACT = "fact"
CLAIM_PROHIBITION = "prohibition"
CLAIM_KINDS = (CLAIM_FACT, CLAIM_PROHIBITION)

#: 시대 차이 판정.
DELTA_YES = "yes"            # 출처가 시대 차이를 지지한다        → research
DELTA_NO = "no"              # 출처가 **차이 없음**을 지지한다     → skip
DELTA_UNRESOLVED = "unresolved"  # 못 찾음·충돌·출처 없음·시간 초과 → 하류 봉인

#: ★claim 이 **시대 차이에 대해 무엇을 말하는가**. `kind`·`required` 와 **독립**이다.
#:
#: 왜 bool 로는 부족한가 — `supports_no_delta: bool` 만 두면 ①차이와 무관한
#: required fact 가 그냥 `yes` 가 되고 ②차이 있음 claim 과 차이 없음 claim 이
#: 같이 와도 `yes` 가 되며 ③한 claim 이 둘 다일 수 있다. 계약의 「충돌은
#: unresolved」가 깨진다 (Codex 실측 반례 셋).
DELTA_EFFECT_DIFF = "supports_difference"
DELTA_EFFECT_NO_DIFF = "supports_no_difference"
DELTA_EFFECT_NEUTRAL = "neutral"
DELTA_EFFECTS = (DELTA_EFFECT_DIFF, DELTA_EFFECT_NO_DIFF, DELTA_EFFECT_NEUTRAL)

#: gap 사유.
GAP_NOT_FOUND = "not_found"
GAP_NO_SOURCE = "no_source"
GAP_CONFLICT = "conflict"
#: ★★★**`time_capped` 는 stopwatch 이름이 아니라 bounded-run umbrella 다**
#:  (Codex). 이 주행에 걸어 둔 **어떤 제한 때문에** 결론을 못 냈다는 뜻이고,
#:  그 제한이 시간인지 횟수인지는 아래 `limit_kind` 가 말한다.
#:  ★네 갈래 전부 하류에서 같은 처리를 받는다 — `unresolved` · retryable.
#:   그래서 `GAP_REASONS` 를 **안 늘린다**. 늘리면 분기만 겹친다.
GAP_TIME_CAPPED = "time_capped"
#: ★전송이 깨진 것은 **못 찾은 것과 다르다.** 못 찾은 것은 다 보고 없는 것이고,
#:  이건 아예 못 본 것이다 — 다음 판에서 **다시 산다**.
GAP_TRANSMISSION_FAILED = "transmission_failed"
#: ★`time_capped` 의 **원인**. 감사·재개 분석에는 원인이 필요하다 (Codex).
#:  판정은 같아도 **고칠 곳이 다르다** — 상한을 올릴 자리와 마감을 늘릴
#:  자리가 다르다.
LIMIT_ADMISSION = "admission_limit"        # 팬아웃 전 승인 상한
LIMIT_PER_CALL_DEADLINE = "per_call_deadline"   # 한 호출의 벽시계
LIMIT_TRANSMISSION_BUDGET = "transmission_budget"  # 누계 **물리 전송** 상한
LIMIT_RUN_DEADLINE = "run_deadline"        # 주행 전체 벽시계
LIMIT_KINDS = (LIMIT_ADMISSION, LIMIT_PER_CALL_DEADLINE,
               LIMIT_TRANSMISSION_BUDGET, LIMIT_RUN_DEADLINE)

GAP_REASONS = (GAP_NOT_FOUND, GAP_NO_SOURCE, GAP_CONFLICT, GAP_TIME_CAPPED,
               GAP_TRANSMISSION_FAILED)


#: 출처는 되짚을 수 있는 **주소**여야 한다. 공백이 섞인 것도 안 된다.
_SOURCE_URL_RE = re.compile(r"^https?://[^\s/]+\.[^\s/]+(?:/\S*)?$")


def _norm(s: Any) -> str:
    return re.sub(r"\s+", " ", unicodedata.normalize("NFKC", str(s or ""))).strip()


def claim_id(subject_id: str, statement_native: str, *,
             kind: str = CLAIM_FACT) -> str:
    """안정된 `claim_id`.

    ★참조 생성·검증이 **어느 claim 을 소비했는지** 기록에 남긴다(계약 §6).
    id 가 없으면 「무엇을 근거로 그렸나」를 못 되짚는다.
    ★batch 크기를 바꿔도 같은 id 가 나와야 한다 — 그래야 progressive admission
    이 완료된 것을 다시 안 산다. 그래서 **batch·순서를 안 섞는다.**
    """
    if kind not in CLAIM_KINDS:
        raise ValueError(f"claim kind 는 {CLAIM_KINDS} 중 하나다: {kind!r}")
    body = f"{_norm(subject_id)}\x1f{kind}\x1f{_norm(statement_native)}"
    return "cl_" + hashlib.sha256(body.encode()).hexdigest()[:20]


def validate_claim(raw: Dict[str, Any], *, subject_id: str
                   ) -> Tuple[Optional[Dict[str, Any]], Optional[str]]:
    """한 건을 계약 모양으로 세운다. ``(claim, 거절 사유)``.

    ★**출처 없는 required claim 은 claim 이 아니다.** 그걸 통과시키면
    「조사했다」가 거짓이 된다 — 계획 §2-4 통과 조건이
    「`claims[]` 에 source 없는 required claim 이 **0**」이다.
    """
    stmt = _norm(raw.get("statement_native"))
    if not stmt:
        return None, "문장이 비었다"
    kind = str(raw.get("kind") or CLAIM_FACT)
    if kind not in CLAIM_KINDS:
        return None, f"kind 가 enum 밖이다: {kind!r}"
    required = raw.get("required")
    if not isinstance(required, bool):
        return None, "required 가 bool 이 아니다"
    raw_srcs = raw.get("sources")
    if raw_srcs is None:
        raw_srcs = []
    # ★문자열을 그대로 받으면 **글자 하나하나가 출처**가 된다 —
    #  `"https://x"` → `["h","t","t","p",...]` (Codex 실측). list 만 받는다.
    if not isinstance(raw_srcs, (list, tuple)):
        return None, f"sources 는 목록이어야 한다 (받은 것 {type(raw_srcs).__name__})"
    srcs = [_norm(u) for u in raw_srcs if isinstance(u, str) and _norm(u)]
    if len(srcs) != len(list(raw_srcs)):
        return None, "sources 에 빈 값이나 문자열 아닌 것이 있다"
    # ★출처는 **되짚을 수 있어야** 한다. 「어디서 봤나」가 아무 글자나면
    #  「조사했다」가 거짓이 된다 — 나중에 아무도 확인할 수 없다.
    bad_url = [u for u in srcs if not _SOURCE_URL_RE.match(u)]
    if bad_url:
        return None, f"출처가 URL 이 아니다: {bad_url[:2]}"
    if required and not srcs:
        return None, "출처 없는 required claim"
    fam = _norm(raw.get("discriminator_family"))
    # ★「차이가 없다」는 **따로 표식된 칸**으로 받는다. 자유 문자열에 특별한
    #  값을 박아 두고 그 글자로 뜻을 가르면, 오타·번역·다른 표기 하나에
    #  판정이 무너진다(저장소 규칙: 글자/substring 으로 의미 판단 금지).
    effect = str(raw.get("delta_effect") or DELTA_EFFECT_NEUTRAL)
    if effect not in DELTA_EFFECTS:
        return None, f"delta_effect 가 enum 밖이다: {effect!r}"
    if effect != DELTA_EFFECT_NEUTRAL and not srcs:
        return None, "출처 없이 시대 차이를 말할 수 없다"
    return {
        "contract_version": CLAIMS_CONTRACT_VERSION,
        "claim_id": claim_id(subject_id, stmt, kind=kind),
        "research_subject_id": str(subject_id),
        "kind": kind,
        "required": required,
        "statement_native": stmt,
        "discriminator_family": fam or None,
        "delta_effect": effect,
        "sources": srcs,
    }, None


def build_gap(subject_id: str, *, reason: str, query: str = "",
              discriminator: str = "", note: str = "",
              required_discriminator: bool = False,
              limit_kind: str = "") -> Dict[str, Any]:
    """못 찾은 것. ★`claims[]` 가 아니라 여기로 온다.

    ★``required_discriminator`` — **꼭 알아야 하는 구별점을 못 찾았다**는 뜻.
    그러면 다른 claim 이 몇 개 있든 **아직 다 못 본 것**이라 `unresolved` 다.
    이 표식이 없으면 `not_found` gap 이 있어도 required claim 하나로 `yes` 가
    통과한다 (Codex).
    """
    if reason not in GAP_REASONS:
        raise ValueError(f"gap 사유는 {GAP_REASONS} 중 하나다: {reason!r}")
    # ★`bool("false")` 는 True 다. 이 판에서 `required`·`gradable` 로 이미
    #  두 번 데인 부류다 — **진짜 bool 만** 받는다.
    if not isinstance(required_discriminator, bool):
        raise ValueError(
            "required_discriminator 는 bool 이어야 한다: "
            f"{required_discriminator!r}")
    # ★★`time_capped` 면 **원인을 반드시 적는다** — 안 적으면 「무슨 제한에
    #  걸렸나」가 사라져 고칠 곳을 못 찾는다. 다른 사유에는 못 붙인다.
    kind = str(limit_kind or "")
    if reason == GAP_TIME_CAPPED:
        if kind not in LIMIT_KINDS:
            raise ValueError(
                f"time_capped 은 limit_kind 가 필요하다 — {LIMIT_KINDS} 중 "
                f"하나여야 하는데 {limit_kind!r} 다")
    elif kind:
        raise ValueError(
            f"limit_kind 는 time_capped 에만 붙는다: reason={reason!r}")
    return {
        "contract_version": CLAIMS_CONTRACT_VERSION,
        "research_subject_id": str(subject_id),
        "reason": reason,
        "query": _norm(query),
        "discriminator": _norm(discriminator),
        "note": _norm(note),
        "required_discriminator": required_discriminator,
        "limit_kind": kind,
    }


def decide_delta(claims: Sequence[Dict[str, Any]],
                 gaps: Sequence[Dict[str, Any]] = (), *,
                 subject_id: str) -> Tuple[str, Dict[str, Any]]:
    """A 를 **한 subject 에 대해** 확정한다. ``(delta, 근거)``.

    ★``subject_id`` 는 **필수**다. batch 로 여러 subject 를 한 호출에 넣으므로,
    받은 목록에 남의 행이 섞여 들어오기 쉽다. 그것을 그대로 접으면 **한
    subject 의 출처가 다른 subject 의 route 를 정한다.** 조용히 걸러 내지도
    않는다 — 그건 호출부의 결함을 감추는 것이라 **막고 알린다**.

    ```
    출처 붙은 required claim 이 하나라도 있다        → yes   (조사한다)
    「차이 없음」을 지지하는 출처가 있고 그 밖이 없다 → no    (건너뛴다)
    그 밖 — 못 찾음·충돌·출처 없음·시간 초과         → unresolved
    ```

    ★**「검색에서 아무것도 안 나옴 → no」는 금지다** (계약 §3).
    `no` 는 **긍정적 출처**가 있을 때만 나온다. 아무것도 없으면 `unresolved`
    이고, `unresolved` 는 하류가 외형을 지어내지 못하게 계속 묶어 둔다.
    """
    sid = str(subject_id)
    if not sid.strip():
        raise ValueError("subject_id 는 필수다")
    # ★목록에 `None` 이 섞여도 **AttributeError 로 죽지 않는다.** 죽으면 무엇이
    #  잘못됐는지 안 남고, 호출부는 「조사가 터졌다」로만 본다.
    foreign = ([(c or {}).get("claim_id") for c in claims
                if str((c or {}).get("research_subject_id")) != sid]
               + [(g or {}).get("reason") for g in gaps
                  if str((g or {}).get("research_subject_id")) != sid])
    if foreign:
        return DELTA_UNRESOLVED, {
            "why": f"다른 subject 의 행이 {len(foreign)}개 섞였다 — 호출부 결함",
            "subject_id": sid, "foreign": foreign[:5]}

    # ★**효과로 가른다.** `required` 만 보면 차이와 **무관한** 사실 하나가
    #  그대로 `yes` 가 된다 (Codex 반례).
    sourced = [c for c in claims if (c.get("sources") or [])]
    diff = [c for c in sourced if c.get("delta_effect") == DELTA_EFFECT_DIFF]
    same = [c for c in sourced if c.get("delta_effect") == DELTA_EFFECT_NO_DIFF]

    bad = [c for c in claims
           if c.get("required") and not (c.get("sources") or [])]
    if bad:
        # ★여기까지 왔다는 건 `validate_claim` 을 안 거쳤다는 뜻이다.
        return DELTA_UNRESOLVED, {
            "why": f"출처 없는 required claim {len(bad)}개 — 계약 위반",
            "subject_id": sid,
            "offending": [c.get("claim_id") for c in bad]}
    if any(g.get("reason") == GAP_CONFLICT for g in gaps):
        return DELTA_UNRESOLVED, {"why": "출처가 서로 어긋난다", "subject_id": sid}
    if any(g.get("reason") == GAP_TIME_CAPPED for g in gaps):
        return DELTA_UNRESOLVED, {"why": "시간 상한에 걸려 다 못 봤다",
                                  "subject_id": sid}
    if any(g.get("reason") == GAP_TRANSMISSION_FAILED for g in gaps):
        return DELTA_UNRESOLVED, {"why": "전송이 깨져 아예 못 봤다",
                                  "subject_id": sid}
    # ★**꼭 알아야 하는 구별점을 못 찾았으면 아직 다 못 본 것**이다.
    #  다른 claim 이 몇 개 있든 통과시키지 않는다.
    missing = [g for g in gaps if g.get("required_discriminator")]
    if missing:
        return DELTA_UNRESOLVED, {
            "why": f"꼭 알아야 할 구별점 {len(missing)}개를 못 찾았다",
            "subject_id": sid}

    # ★양쪽 효과가 같이 나오면 **모른다**다 — 한쪽을 골라 주지 않는다.
    if diff and same:
        return DELTA_UNRESOLVED, {
            "why": "출처가 차이 있음과 없음을 동시에 지지한다",
            "subject_id": sid,
            "difference": [c["claim_id"] for c in diff],
            "no_difference": [c["claim_id"] for c in same]}
    if diff:
        return DELTA_YES, {
            "why": f"출처가 시대 차이를 지지한다 — claim {len(diff)}개",
            "subject_id": sid,
            "fact": sum(1 for c in diff if c["kind"] == CLAIM_FACT),
            "prohibition": sum(1 for c in diff
                               if c["kind"] == CLAIM_PROHIBITION),
            "claim_ids": [c["claim_id"] for c in diff]}
    if same:
        return DELTA_NO, {
            "why": "출처가 「차이 없음」을 지지한다", "subject_id": sid,
            "claim_ids": [c["claim_id"] for c in same]}

    return DELTA_UNRESOLVED, {
        "why": "시대 차이를 말한 출처가 없다 — 못 찾은 것을 「차이 없음」으로 안 읽는다",
        "subject_id": sid, "gap_count": len(gaps)}


# ─────────────────────────────────────────────────────────────────────
# admission — ★팬아웃 **전**에 정한다
# ─────────────────────────────────────────────────────────────────────
#: 조사 결과의 상태. ★`unresolved` 하나로 뭉치면 **시간에 잘린 것**과
#: **다 보고도 못 정한 것**이 같아진다 — 앞의 것은 다음 판에서 다시 사야 한다.
STATUS_COMPLETED = "completed"            # 결론이 났다 (yes / no)
STATUS_UNRESOLVED_TERMINAL = "unresolved_terminal"   # 다 보고도 못 정했다
#: 시간 상한(`GAP_TIME_CAPPED`)·전송 실패(`GAP_TRANSMISSION_FAILED`) — **다시 산다**.
#: ★두 갈래 **모두 저장할 수 있어야** 이 상태가 뜻이 있다. 전에는 전송 실패를
#:  `validate_provenance` 가 막아서 이 갈래가 닿을 수 없었다.
STATUS_RETRYABLE = "retryable"
REVISION_STATUSES = (STATUS_COMPLETED, STATUS_UNRESOLVED_TERMINAL,
                     STATUS_RETRYABLE)

#: admission 이 「이미 샀다」로 세는 상태. ★`retryable` 은 **안 든다.**
_ELIGIBLE_DONE = frozenset({STATUS_COMPLETED, STATUS_UNRESOLVED_TERMINAL})


#: ★입력 신원에 **반드시** 들어가는 것. 하나라도 빠지면 그것이 바뀌어도
#:  옛 완료 행을 그대로 재사용한다 — 「조사했다」가 거짓이 된다 (Codex).
INPUT_IDENTITY_FIELDS = (
    "research_subject_id",
    "subject_payload_hash",   # 원문/subject 논리 payload. ★batch 전체 hash 금지
    "era", "region",
    "claims_pack_version",
    "prompt_raw_hash",        # 같은 version 이라도 **바이트**가 바뀌면 다른 조사
    "schema_raw_hash",
    "policy_version",
    "policy_contract_hash",   # version 문자열만으로는 내용 변화를 못 잡는다
    "provider", "model", "model_version",
)

#: ★**hash 모양이어야 하는 칸.** `"ph1"` 같은 자리끼움이 통과하면 「원문이
#:  바뀌었나」를 못 가른다 — 자리끼움끼리는 늘 같아서 **영원히 같은 조사**가 된다.
_HASH_IDENTITY_FIELDS = frozenset({
    "subject_payload_hash", "prompt_raw_hash",
    "schema_raw_hash", "policy_contract_hash"})
_HEX_RE = re.compile(r"^[0-9a-f]{8,64}$")

#: ★**「모른다」를 신원에 접으면 안 된다.** `unknown` 이 든 신원은 서로 같아져,
#:  아무것도 모르는 두 조사가 **한 조사로 접힌다**.
#:  ★부분 일치로 안 본다 — 정확히 이 값일 때만이다. 글자가 든 것으로 뜻을
#:  판단하면 정상 값(`none-slip` 같은 실제 표기)까지 죽인다.
_UNSETTLED_VALUES = frozenset({
    "unknown", "none", "null", "nil", "n/a", "na", "tbd", "todo", "-", "?"})


def canonical_research_input(**parts: Any) -> str:
    """입력 신원 12칸의 **정본 bytes**. hash 와 저장 JSON 이 **이것 하나**를 쓴다.

    ★따로 조립하면 hash 는 12칸인데 저장은 10칸이 되고, 그러면 저장된 행으로
    hash 를 다시 계산할 수 없다 — 합성 hash 는 역산이 안 되므로 그 행이 정말
    그 입력이었는지 **영원히 확인할 수 없다** (Codex).
    `canonical_rows` 가 claims 에 대해 하는 것과 같은 규칙이다.
    """
    _check_input_identity(parts)
    return json.dumps({k: parts[k].strip() for k in INPUT_IDENTITY_FIELDS},
                      ensure_ascii=False, sort_keys=True)


def research_input_hash(**parts: Any) -> str:
    """**무엇을 조사했는가** — 입력의 신원.

    ★**결과는 안 들어간다.** 같은 입력이라도 결과가 다를 수 있다 —
    그건 **다른 시도**이지 다른 조사가 아니다. 결과의 신원은
    `revision_content_hash` 가 따로 진다.
    ★**batch id 도 안 들어간다.** batch 크기를 바꿔도 완료된 subject 를 다시
    사면 안 된다. 그래서 `subject_payload_hash` 는 **subject 마다의 논리
    payload** 여야 하고 batch 전체 payload 여서는 안 된다 (Codex).
    ★**provider/model 도 들어간다.** 빼면 **옛 provider 의 미발견 결과가 새
    provider 검색까지 막는다** — 이전에 못 찾았다고 새 검색기로도 안 찾는다.
    ★`research_subject_id` 가 들어 있으므로 **입력 신원은 subject 마다 다르다.**
    """
    # ★**같은 bytes 에서 계산한다.** 여기서 따로 접으면 저장된 봉투와 hash 가
    #  갈리고, 갈리면 검산이 「늘 통과」거나 「늘 실패」가 된다.
    return hashlib.sha256(
        canonical_research_input(**parts).encode()).hexdigest()[:24]


def _check_input_identity(parts: Dict[str, Any]) -> None:
    """12칸이 **성립하는지**. ★hash 와 정본 bytes 가 같은 검증을 지난다."""
    extra = sorted(set(parts) - set(INPUT_IDENTITY_FIELDS))
    if extra:
        raise ValueError(f"입력 신원에 없는 칸을 넘겼다: {', '.join(extra)}")
    # ★**빈 칸만 막으면 부족하다** (Codex 실측). 12칸을 전부 `"unknown"` 으로
    #  채워도 hash 가 나왔고, `"ph1"` 같은 자리끼움도 통과했다.
    bad: List[str] = []
    for k in INPUT_IDENTITY_FIELDS:
        v = parts.get(k)
        # ★타입부터 본다. `str()` 을 먹이면 dict·list 도 문자열이 되는데,
        #  그 문자열은 원소 순서에 따라 달라져 신원이 흔들린다.
        if not isinstance(v, str):
            bad.append(f"{k}(문자열이 아니다: {type(v).__name__})")
            continue
        t = v.strip()
        if not t:
            bad.append(f"{k}(비었다)")
        elif t.lower() in _UNSETTLED_VALUES:
            bad.append(f"{k}(미확정 값 {t!r})")
        elif k in _HASH_IDENTITY_FIELDS and not _HEX_RE.match(t):
            bad.append(f"{k}(hash 모양이 아니다 {t!r})")
    if bad:
        raise ValueError(f"입력 신원이 성립하지 않는다: {', '.join(bad)}")


def subject_payload_hash(subject: Dict[str, Any]) -> str:
    """한 subject 의 **논리 payload** 지문. ★batch 전체 hash 를 쓰면 안 된다.

    ★★같은 대상을 batch 1 로 사든 8 로 사든 **같은 것을 산 것**이어야 한다.
    batch 전체로 hash 를 내면 묶이는 방식이 바뀔 때마다 다른 조사로 보여 같은
    것을 다시 산다.

    ★여기 하나가 정본이다 — 도구가 따로 계산하면 두 벌이 되고, 그러면 도구가
    잠근 신원과 프로덕션이 저장한 신원이 **갈린다**.
    """
    import hashlib

    body = json.dumps({
        "surface_form": str((subject or {}).get("surface_form") or ""),
        "source_quote": str((subject or {}).get("source_quote") or ""),
        "owner_type": str((subject or {}).get("owner_type") or ""),
    }, ensure_ascii=False, sort_keys=True)
    return hashlib.sha256(body.encode()).hexdigest()[:16]


def plan_admission(subject_ids: Sequence[str], *,
                   subject_payload_hashes: Dict[str, str],
                   research_inputs: Dict[str, Any],
                   done_rows: Sequence[Dict[str, Any]] = (),
                   cap: Optional[int] = None,
                   batch_size: int = 1) -> Dict[str, Any]:
    """이번 주행에서 **무엇을 살지** 정한다. 바깥 호출은 안 한다.

    ```
    ① `research_subject_id` 로 **안정 정렬** — 같은 입력이면 같은 순서
    ② 이미 끝난 것은 **재구매하지 않는다**
    ③ 남은 것 **앞에서부터** 상한만큼 admission
    ④ 초과는 `unresolved` + `time_capped_count`
    ⑤ 다음 resume 은 ③ 의 pending 부터 **결정적으로 이어간다**
    ```

    ★상한만 두고 이어가는 계약이 없으면 **영구 partial** 이 된다 — 매번 앞의
    것만 사고 뒤는 영원히 안 산다 (Codex).

    ★**batch 는 자르기만 한다.** batch 안에서도 subject 마다 판정·claim·출처는
    독립이고, 한 subject 가 미발견이어도 다른 subject 의 답을 공유하지 않는다.
    그래서 `claim_id` 도 batch 를 안 탄다 — batch 크기를 바꿔도 완료된 subject
    를 다시 안 산다.
    """
    # ★판을 안 주면 「무엇을 기준으로 완료인가」가 없다. 그러면 옛 조사가
    #  영원히 완료로 남아, 팩·시대·지역이 바뀌어도 다시 안 산다.
    # `research_inputs` 는 subject 를 뺀 나머지 입력 신원이다. 빈 칸은
    # `research_input_hash` 가 막는다.
    if batch_size < 1:
        raise ValueError(f"batch_size 는 1 이상이어야 한다: {batch_size}")
    if cap is not None and cap < 0:
        raise ValueError(f"cap 은 0 이상이어야 한다: {cap}")

    # ★**subject id 만으로 「완료」를 믿지 않는다.** 같은 대상이라도 claims 팩·
    #  시대·지역이 바뀌면 **다른 조사**다. id 만 보면 옛 결과가 영원히 살아남아
    #  「조사했다」가 거짓이 된다.
    # ★입력 신원은 **subject 마다 다르다** — hash 안에 `research_subject_id`
    #  가 들어 있다. 값 하나를 전부에 대 보면 두 대상이 다 완료된 정상 상태를
    #  표현할 수 없다 (Codex).
    want = {}
    for s in {str(x) for x in subject_ids}:
        ph = (subject_payload_hashes or {}).get(s)
        if not str(ph or "").strip():
            # ★원문이 바뀌었는지 모르면 「같은 조사인가」를 못 정한다.
            raise ValueError(f"subject_payload_hash 가 없다: {s}")
        want[s] = research_input_hash(
            research_subject_id=s, subject_payload_hash=ph, **research_inputs)

    # ★**행 순서에 안 흔들린다.** subject 하나를 hash 하나로 뭉개면 마지막 행이
    #  앞 행을 덮어, DB 가 돌려준 순서에 따라 재구매 여부가 바뀐다 (Codex).
    #  「그 (subject, 기대 입력 hash, 살아 있는 상태) 조합이 **있는가**」를 묻는다.
    seen_ok: Set[Tuple[str, str]] = set()
    seen_any: Set[Tuple[str, str]] = set()
    seen_retry: Set[Tuple[str, str]] = set()
    for r in done_rows or ():
        sid = str((r or {}).get("research_subject_id") or "")
        ih = str((r or {}).get("research_input_hash") or "")
        st = str((r or {}).get("status") or "")
        if not sid or not ih:
            continue
        seen_any.add((sid, ih))
        # ★`retryable`(시간 상한·전송 실패)은 **완료가 아니다.** 다음 판에서
        #  다시 산다 — 「상한에 걸렸다」를 「다 봤다」로 세면 통과가 거짓이 된다.
        if st in _ELIGIBLE_DONE:
            seen_ok.add((sid, ih))
        elif st == STATUS_RETRYABLE:
            seen_retry.add((sid, ih))
    done = {k for k, h in want.items() if (k, h) in seen_ok}
    stale = sorted(
        k for k, h in want.items()
        if (k, h) not in seen_any
        and any(sid == k for sid, _ in seen_any))
    # ★정렬을 안 하면 같은 입력이 판마다 다른 순서로 들어가고, 상한에 걸리는
    #  대상이 판마다 달라진다 — 이어가기가 성립하지 않는다.
    ordered = sorted({str(s) for s in subject_ids})
    pending = [s for s in ordered if s not in done]
    already = [s for s in ordered if s in done]

    admitted = pending if cap is None else pending[:cap]
    capped = [] if cap is None else pending[cap:]
    batches = [admitted[i:i + batch_size]
               for i in range(0, len(admitted), batch_size)]
    return {
        "contract_version": CLAIMS_CONTRACT_VERSION,
        "batches": batches,
        "admitted": admitted,
        "already_done": already,
        "stale_revision": [s for s in stale if s in ordered],
        "expected_input_hashes": {k: want[k] for k in ordered},
        # ★`retryable` 은 **기대 입력 hash 와 정확히 맞으면서 상태가
        #  retryable 인 것**만이다. 판이 다른 것(`stale`)이나 모르는 상태를
        #  여기 섞으면 「다시 살 것」과 「판이 바뀐 것」이 한 칸이 된다 (Codex).
        "retryable": [k for k in ordered if (k, want[k]) in seen_retry],
        "capped": capped,
        "time_capped_count": len(capped),
        # ★잘린 주행은 acceptance 기준선이 아니다. 그 사실을 산출에 남긴다.
        "is_acceptance_eligible": not capped,
        "logical_text_calls": len(batches),
        "batch_size": batch_size,
        "cap": cap,
    }


# ─────────────────────────────────────────────────────────────────────
# durable record — ★정본은 DB 다. 체크포인트는 참조만 한다
# ─────────────────────────────────────────────────────────────────────
#: ★조사 한 건의 **출처 신원**. 없으면 「누가·무엇으로 냈나」를 못 되짚는다.
#:  `revision_content_hash` 에는 **안 넣는다** — 같은 claims 는 생산자가 달라도
#:  같은 semantic hash 가 맞다 (Codex). 별도 좌표로 저장한다.
PROVENANCE_FIELDS = (
    "provider",              # openai / google / …
    "model",                 # 물리 모델 문자열
    "model_version",
    "prompt_pack_version",   # 팩 버전
    "prompt_locator",        # 어느 파일에서 왔나
    "schema_locator",        # ★schema 도 별도 파일이다 — 한쪽만 남기면 못 되짚는다
    "prompt_raw_hash",       # 실효 프롬프트 **바이트**
    "schema_raw_hash",
    "payload_hash",          # 실제 물리 payload
    "local_trace_id",
    "policy_version",
    # ★**전송이 어떻게 끝났는지를 명시 칸으로 받는다.** 전에는 `transmission_error`
    #  가 **비었는지**로 읽었는데, 없는 칸·빈 칸으로 뜻을 판단하면 조용히 절반이
    #  죽는다 — 이 판에서 이미 겪은 부류다.
    "transmission_status",
)

#: 전송 결말. ★`timeout`·`abandoned` 같은 갈래를 지금 만들지 않는다 — 그것을
#:  쓰는 호출자가 아직 없고, 쓸 자리 없는 갈래는 **아무도 안 채우는 칸**이 된다.
#:  사유는 `transmission_error` 자유 문장으로 남고 gap note 로 그대로 흐른다.
TRANSMISSION_OK = "ok"
TRANSMISSION_FAILED = "failed"
#: ★★★**아예 안 보냈다.** 물리 전송이 **0** 인 로컬 상한(크기 hard 상한 등)이다.
#:  ★`failed` 와 갈라야 한다 — `failed` 는 provider 쪽 문제라 고칠 곳이 다르고,
#:   기록도 「provider 장애」로 오독된다(Codex). 이건 **우리가 안 보낸 것**이라
#:   `time_capped` · `limit_kind=admission_limit` 으로 남고 retryable 이다.
TRANSMISSION_NOT_SENT = "not_sent"
TRANSMISSION_STATUSES = (TRANSMISSION_OK, TRANSMISSION_FAILED,
                         TRANSMISSION_NOT_SENT)

#: ★**결과 좌표는 둘 중 적어도 하나**다. 전에는 `provider_request_id` 를 늘
#:  요구해서, 주석은 「전송 실패면 그 사실」이라 적어 놓고 코드는 **전송 실패를
#:  아예 저장할 수 없었다** — `STATUS_RETRYABLE` 의 「전송 실패」 갈래가 닿을 수
#:  없는 코드였다. 규칙을 주석과 코드 두 곳에 적으면 한쪽만 고쳐진다.
#:  둘 다 비면 「불렀는데 어떻게 됐는지 모른다」이고, 그건 되짚을 수 없다.
#:  ★둘 다 있는 것은 **정상**이다 — 응답 ID 를 받은 뒤 스트림이 끊길 수 있다.
PROVENANCE_OUTCOME_FIELDS = ("provider_request_id", "transmission_error")

#: ★★**provenance 와 입력 신원은 같은 호출을 가리킨다.** 겹치는 좌표가 다르면
#:  둘 중 어느 쪽이 정본인지 아무도 모른다 — 실제로 입력에 `provider=openai`,
#:  provenance 에 `provider=other` 를 넣어도 저장됐다 (Codex 실측).
#:  왼쪽이 provenance 의 칸 이름, 오른쪽이 입력 신원의 칸 이름이다.
PROVENANCE_INPUT_BINDING = (
    ("provider", "provider"),
    ("model", "model"),
    ("model_version", "model_version"),
    ("prompt_pack_version", "claims_pack_version"),
    ("prompt_raw_hash", "prompt_raw_hash"),
    ("schema_raw_hash", "schema_raw_hash"),
    ("policy_version", "policy_version"),
)


def validate_provenance(prov: Optional[Dict[str, Any]]
                        ) -> Tuple[Optional[Dict[str, Any]], Optional[str]]:
    """조사 한 건의 출처. ★**빈 칸을 허용하지 않는다.**

    「나중에 채운다」로 두면 빈 provenance 행이 쌓이고, 그러면 그 행들은
    영원히 되짚을 수 없다 (Codex). 검색 배선이 **첫 실제 writer** 다.
    """
    if not isinstance(prov, dict):
        return None, "provenance 는 dict 여야 한다"
    # ★**모르는 칸을 조용히 버리지 않는다.** 전에는 허용 칸만 골라 복사해서,
    #  검색 배선이 `cost_usd` 같은 새 감사 좌표를 넘겨도 오류 없이 사라졌다
    #  (Codex 실측). 넘긴 쪽은 남았다고 믿고, 기록은 없다.
    allowed = set(PROVENANCE_FIELDS) | set(PROVENANCE_OUTCOME_FIELDS)
    unknown = sorted(set(prov) - allowed)
    if unknown:
        return None, ("provenance 에 모르는 칸이 있다: "
                      f"{', '.join(unknown)} — 남기려면 PROVENANCE_FIELDS 에 "
                      "넣고 계약을 올려라. 조용히 버리지 않는다")
    out: Dict[str, Any] = {}
    missing: List[str] = []
    for k in PROVENANCE_FIELDS:
        v = _norm(prov.get(k))
        if not v:
            missing.append(k)
        else:
            out[k] = v
    if missing:
        return None, f"provenance 빈 칸: {', '.join(missing)}"
    # ★전송이 깨졌으면 응답 ID 가 없다. 그때도 **무슨 일이 있었는지는 남는다.**
    #  둘 다 비는 것만 막는다 — 그건 「불렀는데 결과를 모른다」다.
    st = out["transmission_status"]
    if st not in TRANSMISSION_STATUSES:
        return None, (f"transmission_status 가 enum 밖이다: {st!r} — "
                      f"{TRANSMISSION_STATUSES}")
    outcome = {k: _norm(prov.get(k)) for k in PROVENANCE_OUTCOME_FIELDS}
    # ★**결말이 말한 것과 좌표가 맞아야 한다.** 안 맞으면 두 칸이 서로 다른
    #  말을 하고, 그 뒤로는 어느 쪽이 정본인지 아무도 모른다.
    if st == TRANSMISSION_OK and not outcome["provider_request_id"]:
        return None, "전송이 성공이라면 provider_request_id 가 있어야 한다"
    if st == TRANSMISSION_FAILED and not outcome["transmission_error"]:
        return None, "전송이 실패라면 무엇이 깨졌는지가 있어야 한다"
    out.update({k: v for k, v in outcome.items() if v})
    return out, None


def canonical_rows(rows: Sequence[Dict[str, Any]]) -> List[Dict[str, Any]]:
    """정본 순서. ★**hash 도 저장 JSON 도 이것을 쓴다.**

    둘이 따로 정렬하면 hash 는 같은데 저장 bytes 가 달라진다 — 같은 결과가
    다른 바이트로 남아 감사가 갈린다 (Codex).
    ★`sources` 안쪽도 정렬한다. 출처를 적은 **순서**가 결과를 바꾸면 안 된다.
    """
    out: List[Dict[str, Any]] = []
    for r in rows:
        r = dict(r or {})
        if isinstance(r.get("sources"), (list, tuple)):
            r["sources"] = sorted(str(u) for u in r["sources"])
        out.append(r)
    return sorted(out, key=lambda x: json.dumps(x, ensure_ascii=False,
                                                sort_keys=True))


def revision_content_hash(claims: Sequence[Dict[str, Any]],
                          gaps: Sequence[Dict[str, Any]],
                          delta: str) -> str:
    """**무엇이 나왔는가** — 결과의 신원.

    ★입력 신원(`research_input_hash`)과 **따로 둔다.** 같은 입력이라도 결과가
    다를 수 있고, 그건 **다른 시도**이지 다른 조사가 아니다 (Codex).
    """
    # ★**순서에 안 흔들린다.** `canonical_rows` 하나로 정렬한다 —
    #  hash 와 저장 JSON 이 **같은 것을** 소비해야 갈리지 않는다.
    def _rows(xs):
        return [json.dumps(x, ensure_ascii=False, sort_keys=True)
                for x in canonical_rows(xs)]

    # ★`status` 는 **안 넣는다.** 이제 상한도 전송 실패도 canonical gap 으로
    #  남으므로 status 는 (claims, gaps, delta) 에서 **그대로 유도된다** —
    #  넣으면 같은 결과가 다른 semantic hash 를 갖는다.
    #  ★전에는 「안 넣으면 재시도 행이 1차와 같은 키가 되어 저장이 막힌다」고
    #  적어 뒀는데, 그건 `attempt_id` 를 결과에서 뽑던 시절 이야기다. 지금은
    #  unique 가 `(…, research_input_hash, attempt_id)` 라 시도마다 다른 행이다.
    body = json.dumps({"claims": _rows(claims), "gaps": _rows(gaps),
                       "delta": str(delta)},
                      ensure_ascii=False, sort_keys=True)
    return hashlib.sha256(body.encode()).hexdigest()[:24]


def sanitize_and_decide(claims: Sequence[Dict[str, Any]],
                       gaps: Sequence[Dict[str, Any]] = (),
                       *, research_subject_id: str) -> Dict[str, Any]:
    """★**검증 → malformed 처리 → 판정**을 한 벌로 낸다.

    왜 따로 뺐나 — 저장 경로(`build_revision_row`)와 측정 도구가 **각자**
    검증하다가 갈렸다. 도구는 검증 못 거친 줄을 조용히 **버리고** 남은 것만으로
    `yes`/`no` 를 냈고, 저장 경로는 하나라도 깨지면 **통째로 `unresolved`** 로
    닫았다. 그러면 도구가 「맞음」이라고 세는 것을 production 은 미확정으로
    쓴다 — 재는 것과 도는 것이 다르다.

    ★버리는 것과 닫는 것 중 **닫는 쪽**이 계약이다. 오염된 줄이 있다는 것은
    그 대상을 다 못 본 것이고, 다 못 본 것은 「차이가 없다」로 내려갈 수 없다.

    Returns:
        ``{"claims", "gaps", "malformed", "delta", "why"}``
    """
    clean: List[Dict[str, Any]] = []
    malformed: List[str] = []
    for c in claims:
        _re, why_bad = validate_claim({
            "statement_native": (c or {}).get("statement_native"),
            "kind": (c or {}).get("kind", CLAIM_FACT),
            "required": (c or {}).get("required"),
            "sources": (c or {}).get("sources"),
            "delta_effect": (c or {}).get("delta_effect",
                                          DELTA_EFFECT_NEUTRAL),
            "discriminator_family": (c or {}).get("discriminator_family"),
        }, subject_id=(c or {}).get("research_subject_id")
            or research_subject_id)
        if _re is None:
            malformed.append(f"{(c or {}).get('claim_id') or '?'}: {why_bad}")
        elif (c or {}).get("claim_id") and \
                _re["claim_id"] != (c or {}).get("claim_id"):
            # ★id 를 **안 준 것**은 오염이 아니다 — provider 산출에는 원래
            #  없다. 준 뒤 내용과 안 맞는 것만 오염이다.
            malformed.append(
                f"{(c or {}).get('claim_id')}: claim_id 가 내용과 안 맞다")
        else:
            clean.append(_re)

    clean_gaps: List[Dict[str, Any]] = []
    for g in gaps:
        try:
            clean_gaps.append(build_gap(
                (g or {}).get("research_subject_id") or research_subject_id,
                reason=(g or {}).get("reason"),
                query=(g or {}).get("query") or "",
                discriminator=(g or {}).get("discriminator") or "",
                note=(g or {}).get("note") or "",
                required_discriminator=(g or {}).get(
                    "required_discriminator", False),
                # ★★칸을 더했으면 **그 칸을 복사하는 줄**도 고친다. 안 그러면
                #  `time_capped` gap 이 여기서 계약 위반이 되어 **통째로
                #  버려지고**, 그러면 「시간에 잘렸다」가 기록에서 사라진다.
                limit_kind=(g or {}).get("limit_kind") or ""))
        except (ValueError, TypeError) as exc:
            malformed.append(f"gap: {exc}")

    if malformed:
        delta, why = DELTA_UNRESOLVED, {
            "why": f"검증을 못 거친 claim {len(malformed)}개 — 정본에 안 넣는다",
            "malformed": malformed[:5]}
    else:
        delta, why = decide_delta(clean, clean_gaps,
                                  subject_id=research_subject_id)
    return {"claims": clean, "gaps": clean_gaps, "malformed": malformed,
            "delta": delta, "why": why}


def build_revision_row(*, project_id: str, episode_id: str,
                       research_subject_id: str,
                       subject_payload_hash: str,
                       research_inputs: Dict[str, Any],
                       claims: Sequence[Dict[str, Any]],
                       gaps: Sequence[Dict[str, Any]] = (),
                       provenance: Dict[str, Any],
                       attempt_id: str,
                       audit: Optional[Dict[str, Any]] = None,
                       time_capped: bool = False,
                       created_at: str) -> Dict[str, Any]:
    """저장할 한 행. ★판정을 **여기서 한 번** 하고 그 결과를 같이 담는다.

    ★호출부가 `decide_delta` 를 따로 부르고 결과만 넘기면, 저장된 `delta` 와
    저장된 `claims` 가 갈릴 수 있다. 갈리면 어느 쪽이 정본인지 아무도 모른다.
    """
    # ★`bool("false")` 는 True 다. 이 판에서 `required`·`gradable`·
    #  `required_discriminator` 로 이미 **세 번** 데인 부류다.
    if not isinstance(time_capped, bool):
        raise ValueError(f"time_capped 는 bool 이어야 한다: {time_capped!r}")
    # ★언제 조사했는지 없는 기록은 되짚을 수 없다. 빈 값을 허용하면
    #  「그때 뭐가 최신이었나」를 영원히 못 안다.
    if not _norm(created_at):
        raise ValueError("created_at 은 필수다")
    prov, prov_bad = validate_provenance(provenance)
    if prov is None:
        raise ValueError(f"provenance 가 성립하지 않는다 — {prov_bad}")
    # ★**같은 호출을 가리키는지 확인한다.** 겹치는 좌표가 갈리면 「무엇으로
    #  조사했나」와 「무엇을 조사했나」가 서로 다른 말을 하고, 그 뒤로는 어느
    #  쪽이 정본인지 아무도 모른다.
    drift = [
        f"{pk}: provenance={prov.get(pk)!r} vs 입력={research_inputs.get(ik)!r}"
        for pk, ik in PROVENANCE_INPUT_BINDING
        if _norm(prov.get(pk)) != _norm(research_inputs.get(ik))]
    if drift:
        raise ValueError("provenance 가 입력 신원과 다른 호출을 가리킨다 — "
                         + " · ".join(drift))

    # ★입력 신원은 **한 함수**가 만든다. 여기서 따로 조립하면 admission 이
    #  보는 것과 저장되는 것이 갈린다.
    envelope = canonical_research_input(
        research_subject_id=research_subject_id,
        subject_payload_hash=subject_payload_hash, **research_inputs)
    ih = hashlib.sha256(envelope.encode()).hexdigest()[:24]

    # ★**정본에 검증 안 거친 것을 넣지 않는다.** 검증·오염 처리·판정은
    #  `sanitize_and_decide` **한 벌**이 한다 — 측정 도구도 같은 것을 부른다.
    _fin = sanitize_and_decide(claims, gaps,
                               research_subject_id=research_subject_id)
    clean, clean_gaps = _fin["claims"], _fin["gaps"]
    malformed = _fin["malformed"]
    delta, why = _fin["delta"], _fin["why"]

    # ★상태를 셋으로 가른다. `unresolved` 하나로 뭉치면 **시간에 잘린 것**과
    #  **다 보고도 못 정한 것**이 같아진다 — 앞의 것은 다시 사야 한다.
    capped = time_capped or any(
        g.get("reason") == GAP_TIME_CAPPED for g in clean_gaps)
    # ★전송이 깨진 것은 **provenance 가 말한다** — 호출부의 별도 플래그를 또
    #  받으면 두 곳이 갈리고, 갈리면 어느 쪽이 정본인지 모른다.
    failed = prov.get("transmission_status") == TRANSMISSION_FAILED
    # ★★안 보낸 것은 **provider 실패가 아니다.** 우리 상한에 걸린 것이라
    #  `time_capped`(admission_limit)로 접는다 — 고칠 곳이 다르다.
    if prov.get("transmission_status") == TRANSMISSION_NOT_SENT:
        capped = True
    if failed:
        # ★깨진 전송에서 나온 부분 산출은 **정본이 아니다.** 그것을 그대로
        #  접으면 「다 보고 정했다」가 거짓이 된다.
        if not any(g.get("reason") == GAP_TRANSMISSION_FAILED
                   for g in clean_gaps):
            clean_gaps.append(build_gap(
                research_subject_id, reason=GAP_TRANSMISSION_FAILED,
                note=_norm(prov.get("transmission_error"))[:200],
                required_discriminator=True))
        delta, why = decide_delta(clean, clean_gaps,
                                  subject_id=research_subject_id)
        status = STATUS_RETRYABLE
    elif capped:
        # ★**시간 초과는 계약상 `unresolved`** 다. 상한에 걸렸는데 `delta=yes`
        #  로 저장되면, 「다 보고 조사하기로 했다」가 거짓이 된다 —
        #  아직 다 안 봤다. canonical gap 을 넣고 **판정을 다시 낸다**.
        if not any(g.get("reason") == GAP_TIME_CAPPED for g in clean_gaps):
            clean_gaps.append(build_gap(
                research_subject_id, reason=GAP_TIME_CAPPED,
                note="상한에 걸려 다 못 봤다", required_discriminator=True,
                # ★여기는 **팬아웃 전 승인 상한**이다. 호출 마감·전송 예산·
                #  주행 마감과 판정은 같지만 **고칠 곳이 다르다**.
                limit_kind=LIMIT_ADMISSION))
        delta, why = decide_delta(clean, clean_gaps,
                                  subject_id=research_subject_id)
        status = STATUS_RETRYABLE
    elif delta == DELTA_UNRESOLVED:
        status = STATUS_UNRESOLVED_TERMINAL
    else:
        status = STATUS_COMPLETED

    rch = revision_content_hash(clean, clean_gaps, delta)
    # ★**시도 신원은 호출 전에 발급한 명시값**이어야 한다. 결과에서 뽑으면
    #  같은 결과를 **두 번 실제 호출**했을 때 비용·trace 가 다른데 한 행으로
    #  접혀 두 번째 기록이 사라진다 (Codex).
    att = _norm(attempt_id)
    if not att:
        raise ValueError("attempt_id 는 필수다 — 호출 전에 발급한다")
    # ★row id 는 **전체**를 hash 한다. 앞 8자만 쓰면 `attempt-111A` 와
    #  `attempt-111B` 가 PK 충돌한다.
    _att_h = hashlib.sha256(att.encode()).hexdigest()[:16]
    return {
        "id": f"grr_{ih}_{rch[:8]}_{_att_h}",
        "project_id": str(project_id),
        "episode_id": str(episode_id),
        "research_subject_id": str(research_subject_id),
        "research_input_hash": ih,
        "revision_content_hash": rch,
        "attempt_id": att,
        "era": _norm(research_inputs.get("era")),
        "region": _norm(research_inputs.get("region")),
        "claims_pack_version": _norm(research_inputs.get("claims_pack_version")),
        "policy_version": _norm(research_inputs.get("policy_version")),
        # ★★입력 신원 12칸을 **정본 bytes 그대로** 남긴다. 합성 hash 는
        #  역산이 안 되므로, 이게 없으면 이 행이 정말 그 입력이었는지 영원히
        #  확인할 수 없다. hash 도 **이 문자열에서** 계산된다 (Codex).
        "research_input_json": envelope,
        "delta": delta,
        "delta_reason": str(why.get("why") or ""),
        "status": status,
        # ★**검증을 통과한 정규화 행만.** 원본을 그대로 넣으면 「정본에
        #  안 넣는다」가 거짓말이 된다 — 다음 사람이 이 칸을 정본으로 읽는다.
        # ★**hash 와 같은 canonicalizer 를 쓴다.** 따로 정렬하면 hash 는
        #  같은데 저장 bytes 가 달라져, 같은 결과가 다른 바이트로 남는다.
        "claims_json": json.dumps(canonical_rows(clean), ensure_ascii=False,
                                  sort_keys=True),
        "gaps_json": json.dumps(canonical_rows(clean_gaps),
                                ensure_ascii=False, sort_keys=True),
        # ★거절 **사유**를 남긴다. ★거절된 **원본**은 호출부가 `audit` 로
        #  넘겨야 보존된다 — 여기서 자동으로 안 담는다.
        "audit_json": json.dumps(
            {**dict(audit or {}),
             **({"rejected": malformed} if malformed else {})},
            ensure_ascii=False, sort_keys=True),
        "provenance_json": json.dumps(prov, ensure_ascii=False,
                                      sort_keys=True),
        "created_at": str(created_at),
    }