"""GROUNDING-V2 §4b — batch 실험 **채점**. ★유료 호출 전에 못박는다.

★조사와 채점을 **다른 모듈**에 둔다. 같은 함수에 있으면 채점이 조사를 봐준다.

## 무엇을 증명하나 (Codex 2026-08-30)

    foreign source binding 0 · unsupported cross-subject evidence 0

★**그 이상은 아니다.** 「의미상 교묘한 혼입이 전혀 없다」까지 기계적으로
증명했다고 쓰면 안 된다. LLM 판정기를 더하지 않는다 — 그 판정기가 또 흔들리는
축이 된다.

## 세 갈래로 센다

    어긋남   claim 이 **남의 subject 에 붙은 출처**를 인용했다 · 누락 · 중복
    미확정   인용을 확인할 수 없다 · 남의 고유 표면형이 근거 없이 나타났다
    맞음     그 밖

★**둘 다 0일 때만** 그 batch 가 후보다. ★**개수 그대로** 센다 — 9개 표본에
비율을 쓰면 한 건이 11%p 로 흔들린다.

★육안은 **진단·거부권**으로만 쓴다. 사람이 「괜찮아 보인다」며 미확정을 통과로
바꿀 수 없다 (Codex).
"""
from __future__ import annotations

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

#: 채점 갈래. ★「문제없음」이 아니라 **미확정**이 기본이다 — 확인 못 한 것을
#:  통과로 읽으면 그 배치가 거짓으로 이긴다.
V_WRONG = "어긋남"
V_UNSURE = "미확정"
V_OK = "맞음"
#: ★★★**아무것도 확정 못 한 줄.** claim 이 0 이면 남의 표기가 나올 수도 없어서
#:  세 축을 전부 그냥 지나간다 — 실측(2026-08-30 진단 1회): 83개 주소를 훑고
#:  claim 0 · gap 4 로 돌아온 주행이 「맞음 1」로 세어졌다.
#:  **못 찾은 것과 제대로 찾은 것을 같은 칸에 넣으면** 큰 batch 가 빈손일수록
#:  이긴다. 갈라 센다.
V_EMPTY = "빈손"
VERDICTS = (V_WRONG, V_UNSURE, V_OK, V_EMPTY)


def _norm(text: Any) -> str:
    """비교용 정규화 — NFKC + 소문자 + 공백 축약. ★뜻으로 묶지 않는다."""
    return " ".join(unicodedata.normalize("NFKC", str(text or "")).lower().split())


def _sources_of(batch: Dict[str, Any]) -> set:
    return {_norm(s.get("url")) for s in (batch.get("sources") or [])
            if _norm(s.get("url"))}


def _snippets_by_url(batch: Dict[str, Any]) -> Dict[str, List[str]]:
    """주소 → 그 주소에서 포착한 본문 조각들.

    ★★**주소별로** 묶는다. 호출 전체를 뭉쳐 놓고 「어딘가에 있으면 통과」로
    보면, claim 이 A 를 인용해 놓고 문장은 **B 에서 베껴 와도** 통과한다.
    """
    out: Dict[str, List[str]] = {}
    for src in (batch.get("sources") or []):
        u = _norm(src.get("url"))
        if u:
            out.setdefault(u, []).append(_norm(src.get("snippet")))
    return out


