"""유료 주행 **뒤**의 보고. ★산 것을 판정보다 먼저 저장한 다음 읽는다.

Codex 가 요구한 넷 (2026-08-31) —

> 결과 보고에는 **bootstrap/pipeline 실제 counted** · **provider raw 시도** ·
> **Opik↔장부 결속** · **원본 DB/projects sentinel 전후**를 같이 올려 주세요.

    python tools/grounding_audit/canary_report.py <run_id>
"""
from __future__ import annotations

import hashlib
import json
import subprocess
import sys
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any, Dict, List, Optional

#: 닫힌 판. ★열린 판은 얼마 썼는지 모르므로 앞 누계에 안 넣는다.
TERMINAL = ("completed", "stopped")


def _step_name(x: Any) -> str:
    """`plan.skipped` 한 칸의 스텝 이름. ★두 모양을 다 받는다."""
    return str(x["step"] if isinstance(x, dict) else x)

BACKEND = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(BACKEND))

from tools.grounding_audit import canary_isolation as ci  # noqa: E402

KST = timezone(timedelta(hours=9))


def sentinel_now() -> Dict[str, Any]:
    """원본 자리의 지문. ★주행 **전후**를 견주려고 같은 셈을 쓴다."""
    prod = ci.REPO / "projects"
    files = sorted(p for p in prod.rglob("*") if p.is_file())
    fp = hashlib.sha256("\n".join(
        f"{p.relative_to(prod)}:{p.stat().st_size}:{int(p.stat().st_mtime)}"
        for p in files).encode()).hexdigest()
    q = ("SELECT 'project_registry',count(*) FROM project_registry "
         "UNION ALL SELECT 'episode',count(*) FROM episode "
         "UNION ALL SELECT 'scene_still',count(*) FROM scene_still "
         "UNION ALL SELECT 'entity_canon',count(*) FROM entity_canon "
         "UNION ALL SELECT 'user_account',count(*) FROM user_account;")
    from sqlalchemy.engine import make_url

    u = make_url(ci.template_url())
    out = subprocess.run(
        ["psql", "-h", str(u.host), "-p", str(u.port or 5432),
         "-U", str(u.username), "-d", "theroad", "-t", "-A", "-F", "|",
         "-c", q],
        env={"PGPASSWORD": str(u.password or ""),
             "PATH": "/usr/bin:/bin:/usr/local/bin:/opt/homebrew/bin"},
        capture_output=True, text=True)
    rows = dict(l.split("|") for l in out.stdout.strip().splitlines()
                if "|" in l)
    return {"prod_files": len(files), "fingerprint": fp, "db_rows": rows}


def _opik_get(path: str, **kw) -> Dict[str, Any]:
    import urllib.parse
    import urllib.request

    sys.path.insert(0, str(ci.REPO / "scratchpad"))
    from _opik_env import opik_target                      # noqa: E402

    base, ws, proj = opik_target()
    q = urllib.parse.urlencode({"project_name": proj, **kw})
    req = urllib.request.Request(f"{base}/v1/private/{path}?{q}",
                                 headers={"Comet-Workspace": ws})
    with urllib.request.urlopen(req, timeout=30) as r:
        return json.loads(r.read().decode("utf-8"))


