"""앞 화 명부로 **같은 것을 같은 신원으로** 잇는다 (2026-09-04).

번호를 프로젝트 장부에서 발급하면 덮어쓰기는 없어지지만, 같은 인물이 화마다
**새 번호**를 받는다 — 실측 `da049582` 에서 같은 사람이 `C02 기사 최씨`(3화)와
`C06 최씨`(2화)로 두 행이 됐다.

## 여기서 재는 것

1. 명부가 비면(첫 화) **아무것도 안 바뀐다** — 프롬프트도 스키마도
2. 모델이 명부의 것을 고르면 그 신원을 **물려받는다**
3. 명부 **밖** 값은 인정 안 한다 → 새것
4. 둘이 같은 것을 주장하면 **둘 다 새것**(fail-closed) — 잘못 합치지 않는다
5. 명부에 **별명**과 **구조 앵커**가 실린다
6. `O00` 은 명부에 안 실린다 (예약값)
7. 지시문이 **팩**에서 온다 — 소스에 안 박는다

Lane: ``-m pg``.
"""
from __future__ import annotations

import uuid

import pytest
from sqlalchemy import text as sql_text

pytestmark = pytest.mark.pg


def _seed(session, pid: str) -> None:
    uid = f"carry-{uuid.uuid4()}"
    session.execute(sql_text(
        "INSERT INTO user_account (id, username, display_name, password_hash, "
        "role, is_active, created_at, updated_at) VALUES "
        "(:uid, :un, 't', 'x', 'creator', 1, '2026-01-01', '2026-01-01')"
    ), {"uid": uid, "un": f"u_{uid}"})
    session.execute(sql_text(
        "INSERT INTO project_registry (id, name, created_by, created_at, updated_at) "
        "VALUES (:pid, 'carry', :uid, '2026-01-01', '2026-01-01')"
    ), {"pid": pid, "uid": uid})
    session.commit()


def _canon(session, pid: str, short_id: str, name: str, etype: str,
           desc: str = "") -> str:
    cid = str(uuid.uuid4())
    session.execute(sql_text(
        "INSERT INTO entity_canon (id, project_id, short_id, name, entity_type, "
        "description, stable_traits, metadata_json, t2i_prompt, status, "
        "created_at, updated_at) VALUES "
        "(:cid, :pid, :s, :n, :e, :d, '[]', '{}', '', 'active', "
        "'2026-01-01', '2026-01-01')"
    ), {"cid": cid, "pid": pid, "s": short_id, "n": name, "e": etype, "d": desc})
    return cid


@pytest.fixture
def proj(pg_session):
    pid = f"p-{uuid.uuid4()}"
    _seed(pg_session, pid)
    return pid


# ── 명부 ─────────────────────────────────────────────────────────────


def test_첫_화는_한_바이트도_안_바뀐다(pg_session, proj):
    """★명부가 비면 블록도 스키마 패치도 통째로 안 붙는다."""
    from app.modules.pipeline.episode_carry import (build_roster_block,
                                                    patch_schema_with_prior_ids)

    block, allowed = build_roster_block(pg_session, proj, "character")
    assert block == "" and allowed == []

    schema = {"properties": {"characters": {"type": "array",
                                            "items": {"properties": {"name": {}},
                                                      "required": ["name"]}}}}
    import copy

    before = copy.deepcopy(schema)
    assert patch_schema_with_prior_ids(schema, allowed) == before, "스키마가 바뀌었다"


