"""pipeline_graph_service 결정론 단위 테스트.

이미지 생성 파이프라인 캔버스(설계: docs/w21b-pipeline-canvas-ui-20260629/)의
read-only 그래프 조립 로직 검증. 노드/엣지 복원은 DB·checkpoint 구조키 기반의
결정론 로직이므로 단위 테스트 대상(캔버스 시각 품질은 별도 육안 검증).
"""

import json
import uuid
from datetime import datetime, timezone

import app.services.pipeline_graph_service as pgs
from app.core.database import SessionLocal
from app.models.catalog import ProjectRegistry, UserAccount
from app.models.project import Episode, EntityCanon, ImageAsset, LLMCallLog, SceneStill
from app.services.pipeline_graph_service import build_pipeline_graph, list_episodes

# 스키마 생성은 conftest 의 session-scoped autouse `_ensure_d6_test_db_schema` 가 담당.
# 테스트 간 격리는 매 테스트가 고유 uuid 프로젝트를 만들고 모든 조회가 project_id
# 스코프이므로 보장된다(행 cleanup 불필요 — drop_all 은 다른 테스트 스키마 전염 위험).


def _now() -> str:
    return datetime.now(timezone.utc).isoformat()


def _seed_base(db) -> tuple[str, str]:
    """user + project + 1 episode 생성. (project_id, episode_id) 반환."""
    uid = str(uuid.uuid4())
    db.add(UserAccount(
        id=uid, username=f"u_{uid[:8]}", display_name="t", password_hash="x",
        role="admin", created_at=_now(), updated_at=_now(),
    ))
    db.commit()  # FK(created_by) 충족을 위해 user 먼저 커밋 (ORM 관계 미선언이라 flush 순서 보장 X)
    pid = str(uuid.uuid4())
    db.add(ProjectRegistry(
        id=pid, name="Pipeline Test", created_by=uid,
        created_at=_now(), updated_at=_now(),
    ))
    eid = str(uuid.uuid4())
    db.add(Episode(
        id=eid, project_id=pid, episode_number=1, title="Pilot",
        source_filename="p.pdf", source_path="p.pdf",
        created_at=_now(), updated_at=_now(),
    ))
    db.commit()
    return pid, eid


def _add_entity(db, pid, short_id, entity_type, name) -> str:
    eid = str(uuid.uuid4())
    db.add(EntityCanon(
        id=eid, project_id=pid, short_id=short_id, entity_type=entity_type,
        name=name, created_at=_now(), updated_at=_now(),
    ))
    db.commit()
    return eid


def _add_still(db, pid, episode_id, *, scene_index, shot_index, still_index=1, **kw) -> str:
    sid = str(uuid.uuid4())
    db.add(SceneStill(
        id=sid, project_id=pid, episode_id=episode_id, still_index=still_index,
        scene_index=scene_index, shot_index=shot_index, created_at=_now(), **kw,
    ))
    db.commit()
    return sid


def _add_asset(db, pid, *, asset_type, **kw) -> str:
    aid = kw.pop("id", str(uuid.uuid4()))
    defaults = dict(
        id=aid, project_id=pid, asset_type=asset_type,
        file_path=f"{aid}.png", prompt_used=f"prompt for {asset_type}",
        created_at=_now(),
    )
    defaults.update(kw)
    db.add(ImageAsset(**defaults))
    db.commit()
    return aid


# ── Task 1: 노드 조립 ─────────────────────────────────────────

def test_empty_project_returns_empty_graph():
    db = SessionLocal()
    try:
        pid, _eid = _seed_base(db)
        # 에피소드만 있고 자산 없음
        graph = build_pipeline_graph(db, pid)
        assert graph.nodes == []
        assert graph.edges == []
        assert len(graph.episodes) == 1
        assert graph.episode_id == graph.episodes[0].id  # 첫 에피소드 default
    finally:
        db.close()


