"""C(c) 산출이 **DB 까지** 간다. ★유료 0 · 공개 끝점.

Codex (2026-09-01) — 「새 결과가 `entity_merge` → `entity_t2i` → DB, 판별,
관계 저장까지 도달하는 **세로 시험**이 필요합니다.」

★각 단을 따로 잠그면 「옛 것이 꺼졌다」만 재고 **대체 산출이 없는 것**을
못 본다. 여기서는 producer 산출 하나를 **끝까지** 흘린다 —

    grounding_chunk CP
      → EntityMergeStep._chunk_projection      (호출 0)
      → EntityT2iStep 의 표식 나르기            (호출 0)
      → EntitySyncService.sync_from_checkpoint (DB)
"""
from __future__ import annotations

import json
import uuid
from pathlib import Path

import pytest

from app.modules.pipeline.grounding_entity_sync_ext import (
    CHUNK_SCHEMA_MARKER, CHUNK_SCHEMA_VERSION)

#: 인물·장소부분 — 두 칸이 다 비어 있어도 되는 갈래.
NEUTRAL = {"location": None, "visual_identity": None}
#: 소품 — `visual_identity` 가 **있어야** 한다(검증기가 그렇게 말한다).
PROP_META = {"location": None,
             "visual_identity": {"reference_required": True}}


#: ★★값을 **손으로 조립하지 않는다** — `test_location_part_sync_endpoint`
#:  가 쓰는 **온전한 모양** 그대로다. 내가 칸을 하나씩 지어 넣었더니 enum 도
#:  틀리고 필수 칸도 빠졌다(`SpaceProfileError` 두 번).
LOC_META = {"location": {"space_profile": {
    "kind": "single_space", "allowed_space_keys": ["main"],
    "default_space_key": None}}, "visual_identity": None}


def _ent(name, sid, meta):
    return {"name": name, "short_id": sid, "description": "설명",
            "visual_traits": ["가"], "t2i_prompt": "그림 지문",
            "metadata_json": meta}


CHUNK_CP = {"status": "completed", "data": {
    "characters": [_ent("사람", "C01", NEUTRAL)],
    "locations": [_ent("그 장소", "L01", LOC_META)],
    "props": [_ent("물건", "P01", PROP_META)],
    "location_parts": [_ent("그 부분", "LP01", NEUTRAL)],
    "grounding_part_of": [{"part": "LP01", "whole": "L01"}],
    "contracts": {"adapter": "3.202609012800"},
}}


class _Runner:
    """스텝 함수를 **진짜로** 부르기 위한 최소 대역."""

    project_config = {"grounding_mode": "v2_chunk"}
    project_id = "p"
    episode_id = "e"
    db = None

    def __init__(self, cps):
        self._cps = cps

    def _load_prev_checkpoint(self, sid):
        return self._cps.get(sid)

    def _config_hash(self):
        return "h"


def _merge(chunk_cp):
    from app.core.steps.entity_steps import EntityMergeStep as E

    obj = _Runner({"grounding_chunk": chunk_cp})
    obj._chunk_projection = E._chunk_projection.__get__(obj)
    obj._execute = E._execute.__get__(obj)
    return obj._execute()


def _filter_cp(merge_cp):
    """`entity_filter` 산출 — **진짜 스텝**을 태운다(호출 0 투영)."""
    from app.core.steps.entity_steps import EntityFilterStep as F

    obj = _Runner({"entity_merge": merge_cp})
    obj._chunk_projection = F._chunk_projection.__get__(obj)
    obj._execute = F._execute.__get__(obj)
    return {"status": "completed", **obj._execute()}


