"""episode_reference_policy StepRunner — scene_detail 직전 reference
necessity manifest 1회 계산 (deterministic, LLM 없음).

설계: docs/reference-necessity/index.html §6.1 / §7 Phase 2.
"""
from __future__ import annotations

import json
import logging
from pathlib import Path
from typing import Any, Dict, Tuple

from app.core.episode_reference_policy import (
    EPISODE_REFERENCE_POLICY_SCHEMA_VERSION,
    compute_episode_reference_policy,
)
from app.core.step_runner import StepRunner
from app.modules.pipeline import grounding_entity_contract as _gc

logger = logging.getLogger(__name__)


#: 중앙 조사 결과를 내는 스텝. ★이름을 여기서 지어내지 않는다.
_CENTRAL_STEP = "reference_acquisition"


#: ★이 스텝이 **정책 문**을 집행한다 — `required_refs` 로 올린다.
#:  활성화 안전문이 「선언만 있고 구현이 없는 집행자」를 이것으로 본다.
ENFORCES_GATE = _gc.ENFORCE_BY_POLICY


def _audit_row(row, gate: str):
    """감사에 남길 한 줄. ★raw `status` 와 처분을 **그대로** 안고 간다.

    ★접은 뒤(`outcome`)만 남기면 **왜 없는지**가 사라진다.
    """
    return {"final_id": row["final_id"], "owner_type": row["owner_type"],
            "research_subject_id": row.get("research_subject_id"),
            "enforced_by": gate, "outcome": row["outcome"],
            "status": row["status"], "disposition": row["disposition"],
            # ★어느 칸 때문에 이 갈래에 들어왔는지가 보여야 한다
            "fidelity": row.get("fidelity"),
            "why": row["why"], "why_unbought": row["why_unbought"]}


def _short_id_base(sid: str) -> str:
    return sid.split("O")[0] if sid and "O" in sid else (sid or "")


def build_selected_map_or_raise(shot_selection_cp) -> Dict[int, set]:
    """shot_selection checkpoint → {scene_index: set(selected shot_index)}.

    checkpoint 부재/empty 면 AppError fail-fast — episode_reference_policy 는
    scene_detail 직전 SOT 라 selected_map 불완전 시 전체 shot fallback 금지
    (Phase 0 audit 와 동일 fail-closed 원칙).
    """
    from app.core.errors import AppError
    scenes = (shot_selection_cp or {}).get("data", {}).get("scenes")
    if not scenes:
        raise AppError(
            code="step.no_input",
            message="shot_selection 결과 없음 — episode_reference_policy 는 "
                    "selected_map 없이 진행 불가",
            status_code=400,
        )
    selected_map: Dict[int, set] = {}
    for s in scenes:
        selected_map[s.get("scene_index")] = set(
            s.get("selected_shot_indices", []) or []
        )
    return selected_map


def compute_visible_shot_count_from_checkpoints(
    shot_director_data: Dict[str, Any],
    selected_map: Dict[int, set],
) -> Dict[str, int]:
    """shot_director 의 selected shot 에서 entity 등장 횟수 집계.

    selected_map = {scene_index: set(selected shot_index)}.
    fail-fast (fallback 금지) 2종 — false-positive/negative reference 분류 방지:
      - shot_director 의 어떤 scene 이 selected_map 에 없으면 AppError;
      - selected shot index 가 shot_director.shots 에 실재하지 않으면 AppError
        — 그 shot 의 visible 을 못 세 visible_shot_count 가 undercount 되고
        scene_detail 은 scene-level fallback 으로 그 shot 을 계속 만들므로,
        recurring character 가 text_only 로 잘못 강등될 수 있다
        (range review IMPORTANT 1).
    """
    from app.core.errors import AppError
    counts: Dict[str, int] = {}
    for sc in (shot_director_data or {}).get("scenes", []) or []:
        si = sc.get("scene_index")
        if si not in selected_map:
            raise AppError(
                code="step.episode_reference_policy.selected_scene_missing",
                message=f"shot_selection 에 scene {si} 누락 — selected_map "
                        f"불완전 (fallback 금지)",
                status_code=400,
            )
        sel = selected_map[si]
        seen_selected: set = set()
        for sh in sc.get("shots", []) or []:
            shot_idx = sh.get("shot_index")
            if shot_idx not in sel:
                continue
            seen_selected.add(shot_idx)
            for sid in sh.get("visible_entity_ids", []) or []:
                b = _short_id_base(sid)
                if b:
                    counts[b] = counts.get(b, 0) + 1
        missing_shots = sel - seen_selected
        if missing_shots:
            raise AppError(
                code="step.episode_reference_policy.selected_shot_missing",
                message=f"scene {si} 의 selected shot {sorted(missing_shots)} "
                        f"가 shot_director.shots 에 없음 — selected_map 불완전 "
                        f"(visible_shot_count undercount 위험, fallback 금지)",
                status_code=400,
            )
    return counts


