#!/usr/bin/env python3
"""Build the static pipeline atlas data from the repository's current source.

This script is intentionally documentation-only.  It reads production source,
prompt packs and one completed checkpoint set, then writes pipeline-data.js next
to itself.  It imports no application modules, so generation cannot initialize
the DB, model router, tracing, or any paid provider.
"""

from __future__ import annotations

import ast
import hashlib
import json
import re
from collections import Counter, defaultdict
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Iterable


DOC_DIR = Path(__file__).resolve().parent
ROOT = DOC_DIR.parents[1]
BACKEND = ROOT / "backend"
PROMPTS = ROOT / "prompts"
MANIFEST_FILE = BACKEND / "app/core/step_manifest.py"
REGISTRY_FILE = BACKEND / "app/core/steps/__init__.py"
SAMPLE_PROJECT = "da049582-2c6d-492c-979d-f468d61bab6e"
SAMPLE_EPISODE = "fb7a883f-baac-4145-9131-732ce628d474"
SAMPLE_CP_ROOT = (
    ROOT / "projects" / SAMPLE_PROJECT / "checkpoints" / "episodes" / SAMPLE_EPISODE
)


PHASES = [
    ("ingest", "입력·정규화", "Ingest & normalize", "#5d8cff"),
    ("story", "스토리 구조", "Story structure", "#8a77ff"),
    ("entity", "요소·관계", "Entities & relations", "#d26cff"),
    ("grounding", "고증·참조 조사", "Grounding & reference research", "#44c8c4"),
    ("direction", "연출·연속성", "Direction & continuity", "#ff6fae"),
    ("space", "공간·배경 계획", "Spatial & background planning", "#ff9f5d"),
    ("assets", "참조·배경 자산", "Reference & background assets", "#e9cb55"),
    ("final", "최종 스틸", "Final still pipeline", "#55d7b4"),
    ("output", "웹북·익스포트", "Webbook & export", "#55bde9"),
    ("runtime", "실행 기반", "Runtime infrastructure", "#98a6ba"),
]
PHASE_META = {phase[0]: phase for phase in PHASES}

GROUNDING_STEPS = {
    "grounding_a0", "grounding_chunk", "grounding_plan", "grounding_screen",
    "grounding_research", "reference_acquisition", "episode_reference_policy",
}


def rel(path: Path | str) -> str:
    p = Path(path)
    try:
        return p.resolve().relative_to(ROOT.resolve()).as_posix()
    except ValueError:
        return p.as_posix()


def read_text(path: Path, limit: int | None = None) -> str:
    try:
        text = path.read_text(encoding="utf-8")
    except (OSError, UnicodeDecodeError):
        return ""
    return text if limit is None else text[:limit]


def parse(path: Path) -> ast.Module | None:
    try:
        return ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
    except (OSError, SyntaxError, UnicodeDecodeError):
        return None


def literal_assignment(tree: ast.Module, name: str, default: Any) -> Any:
    for node in tree.body:
        target = None
        value = None
        if isinstance(node, ast.Assign) and node.targets:
            target, value = node.targets[0], node.value
        elif isinstance(node, ast.AnnAssign):
            target, value = node.target, node.value
        if isinstance(target, ast.Name) and target.id == name and value is not None:
            try:
                return ast.literal_eval(value)
            except (ValueError, TypeError):
                return default
    return default


def module_to_path(module: str) -> Path | None:
    if not module.startswith("app."):
        return None
    p = BACKEND / (module.replace(".", "/") + ".py")
    return p if p.exists() else None


def line_anchor(path: str, line: int | None) -> str:
    suffix = f"#L{line}" if line else ""
    return f"../../{path}{suffix}"


def source_ref(path: Path, start: int | None = None, end: int | None = None) -> dict[str, Any]:
    rp = rel(path)
    return {
        "path": rp,
        "start": start,
        "end": end,
        "href": line_anchor(rp, start),
    }


def version_key(value: str) -> tuple[int, str]:
    head, _, tail = value.partition(".")
    try:
        return int(head), tail
    except ValueError:
        return 0, value


def phase_for(step_id: str, cfg: dict[str, Any]) -> str:
    if step_id in {"planning_doc_analysis", "text_cleanup", "scene_segmentation", "scene_save", "episode_summary"}:
        return "ingest"
    if step_id in {"visual_world_rules", "scene_split", "entity_character_list", "scene_summary", "beat_extract", "shot_extract", "shot_validator"}:
        return "story"
    if step_id.startswith("entity_") or step_id.startswith("outlook_"):
        return "entity"
    if step_id in GROUNDING_STEPS or step_id.startswith("grounding_"):
        return "grounding"
    if step_id in {
        "shot_selection", "scene_director", "shot_director", "scene_cinematography",
        "scene_camera_flow", "shot_cinematography", "scene_dependency",
        "shot_dependency", "shot_dependency_t2i", "shot_staging",
        "shot_essence_extraction", "scene_consistency", "location_consistency",
        "scene_detail", "scene_verify", "t2i_review", "zoom_continuity_anchor",
        "shot_continuity", "shot_conti_light", "visual_continuity_anchor",
    }:
        return "direction"
    if step_id in {"world_guide", "ref_image_gen", "composite_image_gen", "character_state_variant", "background_render", "background_chain_render", "outdoor_place_canon"}:
        return "assets"
    if step_id == "scene_image_pipeline":
        return "final"
    if cfg.get("category") == "auxiliary" or step_id == "project_summary":
        return "output"
    return "space"


def class_registry() -> dict[str, dict[str, str]]:
    """Return step id -> class name/source path without importing the app."""
    tree = parse(REGISTRY_FILE)
    if tree is None:
        return {}
    imports: dict[str, Path] = {}
    for node in ast.walk(tree):
        if isinstance(node, ast.ImportFrom) and node.module:
            p = module_to_path(node.module)
            if p:
                for alias in node.names:
                    imports[alias.asname or alias.name] = p
    out: dict[str, dict[str, str]] = {}
    for node in ast.walk(tree):
        if not isinstance(node, ast.Assign) or not node.targets:
            continue
        target = node.targets[0]
        if isinstance(target, ast.Name) and target.id == "STEP_CLASSES" and isinstance(node.value, ast.Dict):
            for k, v in zip(node.value.keys, node.value.values):
                if isinstance(k, ast.Constant) and isinstance(k.value, str) and isinstance(v, ast.Name):
                    p = imports.get(v.id)
                    if p:
                        out[k.value] = {"class": v.id, "path": rel(p)}
        if (
            isinstance(target, ast.Subscript)
            and isinstance(target.value, ast.Name)
            and target.value.id == "STEP_CLASSES"
            and isinstance(node.value, ast.Name)
        ):
            key = target.slice
            if isinstance(key, ast.Constant) and isinstance(key.value, str):
                p = imports.get(node.value.id)
                if p:
                    out[key.value] = {"class": node.value.id, "path": rel(p)}
    return out


@dataclass
class SourceAnalysis:
    ref: dict[str, Any]
    doc: str
    excerpt: str
    methods: list[dict[str, Any]]
    calls: list[str]
    imports: list[str]
    prompt_calls: list[dict[str, Any]]
    checkpoint_reads: list[dict[str, Any]]
    model_refs: list[dict[str, Any]]


def call_name(node: ast.AST) -> str:
    if isinstance(node, ast.Name):
        return node.id
    if isinstance(node, ast.Attribute):
        parent = call_name(node.value)
        return f"{parent}.{node.attr}" if parent else node.attr
    return ""


def expr_label(node: ast.AST | None) -> str | None:
    if node is None:
        return None
    try:
        value = ast.literal_eval(node)
        return str(value)
    except (ValueError, TypeError):
        try:
            return ast.unparse(node)
        except Exception:
            return None


def clipped_expr(node: ast.AST | None, limit: int = 220) -> str:
    if node is None:
        return ""
    try:
        value = ast.unparse(node).replace("\n", " ")
    except Exception:
        return ""
    value = re.sub(r"\s+", " ", value).strip()
    return value if len(value) <= limit else value[: limit - 1] + "…"


def parameter_records(args: ast.arguments) -> list[dict[str, Any]]:
    """Losslessly-enough describe a callable's public input contract."""
    positional = list(args.posonlyargs) + list(args.args)
    defaults: list[ast.AST | None] = [None] * (len(positional) - len(args.defaults)) + list(args.defaults)
    out: list[dict[str, Any]] = []
    for index, (arg, default) in enumerate(zip(positional, defaults)):
        if arg.arg in {"self", "cls"}:
            continue
        out.append({
            "name": arg.arg,
            "annotation": clipped_expr(arg.annotation) or "Any",
            "default": clipped_expr(default) if default is not None else None,
            "required": default is None,
            "kind": "positional-only" if index < len(args.posonlyargs) else "positional-or-keyword",
        })
    if args.vararg:
        out.append({
            "name": f"*{args.vararg.arg}",
            "annotation": clipped_expr(args.vararg.annotation) or "Any",
            "default": None,
            "required": False,
            "kind": "var-positional",
        })
    for arg, default in zip(args.kwonlyargs, args.kw_defaults):
        out.append({
            "name": arg.arg,
            "annotation": clipped_expr(arg.annotation) or "Any",
            "default": clipped_expr(default) if default is not None else None,
            "required": default is None,
            "kind": "keyword-only",
        })
    if args.kwarg:
        out.append({
            "name": f"**{args.kwarg.arg}",
            "annotation": clipped_expr(args.kwarg.annotation) or "Any",
            "default": None,
            "required": False,
            "kind": "var-keyword",
        })
    return out


def return_records(fn: ast.FunctionDef | ast.AsyncFunctionDef) -> list[dict[str, Any]]:
    out: list[dict[str, Any]] = []
    for node in ast.walk(fn):
        if not isinstance(node, ast.Return):
            continue
        keys: list[str] = []
        items: dict[str, str] = {}
        if isinstance(node.value, ast.Dict):
            keys = [
                str(key.value) for key in node.value.keys
                if isinstance(key, ast.Constant) and isinstance(key.value, str)
            ]
            for key, value in zip(node.value.keys, node.value.values):
                if isinstance(key, ast.Constant) and isinstance(key.value, str):
                    items[str(key.value)] = clipped_expr(value, 120)
        out.append({"line": node.lineno, "expression": clipped_expr(node.value), "keys": keys, "items": items})
    return sorted(out, key=lambda item: item["line"])[:16]


def statement_flow(fn: ast.FunctionDef | ast.AsyncFunctionDef) -> list[dict[str, Any]]:
    """Turn the callable's top-level control flow into source-backed Korean steps."""
    rows: list[dict[str, Any]] = []
    for stmt in fn.body:
        title = ""
        detail = ""
        if isinstance(stmt, (ast.Assign, ast.AnnAssign, ast.AugAssign)):
            value = getattr(stmt, "value", None)
            target = getattr(stmt, "target", None)
            if target is None and isinstance(stmt, ast.Assign):
                target = stmt.targets[0] if stmt.targets else None
            target_text = clipped_expr(target, 90)
            if isinstance(value, (ast.Call, ast.Await)):
                call = value.value if isinstance(value, ast.Await) else value
                title = "호출 결과 저장"
                detail = f"{call_name(call.func)} 호출 결과를 {target_text or '지역 값'}에 저장"
            else:
                title = "값 구성"
                detail = f"{target_text or '지역 값'} = {clipped_expr(value)}"
        elif isinstance(stmt, ast.If):
            title, detail = "조건 분기", clipped_expr(stmt.test)
        elif isinstance(stmt, (ast.For, ast.AsyncFor)):
            title = "반복 처리"
            detail = f"{clipped_expr(stmt.target, 90)} ← {clipped_expr(stmt.iter)}"
        elif isinstance(stmt, (ast.With, ast.AsyncWith)):
            title = "컨텍스트 경계"
            detail = ", ".join(clipped_expr(item.context_expr, 120) for item in stmt.items)
        elif isinstance(stmt, ast.Try):
            caught = [clipped_expr(handler.type, 80) or "Exception" for handler in stmt.handlers]
            title, detail = "예외 처리", f"포착: {', '.join(caught)}"
        elif isinstance(stmt, ast.Match):
            title, detail = "패턴 분기", clipped_expr(stmt.subject)
        elif isinstance(stmt, ast.Return):
            title, detail = "결과 반환", clipped_expr(stmt.value)
        elif isinstance(stmt, ast.Raise):
            title, detail = "실패 중단", clipped_expr(stmt.exc)
        elif isinstance(stmt, ast.Expr) and isinstance(stmt.value, (ast.Call, ast.Await)):
            call = stmt.value.value if isinstance(stmt.value, ast.Await) else stmt.value
            title, detail = "부수효과 호출", clipped_expr(call)
        if title:
            rows.append({"title": title, "detail": detail, "line": stmt.lineno})
    # Keep the beginning and a real return/failure tail for long orchestrators.
    if len(rows) > 14:
        tail = next((row for row in reversed(rows) if row["title"] in {"결과 반환", "실패 중단"}), None)
        rows = rows[:13] + ([tail] if tail and tail not in rows[:13] else [])
    return rows