def _detail_cp(merge_cp):
    """`entity_detail` 산출 — **진짜 `EntityDetailStep._execute`** 를 태운다.

    ★앞 판은 그 스텝을 import 만 하고 CP 를 **손으로 만들었다**
    (Codex 2026-09-01). 그러면 필터→상세 사이가 증거 없이 비어 있다.
    """
    from app.core.steps.entity_steps import EntityDetailStep as D

    def _stub(*, step, system_prompt, user_prompt, response_schema,
              project_config, schema_name, opik_metadata, **kw):
        """★결정적 대역 — 원문에 있는 것만 되돌려 준다."""
        out = []
        for key, owner in _LANES():
            for e in (merge_cp["data"].get(key) or []):
                out.append({"name": e["name"], "entity_type": owner,
                            "description": e["description"],
                            "visual_traits": e["visual_traits"]})
        return {"entities": out}

    # ★★두 대역이 **같은 이름**을 다툰다 — detail 것을 안 되돌리면 T2I 대역을
    #  덮어 metadata 가 빈 채로 간다(실제로 그랬다). 그래서 이 스텝을 도는
    #  동안만 걸고 바로 되돌린다.
    import app.core.steps.entity_steps as _es

    _saved = _es.call_structured
    _es.call_structured = _stub
    obj = _Runner({"entity_filter": _filter_cp(merge_cp),
                   "entity_merge": merge_cp,
                   "visual_world_rules": {"data": {"era": "", "region": ""}}})
    obj._load_cleaned_text = lambda: "원문"
    obj.build_opik_metadata = lambda *a, **k: {}
    obj.update_progress = lambda *a, **k: None
    obj.step_id = "entity_detail"
    obj._execute = D._execute.__get__(obj)
    try:
        return {"status": "completed", **obj._execute()}
    finally:
        _es.call_structured = _saved


def _LANES():
    from app.modules.pipeline.grounding_carry import ENTITY_KEY_TO_OWNER

    return ENTITY_KEY_TO_OWNER


def _t2i(merge_cp, monkeypatch, *, resume_cp=None, counter=None):
    """★★**진짜 `EntityT2iStep._execute`** 를 태운다.

    앞 판은 `_by_lane` + marker helper 로 CP 를 **손으로 만들어** 실제 스텝의
    이름 열쇠 충돌을 못 봤다 (Codex 2026-09-01).
    """
    from app.core.steps.entity_steps import EntityT2iStep as T

    # ★대역이 **(이름, 갈래)** 로 가른다 — 지문이 그 둘을 다 싣는다
    #  (`fields={"entity_name": …, "entity_type": …}`). 앞 판은 이름만 봐서
    #  같은 이름의 두 갈래에 **같은 metadata** 를 줬다.
    from app.modules.pipeline.grounding_carry import ENTITY_KEY_TO_OWNER

    meta_by = {}
    for key, owner in ENTITY_KEY_TO_OWNER:
        for e in (merge_cp["data"].get(key) or []):
            meta_by[(e["name"], owner)] = e["metadata_json"]

    def _stub(*, step, system_prompt, user_prompt, response_schema,
              project_config, schema_name, opik_metadata, **kw):
        """★결정적 대역 — 의미 모델 경계 **하나만** 바꾼다."""
        # ★`location` 은 `location_part` 의 **부분 문자열**이다 — 그대로
        #  견주면 둘 다 맞는다. 가장 긴 갈래 이름을 고른다.
        hit = sorted(((nm, ow) for (nm, ow) in meta_by
                      if nm in user_prompt and ow in user_prompt),
                     key=lambda x: len(x[1]), reverse=True)
        assert hit, "★대역이 대상을 못 찾는다"
        return {"t2i_prompt": "그림 지문", "metadata_json": meta_by[hit[0]]}

    def _counted(**kw):
        if counter is not None:
            counter.append(kw.get("user_prompt", ""))
        return _stub(**kw)

    detail = _detail_cp(merge_cp)       # ★대역을 걸기 **전에** 만든다
    monkeypatch.setattr("app.core.steps.entity_steps.call_structured",
                        _counted, raising=False)
    obj = _Runner({"entity_merge": merge_cp,
                   "entity_detail": detail,
                   "visual_world_rules": {"data": {"era": "", "region": ""}}})
    obj.load_checkpoint = lambda: resume_cp
    obj.save_checkpoint = lambda *a, **k: None
    obj.update_progress = lambda *a, **k: None
    obj.build_opik_metadata = lambda *a, **k: {}
    obj.step_id = "entity_t2i"
    for n in ("_execute",):
        setattr(obj, n, getattr(T, n).__get__(obj))
    return obj._execute()


def _write_t2i(seeded, t2i):
    """`entity_t2i` CP 를 **한 자리에서** 적는다.

    ★`status` 를 빠뜨리면 sync 가 그 CP 를 통째로 건너뛴다 — 두 자리에서
    따로 적었더니 한쪽만 빠졌다(실제로 그랬다).
    """
    _write(seeded["tmp"], seeded["pid"], seeded["eid"], "entity_t2i",
           {"status": "completed",
            **{k: v for k, v in t2i.items() if k != "data"},
            "data": t2i["data"]})


