"""GROUNDING-V2 §2-3 — `shadow_plan` 재생기.

설계: 계획 §2 표의 3단계 · 계약 §12.

저장된 에피소드의 **기존 체크포인트만 읽어** 후보를 만들고 분류·계획을 세운다.
★**검색도 이미지 생성도 0건.** production 체크포인트는 **한 바이트도 안 건드린다.**

★모집단은 `entity_merge`(13.5) 다 — production 의 `grounding_plan`(13.65)이
읽는 **그 자리**다. 저빈도 필터(13.7) **앞**이라 거기서 걸러질 것도 다 본다.
예전에는 `entity_filter` 를 읽어 **걸러진 뒤만** 봤는데, 걸러진 것이 곧 고증이
지키려는 대상이라 「production 을 쟀다」가 성립하지 않았다(실측 11→7).

★A0(원문 후보 수집)는 §2-3.5 에서 배선됐다. `a0_candidates` 를 주면 A0 가 건진
**원문 문장**을 근거로 잰다 — production 과 같은 입력이다. 안 주면 저장 CP 의
LLM 묘사로 재고, 그 사실이 `quote_sources` 로 산출에 남는다.
**그 판을 「production 을 쟀다」로 쓰면 안 된다.**
"""
from __future__ import annotations

import json
import logging
from pathlib import Path
from typing import Any, Dict, List, Optional, Sequence

from app.core.grounding_mode import (
    GROUNDING_MODE_SHADOW_PLAN, buys_no_research_at_all, buys_v2_research,
    touches_production_fingerprint,
)
from app.modules.pipeline.grounding_planner import plan
from app.modules.pipeline.grounding_subject import build_subject

logger = logging.getLogger(__name__)

SHADOW_CONTRACT_VERSION = 1

#: ★shadow 산출이 앉는 자리. **production step id 와 겹치지 않는다** —
#: 겹치면 하류 의존이 이 파일을 보고 stale 판정을 한다.
SHADOW_STEP_ID = "grounding_shadow_plan"

#: 후보를 어디서 읽는가.
#: ★production 의 `grounding_plan`(13.65) 이 읽는 것과 **같은 모집단**이다.
#: `entity_filter`(13.7)를 읽으면 **거기서 걸러진 것을 아예 못 본다** —
#: 실측으로 한 에피소드에서 11개 중 4개가 사라졌고, 그것들이 고증이 지키려는
#: 바로 그 대상이다. 같은 것을 잰다고 말하려면 같은 자리에서 읽어야 한다.
_SOURCE_STEP = "entity_merge"
#: 옛 자리 — 왜 안 쓰는지 남겨 둔다(다시 바꾸려는 사람을 위해).
_POST_FILTER_STEP = "entity_filter"
_ENTITY_KEYS = ("characters", "locations", "props")
_OWNER_BY_KEY = {"characters": "character", "locations": "location", "props": "prop"}


def _judge_of(classified: Dict[str, Any]) -> Dict[str, Any]:
    """실제로 판정한 모델. ★설정이 아니라 **응답이 말한 것**이다.

    ``call_structured`` 는 Tier 3 에서 다른 provider 로 넘어갈 수 있으므로,
    「Sol 이 판정했다」를 기록으로 주장하려면 이 칸이 있어야 한다.
    """
    for rec in classified.get("records") or []:
        if rec.get("judge_physical_model"):
            return {
                "alias": rec.get("judge_model_alias"),
                "physical_model": rec.get("judge_physical_model"),
            }
    return {"alias": None, "physical_model": None}


_COMPLETED_STATUSES = ("completed", "success")


class ShadowSourceError(RuntimeError):
    """저장된 CP 를 못 읽었다. ★None 으로 삼키지 않는다.

    삼키면 「era 를 못 읽어 빈 문자열로 판정」이 조용히 일어나고, 그것이 바로
    이 단계에서 방금 고친 결함이다. 깨진 에피소드가 섞여도 전체가 통과로 보인다.
    """


