"""이미지 생성 파이프라인 그래프 조립 서비스 (read-only).

설계 SOT: docs/w21b-pipeline-canvas-ui-20260629/design.html (§9 Codex 반영).

- 노드: DB ``image_asset`` (+ 후속: checkpoint virtual guide/sketch).
- 엣지: i2i(source/parent_image_id) / reference lineage / FP→bg→씬 구조키 / prev_scene.
- ``llm_call_log`` 는 검증/라벨 표시 용도 — UUID edge SOT 아님.

메인 파이프라인 무접촉 — 오직 읽기 전용 조회. 신규 모듈(기존 무수정).
"""

from __future__ import annotations

import hashlib
import json
import logging
import re
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple

from sqlalchemy.orm import Session as OrmSession

from app.models.project import EntityCanon, Episode, ImageAsset, SceneStill

logger = logging.getLogger(__name__)

# Phase 7 / Phase 5 공통 shot_id 형식 ``Sxx_Shotyy`` (scene_checkpoint_loaders 와 동일).
_SHOT_ID_RE = re.compile(r"^S(\d+)_Shot(\d+)$")
from app.schemas.pipeline import (
    EpisodeRef,
    PipelineEdge,
    PipelineGraphResponse,
    PipelineNode,
)


def _image_urls(project_id: str, image_id: str) -> tuple[str, str]:
    """(thumb_url, full_url) for an image_asset id (frontend 파일 엔드포인트)."""
    base = f"/api/v1/projects/{project_id}/images/{image_id}/file"
    return f"{base}?thumb=1", base


def list_episodes(db: OrmSession, project_id: str) -> List[EpisodeRef]:
    rows = (
        db.query(Episode)
        .filter(Episode.project_id == project_id)
        .order_by(Episode.episode_number.asc())
        .all()
    )
    return [
        EpisodeRef(id=e.id, label=f"EP{e.episode_number} {e.title or ''}".strip())
        for e in rows
    ]


def _default_episode_id(
    db: OrmSession, project_id: str, episodes: List[EpisodeRef]
) -> Optional[str]:
    """episode_id 미지정 시 default 에피소드.

    첫 에피소드(예: 기획안)는 이미지 자산이 0개라 빈 캔버스가 되기 쉽다 →
    에피소드-스코프 image_asset 이 하나라도 있는 첫 에피소드(번호 순)를 고른다.
    공용 자산(episode_id IS NULL: 인물 ref/FP)은 콘텐츠 판단에서 제외. 없으면 첫 에피소드.
    """
    if not episodes:
        return None
    rows = (
        db.query(ImageAsset.episode_id)
        .filter(ImageAsset.project_id == project_id, ImageAsset.episode_id.isnot(None))
        .distinct()
        .all()
    )
    with_assets = {r[0] for r in rows}
    for ep in episodes:
        if ep.id in with_assets:
            return ep.id
    return episodes[0].id


def _entity_map(db: OrmSession, project_id: str) -> Dict[str, EntityCanon]:
    rows = db.query(EntityCanon).filter(EntityCanon.project_id == project_id).all()
    return {e.id: e for e in rows}


def _still_map(db: OrmSession, project_id: str, episode_id: Optional[str]) -> Dict[str, SceneStill]:
    q = db.query(SceneStill).filter(SceneStill.project_id == project_id)
    if episode_id is not None:
        q = q.filter(SceneStill.episode_id == episode_id)
    return {s.id: s for s in q.all()}