def _write(tmp: Path, pid: str, eid: str, step: str, payload: dict) -> None:
    d = tmp / pid / "checkpoints" / "episodes" / eid / step
    d.mkdir(parents=True, exist_ok=True)
    (d / "manifest.json").write_text(json.dumps(payload), encoding="utf-8")


@pytest.fixture
def seeded(tmp_path, monkeypatch):
    """★판을 차리는 도구다 — `test_location_part_sync_endpoint` 것과 같은 모양.

    fixture 라 import 가 안 돼서 한 벌 더 둔다. 갈래 목록 같은 **계약**이
    아니라 도구이므로 사본이어도 갈릴 것이 없다.
    """
    from datetime import datetime, timezone

    monkeypatch.setattr("app.core.config.settings.projects_dir", str(tmp_path))

    from app.core.database import SessionLocal, init_db
    from app.models.catalog import ProjectRegistry, UserAccount
    from app.models.project import Episode

    init_db()
    session = SessionLocal()
    try:
        now = datetime.now(timezone.utc).isoformat()
        uid = f"test-user-{uuid.uuid4()}"
        pid = f"test-vert-{uuid.uuid4()}"
        eid = f"test-vert-ep-{uuid.uuid4()}"
        session.add(UserAccount(
            id=uid, username=f"u_{uid}", display_name="t",
            password_hash="x", role="creator", is_active=1,
            created_at=now, updated_at=now))
        session.flush()
        session.add(ProjectRegistry(
            id=pid, name="vert-test", description="", created_by=uid,
            created_at=now, updated_at=now))
        session.flush()
        session.add(Episode(
            id=eid, project_id=pid, episode_number=1, title="t",
            source_filename="f.txt", source_path="/tmp/f.txt",
            language="ko", status="uploaded", created_at=now, updated_at=now))
        session.flush()
        yield {"db": session, "pid": pid, "eid": eid, "tmp": tmp_path}
    finally:
        session.rollback()
        session.close()


class TestTheChunkOutputReachesTheDatabase:
    """★★producer 산출 하나를 **실제 스텝들로** 끝까지 흘린다."""

    def _run(self, seeded, monkeypatch, chunk_cp=None):
        from app.services.checkpoint_sync.entity_sync_service import (
            EntitySyncService)

        merged = _merge(chunk_cp or CHUNK_CP)
        t2i = _t2i(merged, monkeypatch)
        # ★실제 저장/읽기를 한 번 거친다 — 최상위 표식이 살아남는지
        _write_t2i(seeded, t2i)
        got = EntitySyncService(
            seeded["db"], seeded["pid"], seeded["eid"]).sync_from_checkpoint()
        seeded["db"].flush()
        return {"merged": merged, "t2i": t2i, "sync": got}

    def _canons(self, seeded):
        from app.models.project import EntityCanon

        return {r.short_id: r.entity_type for r in seeded["db"].query(
            EntityCanon).filter(
                EntityCanon.project_id == seeded["pid"]).all()}

    def test_all_four_lanes_land_in_the_db(self, seeded, monkeypatch):
        self._run(seeded, monkeypatch)
        assert self._canons(seeded) == {
            "C01": "character", "L01": "location",
            "P01": "prop", "LP01": "location_part"}

    def test_the_real_t2i_step_carried_the_marker(self, seeded, monkeypatch):
        got = self._run(seeded, monkeypatch)["t2i"]
        assert got[CHUNK_SCHEMA_MARKER] == CHUNK_SCHEMA_VERSION
        assert CHUNK_SCHEMA_MARKER not in got["data"]

    def test_a_same_named_place_and_part_both_survive(self, seeded,
                                                      monkeypatch):
        """★★★Codex 재현 (09-01) — 열쇠가 **이름 하나**면 같은 이름의 장소와
        그 부분이 서로 덮거나 건너뛴다. sync 는 `(갈래, 이름)` 계약을 지키는데
        그 **앞에서** 무너졌다.
        """
        same = json.loads(json.dumps(CHUNK_CP))
        same["data"]["locations"][0]["name"] = "이발소"
        same["data"]["location_parts"][0]["name"] = "이발소"
        self._run(seeded, monkeypatch, chunk_cp=same)
        got = self._canons(seeded)
        assert got.get("L01") == "location", "★장소가 덮였다"
        assert got.get("LP01") == "location_part", "★장소 부분이 덮였다"

    def test_the_part_of_relation_is_projected_from_the_same_output(self):
        """★관계도 **같은 산출**에서 나온다 — 모델에 다시 안 묻는다."""
        from app.core.steps.entity_relation_step import EntityRelationStep as R

        obj = _Runner({"grounding_chunk": CHUNK_CP})
        obj._chunk_projection = R._chunk_projection.__get__(obj)
        rows = obj._chunk_projection()["data"]["relations"]
        assert len(rows) == 1
        assert {p["short_id"] for p in rows[0]["participants"]} == {"LP01",
                                                                    "L01"}

    def test_the_filter_keeps_all_four_lanes(self):
        """★★필터가 옛 길로 가면 **네 번째 갈래가 사라진다**."""
        from app.core.steps.entity_steps import EntityFilterStep as F

        obj = _Runner({"entity_merge": _merge(CHUNK_CP)})
        obj._chunk_projection = F._chunk_projection.__get__(obj)
        obj._execute = F._execute.__get__(obj)
        got = obj._execute()["data"]["filtered_entities"]
        assert sorted(got) == ["characters", "location_parts", "locations",
                               "props"]
        assert [e["short_id"] for e in got["location_parts"]] == ["LP01"]