def test_명부에_별명과_구조_앵커가_실린다(pg_session, proj):
    """이름만으로는 같은 이름의 「문」·「바닥」이 안 갈린다.

    ★앵커는 **화 범위**다 (Codex BLOCK 2026-09-04) — 그러니 이 시험도 화와
     링크를 갖춰야 한다. 안 갖추면 앵커가 비는 것이 **맞는 동작**이다.
    """
    from app.modules.pipeline.episode_carry import build_prior_roster

    # ★명부는 **뒤 화가 앞 화를 보는 것**이다 — 1화에 심고 2화에서 본다.
    eid, nxt = str(uuid.uuid4()), str(uuid.uuid4())
    for _e, _n in ((eid, 1), (nxt, 2)):
        pg_session.execute(sql_text(
            "INSERT INTO episode (id, project_id, title, episode_number, "
            "source_filename, source_path, created_at, updated_at) VALUES "
            "(:e, :p, :t, :n, 'x', 'x', 'x', 'x')"),
            {"e": _e, "p": proj, "t": f"{_n}화", "n": _n})
    char = _canon(pg_session, proj, "C01", "최씨", "character")
    ol = _canon(pg_session, proj, "O01", "작업점퍼", "outlook", "낡은 점퍼")
    for _c in (char, ol):
        pg_session.execute(sql_text(
            "INSERT INTO entity_episode_link (id, canon_id, project_id, "
            "episode_id) VALUES (:l, :c, :p, :e)"),
            {"l": str(uuid.uuid4()), "c": _c, "p": proj, "e": eid})
    pg_session.execute(sql_text(
        "INSERT INTO entity_alias (id, canon_id, alias) VALUES (:i, :c, :a)"
    ), {"i": str(uuid.uuid4()), "c": ol, "a": "낡은작업점퍼"})
    pg_session.execute(sql_text(
        "INSERT INTO character_outlook (id, project_id, episode_id, character_id, "
        "outlook_id, created_at) VALUES (:i, :p, :e, :c, :o, '2026-01-01')"
    ), {"i": str(uuid.uuid4()), "p": proj, "e": eid, "c": char, "o": ol})
    pg_session.commit()

    lines, allowed = build_prior_roster(pg_session, proj, "outlook", nxt)
    assert allowed == ["O01"]
    line = lines[0]
    assert "낡은작업점퍼" in line, f"별명이 안 실렸다: {line}"
    assert "C01" in line, f"구조 앵커(입는 인물)가 안 실렸다: {line}"
    assert "낡은 점퍼" in line, f"설명이 안 실렸다: {line}"


def test_O00_은_명부에_안_실린다(pg_session, proj):
    """예약값은 실체가 아니라 「배정 없음」 표식이다."""
    from app.core.entity_identity import NULL_OUTLOOK_SHORT_ID
    from app.modules.pipeline.episode_carry import build_prior_roster

    _canon(pg_session, proj, NULL_OUTLOOK_SHORT_ID, "Null Outlook", "outlook")
    _canon(pg_session, proj, "O01", "작업점퍼", "outlook")
    pg_session.commit()

    _lines, allowed = build_prior_roster(pg_session, proj, "outlook")
    assert allowed == ["O01"], f"예약값이 명부에 실렸다: {allowed}"


def test_지시문이_팩에서_온다(pg_session, proj):
    """소스에 박으면 버전도 hash 도 audit 도 없다."""
    from app.modules.pipeline.episode_carry import (PROMPT_PACK_VERSION,
                                                    build_roster_block,
                                                    pack_fingerprint)

    _canon(pg_session, proj, "C01", "정임", "character")
    pg_session.commit()

    block, allowed = build_roster_block(pg_session, proj, "character")
    assert allowed == ["C01"] and block.strip(), "명부가 안 붙었다"
    fp = pack_fingerprint()
    assert fp["carry_pack"] == PROMPT_PACK_VERSION
    assert len(fp["carry_pack_hash"]) == 16, "팩 지문이 없다"


# ── 대조 ─────────────────────────────────────────────────────────────


def test_명부의_것을_고르면_신원을_물려받는다():
    from app.modules.pipeline.episode_carry import (CARRY_REUSED, FIELD,
                                                    apply_prior_ids)

    rows = [{"name": "기사 최씨", FIELD: "C02"}]
    out = apply_prior_ids(rows, ["C01", "C02"])

    assert rows[0]["short_id"] == "C02", "앞 화 신원을 안 물려받았다"
    assert FIELD not in rows[0], "모델 응답 칸이 산출에 남았다"
    assert out["counts"][CARRY_REUSED] == 1