def test_image_asset_nodes_mapping():
    db = SessionLocal()
    try:
        pid, eid = _seed_base(db)
        loc = _add_entity(db, pid, "L03", "location", "옥탑방")
        ch4 = _add_entity(db, pid, "C04", "character", "민숙")
        still = _add_still(db, pid, eid, scene_index=12, shot_index=8)

        a_scene = _add_asset(
            db, pid, asset_type="scene", episode_id=eid, still_id=still,
            generation_model="gemini-image", stage="scene_image_pipeline",
            generation_call_id="call-123", candidate_index=2, attempt_index=1,
            pipeline_metadata_json=json.dumps({"producer_stage": "scene_image_pipeline", "group_id": "g1"}),
        )
        a_ref = _add_asset(db, pid, asset_type="reference", entity_id=ch4)  # episode 무관(NULL)
        a_fp = _add_asset(db, pid, asset_type="floor_plan", entity_id=loc, episode_id=eid)
        a_bg = _add_asset(db, pid, asset_type="chain_bg", episode_id=eid,
                          entity_id=loc, variant_type="L03B01", variant_label="L03B01")

        graph = build_pipeline_graph(db, pid, eid)
        by_id = {n.id: n for n in graph.nodes}
        assert set(by_id) == {a_scene, a_ref, a_fp, a_bg}

        # 모든 노드는 image_asset origin + url 매핑
        for nid, n in by_id.items():
            assert n.node_origin == "image_asset"
            assert n.thumb_url == f"/api/v1/projects/{pid}/images/{nid}/file?thumb=1"
            assert n.full_url == f"/api/v1/projects/{pid}/images/{nid}/file"
            assert n.prompt and n.prompt.startswith("prompt for")

        # 타입 매핑
        assert by_id[a_scene].asset_type == "scene"
        assert by_id[a_bg].asset_type == "chain_bg"
        # scene 노드는 still 에서 scene/shot index 회수
        assert by_id[a_scene].scene_index == 12
        assert by_id[a_scene].shot_index == 8
        assert by_id[a_scene].generation_model == "gemini-image"
        assert by_id[a_scene].stage == "scene_image_pipeline"
        assert by_id[a_scene].generation_call_id == "call-123"
        assert by_id[a_scene].candidate_index == 2
        assert by_id[a_scene].attempt_index == 1
        assert by_id[a_scene].pipeline_metadata == {
            "producer_stage": "scene_image_pipeline", "group_id": "g1"
        }
        # reference 노드는 entity 정보 enrich
        assert by_id[a_ref].entity_short_id == "C04"
        assert by_id[a_ref].entity_type == "character"
        assert by_id[a_fp].entity_short_id == "L03"
        # chain_bg variant_label 보존(bg_id)
        assert by_id[a_bg].variant_label == "L03B01"
    finally:
        db.close()


def test_invalid_pipeline_metadata_is_non_fatal():
    db = SessionLocal()
    try:
        pid, eid = _seed_base(db)
        asset = _add_asset(
            db, pid, asset_type="generated", episode_id=eid,
            pipeline_role="structure_form_reference",
            pipeline_metadata_json="{not-json",
        )
        by_id = {n.id: n for n in build_pipeline_graph(db, pid, eid).nodes}
        assert by_id[asset].pipeline_role == "structure_form_reference"
        assert by_id[asset].pipeline_metadata is None
    finally:
        db.close()


def test_episode_filter_excludes_other_episode_scenes():
    db = SessionLocal()
    try:
        pid, eid1 = _seed_base(db)
        eid2 = str(uuid.uuid4())
        db.add(Episode(
            id=eid2, project_id=pid, episode_number=2, title="Two",
            source_filename="p.pdf", source_path="p.pdf",
            created_at=_now(), updated_at=_now(),
        ))
        db.commit()
        a1 = _add_asset(db, pid, asset_type="scene", episode_id=eid1)
        _a2 = _add_asset(db, pid, asset_type="scene", episode_id=eid2)
        a_shared = _add_asset(db, pid, asset_type="reference")  # episode NULL = 공용

        graph = build_pipeline_graph(db, pid, eid1)
        ids = {n.id for n in graph.nodes}
        assert a1 in ids
        assert a_shared in ids  # 프로젝트 공용 ref 포함
        assert _a2 not in ids   # 다른 에피소드 scene 제외
        assert len(graph.episodes) == 2
    finally:
        db.close()


# ── Task 2: i2i + reference lineage 엣지 ─────────────────────

def _edge_set(graph, kind):
    return {(e.source, e.target) for e in graph.edges if e.kind == kind}