def score_batch(batch: Dict[str, Any], *, expected_ids: Sequence[str],
                unique_forms: Dict[str, Sequence[str]],  # ★비우면 축 하나가 죽는다
                schema: Optional[Dict[str, Any]] = None,
                baseline: Optional[Dict[str, set]] = None) -> Dict[str, Any]:
    """호출 하나를 채점한다. ★바깥 호출 없음.

    Args:
        batch: `search_claims` 가 낸 한 줄.
        expected_ids: 그 호출에 **넣은** subject id. 누락·중복을 이걸로 센다.
        unique_forms: subject id → **그 대상에만 있는 표면형**. 남의 것이
            근거 없이 claim 에 나타나면 미확정이다 (Codex ④).
            ★이건 **글자 비교**다. 그 값으로 **통과를 주지 않는다** — 미확정
            표시로만 쓴다.
    """
    rows: List[Dict[str, Any]] = []
    want = [str(x) for x in expected_ids]
    # ★★`unique_forms` 를 비워 부르면 **결속 축이 통째로 죽는다.** 실측에서
    #  그럴듯한 자가보고 오결속이 그 상태로 「맞음」이 됐다 — 비우고 부르는
    #  것을 허용하면 언젠가 그렇게 부른다. 명시로 비우려면 `{}` 가 아니라
    #  **`allow_no_forms=True`** 를 준다.
    if batch.get("not_sent"):
        # ★안 보낸 것은 「조사했다」에 안 든다. 어긋남도 아니다 — 미확정이다.
        return {"verdicts": [{"research_subject_id": s, "verdict": V_UNSURE,
                              "why": "hard 상한을 넘어 보내지 않았다"}
                             for s in want],
                "counts": {V_WRONG: 0, V_UNSURE: len(want), V_OK: 0,
                           V_EMPTY: 0}}
    if batch.get("error"):
        return {"verdicts": [{"research_subject_id": s, "verdict": V_UNSURE,
                              "why": f"전송이 깨졌다: {batch['error'][:80]}"}
                             for s in want],
                "counts": {V_WRONG: 0, V_UNSURE: len(want), V_OK: 0,
                           V_EMPTY: 0}}

    parsed = batch.get("parsed")
    if not isinstance(parsed, dict) or not isinstance(
            parsed.get("results"), list):
        return {"verdicts": [{"research_subject_id": s, "verdict": V_UNSURE,
                              "why": "응답을 읽을 수 없다"} for s in want],
                "counts": {V_WRONG: 0, V_UNSURE: len(want), V_OK: 0,
                           V_EMPTY: 0}}

    # ★★**응답 전체가 schema 를 지키나.** `strict=true` 로 보내지만 그것과
    #  **별개로** 우리가 다시 막는다 — provider 계약 하나에만 기대면, 그 계약이
    #  바뀌거나 안 지켜진 판을 아무도 안 잡는다. 독립 fail-closed 방어다
    #  (안 막으면 malformed gap·빈 id 행·추가 칸이 통과한다, Codex).
    #  하나라도 어긋나면 그 호출은 **통째로 미확정**이다: 어느 줄이 오염됐는지
    #  모르는 채로 일부만 통과시키면 그 통과가 거짓이 된다.
    bad_schema = schema_violations(batch, schema=schema)
    if bad_schema:
        return {"verdicts": [{"research_subject_id": s, "verdict": V_UNSURE,
                              "why": f"응답이 schema 를 어겼다: {bad_schema[:2]}"}
                             for s in want],
                "counts": {V_WRONG: 0, V_UNSURE: len(want), V_OK: 0,
                           V_EMPTY: 0}}

    got = [r for r in parsed["results"] if isinstance(r, dict)]
    by_id: Dict[str, List[Dict[str, Any]]] = {}
    for r in got:
        by_id.setdefault(str(r.get("research_subject_id") or ""), []).append(r)

    seen_urls = _sources_of(batch)
    by_url = _snippets_by_url(batch)

    for sid in want:
        mine = by_id.get(sid) or []
        if not mine:
            # ★누락은 **어긋남**이다 — batch 가 대상을 잃은 것이다.
            rows.append({"research_subject_id": sid, "verdict": V_WRONG,
                         "why": "결과에 그 대상 줄이 없다"})
            continue
        if len(mine) > 1:
            rows.append({"research_subject_id": sid, "verdict": V_WRONG,
                         "why": f"그 대상 줄이 {len(mine)}개다 — 중복"})
            continue
        row = mine[0]
        verdict, why = V_OK, ""
        for c in (row.get("claims") or []):
            if not isinstance(c, dict):
                verdict, why = V_WRONG, "claim 이 dict 가 아니다"
                break
            # ★★**대상 결속을 먼저 본다.** 인용이 참이어도 **다른 대상의
            #  사실**이면 정본이 오염된다 — 첫 유료 호출이 그랬다.
            #  ★HTTP support 검증은 「인용이 참인가」만 보므로 이걸 못 잡는다.
            #
            # ★★★**이것이 무엇을 증명하고 무엇은 못 하나** — 정직하게 적는다.
            #  증명한다  : 모델이 **스스로** 「이건 대상 것이다」라고 말했고,
            #             그 근거로 든 말이 **자기 문장 안에 실제로 있다**
            #  못 한다   : 그 말이 **정말 그 대상을 가리키는지**. 「차고」와
            #             「차고에 세워 둔 버스」처럼 한쪽이 다른 쪽 이름을
            #             포함하는 판은 글자로 못 가른다
            #  그래서 이건 **자가보고 + 자기모순 검사**이고, 의미 결속의
            #  증명이 아니다. 남는 몫은 고정 회귀 fixture 와 사람의 **거부권**
            #  이 진다(사람은 통과를 못 준다).
            tb = c.get("target_binding")
            if not isinstance(tb, dict):
                verdict, why = V_WRONG, "target_binding 이 없다 — 결속을 못 본다"
                break
            if tb.get("is_about_target") is not True:
                verdict = V_WRONG
                why = ("대상 자체의 사실이 아니라고 스스로 표시했다 — "
                       f"{str(tb.get('why'))[:60]}")
                break
            words = _norm(tb.get("target_words_in_statement"))
            if not words:
                verdict, why = V_UNSURE, "대상을 가리키는 말을 안 적었다"
                continue
            if words not in _norm(c.get("statement_native")):
                # ★적어 놓고 문장에 없으면 **자기 보고가 자기와 안 맞는다**.
                verdict = V_WRONG
                why = f"대상을 가리킨다는 말이 문장에 없다: {words[:24]!r}"
                break
            raw_src = c.get("sources")
            if raw_src is not None and not isinstance(raw_src, list):
                # ★★목록이 아니면 **명시로 잡는다.** 문자열을 그냥 돌면 글자
                #  하나하나가 주소로 세어져 **우연히** 걸린다 — 우연히 맞는
                #  것은 다음 판에 우연히 틀린다.
                verdict = V_WRONG
                why = f"sources 가 목록이 아니다: {type(raw_src).__name__}"
                break
            urls = [_norm(u) for u in (raw_src or [])]
            if not urls:
                verdict, why = V_UNSURE, "출처 없는 claim 이 있다"
                continue
            # ★★**남의 출처를 인용했나** — 이 호출이 실제로 본 주소 밖이면
            #  그 claim 은 근거를 못 댄다. 즉시 어긋남 (Codex ②).
            outside = [u for u in urls if u not in seen_urls]
            if outside:
                verdict = V_WRONG
                why = f"이 호출이 안 본 주소를 인용했다: {outside[:2]}"
                break
            # ★★**기준선과 대 본다** (Codex). `batch=1` 은 대상마다 호출이
            #  하나라 그 주소가 누구 것인지 **정확히** 안다. 그걸 기준으로:
            #    자기 것       → 통과
            #    남의 것       → **어긋남** (같은 호출 안이라도)
            #    어느 쪽도 아님 → 미확정 (검색이 비결정적이라 새 주소일 수 있다)
            if baseline:
                mine_base = baseline.get(sid) or set()
                theirs = {u for k, v in baseline.items() if k != sid
                          for u in v} - mine_base
                stolen = [u for u in urls if u in theirs]
                if stolen:
                    verdict = V_WRONG
                    why = f"남의 대상이 찾은 주소를 인용했다: {stolen[:2]}"
                    break
                unknown = [u for u in urls
                           if u not in mine_base and u not in theirs]
                if unknown:
                    verdict = V_UNSURE
                    why = ("기준선에 없는 주소다 — 검색이 비결정적이라 새 "
                           f"주소일 수 있다: {unknown[:2]}")
                    continue
            span = _norm(c.get("evidence_span"))
            if not span:
                verdict, why = V_UNSURE, "인용 조각이 없다 — 확인할 수 없다"
                continue
            # ★★인용이 그 페이지에 **실제로 있나**는 여기서 안 본다 —
            #  공식 stable API 가 본문을 안 주므로 provider snippet 으로는
            #  못 잰다. 그 축은 `grounding_claims_support` 가 **그 URL 을 직접
            #  열어서** 진다 (Codex).
            #  ★다만 provider 가 본문을 준 판(beta `results`)에서는 여기서도
            #   본다 — 공짜로 얻는 신호를 버리지 않는다.
            mine_snips = [t for u in urls for t in by_url.get(u, []) if t]
            if mine_snips and not any(span in t for t in mine_snips):
                verdict = V_UNSURE
                why = "인용 조각이 **자기가 인용한 주소의 본문**에 없다"
                continue
            # ★남의 고유 표면형이 **자기 근거 없이** 나타났나 (Codex ④).
            #  ★글자 비교다 — 미확정 표시로만 쓰고 통과를 주지 않는다.
            stmt = _norm(c.get("statement_native"))
            for other, forms in unique_forms.items():
                if other == sid:
                    continue
                hit = [f for f in forms if _norm(f) and _norm(f) in stmt
                       and _norm(f) not in span]
                if hit:
                    verdict = V_UNSURE
                    why = f"남의 고유 표기가 근거 없이 나타났다: {hit[:2]}"
                    break
            if verdict != V_OK:
                continue
        if verdict == V_OK and not (row.get("claims") or []):
            # ★★확정한 문장이 **하나도 없다.** 세 축을 지나간 것이 아니라
            #  **잴 것이 없었던** 것이다 — 「맞음」으로 세면 다음 판에 그 수를
            #  그대로 성과로 읽는다.
            gaps = len(row.get("gaps") or [])
            verdict = V_EMPTY
            why = f"claim 이 하나도 없다 — 확정한 것이 없다(gap {gaps}개)"
        rows.append({"research_subject_id": sid, "verdict": verdict,
                     "why": why})

    # ★넣지 않은 대상이 결과에 있으면 그것도 어긋남이다 — 남의 줄이 섞였다.
    extra = sorted(set(by_id) - set(want) - {""})
    for sid in extra:
        rows.append({"research_subject_id": sid, "verdict": V_WRONG,
                     "why": "넣지 않은 대상의 줄이 왔다"})

    return {"verdicts": rows,
            "counts": {v: sum(1 for r in rows if r["verdict"] == v)
                       for v in VERDICTS}}