def callable_record(fn: ast.FunctionDef | ast.AsyncFunctionDef, *, entry: bool = False) -> dict[str, Any]:
    calls = sorted({call_name(n.func) for n in ast.walk(fn) if isinstance(n, ast.Call) and call_name(n.func)})
    return {
        "name": fn.name,
        "signature": ast.unparse(fn.args),
        "start": fn.lineno,
        "end": getattr(fn, "end_lineno", fn.lineno),
        "doc": ast.get_docstring(fn) or "",
        "async": isinstance(fn, ast.AsyncFunctionDef),
        "entry": entry,
        "parameters": parameter_records(fn.args),
        "returnAnnotation": clipped_expr(fn.returns) or "명시 없음",
        "returns": return_records(fn),
        "calls": calls[:60],
        "flow": statement_flow(fn),
    }


def checkpoint_reads_in_nodes(nodes: Iterable[ast.AST]) -> list[dict[str, Any]]:
    rows: list[dict[str, Any]] = []
    seen: set[tuple[str, int]] = set()
    for root in nodes:
        for node in ast.walk(root):
            if not isinstance(node, ast.Call):
                continue
            name = call_name(node.func)
            if not any(token in name for token in ("load_prev_checkpoint", "load_checkpoint")):
                continue
            step = expr_label(node.args[0]) if node.args else None
            if not step:
                continue
            step = step.strip("'\"")
            key = (step, node.lineno)
            if key not in seen:
                rows.append({"step": step, "loader": name, "line": node.lineno})
                seen.add(key)
    return sorted(rows, key=lambda item: item["line"])


def reachable_module_functions(
    tree: ast.Module,
    target: ast.FunctionDef | ast.AsyncFunctionDef,
    *,
    depth: int = 2,
) -> list[ast.FunctionDef | ast.AsyncFunctionDef]:
    functions = {
        node.name: node for node in tree.body
        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
    }
    reached = [target]
    seen = {target.name}
    frontier = [target]
    for _ in range(depth):
        next_frontier: list[ast.FunctionDef | ast.AsyncFunctionDef] = []
        for current in frontier:
            for call in ast.walk(current):
                if not isinstance(call, ast.Call):
                    continue
                name = call_name(call.func).split(".")[-1]
                helper = functions.get(name)
                if helper and helper.name not in seen:
                    seen.add(helper.name)
                    reached.append(helper)
                    next_frontier.append(helper)
        frontier = next_frontier
    return reached


def model_refs_in_nodes(tree: ast.Module, nodes: Iterable[ast.AST]) -> list[dict[str, Any]]:
    referenced = {node.id for root in nodes for node in ast.walk(root) if isinstance(node, ast.Name)}
    assignments: dict[str, tuple[str, int]] = {}
    for node in tree.body:
        target = None
        value = None
        if isinstance(node, ast.Assign) and len(node.targets) == 1:
            target, value = node.targets[0], node.value
        elif isinstance(node, ast.AnnAssign):
            target, value = node.target, node.value
        if not isinstance(target, ast.Name) or not isinstance(value, ast.Constant) or not isinstance(value.value, str):
            continue
        if "MODEL" not in target.id or "VERSION" in target.id:
            continue
        assignments[target.id] = (value.value, node.lineno)
    rows = [
        {"role": name, "alias": value, "line": line}
        for name, (value, line) in assignments.items()
        if name in referenced
    ]
    # Also retain literal model= aliases local to the actual call path.
    for root in nodes:
        for node in ast.walk(root):
            if not isinstance(node, ast.Call):
                continue
            for kw in node.keywords:
                if kw.arg == "model" and isinstance(kw.value, ast.Constant) and isinstance(kw.value.value, str):
                    rows.append({"role": f"{call_name(node.func)} model", "alias": kw.value.value, "line": node.lineno})
    unique: dict[tuple[str, str], dict[str, Any]] = {}
    for row in rows:
        unique[(row["role"], row["alias"])] = row
    return sorted(unique.values(), key=lambda item: item["line"])


def analyze_class(path: Path, class_name: str) -> SourceAnalysis:
    tree = parse(path)
    empty = SourceAnalysis(source_ref(path), "", "", [], [], [], [], [], [])
    if tree is None:
        return empty
    cls = next((n for n in tree.body if isinstance(n, ast.ClassDef) and n.name == class_name), None)
    if cls is None:
        return empty
    method_nodes: list[ast.FunctionDef | ast.AsyncFunctionDef] = []
    preferred = {"_execute", "execute", "run", "generate", "_config_hash", "check_gate", "run_steps_batch"}
    for fn in cls.body:
        if isinstance(fn, (ast.FunctionDef, ast.AsyncFunctionDef)) and (fn.name in preferred or fn.name.startswith("_load_")):
            method_nodes.append(fn)
    if not method_nodes:
        method_nodes = [fn for fn in cls.body if isinstance(fn, (ast.FunctionDef, ast.AsyncFunctionDef)) and not fn.name.startswith("__")]
    key_fn = next((fn for fn in method_nodes if fn.name in {"_execute", "execute", "run", "generate"}), method_nodes[0] if method_nodes else None)
    methods = [callable_record(fn, entry=fn is key_fn) for fn in method_nodes]
    calls = sorted({call_name(n.func) for n in ast.walk(cls) if isinstance(n, ast.Call) and call_name(n.func)})
    imports = sorted({
        n.module for n in ast.walk(tree)
        if isinstance(n, ast.ImportFrom) and n.module and n.module.startswith("app.")
    })
    prompt_calls = prompt_calls_in_nodes([cls])
    lines = read_text(path).splitlines()
    if key_fn:
        excerpt = "\n".join(lines[key_fn.lineno - 1:min(getattr(key_fn, "end_lineno", key_fn.lineno), key_fn.lineno + 54)])
    else:
        excerpt = "\n".join(lines[cls.lineno - 1:min(getattr(cls, "end_lineno", cls.lineno), cls.lineno + 54)])
    return SourceAnalysis(
        source_ref(path, cls.lineno, getattr(cls, "end_lineno", cls.lineno)),
        ast.get_docstring(cls) or ast.get_docstring(tree) or "",
        excerpt,
        methods[:18],
        calls[:80],
        imports,
        prompt_calls,
        checkpoint_reads_in_nodes([cls]),
        model_refs_in_nodes(tree, [cls]),
    )


def prompt_calls_in_nodes(nodes: Iterable[ast.AST]) -> list[dict[str, Any]]:
    out: list[dict[str, Any]] = []
    for root in nodes:
        for n in ast.walk(root):
            if not isinstance(n, ast.Call):
                continue
            name = call_name(n.func).split(".")[-1]
            if name not in {"load_prompt", "load_schema", "load_prompt_with_source", "get_effective_source"}:
                continue
            module = expr_label(n.args[0]) if n.args else None
            stem = expr_label(n.args[1]) if len(n.args) > 1 else None
            version = None
            for kw in n.keywords:
                if kw.arg == "version":
                    version = expr_label(kw.value)
            out.append({"loader": name, "module": module, "stem": stem, "versionExpr": version, "line": n.lineno})
    return out


def module_source_bundle(source_path: Path) -> tuple[list[Path], list[dict[str, Any]]]:
    """Include directly imported pipeline modules where prompt assembly usually lives."""
    paths = [source_path]
    all_calls: list[dict[str, Any]] = []
    tree = parse(source_path)
    if tree is None:
        return paths, all_calls
    all_calls.extend(prompt_calls_in_nodes([tree]))
    for n in ast.walk(tree):
        if isinstance(n, ast.ImportFrom) and n.module and n.module.startswith("app.modules"):
            p = module_to_path(n.module)
            if p and p not in paths:
                paths.append(p)
                imported_tree = parse(p)
                if imported_tree:
                    all_calls.extend(prompt_calls_in_nodes([imported_tree]))
    return paths, all_calls


def constants_and_maps(paths: Iterable[Path]) -> tuple[dict[str, Any], dict[str, dict[str, str]]]:
    constants: dict[str, Any] = {}
    maps: dict[str, dict[str, str]] = {}
    for path in paths:
        tree = parse(path)
        if tree is None:
            continue
        for node in tree.body:
            target = None
            value = None
            if isinstance(node, ast.Assign) and len(node.targets) == 1:
                target, value = node.targets[0], node.value
            elif isinstance(node, ast.AnnAssign):
                target, value = node.target, node.value
            if not isinstance(target, ast.Name) or value is None:
                continue
            try:
                v = ast.literal_eval(value)
            except (ValueError, TypeError):
                continue
            if isinstance(v, str) and ("VERSION" in target.id or target.id == "_MODULE"):
                constants[target.id] = v
            if isinstance(v, dict) and "VERSION" in target.id and all(isinstance(k, str) and isinstance(x, str) for k, x in v.items()):
                maps[target.id] = v
    return constants, maps


def normalize_module_expr(value: str | None, constants: dict[str, Any]) -> str | None:
    if not value:
        return None
    if value in constants and isinstance(constants[value], str):
        return constants[value]
    if value.startswith(("'", '"')):
        try:
            parsed = ast.literal_eval(value)
            return parsed if isinstance(parsed, str) else value
        except Exception:
            return value
    return value if re.fullmatch(r"[a-zA-Z0-9_]+", value) else None


def resolve_version_expr(expr: str | None, constants: dict[str, Any], maps: dict[str, dict[str, str]]) -> list[str]:
    if not expr:
        return []
    raw = constants.get(expr, expr)
    if not isinstance(raw, str):
        return []
    direct = [raw] if re.fullmatch(r"\d+(?:\.\d+)?", raw) else []
    found: list[str] = []
    for mapping in maps.values():
        if raw in mapping:
            found.append(mapping[raw])
        if raw in mapping.values():
            found.append(raw)
    return sorted(set(direct + found), key=version_key)


def prompt_pack_info(module: str, source_paths: list[Path], calls: list[dict[str, Any]]) -> dict[str, Any] | None:
    root = PROMPTS / "_base" / module
    if not root.is_dir():
        return None
    versions = sorted([p.name for p in root.iterdir() if p.is_dir()], key=version_key)
    constants, maps = constants_and_maps(source_paths)
    selectors = {k: str(v) for k, v in constants.items() if "VERSION" in k}
    detected: set[str] = set()
    relevant_calls = [c for c in calls if normalize_module_expr(c.get("module"), constants) == module]
    for c in relevant_calls:
        detected.update(resolve_version_expr(c.get("versionExpr"), constants, maps))
    for name, value in selectors.items():
        if module.upper() in name or module == "still_recipe" or len(maps) == 1:
            detected.update(resolve_version_expr(value, constants, maps))
    # Unpinned load follows prompt_loader's per-stem latest behavior.
    effective_by_stem: dict[str, str] = {}
    for version in versions:
        for f in (root / version).iterdir():
            if f.is_file() and f.suffix in {".md", ".json"}:
                current = effective_by_stem.get(f.name)
                if current is None or version_key(version) > version_key(current):
                    effective_by_stem[f.name] = version
    requested_stems = {
        str(c["stem"]).strip("'\"") for c in relevant_calls
        if c.get("stem") and re.fullmatch(r"[A-Za-z0-9_.'\"-]+", str(c["stem"]))
    }
    if not detected:
        for stem in requested_stems:
            filename = stem if stem.endswith((".md", ".json")) else f"{stem}.md"
            if filename in effective_by_stem:
                detected.add(effective_by_stem[filename])
    if not detected and versions:
        detected.add(versions[-1])
    pack_files: list[dict[str, Any]] = []
    for version in sorted((v for v in detected if (root / v).is_dir()), key=version_key):
        for f in sorted((root / version).iterdir()):
            if not f.is_file() or f.suffix not in {".md", ".json"}:
                continue
            content = read_text(f, 24000)
            pack_files.append({
                "name": f.name,
                "version": version,
                "path": rel(f),
                "href": line_anchor(rel(f), 1),
                "sha256": hashlib.sha256(f.read_bytes()).hexdigest()[:16],
                "bytes": f.stat().st_size,
                "truncated": f.stat().st_size > len(content.encode("utf-8")),
                "content": content,
            })
    return {
        "module": module,
        "availableVersions": versions,
        "detectedVersions": sorted(detected, key=version_key),
        "selectors": selectors,
        "effectiveByStem": effective_by_stem,
        "calls": relevant_calls,
        "files": pack_files,
        "selectionNote": (
            "명시 selector/호출식에서 감지한 팩" if relevant_calls or selectors
            else "명시 selector를 찾지 못해 stem별 최신 파일을 표시"
        ),
    }