def opik_traces(run_id: str, *, project_id: str = "", episode_id: str = "",
                attempts: Optional[List[Dict[str, Any]]] = None
                ) -> Dict[str, Any]:
    """이 판의 실제 호출. ★**로그 SOT** 는 Opik 이다 — 장부와 견준다.

    ★★★두 번 틀렸다 (2026-09-01) —

        ①canary 태그로만 찾아 **1건**을 냈다. 유료 호출 trace 는 그 태그를
          안 달고 `op:<step>` 을 단다. 결속 고리는 canary 의 **project_id** 다.
        ②top-level trace 만 셌다. 스텝 대부분은 LLM 호출을 **span** 으로 넣고
          일부만 따로 `chat.completion` trace 로 뜬다. trace 만 세면 8, span 을
          세면 **26** 이었다.

    ★★★그리고 **Opik 은 성공한 호출만 남긴다.** litellm 의 `OpikLogger` 에는
    `log_success_event` 뿐이고 실패 훅은 부모의 `pass` 다. 즉 —

        **「Opik 에 없다」는 「안 샀다」가 아니다.**

    돈이 나간 수의 정본은 **문에서 센 장부**이고, Opik 은 그 **부분집합**이다.
    """
    try:
        rows: List[Dict[str, Any]] = []
        for page in (1, 2, 3):
            got = (_opik_get("traces", size=500, page=page).get("content")
                   or [])
            if not got:
                break
            rows += got
    except Exception as exc:                    # noqa: BLE001
        return {"ok": False, "why": f"{type(exc).__name__}: {exc}",
                "★means": "못 읽은 것이지 「없다」가 아니다"}

    keys = [k for k in (project_id, episode_id, run_id) if k]
    if not keys:
        return {"ok": False, "why": "결속 고리(project_id)를 못 얻었다",
                "★means": "못 읽은 것이지 「없다」가 아니다"}
    mine = [t for t in rows
            if any(k in json.dumps(t, default=str) for k in keys)]

    def _at(t: Dict[str, Any]) -> Optional[datetime]:
        """Opik 의 `start_time` 은 `…T02:21:56.123456Z` 꼴이다.

        ★그냥 `+00:00` 을 붙이면 `Z` 와 겹쳐 **못 읽고**, 그러면 창 안에 아무
        것도 안 들어와 **0** 이 나온다 — 결함이 아니라 재는 오류다.
        """
        raw = str(t.get("start_time") or "")[:19]
        return _moment(raw + "+00:00") if len(raw) == 19 else None

    per_attempt = []
    for a in attempts or []:
        st, fi = _moment(a.get("started_kst") or ""), _moment(
            a.get("finished_kst") or "")
        if st is None or fi is None:
            per_attempt.append({"attempt_id": a.get("attempt_id"),
                                "ledger": a.get("used"), "opik_llm": None,
                                "★why": "장부에 시각이 없어 못 나눈다"})
            continue
        win = [t for t in mine
               if (w := _at(t)) is not None and st <= w <= fi]
        n = 0
        for t in win:
            sp = (_opik_get("spans", size=500,
                            trace_id=t["id"]).get("content") or [])
            n += sum(1 for x in sp if str(x.get("type") or "").lower() == "llm")
        got = {"attempt_id": a.get("attempt_id"), "ledger": a.get("used"),
               "opik_llm": n, "traces": len(win),
               "★agrees": (int(a.get("used") or 0) == n)}
        if not got["★agrees"]:
            got["★gap"] = int(a.get("used") or 0) - n
            got["★likely"] = ("Opik 은 **성공만** 남긴다 — 차이는 실패한 전송일 "
                              "수 있다. 창이 비어 0 이면 **재는 오류**다")
        per_attempt.append(got)
    return {
        "ok": True, "bound_by": keys, "★scanned": len(rows),
        "matched": len(mine), "per_attempt": per_attempt,
        "★blind_spot": ("litellm 의 `OpikLogger` 는 `log_success_event` 만 "
                        "가진다 — 부모의 실패 훅은 `pass` 다. **실패한 유료 "
                        "호출은 Opik 에 안 남는다.** 그래서 Opik 수가 장부보다 "
                        "적을 수 있고, 그 차이는 **실패한 전송**이다"),
        "★sot": "돈이 나간 수의 정본은 **문에서 센 장부**다 — Opik 은 부분집합",
    }


def journal_rows(root: Path) -> List[Dict[str, Any]]:
    out = []
    for p in sorted(root.rglob("*journal*.json")):
        try:
            d = json.loads(p.read_text(encoding="utf-8"))
        except Exception:                       # noqa: BLE001
            continue
        for e in d.get("calls") or []:
            out.append({"file": p.name, "identity": e.get("identity"),
                        "status": e.get("status"),
                        "trace_id": e.get("trace_id"),
                        "run_id": e.get("run_id")})
    return out