def test_명부_밖_값은_인정하지_않는다():
    """모델이 없는 ID 를 지어내도 남의 행을 덮으면 안 된다."""
    from app.modules.pipeline.episode_carry import (CARRY_REJECTED, FIELD,
                                                    apply_prior_ids)

    rows = [{"name": "누구", FIELD: "C99"}]
    out = apply_prior_ids(rows, ["C01"])

    assert not rows[0].get("short_id"), (
        f"명부에 없는 C99 를 신원으로 받았다: {rows[0]}")
    assert out["counts"][CARRY_REJECTED] == 1


def test_둘이_같은_것을_주장하면_둘_다_새것(pg_session):
    """★fail-closed — 잘못 합치는 것보다 잘못 쪼개는 편이 안전하다."""
    from app.modules.pipeline.episode_carry import (CARRY_CONTESTED, FIELD,
                                                    apply_prior_ids)

    rows = [{"name": "최씨", FIELD: "C02"}, {"name": "기사 최씨", FIELD: "C02"}]
    out = apply_prior_ids(rows, ["C02"])

    assert not rows[0].get("short_id") and not rows[1].get("short_id"), (
        f"다툰 신원을 한쪽에 줬다: {rows}")
    assert out["counts"][CARRY_CONTESTED] == 1
    assert out["ledger"]["C02"] == CARRY_CONTESTED


def test_NEW_는_새것이다():
    from app.modules.pipeline.episode_carry import (CARRY_NEW, FIELD,
                                                    NEW_SENTINEL,
                                                    apply_prior_ids)

    rows = [{"name": "처음보는사람", FIELD: NEW_SENTINEL}]
    out = apply_prior_ids(rows, ["C01"])

    assert not rows[0].get("short_id")
    assert out["counts"][CARRY_NEW] == 1


def test_스키마에_명부_외의_값은_못_들어간다():
    """runtime enum — 허용 목록은 그 호출에 실린 명부 + NEW 뿐이다."""
    from app.modules.pipeline.episode_carry import (FIELD, NEW_SENTINEL,
                                                    patch_schema_with_prior_ids)

    schema = {"properties": {"characters": {"type": "array",
                                            "items": {"properties": {"name": {}},
                                                      "required": ["name"]}}}}
    out = patch_schema_with_prior_ids(schema, ["C01", "C02"])
    enum = out["properties"]["characters"]["items"]["properties"][FIELD]["enum"]

    assert enum == ["C01", "C02", NEW_SENTINEL]
    assert FIELD in out["properties"]["characters"]["items"]["required"]


# ── 끝점: 물려받은 신원은 발급기가 안 덮는다 ─────────────────────────


def test_물려받은_신원_위에_발급기가_새_번호를_안_준다(pg_session, proj):
    """★두 단계가 이어져야 뜻이 있다 — 대조 뒤에 발급이 돈다."""
    from app.core.entity_identity import assign_short_ids
    from app.modules.pipeline.episode_carry import FIELD, apply_prior_ids

    _canon(pg_session, proj, "C01", "정임", "character")
    _canon(pg_session, proj, "C02", "최씨", "character")
    pg_session.commit()

    rows = [{"name": "기사 최씨", FIELD: "C02"}, {"name": "새 인물", FIELD: "NEW"}]
    apply_prior_ids(rows, ["C01", "C02"])
    assign_short_ids(pg_session, proj, "character", rows)

    assert rows[0]["short_id"] == "C02", "물려받은 신원을 발급기가 덮었다"
    assert rows[1]["short_id"] == "C03", f"새것이 이어서 안 나왔다: {rows[1]}"


# ── 아웃룩도 같은 계약 (Codex 2026-09-04) ────────────────────────────