def compact_sample(value: Any, *, depth: int = 0) -> Any:
    """Keep examples useful without copying an entire episode into every node."""
    if depth >= 4:
        if isinstance(value, dict):
            return f"<object · {len(value)} keys>"
        if isinstance(value, list):
            return f"<array · {len(value)} items>"
    if isinstance(value, dict):
        items = list(value.items())
        out = {str(k): compact_sample(v, depth=depth + 1) for k, v in items[:12]}
        if len(items) > 12:
            out["…"] = f"{len(items) - 12} keys omitted"
        return out
    if isinstance(value, list):
        out = [compact_sample(item, depth=depth + 1) for item in value[:2]]
        if len(value) > 2:
            out.append(f"<{len(value) - 2} more items>")
        return out
    if isinstance(value, str):
        value = value.replace("\x00", "")
        return value if len(value) <= 320 else value[:319] + "…"
    if isinstance(value, (int, float, bool)) or value is None:
        return value
    return repr(value)[:240]


def runtime_evidence(step_id: str) -> dict[str, Any] | None:
    step_root = SAMPLE_CP_ROOT / step_id
    path = step_root / "manifest.json"
    if not path.exists():
        archives = sorted(step_root.glob("manifest_*.json"), reverse=True) if step_root.is_dir() else []
        path = archives[0] if archives else path
    if not path.exists():
        return None
    try:
        obj = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError):
        return None
    data = obj.get("data")
    shape: dict[str, Any] = {"type": type(data).__name__}
    if isinstance(data, dict):
        shape["keys"] = list(data)[:24]
        shape["keyCount"] = len(data)
        for key in ("scenes", "entities", "shots", "items", "results", "images"):
            value = data.get(key)
            if isinstance(value, list):
                shape[f"{key}Count"] = len(value)
    elif isinstance(data, list):
        shape["length"] = len(data)
    return {
        "sampleProject": SAMPLE_PROJECT,
        "sampleEpisode": SAMPLE_EPISODE,
        "path": rel(path),
        "href": line_anchor(rel(path), 1),
        "status": obj.get("status"),
        "resolvedModel": obj.get("resolved_model") or obj.get("model"),
        "startedAt": obj.get("started_at"),
        "completedAt": obj.get("completed_at"),
        "schemaVersion": obj.get("schema_version"),
        "configHash": obj.get("config_hash"),
        "inputHash": obj.get("input_hash"),
        "outputShape": shape,
        "dataSample": compact_sample(data),
        "sampleKind": "완주 판 checkpoint의 실제 data를 길이 제한해 발췌",
    }


def scan_step_nodes() -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
    tree = parse(MANIFEST_FILE)
    if tree is None:
        raise RuntimeError(f"Cannot parse {MANIFEST_FILE}")
    manifest = literal_assignment(tree, "STEP_MANIFEST", {})
    registry = class_registry()
    nodes: list[dict[str, Any]] = []
    edges: list[dict[str, Any]] = []
    prompt_dirs = {p.name for p in (PROMPTS / "_base").iterdir() if p.is_dir()}
    for step_id, cfg in manifest.items():
        reg = registry.get(step_id, {})
        source_path = ROOT / reg["path"] if reg.get("path") else MANIFEST_FILE
        analysis = analyze_class(source_path, reg.get("class", "")) if reg else SourceAnalysis(source_ref(MANIFEST_FILE), "", "", [], [], [], [], [], [])
        source_paths, prompt_calls = module_source_bundle(source_path) if reg else ([MANIFEST_FILE], [])
        prompt_calls = analysis.prompt_calls + prompt_calls
        constants, _ = constants_and_maps(source_paths)
        modules: set[str] = set()
        if step_id in prompt_dirs:
            modules.add(step_id)
        for call in prompt_calls:
            module = normalize_module_expr(call.get("module"), constants)
            if module in prompt_dirs:
                modules.add(module)
        prompt_packs = [p for p in (prompt_pack_info(m, source_paths, prompt_calls) for m in sorted(modules)) if p]
        phase = phase_for(step_id, cfg)
        lifecycle = cfg.get("lifecycle", "active")
        applicability = cfg.get("applicability", "always")
        description = analysis.doc.strip() or f"{cfg.get('label', step_id)} 단계. STEP_MANIFEST의 의존성과 실행 계약을 따른다."
        node = {
            "id": f"step:{step_id}",
            "kind": "manifest-step",
            "pipelineId": step_id,
            "title": cfg.get("label", step_id),
            "subtitle": step_id,
            "phase": phase,
            "phaseLabel": PHASE_META[phase][1],
            "category": cfg.get("category", "analysis"),
            "order": cfg.get("order", 0),
            "model": cfg.get("default_model", "-"),
            "provider": cfg.get("provider", "-"),
            "lifecycle": lifecycle,
            "applicability": applicability,
            "stepType": cfg.get("step_type", "transform"),
            "fanOut": bool(cfg.get("fan_out")),
            "dependsOn": list(cfg.get("depends_on", [])),
            "replacedBy": cfg.get("replaced_by"),
            "schemaVersion": cfg.get("schema_version"),
            "resumeSensitive": bool(cfg.get("resume_sensitive")),
            "allowPartialDownstream": cfg.get("allow_partial_downstream", True),
            "modifiesCheckpoints": cfg.get("modifies_checkpoints", []),
            "description": description,
            "source": analysis.ref,
            "sourceModule": analysis.ref.get("path", rel(MANIFEST_FILE)),
            "className": reg.get("class"),
            "implemented": bool(reg and analysis.methods),
            "analysisNote": (
                "manifest class의 실행 진입점과 checkpoint/model 참조를 AST로 추출"
                if reg and analysis.methods else
                "현재 STEP_CLASSES registry에 구현 심볼이 없음; manifest 선언만 표시"
            ),
            "methods": analysis.methods,
            "calls": analysis.calls,
            "imports": analysis.imports,
            "checkpointReads": analysis.checkpoint_reads,
            "modelRefs": analysis.model_refs,
            "codeExcerpt": analysis.excerpt,
            "prompts": prompt_packs,
            "runtime": runtime_evidence(step_id),
            "searchText": " ".join([
                step_id, str(cfg.get("label", "")), description,
                analysis.ref.get("path", ""), " ".join(modules),
            ]).lower(),
        }
        nodes.append(node)
        for dep in cfg.get("depends_on", []):
            edges.append({
                "id": f"dep:{dep}>{step_id}",
                "source": f"step:{dep}",
                "target": f"step:{step_id}",
                "kind": "dependency",
                "label": "depends on",
            })
    return nodes, edges


def symbol_analysis(path_str: str, symbol: str) -> dict[str, Any]:
    path = ROOT / path_str
    tree = parse(path)
    if tree is None:
        return {
            "source": source_ref(path), "description": "", "codeExcerpt": "", "calls": [],
            "methods": [], "checkpointReads": [], "modelRefs": [], "implemented": False,
            "analysisNote": "Python AST 파싱 실패",
        }
    target = next((n for n in ast.walk(tree) if isinstance(n, (ast.ClassDef, ast.FunctionDef, ast.AsyncFunctionDef)) and n.name == symbol), None)
    if target is None:
        return {
            "source": source_ref(path), "description": ast.get_docstring(tree) or "", "codeExcerpt": "",
            "calls": [], "methods": [], "checkpointReads": [], "modelRefs": [], "implemented": False,
            "analysisNote": f"{symbol} 심볼이 현재 파일에 없음; manifest에만 남은 선언",
        }
    lines = read_text(path).splitlines()
    end = getattr(target, "end_lineno", target.lineno)
    excerpt = "\n".join(lines[target.lineno - 1:min(end, target.lineno + 70)])
    scope_nodes: list[ast.AST] = [target]
    method_nodes: list[ast.FunctionDef | ast.AsyncFunctionDef] = []
    if isinstance(target, ast.ClassDef):
        method_nodes = [fn for fn in target.body if isinstance(fn, (ast.FunctionDef, ast.AsyncFunctionDef))]
        entry = next((fn for fn in method_nodes if fn.name in {"_execute", "execute", "run", "generate", "render", "validate"}), method_nodes[0] if method_nodes else None)
    else:
        reached = reachable_module_functions(tree, target)
        method_nodes = reached
        scope_nodes = list(reached)
        entry = target
    methods = [callable_record(fn, entry=fn is entry) for fn in method_nodes[:24]]
    calls = sorted({call_name(n.func) for root in scope_nodes for n in ast.walk(root) if isinstance(n, ast.Call) and call_name(n.func)})
    return {
        "source": source_ref(path, target.lineno, end),
        "description": ast.get_docstring(target) or ast.get_docstring(tree) or "",
        "codeExcerpt": excerpt,
        "calls": calls[:90],
        "methods": methods,
        "checkpointReads": checkpoint_reads_in_nodes(scope_nodes),
        "modelRefs": model_refs_in_nodes(tree, scope_nodes),
        "implemented": True,
        "analysisNote": "entry callable과 같은 모듈의 직접 helper를 2-hop까지 AST 추적",
    }