def _asset_to_node(
    project_id: str,
    asset: ImageAsset,
    entities: Dict[str, EntityCanon],
    stills: Dict[str, SceneStill],
) -> PipelineNode:
    thumb_url, full_url = _image_urls(project_id, asset.id)
    ent = entities.get(asset.entity_id) if asset.entity_id else None
    still = stills.get(asset.still_id) if asset.still_id else None
    scene_index = still.scene_index if still is not None else None
    shot_index = still.shot_index if still is not None else asset.shot_index
    pipeline_metadata: Optional[Dict[str, Any]] = None
    if asset.pipeline_metadata_json:
        try:
            parsed = json.loads(asset.pipeline_metadata_json)
            if isinstance(parsed, dict):
                pipeline_metadata = parsed
        except (TypeError, ValueError):
            logger.warning(
                "pipeline_graph: invalid pipeline_metadata_json asset=%s", asset.id
            )
    return PipelineNode(
        id=asset.id,
        node_origin="image_asset",
        asset_type=asset.asset_type,
        entity_id=asset.entity_id,
        entity_short_id=ent.short_id if ent else None,
        entity_name=ent.name if ent else None,
        entity_type=ent.entity_type if ent else None,
        still_id=asset.still_id,
        scene_index=scene_index,
        shot_index=shot_index,
        episode_id=asset.episode_id,
        variant_label=asset.variant_label,
        thumb_url=thumb_url,
        full_url=full_url,
        prompt=asset.prompt_used,
        status=asset.status,
        created_at=asset.created_at,
        generation_model=asset.generation_model,
        stage=asset.stage,
        generation_call_id=asset.generation_call_id,
        candidate_index=asset.candidate_index,
        attempt_index=asset.attempt_index,
        pipeline_metadata=pipeline_metadata,
        # Wave3: 중간물/lineage 시각화 (색·필터·badge). is_intermediate 는 NOT NULL.
        pipeline_role=asset.pipeline_role,
        disposition=asset.disposition,
        is_intermediate=bool(asset.is_intermediate),
    )


def _query_image_assets(
    db: OrmSession,
    project_id: str,
    episode_id: Optional[str],
) -> List[ImageAsset]:
    """이 프로젝트의 image_asset row.

    episode_id 가 주어지면 해당 에피소드 + 에피소드 무관(프로젝트 공용: 인물 ref/FP 등
    episode_id IS NULL) 자산을 함께 포함한다.
    """
    q = db.query(ImageAsset).filter(ImageAsset.project_id == project_id)
    if episode_id is not None:
        q = q.filter(
            (ImageAsset.episode_id == episode_id) | (ImageAsset.episode_id.is_(None))
        )
    return q.all()


def _parse_id_list(raw: Optional[str]) -> List[str]:
    """reference_image_ids 같은 JSON 문자열 배열을 안전하게 파싱(실패 시 빈 리스트)."""
    if not raw:
        return []
    try:
        val = json.loads(raw)
    except (ValueError, TypeError):
        return []
    if not isinstance(val, list):
        return []
    return [str(x) for x in val if isinstance(x, (str, int))]


def _reconstruct_image_asset_edges(
    assets: List[ImageAsset],
    node_ids: set[str],
) -> List[PipelineEdge]:
    """DB image_asset 컬럼에서 직접 복원되는 엣지(i2i + reference lineage).

    - i2i: source_image_id / parent_image_id (이미지 UUID 직접 FK). confidence=direct_fk.
    - reference: reference_image_ids (lineage — visible 인물/소품 ref UUID). confidence=lineage_reference.

    노드 집합에 없는 끝점(dangling)은 엣지를 만들지 않는다. (source,target,kind) 중복 제거.
    """
    edges: List[PipelineEdge] = []
    seen: set[tuple[str, str, str]] = set()

    def _add(source: str, target: str, kind: str, confidence: str,
             verification: str) -> None:
        if source not in node_ids or target not in node_ids or source == target:
            return
        key = (source, target, kind)
        if key in seen:
            return
        seen.add(key)
        edges.append(
            PipelineEdge(
                source=source,
                target=target,
                kind=kind,
                confidence=confidence,
                verification_status=verification,
            )
        )

    for a in assets:
        # i2i / 편집 체인 — 부모 이미지 → 이 자산. 진짜 이미지 UUID FK → verified_direct.
        for parent in (a.source_image_id, a.parent_image_id):
            if parent:
                _add(parent, a.id, "i2i", "direct_fk", "verified_direct")
        # reference lineage — 인물/소품 ref → 이 자산. 의도된 lineage(구조 추론).
        for ref_id in _parse_id_list(a.reference_image_ids):
            _add(ref_id, a.id, "reference", "lineage_reference", "structural_inferred")

    return edges