class TestThePartOfReachesTheDatabaseToo:
    """★★세로의 **마지막 한 단** — 관계 투영이 DB 의 `relation_fact` 까지.

    앞 판은 투영 **산출만** 보고 DB 는 안 봤다. 「관계가 생긴다」와
    「관계가 저장된다」는 다른 말이다.
    """

    def _run(self, seeded, monkeypatch, chunk_cp=None):
        from app.core.steps.entity_relation_step import EntityRelationStep as R
        from app.services.checkpoint_sync.entity_sync_service import (
            EntitySyncService)
        from app.services.checkpoint_sync.relation_sync_service import (
            RelationSyncService)

        cp = chunk_cp or CHUNK_CP
        merged = _merge(cp)
        t2i = _t2i(merged, monkeypatch)
        _write_t2i(seeded, t2i)
        EntitySyncService(seeded["db"], seeded["pid"],
                          seeded["eid"]).sync_from_checkpoint()
        seeded["db"].flush()

        # ★관계도 **진짜 스텝**이 낸 것을 쓴다 — 손으로 안 만든다
        obj = _Runner({"grounding_chunk": cp})
        obj._chunk_projection = R._chunk_projection.__get__(obj)
        obj._execute = R._execute.__get__(obj)
        rel = obj._execute()
        _write(seeded["tmp"], seeded["pid"], seeded["eid"], "entity_relation",
               {"status": "completed", "data": rel["data"]})
        got = RelationSyncService(seeded["db"], seeded["pid"],
                                  seeded["eid"]).sync_from_checkpoint()
        seeded["db"].flush()
        return got

    def _count(self, seeded, rtype="part_of"):
        import sqlalchemy as sa

        return seeded["db"].execute(sa.text(
            "SELECT count(*) FROM relation_fact WHERE project_id = :pid "
            "AND relation_type = :rt"),
            {"pid": seeded["pid"], "rt": rtype}).scalar()

    def test_the_relation_lands_in_the_db(self, seeded, monkeypatch):
        self._run(seeded, monkeypatch)
        assert self._count(seeded) == 1

    def test_it_joins_the_right_two_rows(self, seeded, monkeypatch):
        import sqlalchemy as sa

        self._run(seeded, monkeypatch)
        # ★칼럼 이름을 **조립부에서 그대로** 가져온다 — 지어내면 못 찾은
        #  것을 「없다」로 읽게 된다.
        rows = seeded["db"].execute(sa.text(
            "SELECT c.short_id, rp.participant_role FROM relation_fact rf "
            "JOIN relation_participant rp ON rp.relation_id = rf.id "
            "JOIN entity_canon c ON c.id = rp.canon_id "
            "WHERE rf.project_id = :pid AND rf.relation_type = 'part_of'"),
            {"pid": seeded["pid"]}).fetchall()
        assert {(r[0], r[1]) for r in rows} == {("LP01", "part"),
                                                ("L01", "whole")}

    def test_a_same_named_pair_still_joins_correctly(self, seeded,
                                                     monkeypatch):
        """★★같은 이름이어도 **신원으로** 잇는다 — 이름으로 안 잇는다."""
        same = json.loads(json.dumps(CHUNK_CP))
        same["data"]["locations"][0]["name"] = "이발소"
        same["data"]["location_parts"][0]["name"] = "이발소"
        self._run(seeded, monkeypatch, chunk_cp=same)
        assert self._count(seeded) == 1