MANUAL_SPECS: list[dict[str, Any]] = [
    {"id": "api:episode_create", "title": "에피소드 업로드 API", "pipelineId": "POST /projects/{project_id}/episodes/", "phase": "ingest", "kind": "entry", "path": "backend/app/api/v1/episodes.py", "symbol": "create_episode", "description": "multipart 시나리오 파일과 에피소드 메타데이터를 받아 EpisodeService의 저장·추출 경계로 넘긴다."},
    {"id": "input:upload", "title": "시나리오 업로드", "pipelineId": "episode.create", "phase": "ingest", "kind": "entry", "path": "backend/app/services/episode_service.py", "symbol": "create_episode", "description": "PDF 원본을 프로젝트 자산에 저장하고 텍스트를 추출·언어 판별한 뒤 Episode 레코드를 만든다."},
    {"id": "api:run_steps", "title": "분석 실행 API", "pipelineId": "POST /steps/run-all", "phase": "runtime", "kind": "runtime", "path": "backend/app/api/v1/steps.py", "symbol": "run_all_steps", "description": "클라이언트가 선택한 lifecycle·mode·force 범위를 실행 서비스와 category dispatcher에 전달한다."},
    {"id": "runtime:dispatch", "title": "실행 계획·배치 디스패치", "pipelineId": "dispatch_category_run", "phase": "runtime", "kind": "runtime", "path": "backend/app/services/analysis_dispatch_service.py", "symbol": "run_steps_batch", "description": "요청된 step 목록을 공식 순서로 실행하며 blocked/partial cascade, 예산, force/resume 정책을 조정한다."},
    {"id": "runtime:runner", "title": "체크포인트 재개·게이트", "pipelineId": "StepRunner.run", "phase": "runtime", "kind": "runtime", "path": "backend/app/core/step_runner.py", "symbol": "StepRunner", "description": "입력·설정·schema 지문과 의존 상태를 비교해 skip/rerun/block을 결정하고 manifest를 원자적으로 남긴다."},
    {"id": "runtime:checkpoint", "title": "체크포인트 파일 저장", "pipelineId": "StepRunner.save_checkpoint", "phase": "runtime", "kind": "runtime", "path": "backend/app/core/step_runner.py", "symbol": "save_checkpoint", "description": "각 단계 결과와 상태, 모델, 지문을 checkpoints/episodes/.../manifest*.json에 원자적으로 영속한다."},
    {"id": "runtime:projection", "title": "DB 프로젝션 동기화", "pipelineId": "checkpoint_sync.orchestrate_full_sync", "phase": "runtime", "kind": "runtime", "path": "backend/app/services/checkpoint_sync/orchestrator.py", "symbol": "orchestrate_full_sync", "description": "체크포인트를 정본으로 읽어 entity/relation/scene-still/outlook/episode read model을 한 트랜잭션으로 DB에 투영한다."},
    {"id": "runtime:prompt", "title": "프롬프트 팩 해석", "pipelineId": "prompt_loader.load_prompt", "phase": "runtime", "kind": "runtime", "path": "backend/app/modules/prompt_loader.py", "symbol": "load_prompt", "description": "DB active row를 우선하고 파일 팩을 fallback으로 사용한다. 명시 버전 또는 stem별 최신 버전을 해석한다."},
    {"id": "runtime:llm", "title": "모델 alias·호출 라우팅", "pipelineId": "llm_client", "phase": "runtime", "kind": "runtime", "path": "backend/app/modules/llm/llm_client.py", "symbol": "call_structured", "description": "manifest의 모델 alias를 물리 모델로 풀고 구조화 출력, 재시도, 사용량·추적 메타데이터를 통합한다."},
    {"id": "runtime:opik", "title": "Opik trace·span 연결", "pipelineId": "opik_trace.open_trace", "phase": "runtime", "kind": "runtime", "path": "backend/app/modules/llm/opik_trace.py", "symbol": "open_trace", "description": "StepRunner의 trace UID·episode thread·축 태그를 LLM과 이미지 provider span에 연결한다. 체크포인트는 병렬 worker 관찰 누락 때의 무료 정본 통로다."},
    {"id": "grounding:ledger", "title": "중앙 고증 장부 조립", "pipelineId": "grounding_central_inputs.build_ledger", "phase": "grounding", "kind": "grounding-internal", "path": "backend/app/modules/pipeline/grounding_central_inputs.py", "symbol": "build_ledger", "description": "screen·plan·outlook 결과를 final_id 기준 중앙 장부로 합쳐 다섯 owner 갈래가 같은 대상 신원을 사용하게 한다."},
    {"id": "grounding:obligations", "title": "참조 의무 계획", "pipelineId": "grounding_reference_obligations.plan", "phase": "grounding", "kind": "grounding-internal", "path": "backend/app/modules/pipeline/grounding_reference_obligations.py", "symbol": "plan", "description": "장부에서 실제로 참조를 살 대상과 목적을 정하고, 아웃룩 조각은 final_id별 한 벌 의무로 접는다."},
    {"id": "grounding:acquire", "title": "대상별 병렬 획득 조정", "pipelineId": "grounding_central_acquisition.run", "phase": "grounding", "kind": "grounding-internal", "path": "backend/app/modules/pipeline/grounding_central_acquisition.py", "symbol": "run", "description": "모든 대상의 1차를 먼저 병렬 수행하고 미해결 대상만 2차로 보내며, 구매 장부·재개·결과 순서를 결정적으로 유지한다."},
    {"id": "grounding:target_research", "title": "대상 뼈대 웹 조사", "pipelineId": "grounding_target_research.make_writer", "phase": "grounding", "kind": "grounding-internal", "path": "backend/app/modules/pipeline/grounding_target_research.py", "symbol": "make_writer", "description": "원고 근거와 시대·지역 좌표를 읽어 대상 전체 뼈대의 판정 기준, 좁은 질의와 넓은 질의를 한 번 조사하고 재개 가능한 지문을 만든다.", "promptModules": ["grounding_target_research", "coarse_type_pick"]},
    {"id": "grounding:rounds", "title": "검색·다운로드·자동 선택 라운드", "pipelineId": "reference_acquisition_rounds.acquire_one", "phase": "grounding", "kind": "grounding-internal", "path": "backend/app/modules/pipeline/reference_acquisition_rounds.py", "symbol": "acquire_one", "description": "좌표가 실린 질의로 후보를 받고 파일을 내려받아 판정한다. 1차가 없거나 약하면 더 포괄적인 2차로 넓히고 후보가 있으면 자동으로 가장 가까운 한 장을 남긴다."},
    {"id": "grounding:projection", "title": "획득 결과 정본 투영", "pipelineId": "grounding_central_acquisition.acquisition_projection", "phase": "grounding", "kind": "grounding-internal", "path": "backend/app/modules/pipeline/grounding_central_acquisition.py", "symbol": "acquisition_projection", "description": "중앙 획득 체크포인트의 유효 행만 소비자 계약 모양으로 접어 정책·sidecar·야외 경로가 같은 결과를 읽게 한다."},
    {"id": "grounding:sidecar", "title": "샷별 고증 참조 결속", "pipelineId": "grounding_sidecar_writer.write_for_shot", "phase": "grounding", "kind": "grounding-internal", "path": "backend/app/modules/pipeline/grounding_sidecar_writer.py", "symbol": "write_for_shot", "description": "샷의 visible entity와 중앙 투영을 교차해 실제 필요한 장소·부분·아웃룩 참조만 role·SHA·파일 좌표와 함께 scene_detail 카드에 원자적으로 붙인다."},
    {"id": "grounding:outdoor_supplement", "title": "야외 형태 참조 보충", "pipelineId": "grounding_outdoor_supplement.structure_form_obligations", "phase": "grounding", "kind": "grounding-internal", "path": "backend/app/modules/pipeline/grounding_outdoor_supplement.py", "symbol": "structure_form_obligations", "description": "실외 그룹이 요구하지만 중앙 결과에 없는 structure_form만 같은 자동 획득 경계로 보충하고, 별도 체크포인트를 중앙 결과와 충돌 없이 합친다."},
    {"id": "legacy:prompt_translation", "title": "T2I 문안 번역", "pipelineId": "prompt_translation", "phase": "final", "kind": "image-substep", "path": "backend/app/services/prompt_service.py", "symbol": "translate_if_korean", "description": "레거시 compound 경로에서 한국어가 남은 T2I 문안을 reference role 표현을 보존한 채 영어로 번역한다.", "promptModules": ["scene_image"], "model": "gpt-mini", "provider": "gemini"},
    {"id": "legacy:scene_t2i_gen", "title": "단일 씬 T2I 생성", "pipelineId": "scene_t2i_gen", "phase": "final", "kind": "image-substep", "path": "backend/app/modules/pipeline/scene_image_pipeline.py", "symbol": "generate_and_validate_scene", "description": "레거시 compound 경로의 기본 이미지를 생성하고 moderation retry와 단일 프레임 readback을 조정한다.", "model": "gemini-image", "provider": "gemini"},
    {"id": "legacy:scene_t2i_validation", "title": "씬 이미지 품질 검증", "pipelineId": "scene_t2i_validation", "phase": "final", "kind": "image-substep", "path": "backend/app/modules/image_validator.py", "symbol": "validate_scene_image", "description": "생성된 한 장을 비전 모델로 읽어 score, pass 여부와 구체 issues를 구조화한다.", "promptModules": ["image_validation"], "model": "gpt", "provider": "openai"},
    {"id": "legacy:prompt_sanitize", "title": "거절 문안 안전화", "pipelineId": "prompt_sanitize", "phase": "final", "kind": "image-substep", "path": "backend/app/modules/prompt_sanitizer.py", "symbol": "PromptSanitizer", "description": "provider moderation에 걸린 문안을 3단계 전략으로 다시 쓰고 semantic constraint를 결정론적으로 재부착한다.", "promptModules": ["prompt_sanitizer"], "model": "gpt", "provider": "openai"},
    {"id": "legacy:angle_recommend", "title": "카메라 앵글 추천", "pipelineId": "angle_recommend", "phase": "final", "kind": "image-substep", "path": "backend/app/services/fal_angle_helpers.py", "symbol": "select_and_recommend_angle", "description": "후보 중 앵글 조정 가치가 큰 이미지를 골라 horizontal, vertical, zoom 변환값을 추천한다.", "model": "gpt", "provider": "openai"},
    {"id": "legacy:fal_angle_apply", "title": "fal.ai 앵글 변환", "pipelineId": "fal_angle_apply", "phase": "final", "kind": "image-substep", "path": "backend/app/services/fal_angle_helpers.py", "symbol": "apply_fal_angle", "description": "추천된 카메라 회전·zoom을 이미지에 적용해 비교용 변형본을 만든다.", "model": "fal-ai", "provider": "fal"},
    {"id": "legacy:final_select", "title": "최종 대표 이미지 선택", "pipelineId": "final_select", "phase": "final", "kind": "image-substep", "path": "backend/app/services/fal_angle_helpers.py", "symbol": "select_final_best", "description": "원본 후보와 앵글 변형본 가운데 장면의 대표 hero image를 비전 판정으로 선택한다.", "model": "gpt", "provider": "openai"},
    {"id": "image:scope", "title": "샷 범위·재개 판정", "pipelineId": "still.scope_resume", "phase": "final", "kind": "image-substep", "path": "backend/app/services/still_recipe_service.py", "symbol": "run_still_recipe_generation", "description": "force/resume, 선택 샷, 기존 산출과 records 상태를 읽어 이번 호출에서 살 샷과 재사용할 샷을 나눈다."},
    {"id": "image:refs", "title": "참조 자산 조립", "pipelineId": "still_recipe.build_still_refs", "phase": "final", "kind": "image-substep", "path": "backend/app/modules/pipeline/still_recipe.py", "symbol": "build_still_refs", "description": "배경 plate, 콘티, 이전 선정본, 인물/상태 참조를 역할 라벨과 함께 정렬하고 첨부 권위를 제한한다."},
    {"id": "image:prompt", "title": "최종 스틸 문안 조립", "pipelineId": "still_recipe.build_still_prompt", "phase": "final", "kind": "image-substep", "path": "backend/app/modules/pipeline/still_recipe.py", "symbol": "build_still_prompt", "description": "SHOT TEXT·장소·카메라·조명·인물·소품·연속성·텍스트 정책 절을 샷 상태에 맞춰 조립한다.", "promptModules": ["still_recipe"]},
    {"id": "image:era", "title": "시대·장소 레퍼런스 조사", "pipelineId": "era_research", "phase": "final", "kind": "image-substep", "path": "backend/app/modules/pipeline/era_research.py", "symbol": "research_reference", "description": "조사 대상을 분해하고 검색 후보를 만든 뒤 VLM으로 사용할 시대·장소 사진을 선택하고 감사 기록을 남긴다.", "promptModules": ["era_research"], "modelRoles": [{"role": "ASSESS_MODEL · 조사 대상 판별", "alias": "gemini-flash", "line": 67}, {"role": "PICK_MODEL · 후보 사진 선택", "alias": "gemini-pro", "line": 73}]},
    {"id": "image:signage", "title": "대본 근거 표기 저작", "pipelineId": "signage_author.author_inscriptions", "phase": "final", "kind": "image-substep", "path": "backend/app/modules/pipeline/signage_author.py", "symbol": "author_inscriptions", "description": "대본·world facts가 실제로 요구하는 읽을 문자만 구조화해 최종 프롬프트에 공급한다.", "promptModules": ["signage_author"], "modelRoles": [{"role": "AUTHOR_MODEL · 근거 판별·짧은 저작", "alias": "gemini-flash", "line": 51}]},
    {"id": "image:rolls", "title": "N개 후보 이미지 생성", "pipelineId": "multiroll.run_multiroll_select", "phase": "final", "kind": "image-substep", "path": "backend/app/modules/pipeline/multiroll_select.py", "symbol": "run_multiroll_select", "description": "같은 문안·참조로 A..E 후보를 생성한다. 존재하는 roll 파일은 재구매하지 않고 재사용한다."},
    {"id": "image:judge", "title": "정·역순 후보 판정", "pipelineId": "multiroll.judge_abba", "phase": "final", "kind": "image-substep", "path": "backend/app/modules/pipeline/multiroll_select.py", "symbol": "run_multiroll_select", "description": "후보 위치 편향을 막기 위해 순서와 역순 판정을 교차하고 hard violation, 방향·공간·엔티티·물리를 읽어 승자를 고른다.", "promptModules": ["multiroll_judge"]},
    {"id": "image:critique", "title": "선정본 결함 비평", "pipelineId": "multiroll.critique", "phase": "final", "kind": "image-substep", "path": "backend/app/modules/pipeline/multiroll_select.py", "symbol": "_critique_and_fix", "description": "선정본 하나를 계약 축으로 검사해 수정이 필요한 구체 결함과 severity를 구조화한다.", "promptModules": ["multiroll_judge"]},
    {"id": "image:fix", "title": "결함 수정 i2i", "pipelineId": "multiroll.fix", "phase": "final", "kind": "image-substep", "path": "backend/app/modules/pipeline/multiroll_select.py", "symbol": "build_fix_prompt", "description": "비평에서 고른 결함만 겨냥한 수정 문안을 저작하고 선정본 단독 참조로 i2i 수정 후보를 만든다.", "promptModules": ["multiroll_judge"]},
    {"id": "image:rejudge", "title": "원본·수정본 재판정", "pipelineId": "multiroll.fix_rejudge", "phase": "final", "kind": "image-substep", "path": "backend/app/modules/pipeline/multiroll_select.py", "symbol": "_critique_and_fix", "description": "수정 전후를 다시 양방향 비교해 수정이 실제 개선일 때만 최종 _sel로 승격한다.", "promptModules": ["multiroll_judge"]},
    {"id": "image:cine", "title": "시네마틱 최종 변환", "pipelineId": "cine_transform.resolve_or_run_cine_transform", "phase": "final", "kind": "image-substep", "path": "backend/app/modules/pipeline/cine_transform.py", "symbol": "resolve_or_run_cine_transform", "description": "확정된 스틸을 provider i2i로 변환하되 소스의 장소·인물·물건·광원 정체를 보존하고 연출 재료만 전달한다.", "promptModules": ["still_recipe"]},
    {"id": "image:cine_verify", "title": "시네마틱 보존 관문", "pipelineId": "cine_verify", "phase": "final", "kind": "image-substep", "path": "backend/app/modules/pipeline/cine_verify.py", "symbol": "verify_cine_result", "description": "원본/변환본을 좌우 교대 2회 비교하고 둘 다 동의한 보존 축 위반만 기각한다. framing은 측정만 한다.", "promptModules": ["cine_verify"]},
    {"id": "image:persist", "title": "선정 자산·감사 기록 영속", "pipelineId": "still.persist", "phase": "final", "kind": "image-substep", "path": "backend/app/services/still_recipe_service.py", "symbol": "run_still_recipe_generation", "description": "최종 PNG, records.json, review_notes와 checkpoint 자산 메타를 영속하고 신선한 decline/reject도 투영 대상으로 표시한다."},
    {"id": "api:webbook", "title": "웹북 생성 API", "pipelineId": "POST /episodes/{episode_id}/generate-webbook", "phase": "output", "kind": "output", "path": "backend/app/api/v1/exports.py", "symbol": "generate_webbook", "description": "웹북 패키지 생성 요청을 ExportService에 전달하고 생성된 package metadata를 응답한다."},
    {"id": "output:webbook", "title": "웹북 패키지 생성", "pipelineId": "webbook.generate", "phase": "output", "kind": "output", "path": "backend/app/modules/webbook_generator.py", "symbol": "WebbookGenerator", "description": "에피소드 분석·씬·요소·이미지 정본을 읽어 웹북 구조 JSON과 패키지 레코드를 생성한다.", "promptModules": ["prototype_prompts"]},
    {"id": "api:pdf", "title": "PDF 렌더 API", "pipelineId": "POST /episodes/{episode_id}/render-pdf", "phase": "output", "kind": "output", "path": "backend/app/api/v1/exports.py", "symbol": "render_pdf", "description": "선택 에피소드 또는 프로젝트 결합 PDF 렌더 요청을 서비스에 전달한다."},
    {"id": "output:pdf", "title": "PDF 렌더", "pipelineId": "export.render_pdf", "phase": "output", "kind": "output", "path": "backend/app/services/export_service.py", "symbol": "render_pdfs", "description": "웹북 패키지와 scene primary/latest 이미지를 조합해 episode PDF와 결합본을 렌더한다."},
    {"id": "api:validate", "title": "PDF 검증 API", "pipelineId": "POST /exports/{filename}/validate", "phase": "output", "kind": "output", "path": "backend/app/api/v1/exports.py", "symbol": "validate_pdf", "description": "생성된 PDF 파일을 검증 모듈에 전달하고 구조화된 validation result를 반환한다."},
    {"id": "output:validate", "title": "PDF 산출 검증", "pipelineId": "export.validate_pdf", "phase": "output", "kind": "output", "path": "backend/app/modules/pdf_validator.py", "symbol": "PDFValidator", "description": "페이지·텍스트·이미지 등 PDF 산출 계약을 검사하고 검증 결과를 반환한다."},
    {"id": "api:html", "title": "HTML ZIP API", "pipelineId": "POST /episodes/{episode_id}/export-html-zip", "phase": "output", "kind": "output", "path": "backend/app/api/v1/exports.py", "symbol": "export_html_zip", "description": "웹용 HTML과 이미지 번들 생성 요청을 서비스에 전달하고 다운로드 경로를 반환한다."},
    {"id": "output:html", "title": "HTML·이미지 ZIP", "pipelineId": "export.html_zip", "phase": "output", "kind": "output", "path": "backend/app/services/export_service.py", "symbol": "generate_html_zip", "description": "에피소드별 HTML과 PNG/JPEG 자산을 묶어 브라우저용 ZIP을 만든다."},
    {"id": "api:project", "title": "프로젝트 JSON API", "pipelineId": "GET /export", "phase": "output", "kind": "output", "path": "backend/app/api/v1/exports.py", "symbol": "export_project_json", "description": "프로젝트 전체 정본을 외부 이관용 JSON 응답으로 직렬화한다."},
    {"id": "output:project", "title": "프로젝트 원본 익스포트", "pipelineId": "project.export", "phase": "output", "kind": "output", "path": "backend/app/services/project_export_service.py", "symbol": "ProjectExportService", "description": "프로젝트 데이터와 관련 자산을 외부 보관·이관 가능한 패키지로 묶는다."},
]