def _reconstruct_input_image_edges(
    assets: List[ImageAsset],
    node_ids: set[str],
) -> Tuple[List[PipelineEdge], int]:
    """persist-all Wave1/2 ``input_image_ids`` 에서 복원되는 입력 lineage 엣지.

    이 컬럼은 생성물 row 에 **실제 기록된 입력 이미지 UUID**(composite=face+outfit,
    bg=floor_plan, i2i 변형=source 등). 구조적으로 추론된 reference/fp/bg/prev_scene
    엣지와 **중복될 수 있으나 dedup 하지 않는다**(Codex): 사용자가 "구조적 계획 연결"과
    "생성물에 기록된 실제 입력 UUID"를 둘 다 봐야 한다 → 별도 kind=``generated_input``.

    confidence=``input_uuid``(DB UUID 직접 참조), verification=``recorded_input``.
    dangling(노드에 없는 끝점)은 조용히 엣지로 만들지 않고 skip + count 반환(진단 투명성).

    Returns ``(edges, dangling_count)``.
    """
    edges: List[PipelineEdge] = []
    seen: set[Tuple[str, str]] = set()
    dangling = 0
    for a in assets:
        for src in _parse_id_list(a.input_image_ids):
            if src == a.id:
                continue
            if src not in node_ids or a.id not in node_ids:
                dangling += 1
                continue
            key = (src, a.id)
            if key in seen:
                continue
            seen.add(key)
            edges.append(
                PipelineEdge(
                    source=src,
                    target=a.id,
                    kind="generated_input",
                    confidence="input_uuid",
                    verification_status="recorded_input",
                )
            )
    return edges, dangling


def _checkpoint_manifest_path(project_id: str, episode_id: str, step_name: str) -> Path:
    """체크포인트 manifest 경로(읽기 전용). scene_checkpoint_loaders._ep_checkpoint_path 미러.

    기존 모듈 무접촉을 위해 private helper 를 import 하지 않고 동일 구조를 재현한다.
    """
    from app.core.config import settings

    return (
        Path(settings.projects_dir)
        / project_id
        / "checkpoints"
        / "episodes"
        / episode_id
        / step_name
        / "manifest.json"
    )


def _ingest_groups_shape(
    out: Dict[str, Dict[str, Any]],
    data: Dict[str, Any],
    projects_root: Path,
) -> None:
    """Phase 7 / Phase 5 공통 shape ``data.groups[bg_id]`` 를 out 에 ingest(첫 매칭 보존).

    이미지 바이트는 읽지 않는다(그래프 조립 전용). ``load_background_chain_bg_map`` 의
    소비자 키와 동일한 구조 필드(bg_id/shot_ids/location_id)만 사용한다.
    """
    groups = (data or {}).get("groups") or {}
    if not isinstance(groups, dict):
        return
    for bg_id, gres in groups.items():
        if not isinstance(gres, dict):
            continue
        if gres.get("status") != "ok":
            continue
        png_path = gres.get("png_path") or ""
        if not png_path:
            continue
        if bg_id in out:  # Phase 7 우선 — 첫 매칭 보존
            continue
        try:
            rel = str(Path(png_path).relative_to(projects_root))
        except ValueError:
            rel = png_path
        out[bg_id] = {
            "png_path": rel,
            "shot_ids": [str(s) for s in (gres.get("shot_ids") or [])],
            "location_id": gres.get("location_id") or "",
        }