def test_아웃룩_phase1_도_앞_화_명부를_받는다(pg_session, proj, monkeypatch, tmp_path):
    """★발급기만 바꾸면 충돌은 막아도 **같은 옷의 재사용**은 못 한다.

    화마다 새 번호를 받아 O01 남색작업복(1화)과 O05 남색작업복(2화)이 따로
    생긴다. 명부를 줘야 모델이 「그 옷」이라고 고를 수 있다.
    """
    from app.modules.pipeline import outlook_extractor_v2 as ox
    from app.modules.pipeline.episode_carry import (FIELD, build_roster_block)

    _canon(pg_session, proj, "C01", "최씨", "character")
    ol = _canon(pg_session, proj, "O01", "작업점퍼", "outlook", "낡은 점퍼")
    char = pg_session.execute(sql_text(
        "SELECT id FROM entity_canon WHERE project_id = :p AND short_id = 'C01'"
    ), {"p": proj}).fetchone()[0]
    pg_session.execute(sql_text(
        "INSERT INTO character_outlook (id, project_id, character_id, outlook_id, "
        "created_at) VALUES (:i, :p, :c, :o, '2026-01-01')"
    ), {"i": str(uuid.uuid4()), "p": proj, "c": char, "o": ol})
    pg_session.commit()

    seen = {}

    def _capture(**kw):
        seen.update(kw)
        return {"outlooks": [{"name": "낡은작업점퍼", "description": "d",
                              "character_id": "C01", FIELD: "O01"}]}

    monkeypatch.setattr(ox, "call_structured", _capture)
    monkeypatch.setattr(ox, "load_prompt", lambda *a, **k: "p")
    monkeypatch.setattr(ox, "load_schema", lambda *a, **k: {
        "type": "object",
        "properties": {"outlooks": {"type": "array", "items": {
            "type": "object", "properties": {"name": {"type": "string"}},
            "required": ["name"], "additionalProperties": False}}},
        "required": ["outlooks"], "additionalProperties": False})

    carry: dict = {}
    out = ox.extract_outlooks_phase1(
        segments=[{"scene_index": 1, "text": "t"}],
        characters=[{"short_id": "C01", "name": "최씨"}],
        scene_character_map={1: ["C01"]},
        prior_roster=build_roster_block(pg_session, proj, "outlook"),
        carry_out=carry,
    )

    assert "O01" in seen["user_prompt"], "앞 화 아웃룩 명부가 프롬프트에 안 실렸다"
    assert "C01" in seen["user_prompt"], "입는 인물(구조 앵커)이 안 실렸다"
    assert out["outlooks"][0]["short_id"] == "O01", (
        "이름이 달라졌다고 앞 화 옷을 새것으로 만들었다")
    assert carry["counts"]["reused"] == 1


# ── 지문 (Codex ㉱2) ──────────────────────────────────────────────────


def test_첫_화는_지문이_안_움직인다(pg_session, proj, tmp_path, monkeypatch):
    """★명부가 비면 프롬프트도 스키마도 안 바뀌니 지문도 그대로여야 한다.

    움직이면 「아무것도 안 바뀐다」가 거짓이 되고 멀쩡한 옛 체크포인트를
    전부 다시 태운다.
    """
    monkeypatch.setattr("app.core.config.settings.projects_dir", str(tmp_path))
    from app.core.step_runner import compute_config_hash
    from app.core.steps import STEP_CLASSES

    runner = STEP_CLASSES["entity_all_character"](
        step_id="entity_all_character", project_id=proj, episode_id="e1",
        db=pg_session, project_config={})
    assert runner._config_hash() == compute_config_hash({}), (
        "명부가 없는데 지문이 움직였다")