def test_input_image_ids_generated_input_edges():
    """Wave3: input_image_ids → generated_input 엣지(별도 kind, confidence=input_uuid)."""
    db = SessionLocal()
    try:
        pid, eid = _seed_base(db)
        face = _add_asset(db, pid, asset_type="reference", pipeline_role="reference_face")
        outfit = _add_asset(db, pid, asset_type="reference", pipeline_role="reference_outlook")
        comp = _add_asset(
            db, pid, asset_type="reference", pipeline_role="reference_composite",
            input_image_ids=json.dumps([face, outfit]),
        )
        graph = build_pipeline_graph(db, pid, eid)
        gi = _edge_set(graph, "generated_input")
        assert (face, comp) in gi
        assert (outfit, comp) in gi
        e = next(e for e in graph.edges
                 if e.kind == "generated_input" and e.source == face)
        assert e.confidence == "input_uuid"
        assert e.verification_status == "recorded_input"
    finally:
        db.close()


def test_input_edge_coexists_with_structural_edge():
    """Wave3: 같은 source/target 이 structural(i2i)+generated_input 둘 다면 둘 다 유지(dedup 안 함)."""
    db = SessionLocal()
    try:
        pid, eid = _seed_base(db)
        a = _add_asset(db, pid, asset_type="scene", episode_id=eid)
        # b 는 a 를 parent(i2i)로도, input_image_ids 로도 참조.
        b = _add_asset(db, pid, asset_type="scene", episode_id=eid,
                       parent_image_id=a, input_image_ids=json.dumps([a]))
        graph = build_pipeline_graph(db, pid, eid)
        assert (a, b) in _edge_set(graph, "i2i")
        assert (a, b) in _edge_set(graph, "generated_input")
        # 같은 (source,target) 이지만 kind 가 달라 둘 다 보존.
        ab = [e for e in graph.edges if (e.source, e.target) == (a, b)]
        assert {e.kind for e in ab} == {"i2i", "generated_input"}
    finally:
        db.close()


def test_dangling_input_ref_skipped_and_counted():
    """Wave3: 노드에 없는 input UUID 는 엣지 미생성 + diagnostics.dangling_input_refs 증가."""
    db = SessionLocal()
    try:
        pid, eid = _seed_base(db)
        a = _add_asset(db, pid, asset_type="scene", episode_id=eid,
                       input_image_ids=json.dumps(["ghost-uuid-1", "ghost-uuid-2"]))
        graph = build_pipeline_graph(db, pid, eid)
        assert _edge_set(graph, "generated_input") == set()
        assert graph.diagnostics is not None
        assert graph.diagnostics["dangling_input_refs"] == 2
        # 노드 자체는 존재(엣지만 미생성).
        assert a in {n.id for n in graph.nodes}
    finally:
        db.close()


def test_node_pipeline_role_disposition_intermediate():
    """Wave3: node 에 pipeline_role/disposition/is_intermediate 포함(색·필터·badge)."""
    db = SessionLocal()
    try:
        pid, eid = _seed_base(db)
        final = _add_asset(db, pid, asset_type="scene", episode_id=eid,
                           pipeline_role="scene_still", is_intermediate=False)
        rej = _add_asset(db, pid, asset_type="generated", episode_id=eid,
                         pipeline_role="scene_regeneration_candidate",
                         disposition="rejected", is_intermediate=True)
        diag = _add_asset(db, pid, asset_type="generated", episode_id=eid,
                          pipeline_role="angle_camera_diagram",
                          disposition="diagnostic", is_intermediate=True)
        by = {n.id: n for n in build_pipeline_graph(db, pid, eid).nodes}
        assert by[final].pipeline_role == "scene_still"
        assert by[final].is_intermediate is False
        assert by[final].disposition is None
        assert by[rej].disposition == "rejected"
        assert by[rej].is_intermediate is True
        assert by[diag].disposition == "diagnostic"
    finally:
        db.close()