def _read_cp(episode_dir: Path, step: str, *, required: bool) -> Optional[Dict[str, Any]]:
    path = episode_dir / step / "manifest.json"
    if not path.exists():
        if required:
            raise ShadowSourceError(f"{step} 체크포인트가 없다: {path}")
        return None
    try:
        payload = json.loads(path.read_text(encoding="utf-8"))
    except Exception as exc:
        # ★파손은 언제나 fail-closed 다 — required 여부와 무관하다.
        raise ShadowSourceError(f"{step} 체크포인트가 깨졌다: {path} — {exc}") from exc
    if not isinstance(payload, dict):
        raise ShadowSourceError(f"{step} 체크포인트 모양이 dict 가 아니다: {path}")
    return payload


def _require_completed(episode_dir: Path, step: str) -> Dict[str, Any]:
    """★``status`` 가 **정확히 완료**이고 ``data`` 가 dict 인 CP 만 받는다.

    빈 문자열이나 없는 status 를 완료로 봐 주면, 미완 산출로 재생해 놓고
    「후보가 적다」를 결함이 아니라 사실로 읽게 된다. 실물 60개는 전부
    ``completed`` + dict 라 이 기준으로 아무것도 잃지 않는다.
    """
    cp = _read_cp(episode_dir, step, required=True)
    status = cp.get("status")
    if status not in _COMPLETED_STATUSES:
        raise ShadowSourceError(
            f"{step} 이 완료 상태가 아니다: status={status!r} "
            f"(받는 값 {list(_COMPLETED_STATUSES)})")
    if not isinstance(cp.get("data"), dict):
        raise ShadowSourceError(
            f"{step}.data 가 dict 가 아니다 ({type(cp.get('data')).__name__})")
    return cp


def build_subjects_from_saved_episode(
    episode_dir: Path,
    *,
    project_id: str,
    episode_id: str,
    a0_candidates: Optional[Sequence[Dict[str, Any]]] = None,
) -> Dict[str, Any]:
    """저장된 CP 에서 subject 를 만든다. ★읽기만 한다.

    ★결속은 `grounding_carry.build_subjects` 가 한다 — **production 이 쓰는
    그 함수**다. 여기서 따로 만들면 도구가 프로덕션과 다른 입력을 보내게 되고,
    실제로 그랬다: production 은 A0 가 건진 **원문 문장**을 분류기에 넘기는데
    이 함수는 LLM 이 상상해 쓴 ``description`` 을 넘겼다.

    ★``a0_candidates`` 를 안 주면 예전과 같다 — 저장 CP 만 있는 legacy
    에피소드에는 A0 산출이 없다. 그때는 **모든 subject 가
    ``quote_source="entity_description"``** 이고, 그 사실이 반환값에 남는다.
    「production 을 쟀다」로 쓰지 않기 위해서다.

    Returns:
        ``subjects`` · ``unbound`` · ``carry_reasons`` · ``carried`` ·
        ★``quote_sources`` — 무엇을 근거로 쟀는지. 보고에 그대로 옮긴다.
    """
    cp = _require_completed(episode_dir, _SOURCE_STEP)
    data = cp["data"]
    if not isinstance(data, dict):
        # ★모양이 다르면 {} 로 삼키지 않는다 — 후보 0이 「깨끗함」으로 읽힌다.
        raise ShadowSourceError(
            f"{_SOURCE_STEP}.data 가 dict 가 아니다 ({type(data).__name__})")
    entities = {k: data.get(k) or [] for k in _ENTITY_KEYS}
    if not any(isinstance(v, list) and v for v in entities.values()):
        raise ShadowSourceError(
            f"{_SOURCE_STEP}.data 에 엔티티가 하나도 없다 "
            f"(칸: {sorted(data)}) — 0을 「깨끗함」으로 읽게 된다")

    from app.modules.pipeline import grounding_carry as _carry

    built = _carry.build_subjects(
        entities, project_id=project_id, episode_id=episode_id,
        source_step=_SOURCE_STEP, a0_candidates=a0_candidates)

    sources: Dict[str, int] = {}
    for s in built["subjects"]:
        k = s.get("quote_source") or "?"
        sources[k] = sources.get(k, 0) + 1
    return {**built, "quote_sources": sources}