def test_명부가_생기면_지문이_움직인다(pg_session, proj, tmp_path, monkeypatch):
    """★안 움직이면 계약·팩을 고쳐도 resume 이 옛 산출을 그대로 쓴다."""
    monkeypatch.setattr("app.core.config.settings.projects_dir", str(tmp_path))
    from app.core.step_runner import compute_config_hash
    from app.core.steps import STEP_CLASSES

    _canon(pg_session, proj, "C01", "민수", "character")
    pg_session.commit()

    # ★명부는 **화 범위**이고 **앞 화**의 것이다 — 1화에 심고 2화가 본다.
    eid, nxt = str(uuid.uuid4()), str(uuid.uuid4())
    for _e, _n in ((eid, 1), (nxt, 2)):
        pg_session.execute(sql_text(
            "INSERT INTO episode (id, project_id, title, episode_number, "
            "source_filename, source_path, created_at, updated_at) VALUES "
            "(:e, :p, :t, :n, 'x', 'x', 'x', 'x')"),
            {"e": _e, "p": proj, "t": f"{_n}화", "n": _n})
    # ★갈래마다 제 명부를 본다 — 인물 스텝은 인물, 아웃룩 스텝은 아웃룩.
    ol = _canon(pg_session, proj, "O01", "작업점퍼", "outlook")
    cid = pg_session.execute(sql_text(
        "SELECT id FROM entity_canon WHERE project_id = :p AND short_id = 'C01'"
    ), {"p": proj}).fetchone()[0]
    for _c in (cid, ol):
        pg_session.execute(sql_text(
            "INSERT INTO entity_episode_link (id, canon_id, project_id, episode_id) "
            "VALUES (:l, :c, :p, :e)"),
            {"l": str(uuid.uuid4()), "c": _c, "p": proj, "e": eid})
    pg_session.commit()

    for step in ("entity_all_character", "outlook_phase1"):
        runner = STEP_CLASSES[step](
            step_id=step, project_id=proj, episode_id=nxt,
            db=pg_session, project_config={})
        assert runner._config_hash() != compute_config_hash({}), (
            f"{step}: 명부가 있는데 지문이 안 움직였다")


def test_앞_화만_본다_뒤_화_신원은_안_샌다(pg_session, proj):
    """★1화를 다시 분석할 때 2·3화 신원이 **과거로 새면** 안 된다."""
    from app.modules.pipeline.episode_carry import build_prior_roster

    ep1, ep2 = str(uuid.uuid4()), str(uuid.uuid4())
    for eid, num in ((ep1, 1), (ep2, 2)):
        pg_session.execute(sql_text(
            "INSERT INTO episode (id, project_id, title, episode_number, "
            "source_filename, source_path, created_at, updated_at) VALUES "
            "(:e, :p, :t, :n, 'x', 'x', 'x', 'x')"),
            {"e": eid, "p": proj, "t": f"{num}화", "n": num})
    c1 = _canon(pg_session, proj, "C01", "민수", "character")
    c2 = _canon(pg_session, proj, "C02", "정임", "character")
    _lone = _canon(pg_session, proj, "C09", "고아", "character")
    for cid, eid in ((c1, ep1), (c2, ep2)):
        pg_session.execute(sql_text(
            "INSERT INTO entity_episode_link (id, canon_id, project_id, episode_id) "
            "VALUES (:l, :c, :p, :e)"),
            {"l": str(uuid.uuid4()), "c": cid, "p": proj, "e": eid})
    pg_session.commit()

    _l1, a1 = build_prior_roster(pg_session, proj, "character", ep1)
    assert a1 == [], f"1화 명부에 뒤 화·고아·제 것이 샜다: {a1}"

    _l2, a2 = build_prior_roster(pg_session, proj, "character", ep2)
    assert a2 == ["C01"], f"2화 명부가 앞 화만이 아니다: {a2}"

    # ★이 화 자신은 **force 로 다시 뜰 때만** — 늘 넣으면 얼리는 시점에 따라
    #  명부가 달라져서 이미 분석된 화가 재개할 때마다 유료로 다시 산다.
    _l3, a3 = build_prior_roster(pg_session, proj, "character", ep2,
                                 include_self=True)
    assert a3 == ["C01", "C02"], f"force 인데 자기 자신이 빠졌다: {a3}"


# ── 스냅샷 (Codex BLOCK 재지적 2026-09-04) ───────────────────────────


