"""GROUNDING-V2 §2-3.5 — **프로덕션 경로 하나로** 태우는 끝점 시험.

## 왜 이 파일이 따로 있나

`_execute` 를 직접 부르거나 손으로 만든 체크포인트를 먹이는 시험은 **조각마다
초록인데 이어 붙이면 죽는** 결함을 못 잡는다. 이 판에서 실제로 셋이 그랬다:

    ① `STEP_CLASSES` 에 없어 `get_step_runner` 가 404 를 냈다
    ② `_execute` 가 평평하게 돌려줘 하류의 ``data.decided`` 가 늘 비었다
    ③ `grounding_plan` 이 subject 를 다시 발급해 **원문 인용이 끊겼다**

셋 다 손으로 만든 fixture 뒤에 숨어 있었다. 그래서 여기서는

    프로덕션 dispatch(`get_step_runner`) → 공개 `run()` → **디스크에 실제로
    저장된 manifest** → 다음 스텝이 그 파일을 읽음

만 쓴다. LLM 호출만 막고 **그 사이의 배선은 하나도 안 흉내 낸다.**
"""
from __future__ import annotations

import json
from pathlib import Path
from unittest.mock import patch

import pytest
from sqlalchemy import text

from app.core.errors import AppError
from app.services.analysis_dispatch_service import get_step_runner

PID, EID = "p-grounding-chain", "e-grounding-chain"


def _is_connection_failure(exc: Exception) -> bool:
    """★**붙지 못한 것**만 True. deadlock·lock 은 False.

    PostgreSQL SQLSTATE class ``08`` 이 connection exception 이다.
    서버에 아예 못 붙으면 psycopg2 가 코드를 못 받아 ``pgcode`` 가 비어 있다.
    `40P01`(deadlock_detected)·`55P03`(lock_not_available) 은 **간섭 신호**라
    가리면 안 된다.
    """
    code = getattr(getattr(exc, "orig", None), "pgcode", None)
    return code is None or str(code).startswith("08")

#: ★원고 대신 쓰는 짧은 본문. 작품 고유명사를 안 쓴다.
_FULLTEXT = (
    "낡은 차 안. 운전대 옆에 쇠사슬로 묶인 기계식 요금통이 놓여 있다.\n"
    "그 위에 고무줄로 묶인 종이 승차권 뭉치가 얹혀 있다.\n"
    "안내원이 감색 제복에 챙 달린 모자를 쓰고 서 있다.\n"
)

#: A0 가 낼 후보 — 셋 다 **뒤 규칙에 걸려 사라질** 것들이다.
_A0_RAW = [
    {"surface_form": "쇠사슬로 묶인 기계식 요금통",
     "source_anchor": "SEG-001", "owner_type": "prop",
     "source_quote": "운전대 옆에 쇠사슬로 묶인 기계식 요금통이 놓여 있다.",
     "why_candidate": "시대 규격품", "planned_occurrences": 2},
    {"surface_form": "고무줄로 묶인 종이 승차권 뭉치",
     "source_anchor": "SEG-001", "owner_type": "prop",
     "source_quote": "그 위에 고무줄로 묶인 종이 승차권 뭉치가 얹혀 있다.",
     "why_candidate": "시대 인쇄물", "planned_occurrences": 1},
    {"surface_form": "감색 제복에 챙 달린 모자",
     "source_anchor": "SEG-001", "owner_type": "outlook",
     "source_quote": "안내원이 감색 제복에 챙 달린 모자를 쓰고 서 있다.",
     "why_candidate": "직업 복장", "planned_occurrences": 1},
    # ★고정 설비 — base location 으로 **승격하면 안 된다**(producer 는 §2-6.5)
    {"surface_form": "운전대", "source_anchor": "SEG-001",
     "owner_type": "location_part",
     "source_quote": "운전대 옆에 쇠사슬로 묶인 기계식 요금통이 놓여 있다.",
     "why_candidate": "고정 설비", "planned_occurrences": 1},
]