class TestResumeDoesNotRebuy:
    """★★★Codex 재현 (09-01) — 내가 새 열쇠를 넣고 stale 정리를 안 고쳐서,
    정상 재개가 **끝낸 유료 호출을 통째로 다시 샀다**.
    """

    def _first(self, monkeypatch, chunk_cp=None):
        calls = []
        got = _t2i(_merge(chunk_cp or CHUNK_CP), monkeypatch, counter=calls)
        return got, calls

    def test_a_full_resume_buys_nothing(self, monkeypatch):
        done, first = self._first(monkeypatch)
        assert len(first) == 4, "★첫 판이 넷을 안 샀다 — 시험이 죽었다"
        again = []
        _t2i(_merge(CHUNK_CP), monkeypatch, counter=again,
             resume_cp={"data": {"completed": done["data"]["completed"]}})
        assert again == [], f"★재개가 다시 샀다: {len(again)}회"

    def test_a_partial_resume_buys_only_the_rest(self, monkeypatch):
        done, _first = self._first(monkeypatch)
        completed = dict(done["data"]["completed"])
        dropped = sorted(completed)[0]
        completed.pop(dropped)
        again = []
        _t2i(_merge(CHUNK_CP), monkeypatch, counter=again,
             resume_cp={"data": {"completed": completed}})
        assert len(again) == 1, f"★남은 하나만 사야 한다: {len(again)}회"

    def test_a_same_named_partial_reuses_only_the_finished_lane(
            self, monkeypatch):
        """★★같은 이름의 두 갈래 중 **끝낸 쪽만** 되쓴다."""
        same = json.loads(json.dumps(CHUNK_CP))
        same["data"]["locations"][0]["name"] = "이발소"
        same["data"]["location_parts"][0]["name"] = "이발소"
        done, first = self._first(monkeypatch, chunk_cp=same)
        assert len(first) == 4
        completed = {k: v for k, v in done["data"]["completed"].items()
                     if (v or {}).get("entity_type") != "location_part"}
        assert len(completed) == 3, "★같은 이름 둘이 하나로 뭉쳤다"
        again = []
        _t2i(_merge(same), monkeypatch, counter=again,
             resume_cp={"data": {"completed": completed}})
        assert len(again) == 1
        assert "location_part" in again[0]

    def test_an_old_name_keyed_checkpoint_still_resumes(self, monkeypatch):
        """★옛 CP 는 이름만 갖고 있다 — 갈래가 맞으면 되쓴다."""
        done, _first = self._first(monkeypatch)
        legacy = {(v or {}).get("name") or k.split("\u0000")[-1]: v
                  for k, v in done["data"]["completed"].items()}
        again = []
        _t2i(_merge(CHUNK_CP), monkeypatch, counter=again,
             resume_cp={"data": {"completed": legacy}})
        assert again == [], f"★옛 CP 로 재개했더니 다시 샀다: {len(again)}"

    def _legacy_same_named(self, monkeypatch, *, finished):
        """옛 **이름 열쇠** CP 를 만든다 — 같은 이름의 두 갈래 중 `finished` 쪽만
        끝낸 것으로. 옛 CP 에는 이름이 하나뿐이라 나머지 갈래는 들어갈 자리가
        없다.
        """
        same = json.loads(json.dumps(CHUNK_CP))
        same["data"]["locations"][0]["name"] = "이발소"
        same["data"]["location_parts"][0]["name"] = "이발소"
        done, first = self._first(monkeypatch, chunk_cp=same)
        assert len(first) == 4, "★첫 판이 넷을 안 샀다 — 시험이 죽었다"
        legacy = {}
        for _k, v in done["data"]["completed"].items():
            etype = str((v or {}).get("entity_type") or "")
            name = str((v or {}).get("name") or "")
            if name == "이발소" and etype != finished:
                continue                    # ★옛 CP 에는 한 줄만 들어간다
            legacy[name] = v
        assert len(legacy) == 3, f"★옛 CP 를 잘못 지었다: {sorted(legacy)}"
        again = []
        _t2i(_merge(same), monkeypatch, counter=again,
             resume_cp={"data": {"completed": legacy}})
        return again

    def test_a_legacy_name_key_reuses_the_place_and_buys_only_the_part(
            self, monkeypatch):
        """★★★옛 이름 열쇠가 **장소** 완료분을 가리키면 그것을 되쓰고 부분만
        산다.

        Codex 재현 (2026-09-01) — 이름→갈래 **표**로 두면 같은 이름의 뒤엣것이
        앞엣것을 덮어, 멀쩡히 끝낸 장소가 stale 로 지워지고 **둘 다** 다시
        사진다.
        """
        again = self._legacy_same_named(monkeypatch, finished="location")
        assert len(again) == 1, f"★부분 하나만 사야 한다: {len(again)}회"
        assert "location_part" in again[0], "★산 것이 부분이 아니다"

    def test_a_legacy_name_key_reuses_the_part_and_buys_only_the_place(
            self, monkeypatch):
        """★반대 방향 — 옛 열쇠가 **부분** 완료분이면 장소만 산다."""
        again = self._legacy_same_named(monkeypatch, finished="location_part")
        assert len(again) == 1, f"★장소 하나만 사야 한다: {len(again)}회"
        # ★`location` 은 `location_part` 의 부분 문자열이다 — 없는 쪽으로 본다
        assert "location_part" not in again[0], "★산 것이 부분이다"
        assert "location" in again[0], "★산 것이 장소가 아니다"

    def test_a_stale_record_is_still_dropped(self, monkeypatch):
        """★음성 대조 — 지금 대상 밖의 기록은 **여전히 버린다**."""
        done, _first = self._first(monkeypatch)
        completed = dict(done["data"]["completed"])
        completed["prop\u0000없는 것"] = {"entity_type": "prop",
                                          "name": "없는 것"}
        again = []
        got = _t2i(_merge(CHUNK_CP), monkeypatch, counter=again,
                   resume_cp={"data": {"completed": completed}})
        assert again == []
        assert "없는 것" not in json.dumps(got["data"]["completed"],
                                          ensure_ascii=False)