def schema_violations(batch: Dict[str, Any], *,
                      schema: Optional[Dict[str, Any]] = None) -> List[str]:
    """응답 **전체**를 팩 JSON schema 로 본다.

    ★`strict=true` 로 보내지만 그것과 **별개로** 여기서 다시 막는다 —
    provider 계약 하나에만 기대지 않는 독립 fail-closed 방어다.

    ★★전에는 dict 인 claim 만 `validate_claim` 으로 보고 **malformed gap·행·빈
    id·추가 칸을 통째로 무시**했다. Codex 반례: 기대 행에 `claims=[]` +
    `gaps=[reason=bogus]` 를 넣고 빈 id 행을 하나 더 붙이면 **후보로 통과**했다.
    """
    import jsonschema

    if schema is None:
        from app.modules.pipeline.grounding_claims_search import (
            SCHEMA_STEM, load_pack)
        schema = load_pack()["stems"][SCHEMA_STEM]["content"]
    parsed = batch.get("parsed")
    if not isinstance(parsed, dict):
        return ["응답을 읽을 수 없다"]
    out = [f"{'/'.join(str(x) for x in e.path)}: {e.message}"[:120]
           for e in sorted(
               jsonschema.Draft202012Validator(schema).iter_errors(parsed),
               key=lambda e: list(e.path))]
    # ★gap 은 **프로덕션 계약**으로 한 번 더 본다 — schema 의 enum 을 통과해도
    #  `build_gap` 이 거부하는 모양이 있다.
    from app.modules.pipeline.grounding_claims import build_gap

    for row in (parsed.get("results") or []):
        if not isinstance(row, dict):
            continue
        sid = str(row.get("research_subject_id") or "")
        for gp in (row.get("gaps") or []):
            if not isinstance(gp, dict):
                out.append("gap 이 dict 가 아니다")
                continue
            try:
                build_gap(sid or "?", reason=gp.get("reason"),
                          query=gp.get("query") or "",
                          discriminator=gp.get("discriminator") or "",
                          note=gp.get("note") or "",
                          required_discriminator=gp.get(
                              "required_discriminator", False))
            except (ValueError, TypeError) as exc:
                out.append(f"gap: {str(exc)[:80]}")
    return out