@pytest.fixture
def chain_env(tmp_path, monkeypatch):
    """디스크 체크포인트 + step_run row 를 갖춘 실제 실행 환경."""
    from sqlalchemy.exc import OperationalError

    from app.core.config import settings
    from app.core.database import SessionLocal

    monkeypatch.setattr(settings, "projects_dir", str(tmp_path), raising=False)
    # ★`init_db()` 를 **부르지 않는다.** 합산 걷기 한가운데서 raw migration 을
    #  다시 돌리면, 다른 시험이 이미 지운 테이블(`scene_still`·`world_guide`)에
    #  걸려 `DB schema migration failed` 로 선다 — 이 시험과 상관없는 자리다.
    #  스키마는 `tests/conftest.py` 의 세션 fixture 가 세운다. 여기서는
    #  **필요한 테이블 하나**(`step_run`)만 쓴다.
    try:
        db = SessionLocal()
        db.execute(text("SELECT 1 FROM step_run LIMIT 1"))
    except OperationalError as exc:
        # ★붙지도 못하는 환경만 skip 이다. `OperationalError` 를 통째로 삼키면
        #  **deadlock(40P01)·lock timeout 까지 가린다** — 시험 사이 간섭이
        #  조용히 묻힌다. SQLSTATE class 08(connection exception) 과
        #  「코드조차 못 받은 경우」(서버에 아예 못 붙음)만 넘어간다.
        if not _is_connection_failure(exc):
            raise
        pytest.skip(f"test DB 에 못 붙는다 — {exc.__class__.__name__}")
    db.rollback()
    db.execute(text("DELETE FROM step_run WHERE project_id = :p"), {"p": PID})
    # ★이 프로젝트를 **장부에 등록**한다 (2026-09-04). 종전에는 `step_run` 만
    #  심고 `project_registry` 행을 안 만들었다 — 그래도 돌았던 것은 아무도
    #  참조 무결성을 안 봤기 때문이다. 이제 `project_short_id_counter` 가
    #  `project_registry` 를 FK 로 잡으므로, 없는 프로젝트에 번호를 발급하려 들면
    #  **선다**(맞는 동작). fixture 가 실환경 하한을 갖춰야 한다.
    db.execute(text(
        "INSERT INTO user_account (id, username, display_name, password_hash, "
        "role, is_active, created_at, updated_at) VALUES "
        "(:u, :u, 't', 'x', 'creator', 1, 'x', 'x') ON CONFLICT (id) DO NOTHING"),
        {"u": f"u-{PID}"})
    db.execute(text(
        "INSERT INTO project_registry (id, name, created_by, created_at, "
        "updated_at) VALUES (:p, 'grounding-chain', :u, 'x', 'x') "
        "ON CONFLICT (id) DO NOTHING"), {"p": PID, "u": f"u-{PID}"})
    db.commit()

    root = tmp_path / PID / "checkpoints" / "episodes" / EID

    def write_cp(step_id: str, data: dict, status: str = "completed"):
        d = root / step_id
        d.mkdir(parents=True, exist_ok=True)
        (d / "manifest.json").write_text(
            json.dumps({"status": status, "step_id": step_id, "data": data},
                       ensure_ascii=False), encoding="utf-8")
        # ★같은 스텝을 다시 쓰는 시험이 있다 — INSERT 만 하면 두 번째에
        #  primary key 충돌로 **트랜잭션이 통째로 죽는다.**
        db.execute(text(
            "INSERT INTO step_run (id, project_id, episode_id, step_id, status,"
            " created_at, updated_at) VALUES (:i,:p,:e,:s,:st,'t','t')"
            " ON CONFLICT (id) DO UPDATE SET status = EXCLUDED.status"),
            {"i": f"{step_id}-r", "p": PID, "e": EID, "s": step_id, "st": status})
        db.commit()

    # 선행 입력 — 시험 대상이 아니다.
    write_cp("text_cleanup", {"cleaned_text": _FULLTEXT})
    write_cp("visual_world_rules", {"era": "1980년대", "region": "대한민국"})
    write_cp("scene_save", {"segments": [{"scene_index": 1, "content": _FULLTEXT}]})
    write_cp("shot_validator", {"scenes": [
        {"scene_index": 1, "scene_heading": "낡은 차 안",
         "shots": [{"shot_index": 1, "description": "차 안 전경",
                    "characters": []}]}]})
    try:
        yield db, root, write_cp
    finally:
        db.execute(text("DELETE FROM step_run WHERE project_id = :p"), {"p": PID})
        # ★심은 것은 **역순으로** 걷는다 — 계수기 → 프로젝트 → 사람.
        for q in ("DELETE FROM project_short_id_counter WHERE project_id = :p",
                  "DELETE FROM entity_episode_link WHERE project_id = :p",
                  "DELETE FROM entity_canon WHERE project_id = :p",
                  "DELETE FROM project_registry WHERE id = :p"):
            db.execute(text(q), {"p": PID})
        db.execute(text("DELETE FROM user_account WHERE id = :u"),
                   {"u": f"u-{PID}"})
        db.commit()
        db.close()


def _runner(step_id, db, mode="v2"):
    """★프로덕션이 부르는 그 함수. dict 를 들여다보지 않는다."""
    return get_step_runner(step_id=step_id, project_id=PID, episode_id=EID,
                           db=db, project_config={"grounding_mode": mode})


def _stored(root: Path, step_id: str) -> dict:
    """★디스크에 **실제로 저장된** manifest. 반환값이 아니다."""
    return json.loads((root / step_id / "manifest.json").read_text(encoding="utf-8"))