class TestNothingOldIsCalledOnThisPath:
    """★★폭탄 대역 — 옛 모델 경계가 불리면 **터진다**."""

    def test_the_old_filter_provider_is_never_called(self, monkeypatch):
        from app.core.steps.entity_steps import EntityFilterStep as F

        def _boom(*a, **k):
            raise AssertionError("★옛 저빈도 필터를 불렀다")

        monkeypatch.setattr(
            "app.modules.pipeline.entity_filter.filter_low_frequency_entities",
            _boom, raising=False)
        obj = _Runner({"entity_merge": _merge(CHUNK_CP)})
        obj._chunk_projection = F._chunk_projection.__get__(obj)
        obj._execute = F._execute.__get__(obj)
        assert obj._execute()["data"]["filtered_entities"]

    def test_the_old_relation_provider_is_never_called(self, monkeypatch):
        from app.core.steps.entity_relation_step import EntityRelationStep as R

        def _boom(*a, **k):
            raise AssertionError("★옛 관계 추출을 불렀다")

        monkeypatch.setattr(
            "app.modules.pipeline.entity_relation.extract_entity_relations",
            _boom, raising=False)
        obj = _Runner({"grounding_chunk": CHUNK_CP})
        obj._chunk_projection = R._chunk_projection.__get__(obj)
        obj._execute = R._execute.__get__(obj)
        assert obj._execute()["data"]["relations"]


class TestAMissingMarkerFailsClosedInChunkMode:
    """★★★C(c) 판에서 표식 누락을 **정상 legacy 처럼 접으면** LP 만 조용히
    사라진다 (Codex 2026-09-01). 그때는 **서야** 한다.
    """

    def test_it_stops(self):
        from app.core.errors import AppError
        from app.core.steps.entity_steps import _carry_chunk_marker

        merged = _merge(CHUNK_CP)
        merged.pop(CHUNK_SCHEMA_MARKER)
        with pytest.raises(AppError, match="조용히 사라진다"):
            _carry_chunk_marker(_Runner({"entity_merge": merged}))

    def test_a_wrong_version_also_stops(self):
        from app.core.errors import AppError
        from app.core.steps.entity_steps import _carry_chunk_marker

        merged = _merge(CHUNK_CP)
        merged[CHUNK_SCHEMA_MARKER] = "99"
        with pytest.raises(AppError):
            _carry_chunk_marker(_Runner({"entity_merge": merged}))

    def test_legacy_stays_unchanged(self):
        """★음성 대조 — 옛 판은 표식이 없어도 **그대로 지나간다**."""
        from app.core.steps.entity_steps import _carry_chunk_marker

        class _Legacy(_Runner):
            project_config = {"grounding_mode": "legacy"}

        assert _carry_chunk_marker(_Legacy({})) == {}