def contract_survival(batches: Sequence[Dict[str, Any]]) -> Dict[str, Any]:
    """유료로 산 claim 이 **§4a 계약을 통과하나**. ★무료 — 같은 검증기를 쓴다.

    ★채점(`score_run`)과 **다른 것**을 잰다. 채점은 「batch 가 대상을 잃거나
    남의 출처를 붙였나」이고, 이건 「그 claim 을 정본에 넣을 수 있나」다.
    ★그리고 **프로덕션 검증기**(`grounding_claims.validate_claim`)를 그대로
    부른다 — 여기서 따로 판정하면 실험이 프로덕션과 다른 잣대를 쓴다.
    """
    from app.modules.pipeline.grounding_claims import validate_claim

    ok, bad, reasons = 0, 0, {}
    for b in batches:
        parsed = b.get("parsed")
        if not isinstance(parsed, dict):
            continue
        for row in (parsed.get("results") or []):
            if not isinstance(row, dict):
                continue
            sid = str(row.get("research_subject_id") or "")
            for c in (row.get("claims") or []):
                if not isinstance(c, dict):
                    bad += 1
                    reasons["claim 이 dict 가 아니다"] = reasons.get(
                        "claim 이 dict 가 아니다", 0) + 1
                    continue
                got, why = validate_claim(c, subject_id=sid)
                if got is None:
                    bad += 1
                    reasons[str(why)[:60]] = reasons.get(str(why)[:60], 0) + 1
                else:
                    ok += 1
    return {"passed": ok, "rejected": bad, "reasons": reasons,
            # ★비율을 안 쓴다 — 표본이 작아 한 건이 크게 흔들린다.
            "note": "개수 그대로. 통과 0 이어도 batch 결함과는 다른 축이다"}