def test_이_화가_canon_을_만들어도_지문이_안_흔들린다(pg_session, proj, tmp_path,
                                                monkeypatch):
    """★★★명부를 매번 현재 DB 에서 읽으면, 이 화가 제 canon 을 만든 뒤
    같은 체크포인트의 지문이 **스스로 어긋난다** — 그러면 재개가 유료 상류를
    다시 산다. 얼린 것을 봐야 한다.
    """
    monkeypatch.setattr("app.core.config.settings.projects_dir", str(tmp_path))
    from app.modules.pipeline.episode_carry import roster_digest

    eid = str(uuid.uuid4())
    pg_session.execute(sql_text(
        "INSERT INTO episode (id, project_id, title, episode_number, "
        "source_filename, source_path, created_at, updated_at) VALUES "
        "(:e, :p, '1화', 1, 'x', 'x', 'x', 'x')"), {"e": eid, "p": proj})
    pg_session.commit()

    before = roster_digest(pg_session, proj, "character", eid)

    # 이 화가 돌면서 제 canon 과 링크를 만든다.
    cid = _canon(pg_session, proj, "C01", "민수", "character")
    pg_session.execute(sql_text(
        "INSERT INTO entity_episode_link (id, canon_id, project_id, episode_id) "
        "VALUES (:l, :c, :p, :e)"),
        {"l": str(uuid.uuid4()), "c": cid, "p": proj, "e": eid})
    pg_session.commit()

    after = roster_digest(pg_session, proj, "character", eid)
    assert after == before, (
        f"이 화가 canon 을 만들자 지문이 바뀌었다 ({before} → {after}) — "
        "재개가 유료 상류를 다시 산다")


def test_force_면_스냅샷을_다시_뜬다(pg_session, proj, tmp_path, monkeypatch):
    """앞 화가 바뀌었을 때 다시 볼 길은 있어야 한다."""
    monkeypatch.setattr("app.core.config.settings.projects_dir", str(tmp_path))
    from app.modules.pipeline.episode_carry import build_roster_block

    ep1, ep2 = str(uuid.uuid4()), str(uuid.uuid4())
    for eid, num in ((ep1, 1), (ep2, 2)):
        pg_session.execute(sql_text(
            "INSERT INTO episode (id, project_id, title, episode_number, "
            "source_filename, source_path, created_at, updated_at) VALUES "
            "(:e, :p, :t, :n, 'x', 'x', 'x', 'x')"),
            {"e": eid, "p": proj, "t": f"{num}화", "n": num})
    pg_session.commit()

    _b, a0 = build_roster_block(pg_session, proj, "character", ep2)
    assert a0 == []

    cid = _canon(pg_session, proj, "C01", "민수", "character")
    pg_session.execute(sql_text(
        "INSERT INTO entity_episode_link (id, canon_id, project_id, episode_id) "
        "VALUES (:l, :c, :p, :e)"),
        {"l": str(uuid.uuid4()), "c": cid, "p": proj, "e": ep1})
    pg_session.commit()

    _b1, a1 = build_roster_block(pg_session, proj, "character", ep2)
    assert a1 == [], "얼린 것을 안 보고 현재 DB 를 읽었다"

    _b2, a2 = build_roster_block(pg_session, proj, "character", ep2, refresh=True)
    assert a2 == ["C01"], "force 인데도 스냅샷을 안 다시 떴다"