#: ★앞 판에서 **이미 끝나 있던** 스텝. 이번 판은 이것들을 **다시 사면 안 된다**.
#:  ★손으로 적지 않는다 — canary DB 의 `step_run` 에서 읽는다.
def completed_before(run_id: str, *, attempt_started: str) -> List[str]:
    """이번 attempt 가 **시작하기 전에** 이미 완료였던 스텝.

    ★★★**시각을 글자로 견주지 않는다** (Codex 실측 2026-09-01). DB 는 UTC
    (`2026-09-01T02:26+00`)로, 장부는 KST(`2026-09-01T11:21+09`)로 적힌다.
    글자로 견주면 **뒤에 끝난 것이 앞선 것으로** 읽혀, 이번에 산 스텝이
    「재구매」로 잘못 세어진다. 같은 순간으로 바꿔서 견준다.
    """
    started = _moment(attempt_started)
    if started is None:
        raise RuntimeError(f"attempt 시작 시각을 못 읽었다: {attempt_started!r}")
    rows = _psql(run_id,
                 "SELECT step_id, status, updated_at FROM step_run "
                 "ORDER BY step_id;")
    got = []
    for step, status, when in rows:
        if status != "completed":
            continue
        at = _moment(when)
        if at is None:
            raise RuntimeError(f"{step} 의 완료 시각을 못 읽었다: {when!r}")
        if at < started:
            got.append(step)
    return sorted(got)


def _moment(text: str) -> Optional[datetime]:
    """글을 **시각**으로 바꾼다. ★시간대가 없으면 `None` — 짐작하지 않는다."""
    raw = str(text or "").strip().replace(" ", "T")
    if not raw:
        return None
    # ★`+00` 처럼 분이 없는 꼬리를 파이썬이 못 읽는다
    for tail in ("+00", "-00", "+09"):
        if raw.endswith(tail):
            raw += ":00"
            break
    try:
        got = datetime.fromisoformat(raw)
    except ValueError:
        return None
    return got if got.tzinfo is not None else None


def step_run_states(run_id: str) -> Dict[str, str]:
    return {r[0]: r[1] for r in _psql(run_id,
                                      "SELECT step_id, status FROM step_run;")}


def _psql(run_id: str, sql: str) -> List[List[str]]:
    """canary DB 를 읽는다. ★못 읽으면 **빈손이 아니라 예외**다."""
    from sqlalchemy.engine import make_url

    u = make_url(ci.template_url())
    got = subprocess.run(
        ["psql", "-h", str(u.host), "-p", str(u.port or 5432),
         "-U", str(u.username), "-d", ci.db_name(run_id),
         "-t", "-A", "-F", "|", "-c", sql],
        env={"PGPASSWORD": str(u.password or ""),
             "PATH": "/usr/bin:/bin:/usr/local/bin:/opt/homebrew/bin"},
        capture_output=True, text=True)
    if got.returncode != 0:
        raise RuntimeError(f"canary DB 를 못 읽었다: {got.stderr.strip()[:200]}")
    return [l.split("|") for l in got.stdout.strip().splitlines() if "|" in l]