def url_baseline(batches: Sequence[Dict[str, Any]]) -> Dict[str, set]:
    """`batch=1` 주행에서 **대상별로 그 대상이 찾은 주소**를 뽑는다.

    ★★한 호출에 대상이 하나뿐일 때만 소유권이 **정확하다** — 그 호출이 본
    주소는 전부 그 대상 것이다. 그래서 `batch=1` 산출이 **기준선**이 된다
    (Codex).

    ★대상이 둘 이상인 호출은 **안 쓴다** — 거기서 만든 기준선은 이미 섞여 있다.
    """
    out: Dict[str, set] = {}
    for b in batches:
        want = [str(x) for x in (b.get("requested") or [])]
        if len(want) != 1 or b.get("error") or b.get("not_sent"):
            continue
        out.setdefault(want[0], set()).update(
            _norm(s.get("url")) for s in (b.get("sources") or [])
            if _norm(s.get("url")))
    return out


def ownership_strength(batches: Sequence[Dict[str, Any]], *,
                       has_baseline: bool = False) -> Dict[str, Any]:
    """★★출처 소유권을 **얼마나 세게** 잴 수 있나. 정직하게 적는다.

    provider 는 검색 결과를 **호출 단위**로 준다 — 어느 subject 의 질의가 그
    주소를 물어 왔는지 안 알려 준다. 그래서:

    ``subject``  한 호출에 대상이 **하나**뿐 → 그 호출이 본 주소는 전부 그
                 대상 것이다. 소유권을 **정확히** 잰다
    ``call``     대상이 둘 이상 → 같은 주소 풀을 나눠 쓴다. A 를 위해 찾은
                 주소를 B 가 인용해도 **못 잡는다**

    ★그러니 `batch=1` 이 아닌 판의 「foreign source binding 0」은 **호출 밖
    주소를 안 썼다**는 뜻이지 **남의 주소를 안 썼다**는 뜻이 아니다.
    이걸 안 적으면 보고가 증거보다 세진다.
    """
    per = [len(b.get("requested") or []) for b in batches]
    exact = all(n <= 1 for n in per)
    return {
        "level": "subject" if exact else ("baseline" if has_baseline else "call"),
        "max_subjects_per_call": max(per) if per else 0,
        "note": ("한 호출에 대상이 하나뿐이라 출처 소유권을 **정확히** 쟀다"
                 if exact else
                 ("`batch=1` 기준선과 대 봤다 — 남의 대상이 찾은 주소를 쓴 것은 "
                  "**잡는다**. 기준선에 없는 새 주소는 미확정이다"
                  if has_baseline else
                  "한 호출에 대상이 여럿인데 **기준선이 없다** — 소유권은 호출 "
                  "단위까지만 쟀고, 같은 호출 안에서 남의 주소를 쓴 것은 못 잡는다")),
    }


