"""§2-6.5 — `location_part` 가 **DB 까지** 간다. ★유료 0 · 공개 끝점.

Codex BLOCK (2026-09-01) — 「adapter 행 69→89 는 **DB materialization 증거가
아니다**. `EntitySyncService` 가 `location_parts` 를 한 줄도 안 읽는다. 게다가
조회만 넓히면 legacy CP 가 돌 때 stale cleanup 이 **기존 LP link 를 지운다** —
실제 데이터 손실이다.」

여기서 재는 것은 **공개 끝점**이다 — `sync_from_checkpoint()` 를 부르고 DB 를
본다. 조립 함수를 직접 부르지 않는다.
"""
from __future__ import annotations

import json
import uuid
from datetime import datetime, timezone
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}
LOC_META = {"location": {"space_profile": {
    "kind": "single_space", "allowed_space_keys": ["main"],
    "default_space_key": None}}, "visual_identity": None}


def _cp(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")


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


def _marked(data, *, marker=CHUNK_SCHEMA_VERSION):
    got = {"status": "completed", "data": data}
    if marker is not None:
        got[CHUNK_SCHEMA_MARKER] = marker
    return got


def _sync(db, pid, eid):
    from app.services.checkpoint_sync.entity_sync_service import (
        EntitySyncService)

    got = EntitySyncService(db, pid, eid).sync_from_checkpoint()
    db.flush()
    return got


@pytest.fixture
def seeded(tmp_path, monkeypatch):
    """test DB 에 fresh user/project + tmp projects_dir + rollback.

    ★`tests/core` 의 것과 같은 모양이다 — fixture 라 import 가 안 돼서 여기
    한 벌 더 둔다. 갈래 목록 같은 **계약**이 아니라 판을 차리는 도구다.
    """
    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-lp-{uuid.uuid4()}"
        eid = f"test-lp-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="lp-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 session, pid, eid, tmp_path
    finally:
        session.rollback()
        session.close()


class TestALocationPartReachesTheDatabase:
    def test_a_part_only_checkpoint_inserts_a_canon_and_a_link(self, seeded):
        from app.models.project import EntityCanon, EntityEpisodeLink

        db, pid, eid, tmp = seeded
        _cp(tmp, pid, eid, "entity_t2i", _marked({
            "characters": [], "locations": [], "props": [],
            "location_parts": [_ent("회전 간판", "LP01", NEUTRAL)]}))
        got = _sync(db, pid, eid)
        assert got["skipped"] == 0, "★LP 만 있는 CP 가 통째로 건너뛰어졌다"
        row = db.query(EntityCanon).filter_by(project_id=pid,
                                              short_id="LP01").first()
        assert row is not None, "★LP canon 이 안 생겼다"
        assert row.entity_type == "location_part", "★우회 등록됐다"
        link = db.query(EntityEpisodeLink).filter_by(
            project_id=pid, canon_id=row.id).first()
        assert link is not None, "★LP link 가 안 생겼다"

    def test_a_location_and_a_part_may_share_a_name(self, seeded):
        """★★이름 단독 열쇠면 둘이 **조용히 충돌**한다."""
        from app.models.project import EntityCanon

        db, pid, eid, tmp = seeded
        _cp(tmp, pid, eid, "entity_t2i", _marked({
            "characters": [], "props": [],
            "locations": [_ent("같은 이름", "L01", LOC_META)],
            "location_parts": [_ent("같은 이름", "LP01", NEUTRAL)]}))
        _sync(db, pid, eid)
        rows = db.query(EntityCanon).filter_by(
            project_id=pid, name="같은 이름").all()
        assert {r.entity_type for r in rows} == {"location", "location_part"}
        assert {r.short_id for r in rows} == {"L01", "LP01"}

    def test_the_neutral_shape_is_accepted(self, seeded):
        """★validator 가 LP 를 안 받으면 loop 를 여는 순간 즉사한다."""
        from app.core.entity_metadata import assert_matches_sync_owners

        assert_matches_sync_owners()


class TestALegacyCheckpointNeverDeletesAPartLink:
    """★★★데이터 손실 BLOCK — 조회만 넓히고 loop 를 안 넓혔을 때 나던 것.

    LP link 가 `_existing_links` 에는 들어오는데 `_seen_canon_ids` 에는 절대
    안 들어와, 표식 없는 legacy CP 가 **완료**로 돌면 그것을 지웠다.
    """

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

        db, pid, eid, tmp = seeded
        # ①새 CP 로 LP 를 만든다
        _cp(tmp, pid, eid, "entity_t2i", _marked({
            "characters": [], "props": [],
            "locations": [_ent("국밥집", "L01", LOC_META)],
            "location_parts": [_ent("회전 간판", "LP01", NEUTRAL)]}))
        _sync(db, pid, eid)
        lp = db.query(EntityCanon).filter_by(project_id=pid,
                                             short_id="LP01").first()
        assert lp is not None
        assert db.query(EntityEpisodeLink).filter_by(
            project_id=pid, canon_id=lp.id).first() is not None

        # ②표식 **없는** 옛 CP 가 completed 로 돈다
        _cp(tmp, pid, eid, "entity_t2i", {"status": "completed", "data": {
            "characters": [], "props": [],
            "locations": [_ent("국밥집", "L01", LOC_META)]}})
        _sync(db, pid, eid)

        assert db.query(EntityEpisodeLink).filter_by(
            project_id=pid, canon_id=lp.id).first() is not None, (
            "★legacy sync 가 LP link 를 지웠다 — 데이터 손실이다")

    def test_a_legacy_sync_still_cleans_its_own_stale_links(self, seeded):
        """★음성 대조 — 옛 갈래의 stale 정리는 **그대로 돈다**."""
        from app.models.project import EntityCanon, EntityEpisodeLink

        db, pid, eid, tmp = seeded
        _cp(tmp, pid, eid, "entity_t2i", {"status": "completed", "data": {
            "characters": [], "props": [],
            "locations": [_ent("국밥집", "L01", LOC_META),
                          _ent("이발소", "L02", LOC_META)]}})
        _sync(db, pid, eid)
        gone = db.query(EntityCanon).filter_by(project_id=pid,
                                               short_id="L02").first()
        assert gone is not None
        _cp(tmp, pid, eid, "entity_t2i", {"status": "completed", "data": {
            "characters": [], "props": [],
            "locations": [_ent("국밥집", "L01", LOC_META)]}})
        _sync(db, pid, eid)
        assert db.query(EntityEpisodeLink).filter_by(
            project_id=pid, canon_id=gone.id).first() is None


class TestAnUnknownMarkerStops:
    def test_it_refuses_instead_of_falling_back_to_legacy(self, seeded):
        from app.modules.pipeline.grounding_entity_sync_ext import (
            UnknownChunkSchema)

        db, pid, eid, tmp = seeded
        _cp(tmp, pid, eid, "entity_t2i", _marked({
            "characters": [], "locations": [], "props": []}, marker="99"))
        with pytest.raises(UnknownChunkSchema):
            _sync(db, pid, eid)


class TestThePartOfRelationReachesTheDatabase:
    """★facet verifier 의 bindings 는 **in-memory** 증거다 — DB 를 봐야 한다."""

    def _both(self, db, pid, eid, tmp):
        _cp(tmp, pid, eid, "entity_t2i", _marked({
            "characters": [], "props": [],
            "locations": [_ent("국밥집", "L01", LOC_META)],
            "location_parts": [_ent("회전 간판", "LP01", NEUTRAL)]}))
        _sync(db, pid, eid)

    def _relations(self, db, pid, eid, tmp, rows, status="completed"):
        from app.services.checkpoint_sync.relation_sync_service import (
            RelationSyncService)

        _cp(tmp, pid, eid, "entity_relation",
            {"status": status, "data": {"relations": rows}})
        got = RelationSyncService(db, pid, eid).sync_from_checkpoint()
        db.flush()
        return got

    def _part_of_rows(self):
        from app.modules.pipeline.grounding_relation_projection import (
            project_part_of)

        return project_part_of([{"part": "LP01", "whole": "L01"}])

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

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

    def test_it_writes_a_fact_with_part_and_whole_roles(self, seeded):
        """★반환 dict 이 아니라 **DB** 를 본다 — 끝점에서 잰다."""
        db, pid, eid, tmp = seeded
        self._both(db, pid, eid, tmp)
        self._relations(db, pid, eid, tmp, self._part_of_rows())
        assert self._count(db, pid) == 1
        rows = db.execute(__import__("sqlalchemy").text(
            "SELECT rf.relation_type, rp.participant_role, c.short_id "
            "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": pid}).fetchall()
        got_pairs = {(r[1], r[2]) for r in rows}
        assert got_pairs == {("part", "LP01"), ("whole", "L01")}, got_pairs

    def test_running_it_twice_does_not_duplicate(self, seeded):
        db, pid, eid, tmp = seeded
        self._both(db, pid, eid, tmp)
        self._relations(db, pid, eid, tmp, self._part_of_rows())
        first = self._count(db, pid)
        self._relations(db, pid, eid, tmp, self._part_of_rows())
        assert self._count(db, pid) == first == 1, "★두 번 돌려 늘었다"

    def test_part_of_sync_does_not_delete_visual_variants(self, seeded):
        """★★한 타입의 delta 가 다른 타입을 **stale 로 지우면 안 된다**."""
        import sqlalchemy as sa

        db, pid, eid, tmp = seeded
        self._both(db, pid, eid, tmp)
        mixed = self._part_of_rows() + [{
            "base_short_id": "L01", "variant_short_id": "L01",
            "visual_similarity": True, "reason": "그대로 둔다"}]
        # ★visual_variant 는 같은 canon 끼리라도 delta 기구를 지난다
        self._relations(db, pid, eid, tmp, mixed)
        n_vv = db.execute(sa.text(
            "SELECT count(*) FROM relation_fact WHERE project_id = :pid "
            "AND relation_type = 'visual_variant'"), {"pid": pid}).scalar()
        assert n_vv == 1
        # part_of 만 있는 CP 로 다시 돌려도 visual_variant 가 살아 있어야 한다
        self._relations(db, pid, eid, tmp, self._part_of_rows())
        still = db.execute(sa.text(
            "SELECT count(*) FROM relation_fact WHERE project_id = :pid "
            "AND relation_type = 'visual_variant'"), {"pid": pid}).scalar()
        assert still == 1, "★part_of sync 가 visual_variant 를 지웠다"

    def test_a_stale_part_of_is_removed_when_the_cp_speaks_of_it(self,
                                                                 seeded):
        """★그 타입을 **말한** CP 라야 지운다 — 다른 짝으로 바꿔서 잰다."""
        import sqlalchemy as sa

        from app.modules.pipeline.grounding_relation_projection import (
            project_part_of)

        db, pid, eid, tmp = seeded
        _cp(tmp, pid, eid, "entity_t2i", _marked({
            "characters": [], "props": [],
            "locations": [_ent("국밥집", "L01", LOC_META),
                          _ent("이발소", "L02", LOC_META)],
            "location_parts": [_ent("회전 간판", "LP01", NEUTRAL)]}))
        _sync(db, pid, eid)
        self._relations(db, pid, eid, tmp, self._part_of_rows())
        # ★같은 LP 를 **다른 부모**로 옮긴다 — 옛 짝은 stale 이다
        self._relations(db, pid, eid, tmp,
                        project_part_of([{"part": "LP01", "whole": "L02"}]))
        rows = db.execute(sa.text(
            "SELECT c.short_id 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' "
            "AND rp.participant_role = 'whole'"), {"pid": pid}).fetchall()
        assert [r[0] for r in rows] == ["L02"], rows

    def test_a_visual_variant_only_cp_does_not_delete_part_of(self, seeded):
        """★★반대 방향 — visual_variant 만 담은 CP 가 `part_of` 를 지우면 안 된다."""
        import sqlalchemy as sa

        db, pid, eid, tmp = seeded
        self._both(db, pid, eid, tmp)
        self._relations(db, pid, eid, tmp, self._part_of_rows())
        self._relations(db, pid, eid, tmp, [{
            "base_short_id": "L01", "variant_short_id": "L01",
            "visual_similarity": True, "reason": "그대로"}])
        n = db.execute(sa.text(
            "SELECT count(*) FROM relation_fact WHERE project_id = :pid "
            "AND relation_type = 'part_of'"), {"pid": pid}).scalar()
        assert n == 1, "★visual_variant CP 가 part_of 를 지웠다"

    def test_a_partial_checkpoint_never_deletes(self, seeded):
        import sqlalchemy as sa

        db, pid, eid, tmp = seeded
        self._both(db, pid, eid, tmp)
        self._relations(db, pid, eid, tmp, self._part_of_rows())
        self._relations(db, pid, eid, tmp, [], status="partial")
        n = db.execute(sa.text(
            "SELECT count(*) FROM relation_fact WHERE project_id = :pid "
            "AND relation_type = 'part_of'"), {"pid": pid}).scalar()
        assert n == 1


class TestTheStartupMigrationSurvivesATwoLetterPrefix:
    """★★★부팅이 통째로 죽던 자리 (실측 2026-09-01).

    `init_db()` 의 이관 하나가 `SUBSTRING(short_id FROM 2)` 로 **한 글자만**
    뗐다. 접두가 두 글자인 `LP01` 이 있으면 `'P01'` 이 되어 정수 변환이 터지고,
    startup 이관은 **fail-fast** 라 백엔드가 안 뜬다.
    """

    def test_init_db_runs_again_with_a_two_letter_prefix_present(self, seeded):
        from app.core.database import init_db

        db, pid, eid, tmp = seeded
        _cp(tmp, pid, eid, "entity_t2i", _marked({
            "characters": [], "locations": [], "props": [],
            "location_parts": [_ent("회전 간판", "LP01", NEUTRAL)]}))
        _sync(db, pid, eid)
        db.commit()                     # ★이관이 볼 수 있게 실제로 남긴다
        try:
            init_db()                   # ★터지면 부팅이 안 된다
        finally:
            from app.models.project import EntityCanon, EntityEpisodeLink
            for row in db.query(EntityEpisodeLink).filter_by(
                    project_id=pid).all():
                db.delete(row)
            for row in db.query(EntityCanon).filter_by(project_id=pid).all():
                db.delete(row)
            db.commit()

    def test_the_migration_knows_the_two_letter_prefix(self):
        """★`CASE` 에 갈래가 없으면 그 행은 조용히 `short_id = NULL` 이 된다.

        ★★모듈 소스를 글자로 보면 **이 결함을 설명한 주석**이 걸린다 — 실제로
        걸렸다. 그래서 **이관 SQL 목록 자체**를 본다.
        """
        import ast
        import inspect
        import textwrap

        from app.core import database as dbmod

        tree = ast.parse(textwrap.dedent(inspect.getsource(dbmod.init_db)))
        sql = [n.value for n in ast.walk(tree)
               if isinstance(n, ast.Constant) and isinstance(n.value, str)]
        issuing = [q for q in sql if "SET short_id" in q]
        assert issuing, "★short_id 발급 이관을 못 찾았다"
        for q in issuing:
            assert "'location_part' THEN 'LP'" in q, "★갈래가 빠졌다"
            assert "SUBSTRING(short_id FROM 2)" not in q, (
                "★접두가 한 글자라고 가정한다")


class TestTheMigrationChangeDoesNotRenumberExistingOwners:
    """★★Codex 가 짚은 네 번째 경계 (2026-09-01) — 「LP 접두 이관이 기존
    C/L/O/P 번호를 바꾸지 않는지」.

    `SUBSTRING(short_id FROM 2)` → `SUBSTRING(short_id FROM '[0-9]+$')` 는
    **접두가 한 글자면 같은 값**이다. 원본 DB 3825행 전부에서 두 식이 같은
    값을 낸다고 실측했다. 여기서는 그 등가를 **코드로** 잠근다.
    """

    def test_both_expressions_agree_on_every_standard_id(self, seeded):
        import sqlalchemy as sa

        db, pid, eid, tmp = seeded
        ids = ["C01", "L01", "P07", "O12", "C99", "L100", "LP01", "LP42"]
        rows = db.execute(sa.text(
            "SELECT x, SUBSTRING(x FROM 2) AS old, "
            "       SUBSTRING(x FROM '[0-9]+$') AS new "
            "FROM unnest(CAST(:ids AS text[])) AS t(x)"),
            {"ids": ids}).fetchall()
        got = {r[0]: (r[1], r[2]) for r in rows}
        for sid in ("C01", "L01", "P07", "O12", "C99", "L100"):
            assert got[sid][0] == got[sid][1], (sid, got[sid])
        # ★두 글자 접두에서만 갈린다 — 그 자리가 옛 식이 터지던 곳이다
        assert got["LP01"] == ("P01", "01")
        assert got["LP42"] == ("P42", "42")

    def test_the_old_expression_would_have_crashed(self, seeded):
        """★양성 대조 — 옛 식이 실제로 터지는지 본다."""
        import sqlalchemy as sa

        db, pid, eid, tmp = seeded
        with pytest.raises(Exception):
            db.execute(sa.text(
                "SELECT CAST(SUBSTRING('LP01' FROM 2) AS INTEGER)")).scalar()
        db.rollback()
        assert db.execute(sa.text(
            "SELECT CAST(SUBSTRING('LP01' FROM '[0-9]+$') AS INTEGER)")
        ).scalar() == 1

    def test_a_non_numeric_tail_is_skipped_instead_of_crashing(self, seeded):
        """★끝이 숫자가 아니면 **NULL** 이라 `MAX` 가 그냥 건너뛴다."""
        import sqlalchemy as sa

        db, pid, eid, tmp = seeded
        assert db.execute(sa.text(
            "SELECT SUBSTRING('L1_old' FROM '[0-9]+$')")).scalar() is None


class TestTheGuardIsActuallyCalled:
    """★★시험만 부르는 가드는 **가드가 아니다** (Codex 가 `callers 0` 을 짚었다).

    갈래 목록이 어긋나면 그 갈래 **첫 행**에서 죽는데, 그때는 이미 앞 갈래를
    반쯤 쓴 뒤다. 죽기 전에 서야 한다.
    """

    def test_production_calls_it_before_the_loop(self):
        import ast
        import inspect
        import textwrap

        from app.services.checkpoint_sync import entity_sync_service as es

        src = inspect.getsource(es.EntitySyncService.sync_from_checkpoint)
        tree = ast.parse(textwrap.dedent(src))
        calls = [ast.unparse(n) for n in ast.walk(tree)
                 if isinstance(n, ast.Call)]
        assert "assert_matches_sync_owners()" in calls, (
            "★production 이 이 가드를 안 부른다")
        assert (src.index("assert_matches_sync_owners()")
                < src.index("for etype, singular, prefix in _TYPE_DEFS"))

    def test_it_fires_when_the_validator_falls_behind(self, monkeypatch):
        """★양성 대조 — validator 가 갈래를 못 따라가면 **선다**."""
        from app.core import entity_metadata as em

        monkeypatch.setattr(em, "_ALLOWED_ENTITY_TYPES",
                            frozenset({"character", "location", "prop"}))
        with pytest.raises(AssertionError, match="location_part"):
            em.assert_matches_sync_owners()


class TestAMalformedRelationNeverReachesTheDatabase:
    """★★★Codex BLOCK 1 (2026-09-01) — `desired_keys` 가 갈래 짝을 **안 봤다**.

    투영(`project_part_of`)만 검사하고 저장 자리는 참가자 역할·존재만 봐서,
    망가진 CP 의 `C01(part) → P01(whole)` 도 양쪽 canon 만 있으면 DB 에
    `part_of` 로 저장됐다.
    """

    def _seed_all(self, db, pid, eid, tmp):
        _cp(tmp, pid, eid, "entity_t2i", _marked({
            "characters": [_ent("이발사", "C01", NEUTRAL)],
            "props": [_ent("됫박", "P01", {"location": None,
                                          "visual_identity": {
                                              "reference_required": False}})],
            "locations": [_ent("국밥집", "L01", LOC_META)],
            "location_parts": [_ent("회전 간판", "LP01", NEUTRAL)]}))
        _sync(db, pid, eid)

    def _raw(self, part, whole):
        """★투영을 **거치지 않은** 행 — 망가진 CP 를 흉내 낸다."""
        return [{"relation_type": "part_of", "participants": [
            {"short_id": part, "role": "part", "order": 1},
            {"short_id": whole, "role": "whole", "order": 2}]}]

    def _run(self, db, pid, eid, tmp, rows):
        from app.services.checkpoint_sync.relation_sync_service import (
            RelationSyncService)

        _cp(tmp, pid, eid, "entity_relation",
            {"status": "completed", "data": {"relations": rows}})
        return RelationSyncService(db, pid, eid).sync_from_checkpoint()

    def _count(self, db, pid):
        import sqlalchemy as sa

        return db.execute(sa.text(
            "SELECT count(*) FROM relation_fact WHERE project_id = :pid "
            "AND relation_type = 'part_of'"), {"pid": pid}).scalar()

    @pytest.mark.parametrize("part,whole", [
        ("C01", "P01"),      # ★character 는 part_of 로 안 잇는다
        ("LP01", "P01"),     # ★부모가 location 이 아니다
        ("P01", "L01"),      # ★prop 은 part 가 못 된다
        ("L01", "L01"),      # ★제 자신
        ("LPfoo", "L01"),    # ★정본 ID 가 아니다
        ("", "L01"),         # ★빈 것
    ])
    def test_it_is_refused_before_any_row_is_written(self, seeded, part,
                                                      whole):
        from app.modules.pipeline.grounding_relation_projection import (
            RelationProjectionError)

        db, pid, eid, tmp = seeded
        self._seed_all(db, pid, eid, tmp)
        with pytest.raises(RelationProjectionError):
            self._run(db, pid, eid, tmp, self._raw(part, whole))
        db.rollback()
        assert self._count(db, pid) == 0


class TestAMissingCanonStopsBeforeDeleting:
    """★★★Codex BLOCK 2 (2026-09-01) — 경고하고 넘어간 뒤 **멀쩡한 관계를
    지웠다**. 「못 찾았다」를 「없어졌다」로 읽으면 데이터 손실이다."""

    def test_it_raises_and_keeps_the_existing_relation(self, seeded):
        import sqlalchemy as sa

        from app.modules.pipeline.grounding_relation_projection import (
            project_part_of)
        from app.services.checkpoint_sync.relation_sync_service import (
            RelationSyncRefused, RelationSyncService)

        db, pid, eid, tmp = seeded
        _cp(tmp, pid, eid, "entity_t2i", _marked({
            "characters": [], "props": [],
            "locations": [_ent("국밥집", "L01", LOC_META)],
            "location_parts": [_ent("회전 간판", "LP01", NEUTRAL)]}))
        _sync(db, pid, eid)
        rows = project_part_of([{"part": "LP01", "whole": "L01"}])
        _cp(tmp, pid, eid, "entity_relation",
            {"status": "completed", "data": {"relations": rows}})
        RelationSyncService(db, pid, eid).sync_from_checkpoint()
        db.flush()
        before = db.execute(sa.text(
            "SELECT count(*) FROM relation_fact WHERE project_id = :pid "
            "AND relation_type = 'part_of'"), {"pid": pid}).scalar()
        assert before == 1

        # ★없는 canon 을 가리키는 CP — 지우기 전에 서야 한다
        gone = project_part_of([{"part": "LP99", "whole": "L01"}])
        _cp(tmp, pid, eid, "entity_relation",
            {"status": "completed", "data": {"relations": gone}})
        with pytest.raises(RelationSyncRefused, match="canon 이 없다"):
            RelationSyncService(db, pid, eid).sync_from_checkpoint()
        after = db.execute(sa.text(
            "SELECT count(*) FROM relation_fact WHERE project_id = :pid "
            "AND relation_type = 'part_of'"), {"pid": pid}).scalar()
        assert after == 1, "★기존 관계가 지워졌다"