class EpisodeReferencePolicyStep(StepRunner):
    """Step 21.65: episode reference necessity manifest 계산."""

    def _load_prev_checkpoint(self, step_id: str):
        from app.core.config import settings
        cp = (
            Path(settings.projects_dir) / self.project_id
            / "checkpoints" / "episodes" / self.episode_id
            / step_id / "manifest.json"
        )
        if cp.exists():
            return json.loads(cp.read_text(encoding="utf-8"))
        return None

    def _uses_central(self) -> bool:
        """이 판이 **중앙 조사**를 쓰나. ★모드를 여기서 다시 정의하지 않는다."""
        from app.core.grounding_mode import (resolve_grounding_mode,
                                             uses_chunk_producer)

        return uses_chunk_producer(resolve_grounding_mode(self.project_config))

    def _central_forced_short_ids(self):
        """중앙 조사 결과 → **정책 문이 맡은 갈래만** 강제. ★나머지는 감사.

        Returns:
            `(forced, blocked, audit)`.

        ★★★`blocked` 는 **늘 빈 집합**이다. 못 구한 것은 참조 없이 내려간다
        (사용자 확정 2026-08-31: HITL 0). 못 구했다고 여기서 세우면 그것이
        사람 대기다 — `reference_acquisition.downstream_blocked` 한 곳이 이미
        「막지 않는다」로 정해 뒀고, 여기서 다시 정하지 않는다.

        ★★갈래마다 **집행자가 하나**다 (`REFERENCE_ENFORCEMENT_BY_OWNER`).
        정책 문이 아닌 갈래를 `forced` 에 넣으면 배경 묶음·아웃룩 자리가 이미
        붙이는 것을 **두 벌**로 요구하게 된다 (Codex 2026-09-01).

        ★rows 의 dict 를 여기서 다시 해석하지 않는다 — 검증은
        `grounding_central_acquisition.acquisition_projection` 한 곳이 한다.
        """
        from app.core.errors import AppError
        from app.modules.pipeline import grounding_central_acquisition as ca
        from app.modules.pipeline.grounding_entity_contract import (
            ENFORCE_BY_POLICY, REFERENCE_ENFORCEMENT_BY_OWNER,
            enforcement_gate_of)
        from app.modules.pipeline.reference_acquisition import (
            FIDELITY_REJECTED, FIDELITY_VERIFIED, STATUS_SELECTED)

        # ★★`required=True` — 이 판은 중앙 조사에 **직접 의존**한다.
        #  CP 가 없거나 깨졌으면 「대상 0」이 아니라 **기록 손실**이다
        #  (Codex BLOCK 2026-09-02).
        # ★★★HITL 0 (사용자 2026-09-03): 날것 CP — 사람 판정 표는 production 의
        #  의존성이 아니다. 자동 선택 한 장이 그대로 참조다.
        cp = self._load_prev_checkpoint("reference_acquisition")
        rows = ca.acquisition_projection(cp, required=True)
        forced = set()
        elsewhere, unavailable, not_applicable = [], [], []
        #: ★★★coarse 로 고른 것과 **고증이 확인된 것**은 다르다
        #:  (Codex BLOCK 2026-09-02). 심판은 「무엇인가·보이는가」만 봤다 —
        #:  그 사진이 그 시대·그 지역 것인지는 **아무도 안 봤다**. 확인 전에
        #:  붙이면 사람이 못 본 사진이 그림의 근거가 된다. ★2026-09-03 뒤집힘(HITL 0) — 자동
        #:  선택은 그대로 붙는다. 사람 판정은 도구의 override 일 뿐이다.
        #:  ★막지는 않는다 — 참조 **없이** 내려간다(HITL 0).
        unverified = []
        #: ★★사람이 **보고 거절한 것**은 「아무도 안 본 것」과 다른 칸이다
        #:  (Codex NON-BLOCK 2026-09-02 · 내가 확인). 한 칸에 넣으면 읽는
        #:  쪽이 「아직 안 봤다」로 읽어 다시 보게 한다. 둘 다 안 붙고 안 막는다.
        rejected = []
        for r in rows:
            owner = r["owner_type"]
            gate = enforcement_gate_of(owner)
            if gate is None:
                raise AppError(
                    code="episode_reference_policy.unknown_owner",
                    message=(f"갈래 {owner!r} 의 집행자를 모른다 — "
                             f"{sorted(REFERENCE_ENFORCEMENT_BY_OWNER)}"),
                    status_code=409)
            if r["outcome"] is None:
                # ★애초에 살 것이 아니었다 — 「참조 없음」 통계에 안 섞는다
                not_applicable.append(_audit_row(r, gate))
                continue
            if r["outcome"] != STATUS_SELECTED:
                # ★못 구했다 — 막지도 강제하지도 않고 **까닭만 남긴다**
                unavailable.append(_audit_row(r, gate))
                continue
            if r.get("fidelity") == FIDELITY_REJECTED:
                # ★사람이 보고 **아니라고 했다** — 안 붙이고 안 막는다
                rejected.append(_audit_row(r, gate))
                continue
            if r.get("fidelity") != FIDELITY_VERIFIED:
                # ★골랐지만 **고증은 아무도 안 봤다** — 감사에 적되 **붙인다**
                #  (HITL 0 · 사용자 2026-09-03). 사람 판정은 진행 조건이 아니다.
                unverified.append(_audit_row(r, gate))
            if gate == ENFORCE_BY_POLICY:
                forced.add(_short_id_base(str(r["final_id"])))
            else:
                elsewhere.append(_audit_row(r, gate))
        audit = {
            "projection_contract": ca.ACQUISITION_PROJECTION_VERSION,
            "enforced_elsewhere": elsewhere,
            "reference_unavailable": unavailable,
            # ★고른 것 중 **고증 미확인** — 붙이지 않은 까닭이 여기 남는다
            "fidelity_unverified": unverified,
            # ★고른 것 중 사람이 **보고 거절한 것** — 미확인과 다른 칸
            "fidelity_rejected": rejected,
            "not_applicable": not_applicable,
        }
        return forced, set(), audit

    def _research_forced_short_ids(self):
        """정본 revision → **참조를 강제할 short_id** 와 **막을 것**.

        ```
        completed · delta=yes            → 강제한다(고증이 걸렸다)
        completed · delta=no             → 안 한다(차이가 없다고 출처가 말했다)
        unresolved_terminal · retryable  → **막는다**
        ```

        ★`unresolved`/`time_capped` 를 빈 집합으로 바꾸지 않는다 — 그러면
        「조사할 게 없었다」가 되어 고증이 걸린 대상이 참조 없이 지나간다.
        ★override 로 진행할 때만 **완료된 yes 만** 쓰고, 그 사실을 남긴다.
        """
        from app.models.project import GroundingResearchRevision
        from app.modules.pipeline.grounding_claims import (DELTA_YES,
                                                           STATUS_COMPLETED)

        if self.db is None:
            return set(), set()
        # ★★★**이번 판이 기대하는 신원**만 본다 (Codex). 에피소드의 모든 과거
        #  revision 중 최신을 쓰면, 입력이 바뀌었거나 이번 판이
        #  `no_source`·`unbuilt` 인데도 **옛 결과가 섞인다**.
        cp = (self._load_prev_checkpoint("grounding_research") or {}
              ).get("data") or {}
        expected = dict(cp.get("expected_input_hashes") or {})
        # ★이번 판이 조사하려던 대상 중 **결과가 없는 것**은 막을 대상이다 —
        #  인용이 없어 못 산 것도 여기 든다.
        stuck = set(cp.get("no_source") or ()) | set(cp.get("unbuilt") or ())
        if not expected and not stuck:
            return set(), set()
        rows = (self.db.query(GroundingResearchRevision)
                .filter(GroundingResearchRevision.project_id == self.project_id,
                        GroundingResearchRevision.episode_id == self.episode_id)
                .all())
        rows = [r for r in rows
                if str(r.research_input_hash) == expected.get(
                    r.research_subject_id)]
        # ★★여기서 일찍 돌아가면 **기대한 대상의 결과가 없는데 통과**한다 —
        #  그게 「조사할 게 없었다」로 읽히는 자리다. 아래에서 막는다.
        # ★같은 대상의 여러 시도 중 **가장 나중 것**을 본다 — 재개하면 새 행이
        #  붙으므로 첫 행만 보면 옛 실패에 영원히 묶인다.
        latest = {}
        for r in sorted(rows, key=lambda x: str(x.created_at)):
            latest[r.research_subject_id] = r
        short_by_subject = self._short_id_by_subject()
        # ★이번 판이 조사하려 했는데 **행이 아예 없는** 대상도 막는다 —
        #  「결과가 없다」를 「조사할 게 없었다」로 읽지 않는다.
        forced, blocked = set(), set(stuck) | (set(expected) - set(latest))
        for sid, r in latest.items():
            if str(r.status) != STATUS_COMPLETED:
                blocked.add(sid)
                continue
            if str(r.delta) == DELTA_YES:
                short = short_by_subject.get(sid) or ""
                if not short:
                    continue
                # ★★★§4b 가 지원하는 owner 는 **계약이 정한다**
                #  (`REFERENCE_SUPPORTED_OWNERS`). `location_part`·outlook
                #  facet 은 §2-6.5 producer 몫인데, 조사가 「고증이 걸렸다」고
                #  확정한 것을 **참조 없이 통과**시키면 그게 「모른다를 아니다로
                #  닫는」 자리다 (Codex). ★warning 은 defer 가 아니다.
                #  ★접두를 글자로 비교하지 않는다 — `LP01` 이 `"L"` 로도
                #   시작하고 `Pfoo` 도 `"P"` 로 시작한다.
                from app.modules.pipeline.grounding_entity_contract import (
                    reference_owner_of)

                if reference_owner_of(short) is None:
                    blocked.add(sid)
                    continue
                forced.add(short)
        if blocked and self.project_config.get("allow_missing_grounding"):
            # ★알고도 진행한다 — 그래도 **완료된 yes 만** 쓴다.
            logger.warning("episode_reference_policy: 조사 미완 %d개를 "
                           "override 로 통과시킨다", len(blocked))
            blocked = set()
        return forced, blocked

    def _short_id_by_subject(self):
        """subject id → short_id. ★`grounding_plan` 이 남긴 것을 쓴다."""
        cp = self._load_prev_checkpoint("grounding_plan") or {}
        out = {}
        for d in ((cp.get("data") or {}).get("decided") or []):
            sid = d.get("research_subject_id")
            short = d.get("_short_id")
            if sid and short:
                out[sid] = short
        return out

    def _execute(self, mode="resume") -> Dict[str, Any]:
        from app.core.errors import AppError

        sdir_cp = self._load_prev_checkpoint("shot_director")
        if not sdir_cp or not sdir_cp.get("data", {}).get("scenes"):
            raise AppError(
                code="step.no_input",
                message="shot_director 결과 없음",
                status_code=400,
            )
        shot_director_data = sdir_cp["data"]

        # shot_selection fail-fast — selected_map 불완전 시 전체 shot
        # fallback 금지 (Codex v2 BLOCKING1).
        sel_cp = self._load_prev_checkpoint("shot_selection")
        selected_map = build_selected_map_or_raise(sel_cp)

        visible_shot_count = compute_visible_shot_count_from_checkpoints(
            shot_director_data, selected_map,
        )
        entity_types, variant_pole_short_ids = self._load_entity_signals()

        # ★★★GROUNDING-V2 §2-4b — **조사 결과가 이 판단에 들어온다**.
        #  raw plan 의 `route="research"` 는 「살 대상」일 뿐 최종 강제값이
        #  아니다 (Codex). 정본 revision 에서 **completed + delta=yes** 만 올린다.
        # ★★★`v2_chunk` 는 **중앙 조사 CP** 가 정본이다. 옛 갈래
        #  (`grounding_research` CP + revision + `grounding_plan._short_id`)는
        #  그 판에서 **안 돌아** 빈손이 되고, 그러면 조사가 고른 참조가
        #  하나도 의무가 안 된다 — 실측으로 강제 0개였다 (Codex 09-01).
        if self._uses_central():
            forced, blocked, audit = self._central_forced_short_ids()
        else:
            forced, blocked = self._research_forced_short_ids()
            audit = {}
        if blocked:
            from app.core.errors import AppError

            # ★`unresolved`/`time_capped`/`retryable` 을 **빈 집합으로 바꾸지
            #  않는다.** 그러면 「조사할 게 없었다」가 되어 고증이 걸린 대상이
            #  참조 없이 지나간다. 여기서 막고, 운영자가 알고도 진행하려면
            #  project_config['allow_missing_grounding']=True.
            raise AppError(
                code="episode_reference_policy.grounding_incomplete",
                message=(f"조사가 안 끝났거나 §4b 밖 owner 인 대상이 "
                         f"{len(blocked)}개다 — {sorted(blocked)[:5]}. "
                         f"참조 정책을 정할 수 없다"),
                status_code=409)
        manifest = compute_episode_reference_policy(
            visible_shot_count=visible_shot_count,
            entity_types=entity_types,
            variant_pole_short_ids=variant_pole_short_ids,
            research_required_short_ids=forced,
        )
        manifest["grounding"] = {
            "research_required_short_ids": sorted(forced),
            "override_used": bool(self.project_config.get(
                "allow_missing_grounding")),
        }
        # ★★갈래마다 집행자가 하나다 — 정책이 안 맡은 것을 **갈라 적는다**.
        #  세지도 막지도 않지만 「무엇이 어디로 갔나」가 기록에 남아야
        #  나중에 사람이 고칠 수 있다 (Codex 2026-09-01).
        #  ★legacy 판은 이 칸이 **아예 안 생긴다** — byte 단위 비회귀.
        if audit:
            manifest["grounding"].update(audit)

        policy = manifest["policy"]
        text_only_n = sum(1 for p in policy.values() if p["mode"] == "text_only")
        logger.info(
            "episode_reference_policy: %d entities — %d text_only, %d other",
            len(policy), text_only_n, len(policy) - text_only_n,
        )
        return {
            # ★저장과 비교가 **같은 함수**를 본다 — 없으면 runner 가 project_config 지문으로 저장하고 step-local 지문으로 비교해
            #  매 재개마다 어긋나 force → 하류(scene_detail)가 stale 로 다시 구워진다(실측 f7cc45c576c0 plain resume ×3, 각 15~19콜).
            "config_hash": self._config_hash(),
            "completed_count": len(policy),
            "applicable_count": len(policy),
            "failed_count": 0,
            "data": manifest,
            "schema_version": EPISODE_REFERENCE_POLICY_SCHEMA_VERSION,
        }

    def _config_hash(self) -> str:
        """이 스텝의 **처리 지문**. ★재개가 이것으로 옛 CP 를 가른다.

        ★★★왜 필요한가 (Codex BLOCK 2026-09-02): 이 스텝에는 `_config_hash`
        가 **없었다**. 그러면 `step_runner` 가 `compute_config_hash(
        project_config)` 로 떨어지는데, **사람 판정은 `project_config` 를
        안 바꾼다** — 그래서 사람이 `verified` 를 `rejected` 로 정정해도
        완료된 정책 CP 가 그대로 current 로 읽히고, 옛
        `research_required_short_ids` 가 되쓰인다.
        `scene_detail` 쪽은 지문에 접어 다시 도는데 그 **입력**인 이 스텝이
        안 도는, 두 소비자 중 한쪽만 깨지는 자리였다.

        ★★legacy·v2 의 지문은 **한 바이트도 안 바꾼다** — 켠 판에서만
        칸을 더한다. 안 그러면 이 판과 무관한 에피소드가 전부 다시 돈다.
        """
        import hashlib
        import json as _json

        from app.core.grounding_mode import (resolve_grounding_mode,
                                             uses_chunk_producer)
        from app.core.step_runner import compute_config_hash

        base = compute_config_hash(getattr(self, "project_config", None))
        if not uses_chunk_producer(resolve_grounding_mode(
                getattr(self, "project_config", None) or {})):
            return base                     # ★옛 판은 그대로다
        from app.modules.pipeline import grounding_central_acquisition as _ca

        return hashlib.sha256(_json.dumps({
            "base": base,
            # ★중앙 투영 계약 — 붙이는 규칙이 바뀌면 다시 판단한다
            "acquisition_projection": _ca.ACQUISITION_PROJECTION_VERSION,
            # ★HITL 0 (2026-09-03): 사람 판정 지문은 뺐다 — production 지문이 아니다
        }, sort_keys=True).encode("utf-8")).hexdigest()[:16]

    def _load_entity_signals(self) -> Tuple[Dict[str, str], set]:
        """{short_id: entity_type} + variant pole short_id 집합."""
        from app.core.entity_protection import compute_variant_pole_ids
        from app.models.project import (
            EntityCanon, EntityEpisodeLink, RelationFact, RelationParticipant,
        )

        # ★보류(shelved) 제외 — 정본 하나 (2026-09-04).
        from app.core.entity_identity import active_episode_canon_ids

        canon_ids = active_episode_canon_ids(
            self.db, self.project_id, self.episode_id)
        canons = self.db.query(EntityCanon).filter(
            EntityCanon.id.in_(canon_ids)
        ).all() if canon_ids else []
        entity_types: Dict[str, str] = {}
        short_by_canon: Dict[str, str] = {}
        for c in canons:
            if c.short_id:
                base = c.short_id.split("O")[0]
                entity_types[base] = c.entity_type
                short_by_canon[c.id] = base

        relations = self.db.query(RelationFact).filter(
            RelationFact.project_id == self.project_id
        ).all()
        rel_ids = [r.id for r in relations]
        participants = self.db.query(RelationParticipant).filter(
            RelationParticipant.relation_id.in_(rel_ids)
        ).all() if rel_ids else []
        variant_pole_uuids = compute_variant_pole_ids(
            [{"id": c.id, "entity_type": c.entity_type} for c in canons],
            [{"id": r.id, "relation_family": r.relation_family}
             for r in relations],
            [{"relation_id": p.relation_id, "canon_id": p.canon_id}
             for p in participants],
        )
        variant_pole_short_ids = {
            short_by_canon[u] for u in variant_pole_uuids
            if u in short_by_canon
        }
        return entity_types, variant_pole_short_ids