def test_이미_돈_화를_늦게_물어도_지문이_안_움직인다(pg_session, proj, tmp_path,
                                              monkeypatch):
    """★★★이 PR 이전에 **이미 분석된 화**가 재개할 때 유료로 다시 사면 안 된다.

    실측 (2026-09-04): 1화가 73스텝을 다 돌고 난 뒤 재개했더니 10초 만에

        Step entity_all_character error: contract drift detected —
        config_hash mismatch (cp=99914b93…, current=c440c37d…)

    로 섰다. 스냅샷을 **처음 묻는 시점**이 이 화가 돈 *뒤*라서, 명부에 이 화가
    제 손으로 만든 canon 이 들어찼기 때문이다. 스텝이 **제가 쓴 행을 제 지문에**
    넣은 셈이다.

    여기서 재는 것: 한 편짜리 프로젝트에서 canon 이 다 있는 상태로 **처음** 물어도
    명부는 비어 있고 지문은 `""` — 즉 옛 체크포인트가 그대로 산다.
    """
    monkeypatch.setattr("app.core.config.settings.projects_dir", str(tmp_path))
    from app.modules.pipeline.episode_carry import roster_digest

    eid = str(uuid.uuid4())
    pg_session.execute(sql_text(
        "INSERT INTO episode (id, project_id, title, episode_number, "
        "source_filename, source_path, created_at, updated_at) VALUES "
        "(:e, :p, '1화', 1, 'x', 'x', 'x', 'x')"), {"e": eid, "p": proj})
    # 이 화는 **이미 다 돌았다** — canon 과 링크가 다 있다. 스냅샷은 아직 없다.
    for sid, name in (("C01", "민수"), ("C02", "정임"), ("C03", "노인")):
        cid = _canon(pg_session, proj, sid, name, "character")
        pg_session.execute(sql_text(
            "INSERT INTO entity_episode_link (id, canon_id, project_id, episode_id) "
            "VALUES (:l, :c, :p, :e)"),
            {"l": str(uuid.uuid4()), "c": cid, "p": proj, "e": eid})
    pg_session.commit()

    assert roster_digest(pg_session, proj, "character", eid) == "", (
        "이미 돈 화를 늦게 물었더니 제 canon 이 명부에 들어찼다 — "
        "재개가 entity_all_* 과 outlook_phase1 을 유료로 다시 산다")


def test_이미_돈_화의_옛_지문이_그대로_산다(pg_session, proj, tmp_path, monkeypatch):
    """★★끝점 — 스텝이 실제로 계산하는 지문이 legacy 값과 같아야 한다.

    위 시험은 명부만 봤다. 여기서는 `entity_all_character` / `outlook_phase1`
    스텝이 **제 `_config_hash()`** 로 내는 값을 잰다. 이 값이 옛 체크포인트에
    적힌 값과 다르면 runner 가 drift 로 세운다.
    """
    monkeypatch.setattr("app.core.config.settings.projects_dir", str(tmp_path))
    from app.core.step_runner import compute_config_hash
    from app.core.steps import STEP_CLASSES

    eid = str(uuid.uuid4())
    pg_session.execute(sql_text(
        "INSERT INTO episode (id, project_id, title, episode_number, "
        "source_filename, source_path, created_at, updated_at) VALUES "
        "(:e, :p, '1화', 1, 'x', 'x', 'x', 'x')"), {"e": eid, "p": proj})
    for sid, name, et in (("C01", "민수", "character"), ("O01", "외투", "outlook")):
        cid = _canon(pg_session, proj, sid, name, et)
        pg_session.execute(sql_text(
            "INSERT INTO entity_episode_link (id, canon_id, project_id, episode_id) "
            "VALUES (:l, :c, :p, :e)"),
            {"l": str(uuid.uuid4()), "c": cid, "p": proj, "e": eid})
    pg_session.commit()

    for step in ("entity_all_character", "outlook_phase1"):
        runner = STEP_CLASSES[step](
            step_id=step, project_id=proj, episode_id=eid,
            db=pg_session, project_config={})
        assert runner._config_hash() == compute_config_hash({}), (
            f"{step}: 이미 돈 화인데 지문이 움직였다 — 옛 체크포인트가 죽는다")