MANUAL_EDGES = [
    ("api:episode_create", "input:upload", "control", "service 호출"),
    ("input:upload", "api:run_steps", "control", "사용자 실행 요청"),
    ("api:run_steps", "runtime:dispatch", "control", "분석 시작"),
    ("runtime:dispatch", "runtime:runner", "control", "step 실행"),
    ("runtime:runner", "step:planning_doc_analysis", "control", "선택 입력"),
    ("runtime:runner", "step:text_cleanup", "control", "실행"),
    ("runtime:runner", "step:episode_summary", "control", "실행"),
    ("input:upload", "step:text_cleanup", "data", "fulltext"),
    ("input:upload", "step:episode_summary", "data", "fulltext"),
    ("runtime:runner", "runtime:checkpoint", "record", "manifest 기록"),
    ("runtime:checkpoint", "runtime:projection", "record", "read-model sync"),
    ("runtime:prompt", "runtime:llm", "control", "prompt+schema"),
    ("runtime:llm", "runtime:opik", "record", "trace"),
    ("step:grounding_screen", "grounding:ledger", "data", "고증 대상"),
    ("step:grounding_plan", "grounding:ledger", "data", "계획·근거"),
    ("step:outlook_phase3", "grounding:ledger", "data", "아웃룩 정본"),
    ("grounding:ledger", "grounding:obligations", "data", "owner별 장부"),
    ("grounding:obligations", "grounding:acquire", "control", "구매 대상"),
    ("grounding:acquire", "grounding:target_research", "control", "대상 조사"),
    ("grounding:target_research", "grounding:rounds", "data", "좁은·넓은 질의"),
    ("grounding:rounds", "step:reference_acquisition", "record", "선택·미해결 결과"),
    ("step:reference_acquisition", "grounding:projection", "data", "중앙 CP"),
    ("grounding:projection", "step:episode_reference_policy", "data", "canonical 참조"),
    ("grounding:projection", "grounding:sidecar", "data", "location·outlook 참조"),
    ("grounding:sidecar", "step:scene_detail", "data", "샷별 role·bytes"),
    ("grounding:projection", "grounding:outdoor_supplement", "data", "기존 structure_form"),
    ("step:outdoor_place_spec", "grounding:outdoor_supplement", "data", "실외 그룹·형태"),
    ("grounding:outdoor_supplement", "step:outdoor_structure_form_reference", "control", "부족한 장소만"),
    ("step:scene_image_pipeline", "image:scope", "control", "compound 내부"),
    ("step:scene_image_pipeline", "legacy:prompt_translation", "control", "legacy compound"),
    ("step:scene_image_pipeline", "legacy:scene_t2i_gen", "control", "legacy compound"),
    ("step:scene_image_pipeline", "legacy:scene_t2i_validation", "control", "legacy compound"),
    ("step:scene_image_pipeline", "legacy:prompt_sanitize", "control", "legacy retry"),
    ("step:scene_image_pipeline", "legacy:angle_recommend", "control", "declared sub-step"),
    ("step:scene_image_pipeline", "legacy:fal_angle_apply", "control", "declared sub-step"),
    ("step:scene_image_pipeline", "legacy:final_select", "control", "declared sub-step"),
    ("image:scope", "image:refs", "data", "대상 샷"),
    ("image:scope", "image:era", "control", "필요 시"),
    ("image:scope", "image:signage", "control", "기능 플래그"),
    ("image:era", "image:refs", "data", "선택 사진"),
    ("image:signage", "image:prompt", "data", "근거 표기"),
    ("image:refs", "image:prompt", "data", "역할 라벨"),
    ("image:prompt", "image:rolls", "control", "생성 문안"),
    ("image:rolls", "image:judge", "data", "후보 A..E"),
    ("image:judge", "image:critique", "control", "선정본"),
    ("image:critique", "image:fix", "control", "결함 있음"),
    ("image:fix", "image:rejudge", "data", "원본+수정본"),
    ("image:rejudge", "image:cine", "control", "최종 선정본"),
    ("image:critique", "image:cine", "control", "수정 불필요"),
    ("image:cine", "image:cine_verify", "control", "검증 ON"),
    ("image:cine", "image:persist", "data", "검증 OFF/통과"),
    ("image:cine_verify", "image:persist", "data", "통과 또는 원본 복권"),
    ("image:persist", "runtime:checkpoint", "record", "asset manifest"),
    ("image:persist", "output:webbook", "data", "scene stills"),
    ("step:scene_detail", "output:webbook", "data", "씬 상세"),
    ("step:entity_merge", "output:webbook", "data", "요소 정본"),
    ("input:upload", "output:webbook", "data", "대본"),
    ("api:webbook", "output:webbook", "control", "생성 요청"),
    ("output:webbook", "output:pdf", "control", "패키지"),
    ("api:pdf", "output:pdf", "control", "렌더 요청"),
    ("output:pdf", "output:validate", "control", "PDF"),
    ("api:validate", "output:validate", "control", "검증 요청"),
    ("image:persist", "output:html", "data", "이미지"),
    ("output:webbook", "output:html", "data", "구조"),
    ("api:html", "output:html", "control", "번들 요청"),
    ("input:upload", "output:project", "data", "프로젝트"),
    ("runtime:checkpoint", "output:project", "data", "체크포인트"),
    ("api:project", "output:project", "control", "익스포트 요청"),
]