def production_checkpoint_digest(episode_dir: Path) -> Dict[str, str]:
    """production 산출의 **파일 bytes 해시**. 전후로 비교해 불변을 증명한다.

    ★``*/manifest.json`` 만 해시하면 안 된다 — 실제 에피소드 디렉토리에는
    manifest 말고도 json·html·png 가 수백 개 있고, 그것들을 바꿔도
    ``production_unchanged=True`` 가 나온다. 「한 바이트도 안 바뀐다」는
    **모든 파일**을 봐야 할 수 있는 말이다.
    """
    import hashlib

    out: Dict[str, str] = {}
    shadow_dir = episode_dir / SHADOW_STEP_ID
    for path in sorted(episode_dir.rglob("*")):
        if not path.is_file():
            continue
        if shadow_dir in path.parents or path.parent == shadow_dir:
            continue          # shadow 는 바뀌어도 된다 — 그게 이 단계의 산출이다
        rel = str(path.relative_to(episode_dir))
        out[rel] = hashlib.sha256(path.read_bytes()).hexdigest()[:16]
    return out


def write_shadow_checkpoint(episode_dir: Path, payload: Dict[str, Any]) -> Path:
    """★shadow 산출을 **shadow 자리에만** 남긴다.

    production step id 와 안 겹치므로 하류 의존이 이 파일을 보고 stale 판정을
    하지 않는다 (계약 §12).

    ★**앞 판을 조용히 덮지 않는다.** 계약 §8 이 이미 지목한 자리다 —
    「검증 결과가 그 칸을 덮어 감사 기록이 사라졌다」. 두 판을 대조하는 것이
    이 단계의 유일한 무료 측정 수단이라, 덮으면 **비교할 앞 판이 없어진다.**
    기존 판은 저장소 관례대로 ``manifest_<YYYYMMDD_HHMMSS>.json`` 으로 옮긴다.
    """
    from datetime import datetime, timezone

    from app.core.checkpoint_io import atomic_write_json

    out_dir = episode_dir / SHADOW_STEP_ID
    out_dir.mkdir(parents=True, exist_ok=True)
    path = out_dir / "manifest.json"
    if path.exists():
        stamp = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S")
        archive = out_dir / f"manifest_{stamp}.json"
        n = 1
        while archive.exists():          # 같은 초에 두 번 돌아도 안 잃는다
            archive = out_dir / f"manifest_{stamp}_{n}.json"
            n += 1
        path.replace(archive)
    atomic_write_json(path, payload)
    return path