def semantic_deltas(batches: Sequence[Dict[str, Any]]) -> Dict[str, Any]:
    """★**프로덕션 판정기로** subject 마다 결론을 낸다 (Codex).

    도구가 따로 세면 도구와 프로덕션이 갈린다 — 그 부류의 결함을 하루에 네 번
    만났다. 그래서 세 함수를 **그대로 부른다**:
    `validate_claim` → `build_gap` → `decide_delta`.

    ★검색 산출은 **원재료**다. 프로덕션도 그것을 바로 판정에 넣지 않고
    `validate_claim` 으로 세운 뒤에 넣는다 — `claim_id` 도 거기서 붙는다.
    그 단계를 건너뛰고 판정기에 넣으면 터진다(실측).

    `unresolved` 는 「못 찾음·충돌·출처 없음」이다. **빈손은 여기서
    `unresolved` 로 떨어진다** — 「아무것도 안 나옴 → no」는 계약이 금지한다.
    """
    from app.modules.pipeline.grounding_claims import (
        build_gap, decide_delta, validate_claim)

    per: List[Dict[str, Any]] = []
    for b in batches:
        parsed = b.get("parsed")
        got = (parsed or {}).get("results") if isinstance(parsed, dict) else None
        for sid in (b.get("requested") or []):
            sid = str(sid)
            mine = [r for r in (got or []) if isinstance(r, dict)
                    and str(r.get("research_subject_id") or "") == sid]
            if len(mine) != 1:
                # ★없거나 둘이면 **판정 자체가 성립 안 한다** — 미확정이다.
                per.append({"research_subject_id": sid, "delta": "unresolved",
                            "why": {"why": f"그 대상 줄이 {len(mine)}개다"},
                            "rejected": []})
                continue
            claims, rejected = [], []
            for raw in (mine[0].get("claims") or []):
                if not isinstance(raw, dict):
                    rejected.append("claim 이 dict 가 아니다")
                    continue
                c, why = validate_claim(raw, subject_id=sid)
                (claims.append(c) if c else rejected.append(str(why)))
            gaps = []
            for raw in (mine[0].get("gaps") or []):
                if not isinstance(raw, dict):
                    rejected.append("gap 이 dict 가 아니다")
                    continue
                try:
                    gaps.append(build_gap(
                        sid, reason=str(raw.get("reason") or ""),
                        query=str(raw.get("query") or ""),
                        discriminator=str(raw.get("discriminator") or ""),
                        note=str(raw.get("note") or ""),
                        # ★★`bool(...)` 로 감싸면 문자열 "false" 가 **참**이
                        #  된다 — 프로덕션 `build_gap` 은 진짜 bool 을 요구하고
                        #  아니면 거절한다. 여기서 고쳐 주면 **도구가 프로덕션
                        #  보다 무르다**. 원값을 그대로 넘겨 거절하게 둔다.
                        required_discriminator=raw.get(
                            "required_discriminator")))
                except Exception as exc:  # noqa: BLE001 — 사유를 남긴다
                    rejected.append(f"gap 을 못 세운다: {exc}")
            if rejected and not claims and not gaps:
                per.append({"research_subject_id": sid, "delta": "unresolved",
                            "why": {"why": "세울 수 있는 claim·gap 이 없다"},
                            "rejected": rejected})
                continue
            delta, why = decide_delta(claims, gaps, subject_id=sid)
            per.append({"research_subject_id": sid, "delta": delta,
                        "why": why, "rejected": rejected})
    counts = {k: sum(1 for r in per if r["delta"] == k)
              for k in ("yes", "no", "unresolved")}
    return {"per_subject": per, "counts": counts,
            "has_unresolved": counts["unresolved"] > 0,
            "decided_by": ("grounding_claims.validate_claim + build_gap + "
                           "decide_delta (프로덕션 함수)")}