# 코드 흐름 보기의 시각적 제어 구조. 조건은 실제 if/return 계약이 있는
# 자리만 적는다. 단순히 선이 둘이라는 이유로 XOR로 꾸미지 않는다.
FLOW_DECISIONS: list[dict[str, Any]] = [
    {
        "id": "decision:fix_needed",
        "title": "수정할 critical 결함이 있는가?",
        "source": "image:critique",
        "replaceTargets": ["image:fix", "image:cine"],
        "condition": "issues를 editable/deferred/unfixable로 나눈 뒤 editable critical이 1개 이상인지",
        "sourceRef": source_ref(BACKEND / "app/modules/pipeline/multiroll_select.py", 853, 914),
        "branches": [
            {"label": "YES", "condition": "fixable 중 severity=critical ≥ 1", "target": "image:fix"},
            {"label": "NO", "condition": "issues 없음 · 전부 non-editable · critical 없음", "outcome": "outcome:skip_fix", "outcomeTitle": "수정 생략", "target": "decision:cine_enabled"},
        ],
    },
    {
        "id": "decision:fix_winner",
        "title": "재판정 승자는?",
        "source": "image:rejudge",
        "replaceTargets": ["image:cine"],
        "condition": "정순·역순 판정을 합쳐 winner를 고름; 동점 우선순위는 원본 A",
        "sourceRef": source_ref(BACKEND / "app/modules/pipeline/multiroll_select.py", 1003, 1048),
        "branches": [
            {"label": "A", "condition": "winner == A", "outcome": "outcome:keep_original", "outcomeTitle": "원본 유지", "target": "decision:cine_enabled"},
            {"label": "B", "condition": "winner == B", "outcome": "outcome:use_fixed", "outcomeTitle": "수정본 채택", "target": "decision:cine_enabled"},
        ],
    },
    {
        "id": "decision:cine_enabled",
        "title": "시네마틱 변환을 켰는가?",
        "condition": "STILL_CINE_TRANSFORM_ENABLED에서 해석된 cine_on",
        "sourceRef": source_ref(BACKEND / "app/services/still_recipe_service.py", 4559, 4634),
        "branches": [
            {"label": "ON", "condition": "cine_on == true", "target": "image:cine"},
            {"label": "OFF", "condition": "cine_on == false", "outcome": "outcome:no_cine", "outcomeTitle": "선정 원본 사용", "target": "image:persist"},
        ],
    },
    {
        "id": "decision:cine_verify_enabled",
        "title": "보존 검증을 켰는가?",
        "source": "image:cine",
        "replaceTargets": ["image:cine_verify", "image:persist"],
        "condition": "STILL_CINE_VERIFY_ENABLED; OFF이면 판정 호출 없이 변환 결과를 사용",
        "sourceRef": source_ref(BACKEND / "app/modules/pipeline/cine_transform.py", 97, 151),
        "branches": [
            {"label": "ON", "condition": "_verify_on() == true", "target": "image:cine_verify"},
            {"label": "OFF", "condition": "_verify_on() == false", "outcome": "outcome:use_cine_unverified", "outcomeTitle": "변환본 사용", "target": "image:persist"},
        ],
    },
    {
        "id": "decision:cine_verify_result",
        "title": "불변축을 보존했는가?",
        "source": "image:cine_verify",
        "replaceTargets": ["image:persist"],
        "condition": "verify_cine_result의 ok; 판정기 오류·부분 판정은 기각 권한 없음",
        "sourceRef": source_ref(BACKEND / "app/modules/pipeline/cine_transform.py", 134, 170),
        "branches": [
            {"label": "PASS", "condition": "verify.ok == true", "outcome": "outcome:use_cine_verified", "outcomeTitle": "변환본 채택", "target": "image:persist"},
            {"label": "REJECT", "condition": "verify.ok == false · 합의된 축 변경", "outcome": "outcome:restore_original", "outcomeTitle": "원본 복권", "target": "image:persist"},
        ],
    },
]


FLOW_PARALLEL_GROUPS: list[dict[str, Any]] = [
    {
        "id": "parallel:multiroll_candidates",
        "title": "병행 수행 · 누락 후보 Roll A–E",
        "source": "image:rolls",
        "join": "image:judge",
        "condition": "parallel_rolls == true AND 누락 roll 수 > 1",
        "fallback": "조건이 아니면 같은 후보를 canonical label 순서로 순차 생성",
        "sourceRef": source_ref(BACKEND / "app/modules/pipeline/multiroll_select.py", 1485, 1524),
        "tasks": [
            {"id": f"parallel:roll:{label}", "title": f"Roll {label}"}
            for label in ("A", "B", "C", "D", "E")
        ],
    },
]


def prompt_paths_info(paths: list[str]) -> list[dict[str, Any]]:
    files = []
    for path_str in paths:
        path = ROOT / path_str
        if not path.is_file():
            continue
        content = read_text(path, 24000)
        files.append({
            "name": path.name,
            "version": "file",
            "path": rel(path),
            "href": line_anchor(rel(path), 1),
            "sha256": hashlib.sha256(path.read_bytes()).hexdigest()[:16],
            "bytes": path.stat().st_size,
            "truncated": path.stat().st_size > len(content.encode("utf-8")),
            "content": content,
        })
    return files


def manual_nodes() -> list[dict[str, Any]]:
    nodes: list[dict[str, Any]] = []
    for spec in MANUAL_SPECS:
        ana = symbol_analysis(spec["path"], spec["symbol"])
        source_path = ROOT / spec["path"]
        bundle, calls = module_source_bundle(source_path)
        packs = []
        constants, _ = constants_and_maps(bundle)
        prompt_module_names = set(spec.get("promptModules", []))
        for call in calls:
            module = normalize_module_expr(call.get("module"), constants)
            if module and (PROMPTS / "_base" / module).is_dir():
                prompt_module_names.add(module)
        for module in sorted(prompt_module_names):
            info = prompt_pack_info(module, bundle, calls)
            if info:
                packs.append(info)
        external_files = prompt_paths_info(spec.get("promptPaths", []))
        if external_files:
            packs.append({
                "module": "webbook_package",
                "availableVersions": ["file"],
                "detectedVersions": ["file"],
                "selectors": {"DB prompt_version": "v5"},
                "effectiveByStem": {f["name"]: "file" for f in external_files},
                "calls": [],
                "files": external_files,
                "selectionNote": "고정 파일 프롬프트; WebbookPackage 레코드는 prompt_version=v5로 기록",
            })
        phase = spec["phase"]
        description = spec.get("description") or ana["description"]
        nodes.append({
            "id": spec["id"],
            "kind": spec["kind"],
            "pipelineId": spec["pipelineId"],
            "title": spec["title"],
            "subtitle": spec["pipelineId"],
            "phase": phase,
            "phaseLabel": PHASE_META[phase][1],
            "category": spec["kind"],
            "order": 200 + len(nodes),
            "model": spec.get("model", "-"),
            "provider": spec.get("provider", "-"),
            "lifecycle": "active",
            "applicability": "internal",
            "stepType": spec["kind"],
            "fanOut": False,
            "dependsOn": [],
            "description": description,
            "source": ana["source"],
            "sourceModule": spec["path"],
            "className": spec["symbol"],
            "methods": ana["methods"],
            "calls": ana["calls"],
            "imports": [],
            "checkpointReads": ana["checkpointReads"],
            "modelRefs": ana["modelRefs"] + list(spec.get("modelRoles", [])),
            "implemented": ana["implemented"],
            "analysisNote": ana["analysisNote"],
            "codeExcerpt": ana["codeExcerpt"],
            "prompts": packs,
            "runtime": None,
            "searchText": " ".join([spec["title"], spec["pipelineId"], description, spec["path"], " ".join(spec.get("promptModules", []))]).lower(),
        })
    return nodes


def current_model_settings() -> dict[str, dict[str, str]]:
    """Read only whitelisted model names from config defaults and optional .env."""
    wanted = {
        "openai_model", "gemini_text_model", "gemini_flash_model", "gemini_lite_model",
        "gemini_image_model", "openai_image_model", "anthropic_judge_model",
        "anthropic_fable_model", "qwen_vlm_model", "grok_judge_model", "grok_image_model",
    }
    values: dict[str, dict[str, str]] = {}
    config_path = BACKEND / "app/core/config.py"
    tree = parse(config_path)
    if tree:
        for cls in (node for node in tree.body if isinstance(node, ast.ClassDef)):
            for node in cls.body:
                if not isinstance(node, ast.AnnAssign) or not isinstance(node.target, ast.Name):
                    continue
                name = node.target.id
                if name not in wanted or not isinstance(node.value, ast.Constant) or not isinstance(node.value.value, str):
                    continue
                values[name] = {
                    "value": node.value.value,
                    "basis": f"config default · {rel(config_path)}:{node.lineno}",
                }
    env_path = BACKEND / ".env"
    if env_path.is_file():
        env_names = {name.upper(): name for name in wanted}
        for line_no, line in enumerate(read_text(env_path).splitlines(), 1):
            stripped = line.strip()
            if not stripped or stripped.startswith("#") or "=" not in stripped:
                continue
            key, value = stripped.split("=", 1)
            field = env_names.get(key.strip())
            value = value.strip().strip("'\"")
            if field and value:
                values[field] = {
                    "value": value,
                    "basis": f"current .env override · {rel(env_path)}:{line_no}",
                }
    return values


def model_alias_catalog(settings: dict[str, dict[str, str]]) -> dict[str, dict[str, str]]:
    def configured(provider: str, field: str) -> dict[str, str]:
        info = settings.get(field, {"value": field, "basis": "설정 필드의 현재 값을 찾지 못함"})
        return {"provider": provider, "physical": info["value"], "setting": field, "basis": info["basis"]}

    catalog = {
        "gpt": configured("openai", "openai_model"),
        "gpt-mini": configured("gemini", "gemini_flash_model"),
        "gemini-pro": configured("gemini", "gemini_text_model"),
        "gemini-flash": configured("gemini", "gemini_flash_model"),
        "gemini-lite": configured("gemini", "gemini_lite_model"),
        "gemini-image": configured("gemini", "gemini_image_model"),
        "gpt-image-2": configured("openai", "openai_image_model"),
        "claude-opus": configured("anthropic", "anthropic_judge_model"),
        "claude-fable": configured("anthropic", "anthropic_fable_model"),
        "qwen": configured("dashscope", "qwen_vlm_model"),
        "grok": configured("xai", "grok_judge_model"),
        "grok-image": configured("xai", "grok_image_model"),
    }
    for alias, provider, physical in (
        ("gpt-terra", "openai", "gpt-5.6-terra"),
        ("gpt-luna", "openai", "gpt-5.6-sol"),
        ("gpt-nano", "openai", "gpt-5.4-nano"),
        ("gpt-4.1", "openai", "gpt-4.1"),
        ("gpt-4.1-mini", "openai", "gpt-4.1-mini"),
        ("fal-ai", "fal", "provider 작업별 endpoint/model"),
    ):
        catalog[alias] = {
            "provider": provider,
            "physical": physical,
            "setting": "고정/작업별",
            "basis": "llm_client alias table 또는 provider adapter",
        }
    return catalog


def looks_like_model(value: str, catalog: dict[str, dict[str, str]]) -> bool:
    lowered = value.lower()
    return value in catalog or lowered.startswith(("gpt", "gemini", "claude", "qwen", "x-ai/", "grok", "fal"))


def build_model_info(node: dict[str, Any], catalog: dict[str, dict[str, str]]) -> dict[str, Any]:
    rows: list[dict[str, Any]] = []

    def add(role: str, alias: str, *, provider_hint: str | None = None, line: int | None = None, basis: str = "") -> None:
        if not alias or alias == "-":
            return
        if alias == "mixed":
            rows.append({
                "role": role, "alias": alias, "provider": "mixed", "physical": "하위 호출별 상이",
                "setting": "하위 노드 참조", "basis": basis or "manifest/manual contract", "line": line,
            })
            return
        mapped = catalog.get(alias)
        if mapped:
            rows.append({"role": role, "alias": alias, **mapped, "line": line})
            return
        rows.append({
            "role": role,
            "alias": alias,
            "provider": provider_hint or "직접 모델 ID",
            "physical": alias,
            "setting": "literal",
            "basis": basis or "source literal",
            "line": line,
        })

    add("manifest/default", str(node.get("model") or ""), provider_hint=node.get("provider"), basis="STEP_MANIFEST 또는 manual node 선언")
    for ref in node.get("modelRefs", []):
        alias = str(ref.get("alias") or "")
        if looks_like_model(alias, catalog):
            add(str(ref.get("role") or "source model"), alias, line=ref.get("line"), basis="실행 심볼과 2-hop helper가 참조한 source constant")
    unique: list[dict[str, Any]] = []
    seen: set[tuple[str, str]] = set()
    for row in rows:
        key = (row["role"], row["alias"])
        if key not in seen:
            unique.append(row)
            seen.add(key)
    if not unique:
        paid_calls = [
            call for call in node.get("calls", [])
            if call.split(".")[-1] in {"call_structured", "call_text", "generate_image", "completion"}
            or call.endswith(("router.completion", "client.generate_image"))
        ]
        if paid_calls:
            return {
                "display": "동적 설정",
                "models": [{
                    "role": "runtime-selected model",
                    "alias": "동적",
                    "provider": node.get("provider") if node.get("provider") not in {None, "-"} else "router/provider adapter",
                    "physical": "project_config 또는 provider 설정에서 실행 시 결정",
                    "setting": "runtime",
                    "basis": f"AST model call: {', '.join(paid_calls[:4])}",
                    "line": None,
                }],
                "note": "이 노드는 모델 호출을 하지만 alias가 함수 인자·project_config·provider 설정에서 주입되어 정적 snapshot만으로 하나로 고정할 수 없습니다.",
            }
        return {
            "display": "모델 없음",
            "models": [],
            "note": "이 노드의 실행 심볼과 추적한 helper에서 유료 모델 호출 계약을 찾지 못했습니다. 결정론 코드이거나 모델 호출을 하위 노드에 위임합니다.",
        }
    aliases = list(dict.fromkeys(row["alias"] for row in unique))
    display = " + ".join(aliases[:2]) + (f" +{len(aliases) - 2}" if len(aliases) > 2 else "")
    return {
        "display": display,
        "models": unique,
        "note": "alias는 호출 계약, physical은 현재 config default/.env를 읽은 해석값입니다. 실행 중 project override가 있으면 달라질 수 있습니다.",
    }