def test_i2i_chain_edges():
    db = SessionLocal()
    try:
        pid, eid = _seed_base(db)
        a = _add_asset(db, pid, asset_type="scene", episode_id=eid)
        b = _add_asset(db, pid, asset_type="scene", episode_id=eid, parent_image_id=a)
        c = _add_asset(db, pid, asset_type="scene", episode_id=eid, parent_image_id=b)

        graph = build_pipeline_graph(db, pid, eid)
        i2i = _edge_set(graph, "i2i")
        assert (a, b) in i2i
        assert (b, c) in i2i
        assert len(i2i) == 2
        # confidence/verification — 진짜 이미지 UUID FK → verified_direct
        e = next(e for e in graph.edges if (e.source, e.target) == (a, b))
        assert e.confidence == "direct_fk"
        assert e.verification_status == "verified_direct"
    finally:
        db.close()


def test_i2i_source_and_parent_dedup():
    db = SessionLocal()
    try:
        pid, eid = _seed_base(db)
        a = _add_asset(db, pid, asset_type="scene", episode_id=eid)
        # source_image_id == parent_image_id → 엣지 1개만
        b = _add_asset(db, pid, asset_type="composite", episode_id=eid,
                       parent_image_id=a, source_image_id=a)
        graph = build_pipeline_graph(db, pid, eid)
        i2i = [e for e in graph.edges if e.kind == "i2i"]
        assert len(i2i) == 1
        assert (i2i[0].source, i2i[0].target) == (a, b)
    finally:
        db.close()


def test_reference_lineage_edges_and_dangling_filter():
    db = SessionLocal()
    try:
        pid, eid = _seed_base(db)
        ch = _add_entity(db, pid, "C04", "character", "민숙")
        r1 = _add_asset(db, pid, asset_type="reference", entity_id=ch)
        r2 = _add_asset(db, pid, asset_type="reference", entity_id=ch)
        # scene 의 lineage refs = [r1, r2, ghost(노드없음)]
        scene = _add_asset(
            db, pid, asset_type="scene", episode_id=eid,
            reference_image_ids=json.dumps([r1, r2, "ghost-missing-id"]),
        )
        graph = build_pipeline_graph(db, pid, eid)
        ref = _edge_set(graph, "reference")
        assert (r1, scene) in ref
        assert (r2, scene) in ref
        # 노드 집합에 없는 ghost → 엣지 미생성(dangling filter)
        assert all(src in {r1, r2} for (src, _t) in ref)
        assert len(ref) == 2
        e = next(e for e in graph.edges if e.kind == "reference")
        assert e.confidence == "lineage_reference"
    finally:
        db.close()


def test_malformed_reference_ids_no_crash():
    db = SessionLocal()
    try:
        pid, eid = _seed_base(db)
        _add_asset(db, pid, asset_type="scene", episode_id=eid,
                   reference_image_ids="not-json")
        graph = build_pipeline_graph(db, pid, eid)  # 크래시 없이 진행
        assert _edge_set(graph, "reference") == set()
    finally:
        db.close()


def test_list_episodes_ordered_by_number():
    db = SessionLocal()
    try:
        pid, eid1 = _seed_base(db)
        eid0 = str(uuid.uuid4())
        db.add(Episode(
            id=eid0, project_id=pid, episode_number=0, title="Zero",
            source_filename="p.pdf", source_path="p.pdf",
            created_at=_now(), updated_at=_now(),
        ))
        db.commit()
        eps = list_episodes(db, pid)
        assert [e.label.split()[0] for e in eps] == ["EP0", "EP1"]
    finally:
        db.close()


# ── Task 3: FP→배경→씬 + prev_scene 엣지 (checkpoint groups SOT) ──

def test_background_and_fp_edges(monkeypatch):
    db = SessionLocal()
    try:
        pid, eid = _seed_base(db)
        loc = _add_entity(db, pid, "L03", "location", "옥탑방")
        still = _add_still(db, pid, eid, scene_index=12, shot_index=8)
        a_scene = _add_asset(db, pid, asset_type="scene", episode_id=eid, still_id=still)
        a_fp = _add_asset(db, pid, asset_type="floor_plan", entity_id=loc, episode_id=eid)
        a_bg = _add_asset(db, pid, asset_type="chain_bg", episode_id=eid,
                          entity_id=loc, variant_type="L03B01", variant_label="L03B01")
        groups = {"L03B01": {"png_path": "p.png", "shot_ids": ["S12_Shot8"],
                             "location_id": "L03"}}
        monkeypatch.setattr(pgs, "_load_background_groups", lambda p, e: groups)

        graph = build_pipeline_graph(db, pid, eid)
        assert (a_bg, a_scene) in _edge_set(graph, "background")  # bg→scene
        assert (a_fp, a_bg) in _edge_set(graph, "fp")             # fp→bg

        be = next(e for e in graph.edges if e.kind == "background")
        assert be.confidence == "planned_only"
        assert be.verification_status == "unverified"
        assert be.bg_id == "L03B01"
        fe = next(e for e in graph.edges if e.kind == "fp")
        assert fe.confidence == "checkpoint_structural"
    finally:
        db.close()