def replay_shadow_plan(
    episode_dir: Path,
    *,
    project_id: str,
    episode_id: str,
    mode: str = GROUNDING_MODE_SHADOW_PLAN,
    classify_fn=None,
    write_checkpoint: bool = False,
) -> Dict[str, Any]:
    """저장된 에피소드 하나를 shadow 로 재생한다.

    Args:
        mode: ★``shadow_plan`` **만** 받는다. ``legacy`` 는 「이 기구를 안 쓴다」는
            뜻이지 「이 기구를 legacy 로 돌린다」가 아니다 — 받아 주면 무엇을
            잰 판인지 기록에서 갈리지 않는다.
        classify_fn: 분류기. ``None`` 이면 **provider 를 한 번도 안 부르고**
            전부 「분류 미제출」로 둔다. 부를 때는
            ``fn(subjects, era=..., region=...)`` 꼴이어야 한다.
        write_checkpoint: True 면 **shadow 자리에만** 산출을 남긴다.

    Raises:
        ShadowSourceError: 저장 CP 가 없거나 깨졌거나, 분류를 돌리는데 시대가 없을 때.
    """
    if mode != GROUNDING_MODE_SHADOW_PLAN:
        raise ValueError(
            f"replay_shadow_plan 은 mode={GROUNDING_MODE_SHADOW_PLAN!r} 전용이다 "
            f"(받은 값 {mode!r})")
    assert buys_no_research_at_all(mode)
    assert not buys_v2_research(mode)
    assert not touches_production_fingerprint(mode)

    before = production_checkpoint_digest(episode_dir)
    built = build_subjects_from_saved_episode(
        episode_dir, project_id=project_id, episode_id=episode_id)
    subjects = built["subjects"]
    unbound = built["unbound"]

    era = region = ""
    logical_calls = 0
    if classify_fn is None:
        records: List[Dict[str, Any]] = [
            {"research_subject_id": s["research_subject_id"],
             "classifier_missing": True}
            for s in subjects
        ]
        classified = {"records": records, "fingerprint": None, "search_calls": 0}
    else:
        rules = _require_completed(episode_dir, "visual_world_rules")
        rules_data = rules["data"]
        era = str(rules_data.get("era") or "").strip()
        region = str(rules_data.get("region") or "").strip()
        # ★시대가 없으면 **판정하지 않는다.** 빈 값으로 넘기면 분류기가 상상 묘사만
        #   보고 답하고, 그것이 방금 고친 결함 그 자체다.
        if not era or not region:
            raise ShadowSourceError(
                f"visual_world_rules 에 시대/지역이 없다 (era={era!r}, region={region!r}) — "
                "빈 값으로 판정하면 조사 전 상상 묘사만 보고 답하게 된다")
        classified = classify_fn(subjects, era=era, region=region)
        logical_calls = 1 if subjects else 0

    planned = plan(classified.get("records") or [])
    # ★판정을 사람이 읽으려면 **무엇에 대한 판정인지**가 붙어 있어야 한다.
    #   subject 는 id 로만 오가므로 여기서 되붙인다 (route 계산에는 안 쓴다).
    by_id = {s["research_subject_id"]: s for s in subjects}
    for d in planned["decided"]:
        subj = by_id.get(d.get("research_subject_id")) or {}
        d["_surface_form"] = subj.get("surface_form")
        d["_owner_type"] = subj.get("owner_type")
        d["_short_id"] = (subj.get("provenance") or {}).get("short_id")

    after = production_checkpoint_digest(episode_dir)
    out = {
        "contract_version": SHADOW_CONTRACT_VERSION,
        "mode": mode,
        "project_id": project_id,
        "episode_id": episode_id,
        "era": era,
        "region": region,
        "subject_count": len(subjects),
        "counts": planned["counts"],
        "search_calls": int(classified.get("search_calls") or 0) + planned["search_calls"],
        # ★검색은 0이어도 **분류기 호출은 돈이 든다.** 따로 센다 —
        #   「검색 0」만 보고하면 provider 비용이 0으로 읽힌다.
        # ★이름을 **논리 호출**로 못박는다. Router 재시도·fallback 때문에
        #   **실제 전송 수는 이보다 많을 수 있다** — 이 값을 「전송 수」로 읽으면
        #   비용을 과소 보고하게 된다. 물리 모델은 아래 judge 칸이 말한다.
        "classifier_logical_calls": logical_calls,
        "classifier_judge": _judge_of(classified),
        "classifier_fingerprint": classified.get("fingerprint"),
        "production_unchanged": before == after,
        "production_file_count": len(before),
        "decided": planned["decided"] + unbound,
        # ★**무엇을 근거로 쟀는지**를 산출에 남긴다. 저장 CP 만 있는 legacy
        #  에피소드는 A0 산출이 없어 전부 `entity_description` 이다 —
        #  그 판을 「production 을 쟀다」로 쓰면 안 된다.
        "quote_sources": built["quote_sources"],
        "carry_reasons": built["carry_reasons"],
        "a0_carried": built["carried"],
        "unbound_count": len(unbound),
    }
    if write_checkpoint:
        out["shadow_checkpoint_path"] = str(write_shadow_checkpoint(episode_dir, out))
        # ★shadow 를 쓴 뒤에도 production 은 그대로여야 한다.
        out["production_unchanged"] = before == production_checkpoint_digest(episode_dir)
    return out