def placeholder_for_parameter(param: dict[str, Any]) -> Any:
    default = param.get("default")
    if default is not None:
        try:
            return ast.literal_eval(default)
        except (ValueError, SyntaxError):
            return f"<default: {default}>"
    name = str(param.get("name", "")).lstrip("*").lower()
    annotation = str(param.get("annotation", "")).lower()
    if name == "project_id":
        return SAMPLE_PROJECT
    if name == "episode_id":
        return SAMPLE_EPISODE
    if name in {"db", "session"}:
        return "<SQLAlchemy Session>"
    if "path" in name or "path" in annotation:
        return "<filesystem path>"
    if any(token in annotation for token in ("list", "sequence", "tuple", "set")):
        return []
    if any(token in annotation for token in ("dict", "mapping")):
        return {}
    if "bool" in annotation:
        return True
    if "int" in annotation:
        return 1
    if "float" in annotation:
        return 0.5
    if "bytes" in annotation:
        return "<bytes>"
    if "image" in name:
        return "<image path or bytes>"
    if "prompt" in name or "text" in name:
        return "<text>"
    return f"<{param.get('annotation') or 'value'}>"


def inferred_output_example(entry: dict[str, Any] | None, node: dict[str, Any]) -> Any:
    if entry:
        for ret in entry.get("returns", []):
            items = ret.get("items") or {}
            if items:
                return {key: f"<from code: {expr}>" for key, expr in list(items.items())[:12]}
        annotation = str(entry.get("returnAnnotation") or "")
        lowered = annotation.lower()
        if "dict" in lowered or "mapping" in lowered:
            return {"return_type": annotation, "value": "<함수 내부에서 구성한 object>"}
        if "list" in lowered or "sequence" in lowered:
            return {"return_type": annotation, "value": ["<runtime item>"]}
        if "bool" in lowered:
            return True
        if "str" in lowered:
            return "<returned text>"
        if annotation not in {"", "명시 없음", "None"}:
            return {"return_type": annotation, "value": "<runtime value>"}
    return {"result": f"<{node.get('description', 'node result')[:120]}>"}


PLAIN_GLOSSARY: list[tuple[tuple[str, ...], str, str]] = [
    (("checkpoint", "체크포인트"), "checkpoint", "앞 단계 결과를 다시 계산하지 않도록 저장해 둔 중간 결과"),
    (("manifest",), "manifest", "checkpoint의 상태·모델·입출력 표식을 적은 실행 기록 JSON"),
    (("resume", "재개"), "resume", "이미 유효한 결과는 재사용하고 필요한 부분부터 이어서 실행하는 방식"),
    (("force",), "force", "기존 결과를 그대로 믿지 않고 사용자가 명시적으로 다시 실행하는 방식"),
    (("fingerprint", "config hash", "input hash", "지문"), "fingerprint/hash", "입력·설정이 이전 실행과 같은지 비교하는 짧은 식별값"),
    (("fan-out", "병렬"), "fan-out", "씬이나 샷처럼 여러 항목을 나눠 동시에 처리하는 방식"),
    (("prompt pack", "prompt", "프롬프트", "문안"), "prompt pack", "모델에게 역할·금지사항·출력 형식을 알려 주는 지시문 묶음"),
    (("provider",), "provider", "OpenAI·Gemini·fal처럼 실제 모델 호출을 처리하는 제공 서비스"),
    (("alias", "별칭"), "model alias", "코드가 쓰는 짧은 모델 이름. 실행 시 실제 물리 모델 이름으로 풀림"),
    (("vlm", "판정"), "VLM", "이미지를 보고 내용이나 품질을 구조화해 답하는 vision-language model"),
    (("i2i",), "i2i", "기존 이미지를 입력으로 받아 구도·화풍·결함 등을 수정하는 이미지 변환"),
    (("t2i",), "t2i", "글로 쓴 지시문에서 새 이미지를 만드는 text-to-image 생성"),
    (("upstream", "앞 단계"), "upstream", "현재 노드보다 먼저 실행되어 입력을 공급하는 단계"),
    (("downstream", "다음 단계"), "downstream", "현재 노드 결과를 받아 뒤에서 이어서 처리하는 단계"),
    (("schema", "스키마"), "schema", "반환 JSON에 어떤 칸과 자료형이 있어야 하는지 정한 형식 계약"),
]


def plainify_description(text: str) -> str:
    cleaned = re.sub(r"\s+", " ", (text or "").strip())
    cleaned = re.sub(r"^[A-Za-z0-9_]+\s*[—-]\s*Phase\s+[^.]+\.\s*", "", cleaned)
    replacements = (
        ("LLM 1회씩 t2i prompt 생성", "각 항목마다 언어 모델을 한 번 호출해 이미지 생성 지시문을 만듭니다"),
        ("t2i prompt", "이미지 생성 지시문"),
        ("T2I 프롬프트", "이미지 생성 지시문"),
        ("depends_on_fp DAG", "도면 사이의 선행 관계"),
        ("level 병렬 가능", "서로 의존하지 않는 도면은 동시에 처리 가능"),
        ("skip/rerun/block", "이전 결과 재사용·다시 실행·중단"),
        ("manifest", "실행 기록 파일(manifest)"),
        ("schema", "출력 형식(schema)"),
        ("fingerprint", "변경 감지 표식(fingerprint)"),
        ("provider", "모델 제공 서비스(provider)"),
        ("upstream", "앞 단계"),
        ("downstream", "다음 단계"),
    )
    for old, new in replacements:
        cleaned = cleaned.replace(old, new)
    sentences = re.split(r"(?<=[.!?다요])\s+", cleaned)
    shortened = " ".join(sentences[:2]).strip()
    return shortened if len(shortened) <= 360 else shortened[:359] + "…"


def build_plain_guide(
    node: dict[str, Any],
    node_by_id: dict[str, dict[str, Any]],
    entry: dict[str, Any] | None,
    inputs: list[dict[str, Any]],
    outputs: list[dict[str, Any]],
    steps: list[dict[str, Any]],
) -> dict[str, Any]:
    upstream_ids = node.get("incoming", [])
    downstream_ids = node.get("outgoing", [])
    upstream_titles = [node_by_id[item]["title"] for item in upstream_ids if item in node_by_id]
    downstream_titles = [node_by_id[item]["title"] for item in downstream_ids if item in node_by_id]
    input_names = [item["name"] for item in inputs[:6]]
    output_names = [item["name"].replace("checkpoint.data.", "") for item in outputs[:6]]

    receives = (
        f"앞 단계인 {', '.join(upstream_titles[:5])}에서 결과를 받습니다."
        if upstream_titles else
        f"호출할 때 {', '.join(input_names)} 값을 받습니다."
    )
    if len(upstream_titles) > 5:
        receives = receives[:-1] + f" 외 {len(upstream_titles) - 5}개 단계에서 결과를 받습니다."
    if downstream_titles:
        shown = ", ".join(downstream_titles[:5])
        passes = (
            f"{shown} 등 총 {len(downstream_titles)}개 다음 단계가 이 결과를 이어서 사용합니다."
            if len(downstream_titles) > 5 else
            f"정리한 결과를 {shown} 단계가 이어서 사용합니다."
        )
    else:
        passes = "이 결과가 사용자에게 전달되거나 최종 파일·DB 기록으로 남습니다."

    flow_translation = {
        "공식 upstream 계약 확인": "필요한 앞 단계 결과가 모두 준비됐는지 먼저 확인합니다.",
        "실행 진입": "준비된 입력으로 이 노드의 핵심 함수를 시작합니다.",
        "조건 분기": "설정과 입력 상태를 보고 실행할 경로를 고릅니다.",
        "호출 결과 저장": "필요한 하위 기능을 실행하고 돌아온 값을 모읍니다.",
        "값 구성": "다음 판단이나 출력에 필요한 값을 정리합니다.",
        "반복 처리": "여러 씬·샷·항목을 하나씩 또는 병렬로 처리합니다.",
        "컨텍스트 경계": "추적·트랜잭션·파일 같은 실행 범위를 안전하게 엽니다.",
        "예외 처리": "하위 호출이 실패했을 때 기록하거나 중단할 방법을 적용합니다.",
        "결과 반환": "정리된 결과를 호출자에게 돌려줍니다.",
        "출력 경계": "결과를 다음 노드가 읽을 수 있는 형태로 넘깁니다.",
        "실행 구현 없음": "현재 checkout에는 실제 실행 함수가 없어 동작하지 않습니다.",
    }
    how: list[str] = []
    for step in steps:
        translated = flow_translation.get(step["title"])
        if translated and translated not in how:
            how.append(translated)
        if len(how) == 4:
            break
    if not how:
        how = ["입력을 읽고 코드에 정의된 규칙대로 결과를 만듭니다."]

    runtime = node.get("runtime")
    implemented = bool(node.get("implemented"))
    model_display = node.get("modelInfo", {}).get("display", "모델 없음")
    if not implemented:
        walkthrough = "현재 checkout에는 이 노드의 실행 함수가 없습니다. 따라서 입력·출력 예시는 manifest 선언 모양만 보여 주며 실제 실행 결과가 아닙니다."
        example_basis = "미구현 선언 확인"
    elif runtime:
        status = runtime.get("status") or "기록됨"
        model_clause = (
            f"모델 기록은 {runtime.get('resolvedModel') or model_display}이고"
            if model_display != "모델 없음" else
            "직접 모델 호출 없이 코드로 처리됐고"
        )
        walkthrough = (
            f"완주 판에서는 {entry['name'] if entry else node.get('pipelineId')} 실행이 {status} 상태로 끝났습니다. "
            f"{model_clause}, 결과에는 {', '.join(output_names) or 'data'} 값이 실제로 남았습니다."
        )
        example_basis = "실제 완주 checkpoint 기록"
    else:
        model_clause = (
            f"실행 시 {model_display} 모델 경로를 사용합니다."
            if model_display != "모델 없음" else
            "이 노드 자체에서는 모델 호출이 확인되지 않았습니다."
        )
        walkthrough = (
            f"대표 예에서는 {', '.join(input_names) or '호출 컨텍스트'}를 넣으면 "
            f"{', '.join(output_names) or '함수 반환값'} 형태가 나옵니다. {model_clause} "
            "실제 값이 아니라 함수 시그니처와 return 코드에서 읽은 구조 예입니다."
        )
        example_basis = "AST 기반 대표 예시"

    if not implemented:
        why = "현재는 manifest에 이름만 남아 있고 실행 코드가 없어 실제 파이프라인에서는 동작하지 않습니다."
    elif downstream_titles:
        why = f"이 단계가 없으면 {downstream_titles[0]} 단계가 쓸 ‘{output_names[0] if output_names else '처리 결과'}’ 값이 만들어지지 않습니다."
    else:
        why = "파이프라인의 처리 결과를 외부에서 사용할 수 있는 산출물이나 기록으로 마무리하기 위해 필요합니다."
    plain_description = (
        "현재 checkout에는 실제 실행 코드가 없는 선언입니다."
        if not implemented else plainify_description(node.get("description", "코드에 정의된 변환을 수행합니다."))
    )
    one_liner = f"{node['title']}: {plain_description}"

    searchable = " ".join([
        node.get("pipelineId", ""), node.get("description", ""),
        node.get("logicDetail", {}).get("summary", ""), " ".join(node.get("calls", [])),
        json.dumps(node.get("modelInfo", {}), ensure_ascii=False),
    ]).lower()
    terms = [
        {"term": label, "meaning": meaning}
        for needles, label, meaning in PLAIN_GLOSSARY
        if any(needle.lower() in searchable for needle in needles)
    ][:8]
    return {
        "oneLiner": one_liner,
        "why": why,
        "receives": receives,
        "does": plain_description,
        "how": how,
        "passes": passes,
        "walkthrough": walkthrough,
        "exampleBasis": example_basis,
        "terms": terms,
    }