def test_background_virtual_node_when_no_db_chain_bg(monkeypatch):
    db = SessionLocal()
    try:
        pid, eid = _seed_base(db)
        loc = _add_entity(db, pid, "L03", "location", "옥탑방")
        still = _add_still(db, pid, eid, scene_index=12, shot_index=8)
        a_scene = _add_asset(db, pid, asset_type="scene", episode_id=eid, still_id=still)
        a_fp = _add_asset(db, pid, asset_type="floor_plan", entity_id=loc, episode_id=eid)
        # DB 에 chain_bg 자산 없음 → checkpoint virtual 노드로 fallback
        groups = {"L03B02": {"png_path": "images/bg.png", "shot_ids": ["S12_Shot8"],
                             "location_id": "L03"}}
        monkeypatch.setattr(pgs, "_load_background_groups", lambda p, e: groups)

        graph = build_pipeline_graph(db, pid, eid)
        vnodes = [n for n in graph.nodes if n.node_origin == "checkpoint_virtual"]
        assert len(vnodes) == 1
        v = vnodes[0]
        assert v.asset_type == "chain_bg"
        assert v.variant_label == "L03B02"
        assert v.full_url == "images/bg.png"
        assert v.thumb_url is None  # virtual 은 파일 엔드포인트 없음
        assert (v.id, a_scene) in _edge_set(graph, "background")
        assert (a_fp, v.id) in _edge_set(graph, "fp")
    finally:
        db.close()


def test_background_no_matching_shot_yields_no_bg_edge(monkeypatch):
    db = SessionLocal()
    try:
        pid, eid = _seed_base(db)
        loc = _add_entity(db, pid, "L03", "location", "옥탑방")
        still = _add_still(db, pid, eid, scene_index=12, shot_index=8)
        _add_asset(db, pid, asset_type="scene", episode_id=eid, still_id=still)
        _add_asset(db, pid, asset_type="chain_bg", episode_id=eid,
                   entity_id=loc, variant_type="L03B01")
        # shot_ids 가 존재하지 않는 shot 을 가리킴 → bg→scene 엣지 0
        groups = {"L03B01": {"png_path": "p.png", "shot_ids": ["S99_Shot1"],
                             "location_id": "L03"}}
        monkeypatch.setattr(pgs, "_load_background_groups", lambda p, e: groups)

        graph = build_pipeline_graph(db, pid, eid)
        assert _edge_set(graph, "background") == set()
    finally:
        db.close()


def test_prev_scene_edges(monkeypatch):
    db = SessionLocal()
    try:
        pid, eid = _seed_base(db)
        s_prev = _add_still(db, pid, eid, scene_index=11, shot_index=2)
        s_cur = _add_still(db, pid, eid, scene_index=12, shot_index=8,
                           dependent_scene_id=s_prev)
        a_prev = _add_asset(db, pid, asset_type="scene", episode_id=eid, still_id=s_prev)
        a_cur = _add_asset(db, pid, asset_type="scene", episode_id=eid, still_id=s_cur)
        monkeypatch.setattr(pgs, "_load_background_groups", lambda p, e: {})

        graph = build_pipeline_graph(db, pid, eid)
        assert (a_prev, a_cur) in _edge_set(graph, "prev_scene")
        e = next(e for e in graph.edges if e.kind == "prev_scene")
        assert e.confidence == "still_fk_resolved"  # 이미지 self-FK 아닌 still FK 해석
        assert e.dependent_still_id == s_prev
        assert e.source_selection == "is_primary_or_latest"
    finally:
        db.close()