class TestTheWholeChainInProductionShape:
    """A0 원문 후보 → overlay → 계획 → 저장 manifest → 필터 보호 → 완전성."""

    @staticmethod
    def _run_a0(db):
        import app.modules.pipeline.grounding_a0 as a0

        def _call(**k):
            return {"candidates": [dict(c) for c in _A0_RAW]}

        with patch.object(a0, "_call_structured", side_effect=_call):
            return _runner("grounding_a0", db).run(mode="force")

    @staticmethod
    def _run_entity_all(db, sink):
        """소품 리스팅. ★나가는 **user prompt** 를 붙잡는다."""
        import app.modules.pipeline.entity_lister as el

        def _fake(*a, **k):
            sink.append(k.get("user_prompt") or (a[1] if len(a) > 1 else ""))
            return {"props": [
                {"name": "회수권 뭉치", "description": "종이 뭉치"},
                {"name": "요금통", "description": "쇠 상자"},
            ]}

        with patch.object(el, "call_structured", side_effect=_fake):
            return _runner("entity_all_prop", db).run(mode="force")

    @staticmethod
    def _run_plan(db, seen):
        import app.modules.pipeline.grounding_classifier as gc

        # ★`classify_samples` 를 손 반환으로 막으면 **패널 loop 와 지문 조립이
        #  통째로 안 돈다.** 실제로 그래서 지문에 칸 하나가 빠진 것을 이 시험이
        #  못 봤다. 진짜 경계(`_call_structured`)만 막고 나머지는 다 태운다.
        def _call(**kw):
            subjects = kw.get("_subjects_for_test") or []
            sink = kw.get("usage_sink")
            alias = (kw.get("model_alias") or kw.get("alias")
                     or _call.next_alias)
            if sink is not None:
                sink.update({"alias": alias,
                             "physical_model": f"phys/{alias}"})
            return {"classifications": [
                {"research_subject_id": sid,
                 "grounding_class": "externally_grounded",
                 "discriminability": "yes",
                 "referent_specificity": "exact_variant",
                 "difficulty": "hard", "confidence": 0.9,
                 "visibility_intent": "yes", "locale": "KR",
                 "generation": "1980s", "visible_discriminators": ["x"],
                 "likely_failure_modes": ["y"], "rationale": "z",
                 # ★2026-08-30: 보존·참고획득을 여는 칸. 비면 route 가
                 #  미확정으로 닫힌다(그게 계약이다).
                 "generation_difficulty": "not_hard"}
                for sid in _call.subject_ids]}
        _call.next_alias = "gpt"
        _call.subject_ids = []

        real_classify = gc.classify

        def _classify(subjects, **k):
            seen.extend(s for s in subjects if s not in seen)
            _call.subject_ids = [s["research_subject_id"] for s in subjects]
            _call.next_alias = str(
                ((k.get("project_config") or {}).get(gc.STEP_NAME) or {})
                .get("model") or "gpt")
            return real_classify(subjects, **k)

        with patch.object(gc, "_call_structured", _call), \
                patch.object(gc, "classify", _classify):
            return _runner("grounding_plan", db).run(mode="force")

    def test_the_chain_carries_the_candidate_all_the_way(self, chain_env):
        db, root, write_cp = chain_env

        # ① A0 — 원문에서 건진다
        assert self._run_a0(db)["status"] == "completed"
        a0_cp = _stored(root, "grounding_a0")["data"]
        assert len(a0_cp["candidates"]) == 4
        sids = {c["surface_form"]: c["research_subject_id"]
                for c in a0_cp["candidates"]}
        assert all(sids.values()), "A0 가 subject id 를 안 붙였다"

        # ② overlay — 나가는 프롬프트에 **실제로 실렸나**
        prompts: list = []
        self._run_entity_all(db, prompts)
        assert prompts, "리스터를 안 태웠다"
        assert "쇠사슬로 묶인 기계식 요금통" in prompts[0], \
            "A0 후보가 추출 프롬프트까지 안 갔다"
        assert "빼지 마세요" in prompts[0], "제외 규칙 완화 문구가 안 갔다"
        # ★outlook 은 이 갈래가 아니다 — 억지로 prop 에 태우지 않는다
        assert "감색 제복" not in prompts[0]

        # ★고정 설비는 **base location 으로도** 안 올라간다 (producer 는 §2-6.5)
        loc_prompts: list = []
        import app.modules.pipeline.entity_lister as el
        with patch.object(el, "call_structured",
                          side_effect=lambda *a, **k: (
                              loc_prompts.append(k.get("user_prompt")),
                              {"locations": [{"name": "차 안", "description": "좁다"}]})[1]):
            _runner("entity_all_location", db).run(mode="force")
        assert loc_prompts and "운전대" not in loc_prompts[0], \
            "고정 설비가 base location 으로 승격됐다"

        # ③ 계획 — A0 subject 를 **물려받고** 원문 인용을 쓴다
        #  ★추출은 이름을 **줄여서** 내놓는다. 그래도 물려받아야 한다.
        #   `손수건` 은 A0 후보가 없는 쪽 — 다시 발급하는 갈래를 같이 본다.
        write_cp("entity_merge", {
            "characters": [], "locations": [],
            "props": [{"short_id": "P01", "name": "승차권 뭉치",
                       "description": "LLM 이 상상해 쓴 묘사"},
                      {"short_id": "P02", "name": "요금통",
                       "description": "LLM 이 상상해 쓴 묘사"},
                      {"short_id": "P03", "name": "손수건",
                       "description": "LLM 이 상상해 쓴 묘사"}]})
        seen: list = []
        assert self._run_plan(db, seen)["status"] == "completed"
        plan = _stored(root, "grounding_plan")["data"]
        assert plan["a0_carried"] == 2, "A0 후보를 안 물려받고 다시 발급했다"
        assert plan["decided"], "저장된 manifest 의 decided 가 비었다"
        got = {d["_short_id"]: d for d in plan["decided"]}
        assert set(got) == {"P01", "P02", "P03"}
        # ★A0 가 발급한 그 id 여야 한다
        assert got["P01"]["research_subject_id"] == sids["고무줄로 묶인 종이 승차권 뭉치"]
        assert got["P02"]["research_subject_id"] == sids["쇠사슬로 묶인 기계식 요금통"]

        # ★분류기가 본 근거가 **원문 문장**이어야 한다 — 상상 묘사가 아니라.
        by_sid = {s["research_subject_id"]: s for s in seen}
        carried = by_sid[got["P02"]["research_subject_id"]]
        assert carried["quote_source"] == "manuscript"
        assert carried["source_quote"] in _FULLTEXT
        fresh = by_sid[got["P03"]["research_subject_id"]]
        assert fresh["quote_source"] == "entity_description"

        # ④ 완전성 — outlook 은 「없어진 것」이 아니라 「여기서 못 받는 것」
        rep = plan["completeness"]
        assert rep["missing_count"] == 0, rep["missing"]
        # ★outlook 과 location_part 둘 다 「없어진 것」이 아니라 **여기서 못 받는 것**
        assert len(rep["deferred"]) == 2
        assert {c["owner_type"] for c in rep["deferred"]} == {"outlook",
                                                              "location_part"}

        # ⑤ 필터 — 보호 목록에 **실제로** 들어간다
        write_cp("entity_relation", {"relations": []})
        with patch("app.modules.pipeline.entity_filter"
                   ".filter_low_frequency_entities") as f:
            f.return_value = {"filtered": {}}
            _runner("entity_filter", db).run(mode="force")
        assert f.call_count == 1
        assert f.call_args.kwargs["protected_short_ids"] == {"P01", "P02", "P03"}

    def test_the_second_dropping_place_gets_the_overlay_too(self, chain_env):
        """★후보를 버리는 자리는 **한 군데가 아니다**(계획 §1.8).

        `entity_all` 을 살아남아도 `entity_extract` 가 같은 부류를 또 거른다.
        그 프롬프트에도 overlay 가, **제외 기준 뒤**에 실려야 한다.
        """
        db, root, write_cp = chain_env
        self._run_a0(db)
        write_cp("entity_all_prop", {"props": [
            {"short_id": "P01", "name": "승차권 뭉치"}]})

        prompts: list = []
        import app.modules.pipeline.entity_extractor_v4 as ev

        def _fake(**k):
            prompts.append(k["user_prompt"])
            return {"props": [{"name": "승차권 뭉치", "description": "종이"}]}

        with patch.object(ev, "call_structured", side_effect=_fake):
            _runner("entity_extract_prop", db).run(mode="force")

        assert prompts, "추출기를 안 태웠다"
        text = prompts[0]
        assert "이 목록이 우선합니다" in text, \
            "두 번째 거르는 자리에 overlay 가 안 갔다"
        from app.modules.prompt_loader import load_prompt
        tail = [ln for ln in load_prompt("entity_extract_v4", "prop").strip()
                .split("\n") if ln.strip()][-1]
        assert text.index(tail) < text.index("이 목록이 우선합니다"), \
            "overlay 가 제외 기준 앞에 있다"

    def test_an_ambiguous_carry_does_not_slip_through_as_fresh(self, chain_env):
        """★A0 후보가 **있는데 어느 것인지 못 정한 것**을 새로 발급하면
        그 후보의 id 와 원문 인용을 조용히 잃는다.

        완전성 검사는 표면형이 산출에 남아 있으니 `complete=True` 를 낸다 —
        그것만 보면 통과로 지나간다. 「모른다」를 「없었다」로 닫지 않는다.
        """
        db, root, write_cp = chain_env

        import app.modules.pipeline.grounding_a0 as a0
        two = [
            {"surface_form": "종이 승차권", "source_anchor": "SEG-001",
             "owner_type": "prop", "why_candidate": "시대 인쇄물",
             "planned_occurrences": 1,
             "source_quote": "그 위에 고무줄로 묶인 종이 승차권 뭉치가 얹혀 있다."},
            {"surface_form": "종이 승차권 뭉치", "source_anchor": "SEG-002",
             "owner_type": "prop", "why_candidate": "시대 인쇄물",
             "planned_occurrences": 1,
             "source_quote": "그 위에 고무줄로 묶인 종이 승차권 뭉치가 얹혀 있다."},
        ]
        with patch.object(a0, "_call_structured",
                          side_effect=lambda **k: {"candidates": two}):
            _runner("grounding_a0", db).run(mode="force")

        write_cp("entity_merge", {"characters": [], "locations": [],
                                  "props": [{"short_id": "P01",
                                             "name": "종이 승차권 뭉치",
                                             "description": "LLM 이 상상해 쓴 묘사"}]})
        write_cp("entity_relation", {"relations": []})
        seen: list = []
        # ★표면형은 남아 있으니 완전성만 보면 통과다 — 그런데 결속을 못 했다.
        assert self._run_plan(db, seen)["status"] == "partial"
        # ★못 붙인 것에 **유료 분류를 사지 않는다** — 분류기가 아예 안 불린다.
        assert seen == [], "못 붙인 것을 분류기에 보냈다"

        plan = _stored(root, "grounding_plan")["data"]
        assert plan["completeness"]["complete"] is True
        assert plan["carry_reasons"].get("ambiguous") == 1
        assert plan["unresolved_carry_count"] == 1
        assert plan["a0_carried"] == 0

        row = plan["decided"][0]
        assert row["route"] == "unresolved"
        assert row["route_override_reason"] == "a0_carry_ambiguous"
        assert row["sample_count"] == 0
        # ★원 후보의 id·anchor·인용이 **그대로 남아 있어야** 한다
        keep = row["a0_candidates"]
        assert len(keep) == 2
        assert {c["surface_form"] for c in keep} == {"종이 승차권",
                                                     "종이 승차권 뭉치"}
        assert all(c["research_subject_id"] and c["source_anchor"]
                   and c["source_quote"] in _FULLTEXT for c in keep)

        with pytest.raises(AppError) as e:
            _runner("entity_filter", db).run(mode="force")
        assert e.value.code == "gate.blocked"

    def test_a_duplicate_surface_keeps_both_originals(self, chain_env):
        """★같은 `(owner, 표면형)` 후보가 둘이면 **둘 다** 남아야 한다.

        색인이 하나를 덮으면 그 자리에서 다른 후보의 id·anchor·인용이
        사라진다 — 「원 후보를 보존한다」가 거짓이 된다.
        """
        db, root, write_cp = chain_env

        import app.modules.pipeline.grounding_a0 as a0
        same = [
            {"surface_form": "요금통", "source_anchor": "SEG-001",
             "owner_type": "prop", "why_candidate": "시대 규격품",
             "planned_occurrences": 1,
             "source_quote": "운전대 옆에 쇠사슬로 묶인 기계식 요금통이 놓여 있다."},
            {"surface_form": "요금통", "source_anchor": "SEG-002",
             "owner_type": "prop", "why_candidate": "시대 규격품",
             "planned_occurrences": 1,
             "source_quote": "그 위에 고무줄로 묶인 종이 승차권 뭉치가 얹혀 있다."},
        ]
        with patch.object(a0, "_call_structured",
                          side_effect=lambda **k: {"candidates": same}):
            _runner("grounding_a0", db).run(mode="force")
        stored = _stored(root, "grounding_a0")["data"]["candidates"]
        assert len({c["research_subject_id"] for c in stored}) == 2, \
            "A0 가 발급한 id 가 겹쳤다 — anchor 가 다르면 달라야 한다"

        write_cp("entity_merge", {"characters": [], "locations": [],
                                  "props": [{"short_id": "P01", "name": "요금통",
                                             "description": "상상 묘사"}]})
        write_cp("entity_relation", {"relations": []})
        seen: list = []
        assert self._run_plan(db, seen)["status"] == "partial"
        assert seen == [], "못 붙인 것을 분류기에 보냈다"

        plan = _stored(root, "grounding_plan")["data"]
        assert plan["carry_reasons"].get("duplicate_surface") == 1
        keep = plan["decided"][0]["a0_candidates"]
        assert len(keep) == 2, f"후보 하나를 잃었다: {keep}"
        assert {c["source_anchor"] for c in keep} == {"SEG-001", "SEG-002"}
        assert len({c["research_subject_id"] for c in keep}) == 2
        assert all(c["source_quote"] in _FULLTEXT for c in keep)

    def test_counts_cover_every_decided_row(self, chain_env):
        """★집계 모집단이 `decided` 와 같아야 한다.

        분류기가 낸 것만 세면 못 붙여 분류 안 한 `unresolved` 가 빠진다 —
        전부 못 붙인 판에서는 `counts` 가 통째로 빈다.
        """
        db, root, write_cp = chain_env
        write_cp("entity_relation", {"relations": []})

        # ★겹치는 후보 둘 + 안 겹치는 후보 하나
        import app.modules.pipeline.grounding_a0 as a0
        mixed = [
            {"surface_form": "종이 승차권", "source_anchor": "SEG-001",
             "owner_type": "prop", "why_candidate": "시대 인쇄물",
             "planned_occurrences": 1,
             "source_quote": "그 위에 고무줄로 묶인 종이 승차권 뭉치가 얹혀 있다."},
            {"surface_form": "종이 승차권 뭉치", "source_anchor": "SEG-002",
             "owner_type": "prop", "why_candidate": "시대 인쇄물",
             "planned_occurrences": 1,
             "source_quote": "그 위에 고무줄로 묶인 종이 승차권 뭉치가 얹혀 있다."},
            {"surface_form": "기계식 요금통", "source_anchor": "SEG-001",
             "owner_type": "prop", "why_candidate": "시대 규격품",
             "planned_occurrences": 1,
             "source_quote": "운전대 옆에 쇠사슬로 묶인 기계식 요금통이 놓여 있다."},
        ]
        with patch.object(a0, "_call_structured",
                          side_effect=lambda **k: {"candidates": mixed}):
            _runner("grounding_a0", db).run(mode="force")

        # ㉠ 겹치는 둘은 못 붙고, 안 겹치는 하나(요금통)는 **승격**된다
        write_cp("entity_merge", {"characters": [], "locations": [],
                                  "props": [{"short_id": "P01",
                                             "name": "종이 승차권 뭉치",
                                             "description": "상상 묘사"},
                                            {"short_id": "P02",
                                             "name": "종이 승차권 뭉치 하나",
                                             "description": "상상 묘사"}]})
        self._run_plan(db, [])
        plan = _stored(root, "grounding_plan")["data"]
        assert sum(plan["counts"].values()) == len(plan["decided"])
        # ★★「요금통」은 어느 엔티티도 안 불렀지만 base owner 라 **승격**되어
        #  분류된다 — 승격이 없던 시절에는 그 원문 인용이 통째로 버려졌다.
        #  겹치는 둘은 **승격 안 한다**: 「서로 다른 두 대상」인지 「같은 것을
        #  두 번 적은 것」인지 기계적으로 못 가르고, 승격하면 같은 것을 두 번
        #  조사할 수 있다.
        assert plan["counts"] == {"research": 1, "unresolved": 2}, plan["counts"]

        # ㉡ 섞인 판 — 붙은 것 하나(요금통) + 못 붙인 것 하나
        write_cp("entity_merge", {"characters": [], "locations": [],
                                  "props": [{"short_id": "P01",
                                             "name": "기계식 요금통",
                                             "description": "상상 묘사"},
                                            {"short_id": "P02",
                                             "name": "종이 승차권 뭉치",
                                             "description": "상상 묘사"}]})
        self._run_plan(db, [])
        plan = _stored(root, "grounding_plan")["data"]
        assert sum(plan["counts"].values()) == len(plan["decided"])
        assert plan["counts"].get("unresolved"), plan["counts"]
        assert len(plan["counts"]) >= 2, plan["counts"]

    def test_a_lost_candidate_blocks_the_filter(self, chain_env):
        """★A0 가 건진 것이 산출에 안 남으면 **필터가 막힌다.**

        경고만 찍으면 조사해야 할 것이 지워진 채로 계속 간다. 여기서는
        추출이 후보를 **하나도 안 내놓은** 판을 만든다.
        """
        db, root, write_cp = chain_env
        self._run_a0(db)

        # ★A0 가 건진 두 소품이 산출에 없다 — 앞에서 지워진 것이다.
        write_cp("entity_merge", {"characters": [], "locations": [],
                                  "props": [{"short_id": "P09", "name": "손수건",
                                             "description": "천"}]})
        write_cp("entity_relation", {"relations": []})
        seen: list = []
        self._run_plan(db, seen)

        plan = _stored(root, "grounding_plan")["data"]
        # ★★**두 수가 다르다.** 엔티티 행은 없지만(추출이 지웠다) 승격되어
        #  **조사는 계속된다** — 그래서 잃은 것은 0이고 필터도 안 막힌다.
        #  계약 §7 「엔티티 보존과 조사는 별개」가 여기서 갈라진다.
        assert plan["completeness"]["entity_missing_count"] == 2
        assert plan["completeness"]["missing_count"] == 0
        assert plan["completeness"]["promoted_count"] == 2
        assert plan["completeness"]["complete"] is True

        # ★필터는 안 막힌다 — 조사할 것을 잃지 않았기 때문이다.
        _runner("entity_filter", db).run(mode="force")

    def test_the_candidate_ledger_survives_to_disk(self, chain_env):
        """★★문서가 「SOT 는 체크포인트」라고 적어 놓고 **체크포인트에 없었다**.

        조립 자리에서 장부를 만들어 놓고 호출부가 안 담으면, 그 장부는 이
        프로세스 안에서만 살고 다음 사람은 못 본다 — offline replay 도 못 읽는다.
        **디스크에서 다시 읽어** 잰다.
        """
        db, root, write_cp = chain_env
        self._run_a0(db)
        write_cp("entity_merge", {"characters": [], "locations": [],
                                  "props": [{"short_id": "P01",
                                             "name": "요금통",
                                             "description": "상상 묘사"}]})
        write_cp("entity_relation", {"relations": []})
        self._run_plan(db, [])

        led = _stored(root, "grounding_plan")["data"]["candidate_ledger"]
        assert led["total"] == len(_A0_RAW)
        assert sum(led["by_disposition"].values()) == led["total"]
        # ★행마다 **누구인지·어디서 왔는지**가 그대로 남는다
        for r in led["rows"]:
            assert r["research_subject_id"]
            assert r["surface_form"]
            assert r["owner_type"]
            assert r["disposition"] in (
                "carried", "promoted", "deferred", "contested", "unresolved")
        # ★A0 CP 의 후보 id 집합과 **같아야** 한다 — 하나도 안 사라진다
        a0 = _stored(root, "grounding_a0")["data"]["candidates"]
        assert {r["research_subject_id"] for r in led["rows"]} == \
            {c["research_subject_id"] for c in a0}

    def test_the_carry_contract_version_moved_with_the_meaning(self):
        """★★안 올리면 옛 `grounding_plan` CP 를 그대로 재사용해 **승격이
        아무 데도 안 닿는다** — 고친 것이 안 도는 부류다 (Codex).
        """
        from app.modules.pipeline import grounding_carry as carry

        assert carry.CARRY_CONTRACT_VERSION == 3

    def test_the_version_is_actually_folded_into_the_fingerprint(self, chain_env):
        """★상수만 올리고 지문이 안 읽으면 resume 이 안 움직인다."""
        import inspect

        from app.core.steps import grounding_steps

        src = inspect.getsource(grounding_steps.GroundingPlanStep._config_hash)
        assert "CARRY_CONTRACT_VERSION" in src

    def test_a_promoted_rescue_is_reported_not_swallowed(self, chain_env,
                                                        caplog):
        """★★「승격으로 살렸으니 괜찮다」로 넘어가면 **추출 결함이 영원히 안
        드러난다**. 엔티티 행이 사라진 것은 그대로 사실이다.

        ★그리고 아무도 안 읽는 수는 **칸만 있고 뜻이 없다** — `entity_missing_count`
        를 저장만 하고 어디서도 안 꺼내고 있었다.
        """
        import logging

        db, root, write_cp = chain_env
        self._run_a0(db)
        write_cp("entity_merge", {"characters": [], "locations": [],
                                  "props": [{"short_id": "P09", "name": "손수건",
                                             "description": "천"}]})
        write_cp("entity_relation", {"relations": []})
        with caplog.at_level(logging.WARNING):
            self._run_plan(db, [])
        assert any("승격으로" in r.getMessage() for r in caplog.records), \
            "승격으로 살린 사실이 아무 데도 안 남았다"
        # ★수까지 실려야 한다 — 「있었다」만으로는 몇 개인지 모른다
        msg = next(r.getMessage() for r in caplog.records
                   if "승격으로" in r.getMessage())
        assert "entity_missing=2" in msg and "missing=0" in msg

    def test_the_ledger_survives_the_no_subject_path_too(self, chain_env):
        """★★반환이 둘이면 **한쪽만 고쳐진다**.

        「분류할 subject 가 하나도 없는」 조기 반환이 장부 칸을 안 넘겨, 그
        판에서만 장부가 통째로 비었다 — 그런데 그 갈래야말로 **전부 얽혀서
        아무것도 분류 못 한 판**이라 장부가 제일 필요한 자리다 (Codex).
        """
        db, root, write_cp = chain_env
        import app.modules.pipeline.grounding_a0 as a0
        tangled = [
            {"surface_form": "종이 승차권", "source_anchor": "SEG-001",
             "owner_type": "prop", "why_candidate": "시대 인쇄물",
             "planned_occurrences": 1,
             "source_quote": "그 위에 고무줄로 묶인 종이 승차권 뭉치가 얹혀 있다."},
            {"surface_form": "종이 승차권 뭉치", "source_anchor": "SEG-002",
             "owner_type": "prop", "why_candidate": "시대 인쇄물",
             "planned_occurrences": 1,
             "source_quote": "그 위에 고무줄로 묶인 종이 승차권 뭉치가 얹혀 있다."},
        ]
        with patch.object(a0, "_call_structured",
                          side_effect=lambda **k: {"candidates": tangled}):
            _runner("grounding_a0", db).run(mode="force")
        write_cp("entity_merge", {"characters": [], "locations": [],
                                  "props": [{"short_id": "P01",
                                             "name": "종이 승차권 뭉치",
                                             "description": "상상 묘사"}]})
        write_cp("entity_relation", {"relations": []})
        seen: list = []
        self._run_plan(db, seen)
        assert seen == [], "분류기를 불렀다 — 조기 반환 갈래가 아니다"

        led = _stored(root, "grounding_plan")["data"]["candidate_ledger"]
        assert led, "조기 반환이 장부를 버렸다"
        assert led["total"] == 2
        assert led["by_disposition"]["unresolved"] == 2

    def test_a_candidate_lost_without_promotion_still_blocks_the_filter(
            self, chain_env):
        """★★게이트가 **무뎌지지 않았는지**. 승격 안 되는 갈래로 잰다.

        얽혀서 `unresolved` 인 후보는 승격 대상이 아니다 — 그건 여전히
        「잃은 것」이고 필터를 막아야 한다. 이 시험이 없으면 승격이 게이트를
        통째로 죽였는지 알 수 없다.
        """
        db, root, write_cp = chain_env
        # ★두 후보가 **한 엔티티에** 걸리게 만든다 → 얽힘 → 승격 안 함
        import app.modules.pipeline.grounding_a0 as a0
        tangled = [
            {"surface_form": "종이 승차권", "source_anchor": "SEG-001",
             "owner_type": "prop", "why_candidate": "시대 인쇄물",
             "planned_occurrences": 1,
             "source_quote": "그 위에 고무줄로 묶인 종이 승차권 뭉치가 얹혀 있다."},
            {"surface_form": "종이 승차권 뭉치", "source_anchor": "SEG-002",
             "owner_type": "prop", "why_candidate": "시대 인쇄물",
             "planned_occurrences": 1,
             "source_quote": "그 위에 고무줄로 묶인 종이 승차권 뭉치가 얹혀 있다."},
        ]
        with patch.object(a0, "_call_structured",
                          side_effect=lambda **k: {"candidates": tangled}):
            _runner("grounding_a0", db).run(mode="force")
        write_cp("entity_merge", {"characters": [], "locations": [],
                                  "props": [{"short_id": "P01",
                                             "name": "종이 승차권 뭉치",
                                             "description": "상상 묘사"}]})
        write_cp("entity_relation", {"relations": []})
        assert self._run_plan(db, [])["status"] == "partial"

        with pytest.raises(AppError) as e:
            _runner("entity_filter", db).run(mode="force")
        assert e.value.code == "gate.blocked"

    def test_the_operator_can_override_a_lost_candidate(self, chain_env):
        """★알고도 진행하는 길은 남긴다 — 다만 **명시적으로**."""
        db, root, write_cp = chain_env
        self._run_a0(db)
        write_cp("entity_merge", {"characters": [], "locations": [],
                                  "props": [{"short_id": "P09", "name": "손수건",
                                             "description": "천"}]})
        write_cp("entity_relation", {"relations": []})
        self._run_plan(db, [])

        r = get_step_runner(
            step_id="entity_filter", project_id=PID, episode_id=EID, db=db,
            project_config={"grounding_mode": "v2",
                            "allow_missing_grounding": True})
        with patch("app.modules.pipeline.entity_filter"
                   ".filter_low_frequency_entities") as f:
            f.return_value = {"filtered": {}}
            r.run(mode="force")
        assert f.call_count == 1

    def test_shadow_touches_nothing_in_production(self, chain_env):
        """★shadow 는 **파이프라인에 안 들어온다.**

        프롬프트 값만 가리는 것으로는 부족하다 — 이 스텝들은 production
        step id 를 쓰고 DAG 에 하류가 걸려 있어, `force` 로 한 번 돌리면
        `_execute` 전에 그 하류가 통째로 무효화된다. 「shadow 를 켠 것만으로
        하류가 stale 되지 않는다」(§2-3b 통과 조건)가 거짓이 된다.

        여기서 재는 것은 **셋 다**다 — 파일 바이트 · step_run 행 · 프롬프트.
        """
        from app.modules.pipeline.grounding_shadow import (
            production_checkpoint_digest,
        )

        db, root, write_cp = chain_env
        write_cp("entity_merge", {"characters": [], "locations": [],
                                  "props": [{"short_id": "P01", "name": "요금통"}]})
        write_cp("entity_relation", {"relations": []})

        def _rows():
            return sorted(db.execute(text(
                "SELECT step_id, status FROM step_run WHERE project_id = :p"),
                {"p": PID}).fetchall())

        before_files = production_checkpoint_digest(root)
        before_rows = _rows()

        import app.modules.pipeline.grounding_a0 as a0

        for sid in ("grounding_a0", "grounding_plan"):
            with patch.object(a0, "_call_structured",
                              side_effect=AssertionError("shadow 인데 유료 호출")):
                assert _runner(sid, db, mode="shadow_plan").run(
                    mode="force")["status"] == "not_applicable"

        # ★새로 생긴 not_applicable 표식만 빼고 **한 바이트도** 안 달라져야 한다.
        after_files = {k: v for k, v in production_checkpoint_digest(root).items()
                       if not k.startswith(("grounding_a0/", "grounding_plan/"))}
        assert after_files == before_files, "shadow 가 production 파일을 바꿨다"
        after_rows = [r for r in _rows()
                      if r[0] not in ("grounding_a0", "grounding_plan")]
        assert after_rows == before_rows, "shadow 가 하류 step_run 을 건드렸다"

        prompts: list = []
        self._run_entity_all_in_mode(db, prompts, "shadow_plan")
        assert prompts and "원문에서 먼저 건진 고증 후보" not in prompts[0], \
            "shadow 인데 overlay 가 나가는 프롬프트에 실렸다"

    def test_legacy_and_shadow_produce_the_same_thing(self, chain_env):
        """★shadow 를 켜는 것과 legacy 는 **구별이 안 돼야** 한다."""
        db, root, write_cp = chain_env
        out = {}
        for mode in ("legacy", "shadow_plan"):
            assert _runner("grounding_a0", db, mode=mode).run(
                mode="force")["status"] == "not_applicable"
            prompts: list = []
            self._run_entity_all_in_mode(db, prompts, mode)
            out[mode] = prompts[0]
        assert out["legacy"] == out["shadow_plan"]

    @staticmethod
    def _run_entity_all_in_mode(db, sink, mode):
        import app.modules.pipeline.entity_lister as el

        def _fake(*a, **k):
            sink.append(k.get("user_prompt") or (a[1] if len(a) > 1 else ""))
            return {"props": [{"name": "요금통", "description": "쇠 상자"}]}

        with patch.object(el, "call_structured", side_effect=_fake):
            return _runner("entity_all_prop", db, mode=mode).run(mode="force")

    def test_legacy_runs_the_same_chain_with_nothing_added(self, chain_env):
        """★legacy — A0 도 계획도 **안 돌고**, 프롬프트도 안 바뀐다."""
        db, root, write_cp = chain_env

        assert _runner("grounding_a0", db, mode="legacy").run(
            mode="force")["status"] == "not_applicable"

        prompts: list = []
        import app.modules.pipeline.entity_lister as el

        def _fake(*a, **k):
            prompts.append(k.get("user_prompt") or (a[1] if len(a) > 1 else ""))
            return {"props": [{"name": "요금통", "description": "쇠 상자"}]}

        with patch.object(el, "call_structured", side_effect=_fake):
            _runner("entity_all_prop", db, mode="legacy").run(mode="force")
        assert prompts and "원문에서 먼저 건진 고증 후보" not in prompts[0]

        write_cp("entity_merge", {"characters": [], "locations": [],
                                  "props": [{"short_id": "P01", "name": "요금통"}]})
        write_cp("entity_relation", {"relations": []})
        assert _runner("grounding_plan", db, mode="legacy").run(
            mode="force")["status"] == "not_applicable"
        with patch("app.modules.pipeline.entity_filter"
                   ".filter_low_frequency_entities") as f:
            f.return_value = {"filtered": {}}
            _runner("entity_filter", db, mode="legacy").run(mode="force")
        assert f.call_args.kwargs["protected_short_ids"] is None