def _load_background_groups(
    project_id: str, episode_id: str
) -> Dict[str, Dict[str, Any]]:
    """배경 체크포인트에서 ``bg_id -> {png_path, shot_ids, location_id}`` 복원(읽기 전용).

    우선순위(설계 §9.1): background_render(Phase 7) → background_chain_render(Phase 5)
    → legacy ``data.locations[loc].shot_backgrounds[]``. 첫 매칭 보존(Phase 7 우선).
    체크포인트 부재/파싱 실패 시 빈 dict(non-fatal).
    """
    from app.core.config import settings

    out: Dict[str, Dict[str, Any]] = {}
    projects_root = Path(settings.projects_dir).parent

    for step_name in ("background_render", "background_chain_render"):
        cp = _checkpoint_manifest_path(project_id, episode_id, step_name)
        if not cp.exists():
            continue
        try:
            data = json.loads(cp.read_text(encoding="utf-8")).get("data", {})
        except (OSError, ValueError) as exc:
            logger.warning("pipeline_graph: %s checkpoint load failed: %s", step_name, exc)
            continue
        _ingest_groups_shape(out, data, projects_root)

        # Phase 4 LEGACY (background_chain_render 만): locations[].shot_backgrounds[].
        if step_name == "background_chain_render":
            locations = data.get("locations") or {}
            if isinstance(locations, dict):
                for loc_id, loc_data in locations.items():
                    for sb in (loc_data or {}).get("shot_backgrounds", []) or []:
                        if not isinstance(sb, dict):
                            continue
                        si, shi = sb.get("scene_index"), sb.get("shot_index")
                        img_path = sb.get("image_path", "") or ""
                        if si is None or shi is None or not img_path:
                            continue
                        node_id = sb.get("node_id") or f"legacy_{loc_id}_{si}_{shi}"
                        if node_id in out:
                            continue
                        try:
                            rel = str(Path(img_path).relative_to(projects_root))
                        except ValueError:
                            rel = img_path
                        out[node_id] = {
                            "png_path": rel,
                            "shot_ids": [f"S{si}_Shot{shi}"],
                            "location_id": loc_id,
                        }
    return out


def _reconstruct_background_edges(
    groups: Dict[str, Dict[str, Any]],
    *,
    episode_id: Optional[str],
    chain_bg_by_bgid: Dict[str, str],
    fp_ids_by_canon: Dict[str, List[str]],
    location_canon_by_short: Dict[str, str],
    scene_assets_by_shot: Dict[Tuple[int, int], List[str]],
    node_ids: set[str],
) -> Tuple[List[PipelineEdge], List[PipelineNode]]:
    """FP→배경→씬 엣지 복원(설계 §9.1, 구조키만 — 라벨 파싱 금지).

    - bg→scene: groups[bg_id].shot_ids(``Sxx_Shotyy``) → (scene_index, shot_index) →
      그 shot 의 scene asset. confidence=planned_only/verification=unverified(실제 attach
      는 close-framing 등으로 skip 가능 → Task4 call_log 로 승급).
    - fp→bg: floor_plan asset(entity_id == location canon) → bg. confidence=checkpoint_structural.
    - bg 노드: DB chain_bg(variant_type=bg_id) 매칭, 없으면 png_path 기반 virtual 노드
      (node_origin='checkpoint_virtual' — DB sync miss 숨기지 않음).

    반환: (edges, virtual_nodes). virtual_nodes 는 호출부가 nodes/node_ids 에 합친다.
    """
    edges: List[PipelineEdge] = []
    virtual_nodes: List[PipelineNode] = []
    seen_virtual: set[str] = set()
    seen_edge: set[Tuple[str, str, str]] = set()

    def _add(source: str, target: str, kind: str, confidence: str, verification: str,
             bg_id: Optional[str]) -> None:
        if source == target:
            return
        key = (source, target, kind)
        if key in seen_edge:
            return
        seen_edge.add(key)
        edges.append(PipelineEdge(
            source=source, target=target, kind=kind, confidence=confidence,
            verification_status=verification, bg_id=bg_id,
        ))

    for bg_id, g in groups.items():
        png_path = g.get("png_path") or ""
        loc_short = g.get("location_id") or ""
        canon = location_canon_by_short.get(loc_short)

        bg_node_id = chain_bg_by_bgid.get(bg_id)
        if bg_node_id is None:
            # DB sync miss — png_path 로 checkpoint virtual 노드 생성.
            if not png_path:
                continue
            vid = "virtual:" + hashlib.sha1(png_path.encode("utf-8")).hexdigest()
            if vid not in seen_virtual and vid not in node_ids:
                seen_virtual.add(vid)
                virtual_nodes.append(PipelineNode(
                    id=vid,
                    node_origin="checkpoint_virtual",
                    asset_type="chain_bg",
                    entity_id=canon,
                    episode_id=episode_id,
                    variant_label=bg_id,
                    full_url=png_path,
                    status="checkpoint_only",
                ))
            bg_node_id = vid

        for sid in g.get("shot_ids") or []:
            m = _SHOT_ID_RE.match(str(sid))
            if not m:
                continue
            shot_key = (int(m.group(1)), int(m.group(2)))
            for scene_id in scene_assets_by_shot.get(shot_key, []):
                _add(bg_node_id, scene_id, "background", "planned_only", "unverified", bg_id)

        if canon:
            for fp_id in fp_ids_by_canon.get(canon, []):
                if fp_id in node_ids:
                    _add(fp_id, bg_node_id, "fp", "checkpoint_structural",
                         "structural_inferred", bg_id)

    return edges, virtual_nodes