def test_load_background_groups_reads_phase7_manifest(tmp_path, monkeypatch):
    """Phase 7 manifest IO 로더: groups 복원 + status!=ok skip + png_path 상대화."""
    from app.core.config import settings

    monkeypatch.setattr(settings, "projects_dir", str(tmp_path / "projects"))
    pid, eid = "proj-x", "ep-x"
    cp_dir = (tmp_path / "projects" / pid / "checkpoints" / "episodes" / eid
              / "background_render")
    cp_dir.mkdir(parents=True)
    abs_png = str(tmp_path / "projects" / pid / "images" / "bg1.png")
    manifest = {"data": {"groups": {
        "L03B01": {"status": "ok", "png_path": abs_png,
                   "shot_ids": ["S12_Shot8"], "location_id": "L03"},
        "L03B02": {"status": "failed", "png_path": abs_png,
                   "shot_ids": [], "location_id": "L03"},  # status!=ok → skip
    }}}
    (cp_dir / "manifest.json").write_text(json.dumps(manifest), encoding="utf-8")

    groups = pgs._load_background_groups(pid, eid)
    assert set(groups) == {"L03B01"}
    g = groups["L03B01"]
    assert g["shot_ids"] == ["S12_Shot8"]
    assert g["location_id"] == "L03"
    assert g["png_path"] == f"projects/{pid}/images/bg1.png"  # projects_root 기준 상대화


def test_load_background_groups_missing_checkpoint_returns_empty(tmp_path, monkeypatch):
    from app.core.config import settings

    monkeypatch.setattr(settings, "projects_dir", str(tmp_path / "projects"))
    assert pgs._load_background_groups("nope", "nope") == {}


# ── Task 4: call-level overlay (구조 키 조인, 라벨 의미파싱 0) ──────────────

def _add_call_log(db, pid, eid, *, still_id, ref_labels, user_prompt="auth prompt",
                  operation_type="scene_image_gen"):
    db.add(LLMCallLog(
        id=str(uuid.uuid4()), project_id=pid, episode_id=eid,
        operation_type=operation_type, model_name="gemini", user_prompt=user_prompt,
        reference_image_ids=json.dumps(ref_labels),
        metadata_json=json.dumps({"still_id": still_id}),
        status="success", created_at=_now(),
    ))
    db.commit()


def _setup_bg_edge(db, monkeypatch):
    """bg→scene 엣지 1개가 생기는 최소 셋업. (pid, eid, still_id, a_scene, a_bg) 반환."""
    pid, eid = _seed_base(db)
    loc = _add_entity(db, pid, "L03", "location", "옥탑방")
    still = _add_still(db, pid, eid, scene_index=12, shot_index=8)
    a_scene = _add_asset(db, pid, asset_type="scene", episode_id=eid, still_id=still)
    a_bg = _add_asset(db, pid, asset_type="chain_bg", episode_id=eid,
                      entity_id=loc, variant_type="L03B01")
    groups = {"L03B01": {"png_path": "p.png", "shot_ids": ["S12_Shot8"],
                         "location_id": "L03"}}
    monkeypatch.setattr(pgs, "_load_background_groups", lambda p, e: groups)
    return pid, eid, still, a_scene, a_bg


def test_overlay_ref_zero_marks_call_log_no_refs(monkeypatch):
    """호출은 있었으나 ref 0개 → planned bg 미attach 확정(call_log_no_refs)."""
    db = SessionLocal()
    try:
        pid, eid, still, _a_scene, _a_bg = _setup_bg_edge(db, monkeypatch)
        _add_call_log(db, pid, eid, still_id=still, ref_labels=[])
        graph = build_pipeline_graph(db, pid, eid)
        be = next(e for e in graph.edges if e.kind == "background")
        assert be.call_ref_count == 0
        assert be.verification_status == "call_log_no_refs"
    finally:
        db.close()


def test_overlay_refs_present_unresolved(monkeypatch):
    """ref 있었으나 어떤 planned edge인지 UUID 매칭 불가 → refs_present_unresolved."""
    db = SessionLocal()
    try:
        pid, eid, still, _a_scene, _a_bg = _setup_bg_edge(db, monkeypatch)
        _add_call_log(db, pid, eid, still_id=still, ref_labels=["anything", "two"])
        graph = build_pipeline_graph(db, pid, eid)
        be = next(e for e in graph.edges if e.kind == "background")
        assert be.call_ref_count == 2
        assert be.verification_status == "refs_present_unresolved"
        assert be.verification_status != "verified_direct"  # UUID 검증 단정 금지
    finally:
        db.close()