def acceptance(run_id: str, *, report: Dict[str, Any]) -> Dict[str, Any]:
    """Codex 가 못박은 다섯 (2026-09-01). ★세 갈래로 센다.

    ★★**미확정을 합격으로 바꾸지 않는다** — 어긋남·미확정·맞음을 따로 세고,
    앞의 둘이 **모두 0일 때만** 통과다.
    """
    root = ci.root_dir(run_id)
    led = json.loads((root / "pipeline_attempts.json").read_text(
        encoding="utf-8"))
    this = led[-1]
    per = this.get("per_step") or {}
    axes: List[Dict[str, Any]] = []
    # ★★시작값을 **손으로 안 적는다** (Codex 2026-09-01). 15/77 을 박아 두면
    #  다음 판(42/50)은 성공해도 어긋남이 된다. 장부에서 셈한다.
    prior = [a for a in led[:-1] if a.get("status") in TERMINAL]
    want_before = sum(int(a.get("used") or 0) for a in prior)
    ceiling = int(this.get("ceiling") or 0)
    want_cap = ceiling - want_before

    def ax(name: str, ok: Optional[bool], said: str) -> None:
        axes.append({"axis": name, "verdict": ("맞음" if ok is True else
                                               "어긋남" if ok is False else
                                               "미확정"), "잰 것": said})

    # ①시작값 — **장부가 정본**이다
    ax(f"시작값 cumulative {want_before} · scope_cap {want_cap}",
       (this.get("cumulative_before") == want_before
        and this.get("scope_cap") == want_cap),
       f"장부의 앞 terminal 합 {want_before} (정지선 {ceiling}) vs "
       f"이 판이 적은 cumulative_before={this.get('cumulative_before')} · "
       f"scope_cap={this.get('scope_cap')}")

    # ②앞서 끝난 것을 다시 사지 않았다
    try:
        done = completed_before(run_id,
                                attempt_started=this.get("started_kst") or "")
        bought = {s: per[s]["counted"] for s in done
                  if s in per and per[s].get("counted")}
        ax(f"앞서 끝난 {len(done)}개 재구매 0", not bought,
           f"다시 산 것: {bought or '없다'}")
    except Exception as exc:                    # noqa: BLE001
        ax("앞서 끝난 것 재구매 0", None, f"못 읽었다: {exc}")

    # ③적용 안 되는 넷이 **돌아서** not_applicable
    # ★★`canary_run.json` 의 `plan.skipped` 는 **문자열 목록**이다.
    #  앞 판은 dict 로 읽어 실제 산출에서 바로 터졌다 — 내 fixture 가 실제와
    #  **반대 모양**을 지어 결함을 초록으로 잠그고 있었다 (Codex 2026-09-01).
    skipped = [_step_name(x) for x in (json.loads(
        (root / "canary_run.json").read_text(encoding="utf-8")
    ).get("plan") or {}).get("skipped") or []]
    try:
        st = step_run_states(run_id)
        bad = {s: st.get(s, "기록 없음") for s in skipped
               if st.get(s) != "not_applicable"}
        spent = {s: per[s]["counted"] for s in skipped
                 if s in per and per[s].get("counted")}
        ax(f"적용 제외 {len(skipped)}개 = not_applicable · counted 0",
           (not bad) and (not spent),
           f"어긋난 것: {bad or '없다'} · 쓴 것: {spent or '없다'}")
    except Exception as exc:                    # noqa: BLE001
        ax("적용 제외 넷", None, f"못 읽었다: {exc}")

    # ④장부 — 열린 판 0 · 같은 id 가 terminal
    opened = [a["attempt_id"] for a in led if a.get("status") not in TERMINAL]
    ax("열린 판 0 · 이 판이 terminal",
       (not opened) and this.get("status") in TERMINAL,
       f"열린 것 {opened or '없다'} · 이 판 status={this.get('status')} "
       f"(★「provider 앞에서 생긴다」는 급사 시험이 잰다 — 여기서는 못 잰다)")

    # ⑤돈 — 갈라 적고 정지선 아래
    total = sum(int(a.get("used") or 0) for a in led)
    img = sum(int(a.get("image_used") or 0) for a in led)
    ax(f"이 run 누계가 정지선 {ceiling} 이하 · 이미지 0",
       bool(ceiling) and total <= ceiling and img == 0,
       f"누계 {total} (앞 {want_before} + 이번 {this.get('used')}) · "
       f"이미지 {img}")

    # ⑥원본 무변
    sen = report["★sentinel"]
    ax("원본 sentinel 무변",
       sen["files_unchanged"] and sen["fingerprint_unchanged"]
       and sen["db_rows_unchanged"],
       f"파일 {sen['files_unchanged']} · 지문 {sen['fingerprint_unchanged']} "
       f"· 행 {sen['db_rows_unchanged']}")

    tally = {k: sum(1 for a in axes if a["verdict"] == k)
             for k in ("맞음", "어긋남", "미확정")}
    return {
        "axes": axes, "tally": tally,
        # ★수를 **손으로 안 적는다** — 장부에서 판마다 그대로 낸다
        "★cost_split": {
            "판별": [{"attempt_id": a.get("attempt_id"),
                      "source": a.get("source"), "used": a.get("used"),
                      "image_used": a.get("image_used")} for a in led],
            "이번 신규": this.get("used"),
            "이 run 누계": total,
            "정지선": ceiling,
            "이미지": img,
            "이미지 승인": this.get("approved_image_calls"),
            "★옛 1씬 판": ("이 장부 밖이다 — 다른 run 이라 여기서 안 센다"),
        },
        "passed": tally["어긋남"] == 0 and tally["미확정"] == 0,
        "★means": ("**미확정은 합격이 아니다.** 어긋남과 미확정이 **둘 다 0** "
                   "일 때만 통과다"),
    }