class TestOnlyConnectionFailuresAreSkipped:
    """★`OperationalError` 를 통째로 삼키면 **deadlock 까지 가린다.**

    실제 혼합 실행에서 `safe_drop_all` 이 `DeadlockDetected(40P01)` 를 냈고
    그것도 `sqlalchemy.exc.OperationalError` 다 (Codex 실측).
    """

    @staticmethod
    def _err(pgcode):
        from sqlalchemy.exc import OperationalError

        class _Orig(Exception):
            pass

        o = _Orig("boom")
        o.pgcode = pgcode
        return OperationalError("SELECT 1", {}, o)

    @pytest.mark.parametrize("code", ["08000", "08003", "08006", "08001", None])
    def test_connection_classes_are_skipped(self, code):
        assert _is_connection_failure(self._err(code)) is True

    @pytest.mark.parametrize("code", ["40P01", "55P03", "57014", "42P01"])
    def test_deadlock_and_lock_are_not_swallowed(self, code):
        assert _is_connection_failure(self._err(code)) is False

    def test_a_deadlock_reaches_the_test_instead_of_skipping(self, monkeypatch):
        """★끝점 — fixture 가 deadlock 을 **skip 으로 안 바꾼다.**"""
        import app.core.database as dbmod

        boom = self._err("40P01")
        monkeypatch.setattr(dbmod, "SessionLocal",
                            lambda: (_ for _ in ()).throw(boom))
        with pytest.raises(Exception) as e:
            list(chain_env.__wrapped__(Path("/tmp"), monkeypatch))
        assert e.value is boom or "40P01" in str(e.value)