def test_overlay_label_content_invariant(monkeypatch):
    """★절대규칙 검증: 라벨 문자열 내용(bg_id 포함 여부)이 달라도 결과 동일(의미파싱 0)."""
    db = SessionLocal()
    try:
        # A: 라벨에 bg_id 가 박혀 있음
        pidA, eidA, stillA, _sA, _bA = _setup_bg_edge(db, monkeypatch)
        _add_call_log(db, pidA, eidA, still_id=stillA,
                      ref_labels=["background chain ref (L03B01 for L03) — match wall"])
        gA = build_pipeline_graph(db, pidA, eidA)
        beA = next(e for e in gA.edges if e.kind == "background")

        # B: 완전히 무관한 라벨, 같은 개수(1)
        pidB, eidB, stillB, _sB, _bB = _setup_bg_edge(db, monkeypatch)
        _add_call_log(db, pidB, eidB, still_id=stillB, ref_labels=["zzz unrelated label"])
        gB = build_pipeline_graph(db, pidB, eidB)
        beB = next(e for e in gB.edges if e.kind == "background")

        assert beA.call_ref_count == beB.call_ref_count == 1
        assert beA.verification_status == beB.verification_status == "refs_present_unresolved"
    finally:
        db.close()


def test_overlay_prompt_authoritative_on_node_not_edge(monkeypatch):
    """prompt_authoritative 는 scene 노드에만 부착, 엣지 생성에는 미사용."""
    db = SessionLocal()
    try:
        pid, eid, still, a_scene, a_bg = _setup_bg_edge(db, monkeypatch)
        _add_call_log(db, pid, eid, still_id=still, ref_labels=["x"],
                      user_prompt="THE AUTHORITATIVE PROMPT")
        graph = build_pipeline_graph(db, pid, eid)
        scene_node = next(n for n in graph.nodes if n.id == a_scene)
        assert scene_node.prompt_authoritative == "THE AUTHORITATIVE PROMPT"
        assert scene_node.actual_ref_count == 1
        bg_edges = [e for e in graph.edges if e.kind == "background"]
        assert len(bg_edges) == 1  # overlay 가 엣지 개수에 영향 없음
        assert (a_bg, a_scene) in _edge_set(graph, "background")
    finally:
        db.close()


def test_overlay_no_call_log_keeps_unverified(monkeypatch):
    db = SessionLocal()
    try:
        pid, eid, _still, _a_scene, _a_bg = _setup_bg_edge(db, monkeypatch)
        graph = build_pipeline_graph(db, pid, eid)  # call_log 없음
        be = next(e for e in graph.edges if e.kind == "background")
        assert be.verification_status == "unverified"
        assert be.call_ref_count is None
    finally:
        db.close()


def test_edge_ids_stable_unique_deterministic(monkeypatch):
    """모든 엣지에 안정 id(kind:source:target) 부여 + 고유 + 두 번 빌드 동일(React Flow)."""
    db = SessionLocal()
    try:
        pid, eid, still, a_scene, a_bg = _setup_bg_edge(db, monkeypatch)
        # i2i 체인 + reference 도 섞어 여러 kind 검증
        a2 = _add_asset(db, pid, asset_type="scene", episode_id=eid, parent_image_id=a_scene)

        g1 = build_pipeline_graph(db, pid, eid)
        ids = [e.id for e in g1.edges]
        assert all(ids), "모든 엣지에 id 부여"
        assert len(ids) == len(set(ids)), "엣지 id 고유"
        for e in g1.edges:
            assert e.id.startswith(f"{e.kind}:{e.source}:{e.target}")

        # 결정론: 같은 데이터로 다시 빌드 → 동일 id 집합
        g2 = build_pipeline_graph(db, pid, eid)
        assert {e.id for e in g1.edges} == {e.id for e in g2.edges}
    finally:
        db.close()


def test_background_consumer_enabled_flag(monkeypatch):
    db = SessionLocal()
    try:
        pid, eid = _seed_base(db)
        monkeypatch.setattr(pgs, "_load_background_groups", lambda p, e: {})
        graph = build_pipeline_graph(db, pid, eid)
        assert isinstance(graph.background_consumer_enabled, bool)  # settings 반영
    finally:
        db.close()