def build_logic_detail(node: dict[str, Any], node_by_id: dict[str, dict[str, Any]]) -> dict[str, Any]:
    methods = node.get("methods", [])
    entry = next((method for method in methods if method.get("entry")), methods[0] if methods else None)
    source_path = node.get("source", {}).get("path", node.get("sourceModule", ""))

    inputs: list[dict[str, Any]] = []
    seen_inputs: set[tuple[str, str]] = set()

    def add_input(name: str, type_name: str, required: bool, source: str, evidence: str, line: int | None = None) -> None:
        key = (name, source)
        if key in seen_inputs:
            return
        seen_inputs.add(key)
        inputs.append({
            "name": name, "type": type_name, "required": required, "source": source,
            "evidence": evidence, "line": line,
            "href": line_anchor(source_path, line) if line else node.get("source", {}).get("href"),
        })

    for dep in node.get("dependsOn", []):
        dep_node = node_by_id.get(f"step:{dep}")
        add_input(
            dep, "checkpoint manifest data", True, f"step:{dep}",
            f"STEP_MANIFEST depends_on · {dep_node.get('title') if dep_node else dep}",
        )
    for read in node.get("checkpointReads", []):
        add_input(
            str(read["step"]), "checkpoint manifest data", False, str(read["loader"]),
            "실행 심볼 AST에서 checkpoint loader 호출 확인", read.get("line"),
        )
    for param in (entry or {}).get("parameters", [])[:16]:
        add_input(
            str(param["name"]), str(param.get("annotation") or "Any"), bool(param.get("required")),
            "call argument", f"{param.get('kind')} · default={param.get('default') if param.get('default') is not None else '없음'}",
            (entry or {}).get("start"),
        )
    for pack in node.get("prompts", []):
        add_input(
            str(pack["module"]), "prompt pack", True, "prompt_loader",
            f"감지 버전: {', '.join(pack.get('detectedVersions', [])) or 'stem별 latest'}",
        )
    if not inputs:
        add_input("호출 컨텍스트", "runtime context", True, node.get("pipelineId", "entry"), "구현 심볼의 호출 경계")

    outputs: list[dict[str, Any]] = []
    runtime = node.get("runtime")
    if runtime:
        shape = runtime.get("outputShape", {})
        keys = shape.get("keys") or []
        if keys:
            for key in keys:
                count = shape.get(f"{key}Count")
                outputs.append({
                    "name": f"checkpoint.data.{key}", "type": "array" if count is not None else "runtime value",
                    "evidence": f"완주 판 manifest 실측{f' · {count} items' if count is not None else ''}",
                })
        else:
            outputs.append({"name": "checkpoint.data", "type": shape.get("type", "unknown"), "evidence": "완주 판 manifest 실측"})
    elif entry:
        return_keys: list[str] = []
        for ret in entry.get("returns", []):
            return_keys.extend(ret.get("keys", []))
        if return_keys:
            outputs.extend({"name": key, "type": "returned field", "evidence": "entry callable의 dict return AST"} for key in dict.fromkeys(return_keys))
        else:
            outputs.append({
                "name": "return value", "type": entry.get("returnAnnotation", "명시 없음"),
                "evidence": "type annotation과 return expression에서 추론",
            })
    else:
        outputs.append({
            "name": "실행 결과 없음", "type": "unimplemented declaration",
            "evidence": node.get("analysisNote", "현재 checkout에 실행 심볼 없음"),
        })

    steps: list[dict[str, Any]] = []
    if node.get("dependsOn"):
        names = ", ".join(node["dependsOn"])
        steps.append({"title": "공식 upstream 계약 확인", "detail": f"{names} 체크포인트가 준비되어야 실행한다.", "line": None, "href": node.get("source", {}).get("href")})
    if entry:
        steps.append({
            "title": "실행 진입",
            "detail": f"{'async ' if entry.get('async') else ''}{entry['name']}({entry.get('signature', '')})",
            "line": entry.get("start"), "href": line_anchor(source_path, entry.get("start")),
        })
        for flow in entry.get("flow", []):
            steps.append({**flow, "href": line_anchor(source_path, flow.get("line"))})
    else:
        steps.append({
            "title": "실행 구현 없음", "detail": node.get("analysisNote", "manifest 선언만 존재"),
            "line": None, "href": node.get("source", {}).get("href"),
        })
    if not any(step["title"] == "결과 반환" for step in steps):
        steps.append({
            "title": "출력 경계",
            "detail": ", ".join(output["name"] for output in outputs[:8]),
            "line": None, "href": runtime.get("href") if runtime else node.get("source", {}).get("href"),
        })

    args = {
        str(param["name"]): placeholder_for_parameter(param)
        for param in (entry or {}).get("parameters", [])[:12]
    }
    dependency_paths = {
        dep: f"{rel(SAMPLE_CP_ROOT)}/{dep}/manifest.json"
        for dep in node.get("dependsOn", [])
    }
    example_input: dict[str, Any] = {
        "call": f"{node.get('className') or node.get('pipelineId')}.{entry['name']}" if entry and node.get("className") != entry["name"] else (entry["name"] if entry else node.get("pipelineId")),
        "arguments": args,
    }
    if dependency_paths:
        example_input["dependency_checkpoints"] = dependency_paths
    if runtime:
        example_output = {
            "status": runtime.get("status"),
            "resolved_model_alias": runtime.get("resolvedModel"),
            "data": runtime.get("dataSample"),
        }
        example_kind = runtime.get("sampleKind")
    else:
        example_output = inferred_output_example(entry, node)
        example_kind = "실행하지 않고 type annotation·return AST로 만든 대표 shape; 실제 값으로 오해하면 안 됨"

    input_names = ", ".join(item["name"] for item in inputs[:5])
    call_names = [call for call in (entry or {}).get("calls", []) if not call.startswith(("str", "len", "dict", "list", "set", "getattr", "isinstance"))]
    call_summary = ", ".join(call_names[:4]) or "내부 결정론 로직"
    output_names = ", ".join(item["name"] for item in outputs[:5])
    summary = (
        f"{entry['name']}가 {input_names}을(를) 입력으로 받고 {call_summary} 경로를 실행한다. "
        f"결과는 {output_names} 형태로 나간다. 아래 순서는 {source_path}의 실제 AST 줄 순서이며, "
        f"예제 출력은 {'완주 판 checkpoint 실측' if runtime else '실행 없는 구조 추론'}이다."
        if entry else
        f"{node.get('pipelineId')}는 현재 checkout에 실행 심볼이 없는 선언이다. {node.get('analysisNote', '')}"
    )
    detail = {
        "summary": summary,
        "entryCallable": entry,
        "inputs": inputs,
        "steps": steps,
        "outputs": outputs,
        "exampleInput": example_input,
        "exampleOutput": example_output,
        "exampleKind": example_kind,
        "evidence": {
            "source": node.get("source"),
            "runtime": runtime.get("path") if runtime else None,
            "analysis": node.get("analysisNote"),
        },
    }
    detail["plainGuide"] = build_plain_guide(node, node_by_id, entry, inputs, outputs, steps)
    return detail


def main() -> None:
    step_nodes, edges = scan_step_nodes()
    extra_nodes = manual_nodes()
    for i, (source, target, kind, label) in enumerate(MANUAL_EDGES):
        edges.append({"id": f"manual:{i}", "source": source, "target": target, "kind": kind, "label": label})
    nodes = step_nodes + extra_nodes
    ids = {n["id"] for n in nodes}
    edges = [e for e in edges if e["source"] in ids and e["target"] in ids]

    incoming: dict[str, list[str]] = defaultdict(list)
    outgoing: dict[str, list[str]] = defaultdict(list)
    for edge in edges:
        incoming[edge["target"]].append(edge["source"])
        outgoing[edge["source"]].append(edge["target"])
    for node in nodes:
        node["incoming"] = incoming[node["id"]]
        node["outgoing"] = outgoing[node["id"]]

    settings = current_model_settings()
    model_catalog = model_alias_catalog(settings)
    node_by_id = {node["id"]: node for node in nodes}
    for node in nodes:
        node["modelInfo"] = build_model_info(node, model_catalog)
        node["modelLabel"] = node["modelInfo"]["display"]
        node["logicDetail"] = build_logic_detail(node, node_by_id)

    node_counts = Counter(n["kind"] for n in nodes)
    lifecycle_counts = Counter(n["lifecycle"] for n in step_nodes)
    runtime_counts = Counter((n.get("runtime") or {}).get("status", "no-sample") for n in step_nodes)
    prompt_modules = sorted({p["module"] for n in nodes for p in n.get("prompts", [])})
    sources = Counter(n["sourceModule"] for n in nodes)
    unique_prompt_files = {
        f["path"] for n in nodes for p in n.get("prompts", [])
        for f in p.get("files", [])
    }
    snapshot_paths = sorted(
        set(sources)
        | unique_prompt_files
        | {rel(MANIFEST_FILE), rel(REGISTRY_FILE), rel(DOC_DIR / "build_graph.py")}
    )
    source_snapshot = {
        path: hashlib.sha256((ROOT / path).read_bytes()).hexdigest()
        for path in snapshot_paths
        if (ROOT / path).is_file()
    }
    snapshot_digest = hashlib.sha256(
        json.dumps(source_snapshot, sort_keys=True, separators=(",", ":")).encode("utf-8")
    ).hexdigest()
    data = {
        "generatedAt": datetime.now(timezone.utc).isoformat(),
        "repoRoot": str(ROOT),
        "manifestSource": rel(MANIFEST_FILE),
        "sample": {"project": SAMPLE_PROJECT, "episode": SAMPLE_EPISODE, "checkpointRoot": rel(SAMPLE_CP_ROOT)},
        "modelSettings": {
            name: {"value": info["value"], "basis": info["basis"]}
            for name, info in sorted(settings.items())
        },
        "phases": [
            {"id": p[0], "label": p[1], "english": p[2], "color": p[3], "count": sum(1 for n in nodes if n["phase"] == p[0])}
            for p in PHASES
        ],
        "stats": {
            "nodes": len(nodes),
            "manifestSteps": len(step_nodes),
            "edges": len(edges),
            "promptModules": len(prompt_modules),
            "promptFiles": len(unique_prompt_files),
            "sourceFiles": len(sources),
            "nodeKinds": dict(node_counts),
            "lifecycles": dict(lifecycle_counts),
            "sampleStatuses": dict(runtime_counts),
        },
        "promptModules": prompt_modules,
        "sourceModules": [{"path": path, "count": count} for path, count in sorted(sources.items())],
        "sourceSnapshot": {
            "algorithm": "sha256",
            "digest": snapshot_digest,
            "files": source_snapshot,
        },
        "flowDecisions": FLOW_DECISIONS,
        "flowParallelGroups": FLOW_PARALLEL_GROUPS,
        "nodes": nodes,
        "edges": edges,
    }
    payload = json.dumps(data, ensure_ascii=False, separators=(",", ":"))
    out = DOC_DIR / "pipeline-data.js"
    out.write_text(f"window.PIPELINE_DATA={payload};\n", encoding="utf-8")
    print(json.dumps(data["stats"], ensure_ascii=False, indent=2))
    print(f"wrote {out} ({out.stat().st_size:,} bytes)")


if __name__ == "__main__":
    main()