def _representative_scene_asset(scene_assets: List[ImageAsset]) -> Optional[str]:
    """still 한 개의 scene asset 들 중 대표 1개(결정론). is_primary → 최신 → id 순."""
    if not scene_assets:
        return None
    # 안정 정렬 3단계: id(tiebreak) → 최신 created_at → is_primary 우선.
    ordered = sorted(scene_assets, key=lambda a: str(a.id))
    ordered = sorted(ordered, key=lambda a: a.created_at or "", reverse=True)
    ordered = sorted(ordered, key=lambda a: 0 if getattr(a, "is_primary", 0) else 1)
    return ordered[0].id


def _reconstruct_prev_scene_edges(
    stills: Dict[str, SceneStill],
    scene_assets_by_still: Dict[str, List[ImageAsset]],
    node_ids: set[str],
) -> List[PipelineEdge]:
    """prev_scene 엣지: ``scene_still.dependent_scene_id``(직전 의존 still UUID) → 현재 씬.

    의존 still 의 대표 scene asset(is_primary→최신) → 현재 still 의 각 scene asset.
    confidence=still_fk_resolved(이미지 self-FK 아닌 still FK 해석 — Codex 보강),
    verification=structural_inferred. 투명성 위해 dependent_still_id/source_selection 부착.
    """
    edges: List[PipelineEdge] = []
    seen: set[Tuple[str, str]] = set()
    for still_id, still in stills.items():
        dep_id = getattr(still, "dependent_scene_id", None)
        if not dep_id or dep_id == still_id or dep_id not in stills:
            continue
        rep_dep = _representative_scene_asset(scene_assets_by_still.get(dep_id, []))
        if rep_dep is None or rep_dep not in node_ids:
            continue
        for cur in scene_assets_by_still.get(still_id, []):
            if cur.id not in node_ids or cur.id == rep_dep:
                continue
            key = (rep_dep, cur.id)
            if key in seen:
                continue
            seen.add(key)
            edges.append(PipelineEdge(
                source=rep_dep, target=cur.id, kind="prev_scene",
                confidence="still_fk_resolved", verification_status="structural_inferred",
                dependent_still_id=dep_id, source_selection="is_primary_or_latest",
            ))
    return edges


def _assign_edge_ids(edges: List[PipelineEdge]) -> None:
    """React Flow 용 안정·결정론 edge id 부여(in-place). ``kind:source:target``,
    동일 키 중복 시 ``#n`` suffix(Codex 권고 — index 기반 id 회피)."""
    seen: Dict[str, int] = {}
    for e in edges:
        base = f"{e.kind}:{e.source}:{e.target}"
        n = seen.get(base, 0)
        seen[base] = n + 1
        e.id = base if n == 0 else f"{base}#{n}"