def score_run(batches: Sequence[Dict[str, Any]], *,
              unique_forms: Dict[str, Sequence[str]],
              schema: Optional[Dict[str, Any]] = None,
              baseline: Optional[Dict[str, set]] = None) -> Dict[str, Any]:
    """한 batch 크기의 주행 전체. ★**둘 다 0일 때만** 후보다."""
    rows: List[Dict[str, Any]] = []
    for b in batches:
        rows.extend(score_batch(b, expected_ids=b.get("requested") or [],
                                unique_forms=unique_forms,
                                schema=schema, baseline=baseline)["verdicts"])
    counts = {v: sum(1 for r in rows if r["verdict"] == v) for v in VERDICTS}
    semantic = semantic_deltas(batches)
    return {
        "verdicts": rows,
        # ★**개수 그대로.** 9개 표본에 비율을 쓰면 한 건이 11%p 로 흔들린다.
        "counts": counts,
        # ★★★**빈손이 하나라도 있으면 후보가 아니다** (Codex). 어긋남·미확정이
        #  0 이어도 그 줄은 잰 것이 없다 — 그대로 두면 **큰 batch 가 빈손일수록
        #  이긴다**. 사람 검토로도 빈손을 통과로 못 올린다.
        # ★두 축을 **한 칸에 합치지 않는다.** 여기는 **기계 채점**(형식이
        #  어긋났나)이고, 「조사가 됐나」는 아래 `semantic` 이다. 크기를 고를 때
        #  둘 다 요구한다 — `choose_batch_size` 를 보라.
        "is_candidate": (counts[V_WRONG] == 0 and counts[V_UNSURE] == 0
                         and counts[V_EMPTY] == 0 and counts[V_OK] > 0),
        # ★★**연구 의미의 결론**은 프로덕션 `decide_delta` 가 낸다 (Codex).
        #  기계 채점(위 counts)은 「형식이 어긋났나」이고, 이건 「조사가 됐나」다.
        #  둘을 한 칸에 두면 빈손이 성공처럼 읽힌다.
        "semantic": semantic,
        # ★★출처 소유권을 **얼마나 세게** 쟀는지 같이 낸다. batch>1 에서는
        #  호출 단위라, 「남의 주소를 안 썼다」가 아니라 「호출 밖 주소를 안
        #  썼다」까지다.
        "ownership": ownership_strength(batches,
                                        has_baseline=bool(baseline)),
        # ★빈손은 **따로** 낸다 — 「나빴다」도 「좋았다」도 아니다.
        "empty_rows": counts[V_EMPTY],
        "proves": ("빈손 줄은 세 축을 지나간 것이 아니라 잴 것이 없었던 것이다. "
                   "foreign source binding 0 · unsupported cross-subject "
                   "evidence 0 · target_binding 자가보고와 자기모순 0 — "
                   "**그 이상은 아니다**. 「그 말이 정말 그 대상을 가리키나」는 "
                   "글자로 못 가른다(예: 「차고」 vs 「차고에 세워 둔 버스」). "
                   + ownership_strength(
                       batches, has_baseline=bool(baseline))["note"]),
    }


def choose_batch_size(scored: Dict[int, Dict[str, Any]]) -> Dict[str, Any]:
    """후보 중 **가장 큰** 크기. ★후보가 없으면 **기본값을 안 바꾼다**.

    ★「제일 나은 것」이 아니라 「어긋남 0 · 미확정 0」인 것만 후보다 —
    「제일 나은 것」을 고르면 전부 나쁠 때도 하나가 이긴다.
    """
    # ★★기준선 없이 잰 `batch>1` 은 **후보가 못 된다** (Codex). 「호출 밖
    #  주소를 안 썼다」는 「남의 주소를 안 썼다」가 아니고, 그 상태로 크기를
    #  고르면 못 잰 것을 통과로 삼는 것이다.
    # ★★★**「고른다」가 아니라 「기계로 걸러진다」**. 결속을 기계로 다 못
    #  닫았으므로(자가보고+자기모순+남의 고유표기를 다 지나는 오결속이 실재),
    #  여기서 나온 값은 **기본값 변경 근거가 아니다** (Codex).
    # ★★`is_candidate` 안에 이미 빈손 0 · semantic unresolved 없음이 들어
    #  있지만, **여기서도 명시로 본다** — 위쪽이 완화되면 조용히 통과한다.
    ok = sorted(b for b, r in scored.items()
                if r.get("is_candidate")
                and r.get("counts", {}).get(V_EMPTY, 0) == 0
                and not r.get("semantic", {}).get("has_unresolved")
                and (b == 1 or r.get("ownership", {}).get("level")
                     in ("subject", "baseline")))
    return {
        "candidates": ok,
        "chosen": ok[-1] if ok else None,
        "why": ("어긋남 0 · 미확정 0 이고 **소유권을 실제로 잰** 것 중 가장 큰 것"
                if ok else
                "후보가 없다 — 기본값을 바꾸지 않는다(개별 호출 유지)"),
        # ★깨끗한데 **소유권을 못 잰** 것은 따로 적는다 — 후보는 아니지만
        #  「나빴다」도 아니다. 섞어 두면 다음 판에 오독한다.
        "clean_but_unmeasured": sorted(
            b for b, r in scored.items()
            if r.get("is_candidate") and b != 1
            and r.get("ownership", {}).get("level") == "call"),
        "scored": {b: scored[b]["counts"] for b in sorted(scored)},
    }