def build(run_id: str, *, before: Dict[str, Any]) -> Dict[str, Any]:
    root = ci.root_dir(run_id)
    run = json.loads((root / "canary_run.json").read_text(encoding="utf-8"))
    rows = journal_rows(root)
    after = sentinel_now()
    boot = run.get("stages", {}).get("bootstrap") or {}
    _ = boot
    pipe = run.get("stages", {}).get("pipeline") or {}
    per = pipe.get("per_step") or {}
    return {
        "written_at_kst": datetime.now(KST).isoformat(timespec="seconds"),
        "run_id": run_id,
        "★counted": {
            "bootstrap": (boot.get("bootstrap_budget") or {}).get("used"),
            "bootstrap_cap": boot.get("cap"),
            "pipeline_total": sum(int(v.get("counted") or 0)
                                  for v in per.values()),
            "pipeline_per_step": per,
            "pipeline_budget": pipe.get("budget"),
        },
        "★router_lock": {
            k: run.get("stages", {}).get(k)
            for k in ("request_lock", "router_ready",
                      "router_lock_after_bootstrap", "router_lock")},
        "★opik_vs_journal": {
            "opik": opik_traces(
                run_id, project_id=str(boot.get("project_id") or ""),
                episode_id=str(boot.get("episode_id") or ""),
                attempts=json.loads(
                    (root / "pipeline_attempts.json").read_text(
                        encoding="utf-8"))),
            "journal_rows": rows,
            "journal_trace_ids": sorted(
                {r["trace_id"] for r in rows if r.get("trace_id")}),
        },
        "★sentinel": {
            "before": before, "after": after,
            "files_unchanged": before.get("prod_files") == after["prod_files"],
            "fingerprint_unchanged":
                before.get("fingerprint") == after["fingerprint"],
            "db_rows_unchanged": before.get("db_rows") == after["db_rows"],
            "★means": ("원본이 **안 바뀌었다**는 뜻이지 「쓰기 0 증명」이 "
                       "아니다 — sentinel 무변이다"),
        },
        "stopped_at": pipe.get("stopped_at"),
    }


def main() -> int:
    if len(sys.argv) < 2:
        print(__doc__)
        return 2
    rid = sys.argv[1]
    bf = ci.REPO / "artifact" / "20260831_bundle_canary_preflight" \
        / "sentinel_before.json"
    before = json.loads(bf.read_text(encoding="utf-8")) if bf.is_file() else {}
    got = build(rid, before=before)
    got["★acceptance"] = acceptance(rid, report=got)
    out = ci.root_dir(rid) / "canary_report.json"
    out.write_text(json.dumps(got, ensure_ascii=False, indent=1, default=str),
                   encoding="utf-8")
    c = got["★counted"]
    print(f"■ canary {rid}")
    print(f"  counted — 부트스트랩 {c['bootstrap']}/{c['bootstrap_cap']} · "
          f"pipeline {c['pipeline_total']}")
    o = got["★opik_vs_journal"]["opik"]
    if o.get("ok"):
        print("  Opik ↔ 장부 (판별)")
        for r in o["per_attempt"]:
            if r.get("opik_llm") is None:
                mark = f"  — {r.get('★why', '못 나눴다')}"
            elif r.get("★agrees"):
                mark = "  ✓"
            else:
                mark = (f"  ★차이 {r['★gap']} — {r['★likely']}")
            print(f"    {r['attempt_id']:16} 장부 {r['ledger']} · "
                  f"Opik llm {r['opik_llm']}{mark}")
    else:
        print(f"  Opik — {o}")
    s = got["★sentinel"]
    print(f"  원본 무변 — 파일 {s['files_unchanged']} · 지문 "
          f"{s['fingerprint_unchanged']} · 행 {s['db_rows_unchanged']}")
    a = got["★acceptance"]
    print("  ── acceptance")
    for x in a["axes"]:
        print(f"    {x['verdict']:4} {x['axis']} — {x['잰 것']}")
    print(f"  맞음 {a['tally']['맞음']} · 어긋남 {a['tally']['어긋남']} · "
          f"미확정 {a['tally']['미확정']}")
    print(f"  돈 — {a['★cost_split']}")
    print(f"  적었다: {out}")
    # ★어긋남 1 · 미확정 2 · 통과 0 — **미확정을 합격으로 바꾸지 않는다**
    if a["tally"]["어긋남"]:
        return 1
    if a["tally"]["미확정"]:
        return 2
    return 0


if __name__ == "__main__":
    raise SystemExit(main())