def _load_call_log_overlay(
    db: OrmSession, project_id: str, episode_id: str, still_ids: List[str]
) -> Dict[str, Dict[str, Any]]:
    """still 별 scene_image_gen call_log 집계 — call-level overlay(설계 §9.2, Codex 보강).

    ★절대규칙: 라벨 텍스트 의미파싱 0. 구조 키(``metadata_json.still_id`` + operation_type)
    로만 조인하고, ``reference_image_ids`` 는 **원소 개수(len)** 만 센다(어떤 ref인지
    내용은 보지 않음 — UUID 매칭은 증명 불가).

    Returns ``{still_id: {has_call, max_ref_count, any_refs, prompt_authoritative}}``.
    call_log 없거나 쿼리 실패 시 빈 dict(non-fatal).
    """
    out: Dict[str, Dict[str, Any]] = {}
    if not still_ids:
        return out
    from sqlalchemy import text as _text

    sql = _text(
        """
        SELECT (metadata_json::jsonb)->>'still_id' AS sid,
               reference_image_ids, user_prompt
        FROM llm_call_log
        WHERE project_id = :pid
          AND episode_id = :eid
          AND operation_type IN ('single_scene_image_gen', 'scene_image_gen')
          AND (metadata_json::jsonb)->>'still_id' = ANY(:sids)
        ORDER BY created_at DESC
        """
    )
    try:
        rows = db.execute(
            sql, {"pid": project_id, "eid": episode_id, "sids": list(still_ids)}
        ).all()
    except Exception as exc:  # 잘못된 metadata_json cast 등 — non-fatal
        logger.warning("pipeline_graph: call_log overlay lookup failed: %s", exc)
        return out

    for sid, ref_ids_json, user_prompt in rows:
        if not sid:
            continue
        try:
            ref_list = json.loads(ref_ids_json or "[]")
            n = len(ref_list) if isinstance(ref_list, list) else 0
        except (ValueError, TypeError):
            n = 0
        agg = out.get(sid)
        if agg is None:
            # rows DESC → 첫 등장이 최신 호출 = prompt_authoritative.
            out[sid] = {
                "has_call": True,
                "max_ref_count": n,
                "any_refs": n > 0,
                "prompt_authoritative": user_prompt,
            }
        else:
            agg["max_ref_count"] = max(agg["max_ref_count"], n)
            agg["any_refs"] = agg["any_refs"] or n > 0
    return out


def _apply_call_log_overlay(
    nodes: List[PipelineNode],
    edges: List[PipelineEdge],
    overlay: Dict[str, Dict[str, Any]],
    scene_still_by_asset: Dict[str, str],
) -> None:
    """call-level overlay 를 노드(prompt_authoritative/actual_ref_count)와 background
    엣지(verification_status/call_ref_count)에 반영. in-place. 라벨 내용 미파싱.

    - 노드: 그 still 의 scene 호출 user_prompt + ref 개수.
    - background 엣지: ref 0개 → ``call_log_no_refs``(미attach 확정), ref>0 →
      ``refs_present_unresolved``(어떤 ref인지 UUID 매칭 불가). call 없으면 그대로.
    """
    for n in nodes:
        ov = overlay.get(n.still_id) if n.still_id else None
        if ov:
            n.prompt_authoritative = ov.get("prompt_authoritative")
            n.actual_ref_count = ov.get("max_ref_count")
    for e in edges:
        if e.kind != "background":
            continue
        sid = scene_still_by_asset.get(e.target)
        ov = overlay.get(sid) if sid else None
        if not ov:
            continue
        e.call_ref_count = ov["max_ref_count"]
        e.verification_status = (
            "refs_present_unresolved" if ov["any_refs"] else "call_log_no_refs"
        )