def test_앵커도_명부와_같은_범위를_본다(pg_session, proj, tmp_path, monkeypatch):
    """★★★명부에서 이 화를 뺐는데 **앵커만** 이 화를 보면 안 된다 (Codex BLOCK).

    `_carry_anchor_rows` 는 「이 화 또는 앞 화」를 늘 봤다. 그래서 명부 줄에
    붙는 **구조 앵커**(그 아웃룩을 입는 인물)가, 이 화가 돌기 **전에** 얼렸는지
    **뒤에** 얼렸는지에 따라 달라진다 — 방금 잡은 「얼리는 시점과 무관」이
    아웃룩에서 그대로 다시 깨진다.

    재현 (Codex 가 준 것): 1화에서 `O01` 을 `C01` 이 입고, 2화에서 같은 `O01` 을
    `C02` 가 입는다.

        2화가 돌기 전 normal  → 앵커 C01
        2화가 돈 뒤 늦게 normal → 앵커 **C01 뿐이어야 한다** (앞 판은 C01·C02)
        force refresh          → C01·C02 (이 화를 보는 유일한 자리)
        force 바로 뒤 normal   → force 가 저장한 것과 **같아야** 한다
    """
    monkeypatch.setattr("app.core.config.settings.projects_dir", str(tmp_path))
    from app.modules.pipeline.episode_carry import carry_snapshot, roster_digest

    ep1, ep2 = str(uuid.uuid4()), str(uuid.uuid4())
    for _e, _n in ((ep1, 1), (ep2, 2)):
        pg_session.execute(sql_text(
            "INSERT INTO episode (id, project_id, title, episode_number, "
            "source_filename, source_path, created_at, updated_at) VALUES "
            "(:e, :p, :t, :n, 'x', 'x', 'x', 'x')"),
            {"e": _e, "p": proj, "t": f"{_n}화", "n": _n})
    c1 = _canon(pg_session, proj, "C01", "민수", "character")
    c2 = _canon(pg_session, proj, "C02", "정임", "character")
    ol = _canon(pg_session, proj, "O01", "작업점퍼", "outlook")
    for cid, eid in ((c1, ep1), (ol, ep1), (c2, ep2), (ol, ep2)):
        pg_session.execute(sql_text(
            "INSERT INTO entity_episode_link (id, canon_id, project_id, episode_id) "
            "VALUES (:l, :c, :p, :e)"),
            {"l": str(uuid.uuid4()), "c": cid, "p": proj, "e": eid})
    pg_session.execute(sql_text(
        "INSERT INTO character_outlook (id, project_id, episode_id, character_id, "
        "outlook_id, created_at) VALUES (:i, :p, :e, :c, :o, 'x')"
    ), {"i": str(uuid.uuid4()), "p": proj, "e": ep1, "c": c1, "o": ol})
    pg_session.commit()

    # ① 2화가 아직 안 돌았다 — 앵커는 1화의 착용자.
    early = carry_snapshot(pg_session, proj, ep2, "outlook")
    assert "C01" in early["lines"][0], f"앞 화 착용자가 없다: {early['lines']}"
    assert "C02" not in early["lines"][0], f"이 화 착용자가 샜다: {early['lines']}"
    early_digest = early["digest"]

    # ② 2화가 돌면서 제 배정을 만든다. 스냅샷을 지워 **늦게 처음 묻는** 판을 만든다.
    pg_session.execute(sql_text(
        "INSERT INTO character_outlook (id, project_id, episode_id, character_id, "
        "outlook_id, created_at) VALUES (:i, :p, :e, :c, :o, 'x')"
    ), {"i": str(uuid.uuid4()), "p": proj, "e": ep2, "c": c2, "o": ol})
    pg_session.commit()
    (tmp_path / proj / "checkpoints" / "episodes" / ep2
     / "_carry_snapshot.json").unlink()

    late = carry_snapshot(pg_session, proj, ep2, "outlook")
    assert late["digest"] == early_digest, (
        f"늦게 얼렸더니 앵커가 달라졌다 ({early['lines']} → {late['lines']}) — "
        "재개가 outlook_phase1 을 유료로 다시 산다")

    # ③ force 는 이 화를 본다.
    forced = carry_snapshot(pg_session, proj, ep2, "outlook", refresh=True)
    assert "C02" in forced["lines"][0], (
        f"force 인데 이 화 착용자가 안 보인다: {forced['lines']}")

    # ④ force 바로 뒤 normal 은 force 가 저장한 것을 그대로 읽는다.
    assert roster_digest(pg_session, proj, "outlook", ep2) == forced["digest"], (
        "force 뒤 재개가 저장된 스냅샷을 안 읽었다")