def build_pipeline_graph(
    db: OrmSession,
    project_id: str,
    episode_id: Optional[str] = None,
) -> PipelineGraphResponse:
    """프로젝트의 이미지 생성 파이프라인 그래프(에피소드 단위).

    노드 = image_asset (후속: checkpoint virtual). 엣지 = i2i + reference lineage
    (후속: FP→bg→씬 구조키, call_log 검증).
    """
    episodes = list_episodes(db, project_id)
    if episode_id is None and episodes:
        episode_id = _default_episode_id(db, project_id, episodes)

    entities = _entity_map(db, project_id)
    stills = _still_map(db, project_id, episode_id)

    assets = _query_image_assets(db, project_id, episode_id)
    nodes = [_asset_to_node(project_id, a, entities, stills) for a in assets]
    node_ids = {n.id for n in nodes}
    edges = _reconstruct_image_asset_edges(assets, node_ids)

    # ── FP→배경→씬 + prev_scene (구조키 조인) ──────────────────
    location_canon_by_short = {
        ent.short_id: cid
        for cid, ent in entities.items()
        if ent.entity_type == "location" and ent.short_id
    }
    chain_bg_by_bgid: Dict[str, str] = {
        a.variant_type: a.id
        for a in assets
        if a.asset_type == "chain_bg" and a.variant_type and a.episode_id == episode_id
    }
    fp_ids_by_canon: Dict[str, List[str]] = {}
    scene_assets_by_shot: Dict[Tuple[int, int], List[str]] = {}
    scene_assets_by_still: Dict[str, List[ImageAsset]] = {}
    scene_still_by_asset: Dict[str, str] = {}
    for a in assets:
        if a.asset_type == "floor_plan" and a.entity_id:
            fp_ids_by_canon.setdefault(a.entity_id, []).append(a.id)
        elif a.asset_type == "scene" and a.still_id:
            scene_assets_by_still.setdefault(a.still_id, []).append(a)
            scene_still_by_asset[a.id] = a.still_id
            st = stills.get(a.still_id)
            if st is not None and st.scene_index is not None and st.shot_index is not None:
                scene_assets_by_shot.setdefault(
                    (st.scene_index, st.shot_index), []
                ).append(a.id)

    groups = _load_background_groups(project_id, episode_id) if episode_id else {}
    bg_edges, virtual_nodes = _reconstruct_background_edges(
        groups,
        episode_id=episode_id,
        chain_bg_by_bgid=chain_bg_by_bgid,
        fp_ids_by_canon=fp_ids_by_canon,
        location_canon_by_short=location_canon_by_short,
        scene_assets_by_shot=scene_assets_by_shot,
        node_ids=node_ids,
    )
    nodes.extend(virtual_nodes)
    node_ids.update(n.id for n in virtual_nodes)
    edges.extend(bg_edges)
    edges.extend(
        _reconstruct_prev_scene_edges(stills, scene_assets_by_still, node_ids)
    )

    # ── Wave3: input_image_ids 실입력 lineage 엣지(별도 kind, 구조엣지와 공존) ──
    input_edges, dangling_inputs = _reconstruct_input_image_edges(assets, node_ids)
    edges.extend(input_edges)

    # ── Task4: call-level overlay (구조 키 조인, 라벨 의미파싱 없음) ──
    if episode_id:
        overlay = _load_call_log_overlay(
            db, project_id, episode_id, list(scene_assets_by_still.keys())
        )
        if overlay:
            _apply_call_log_overlay(nodes, edges, overlay, scene_still_by_asset)

    _assign_edge_ids(edges)

    from app.core.config import settings

    return PipelineGraphResponse(
        episode_id=episode_id,
        nodes=nodes,
        edges=edges,
        episodes=episodes,
        background_consumer_enabled=bool(
            getattr(settings, "background_chain_enabled", True)
        ),
        diagnostics={"dangling_input_refs": dangling_inputs},
    )