from __future__ import annotations

import asyncio
import time
from collections import deque
from graphlib import TopologicalSorter, CycleError as _GraphlibCycleError
from typing import Any, Callable, Awaitable

from models.graph import GraphNode, GraphEdge, PortValueDict
from services.cache import ExecutionCache
from services.output import resolve_output_ref
from execution.error_classifier import classify_error
from models.events import (
    ExecutionEvent,
    QueuedEvent,
    ExecutingEvent,
    ExecutedEvent,
    ErrorEvent,
    ValidationErrorDetail,
    GraphCompleteEvent,
)

CycleError = _GraphlibCycleError


def _image_input_output(params: dict) -> dict:
    file_path = resolve_output_ref(str(params.get("filePath", "")))
    return {"image": {"type": "Image", "value": file_path}}


async def _maybe_probe_video_output(node: GraphNode, node_outputs: dict[str, Any]) -> None:
    """If a node produced a Video output and we haven't yet probed it,
    run ffprobe and stash sourceDuration / sourceFps / sourceIsVfr on
    node.params. Mirrors the upload-time probe so every video source —
    Veo, Kling, Sora, Seedance, Wan, etc. — opens its downstream editor
    with a pre-populated clip instead of forcing the user to Run the
    edit node first. Best-effort: failures are silent (the editor's
    Run-to-populate fallback still works as graceful degradation)."""
    if node.params.get("sourceDuration") is not None:
        return
    from pathlib import Path
    for v in node_outputs.values():
        if not isinstance(v, dict) or v.get("type") != "Video":
            continue
        value = v.get("value")
        if not isinstance(value, str) or not value:
            continue
        # Outputs are absolute local paths after handlers write to OUTPUT_ROOT.
        # Skip remote URLs and missing files — nothing we can probe.
        if value.startswith(("http://", "https://")):
            continue
        path = Path(value)
        if not path.exists():
            continue
        try:
            from services.ffmpeg import ffprobe_video
            probe = await ffprobe_video(path)
            node.params["sourceDuration"] = probe.duration
            node.params["sourceFps"] = probe.fps
            node.params["sourceIsVfr"] = probe.is_vfr
        except Exception:
            pass
        return

NODE_DEFS: dict[str, dict[str, Any]] = {
    "gpt-image-1-generate": {
        "inputPorts": [{"id": "prompt", "required": True}],
        "outputPorts": [{"id": "image"}],
        "envKeyName": "OPENAI_API_KEY",
    },
    "claude-chat": {
        "inputPorts": [
            {"id": "messages", "required": True},
            {"id": "images", "required": False},
        ],
        "outputPorts": [{"id": "text"}],
        "envKeyName": "ANTHROPIC_API_KEY",
    },
    "runway-gen4-turbo": {
        "inputPorts": [
            {"id": "image", "required": True},
            {"id": "prompt", "required": False},
        ],
        "outputPorts": [{"id": "video"}],
        "envKeyName": "RUNWAY_API_KEY",
    },
    "elevenlabs-tts": {
        "inputPorts": [{"id": "text", "required": True}],
        "outputPorts": [{"id": "audio"}],
        "envKeyName": "ELEVENLABS_API_KEY",
    },
    "flux-1-1-ultra": {
        "inputPorts": [
            {"id": "prompt", "required": True},
            {"id": "image", "required": False},
        ],
        "outputPorts": [{"id": "image"}],
        "envKeyName": "FAL_KEY",
    },
    "text-input": {
        "inputPorts": [],
        "outputPorts": [{"id": "text"}],
        "envKeyName": [],
    },
    "image-input": {
        "inputPorts": [],
        "outputPorts": [{"id": "image"}],
        "envKeyName": [],
    },
    "preview": {
        "inputPorts": [{"id": "input", "required": True}],
        "outputPorts": [],
        "envKeyName": [],
    },
    "combine-text": {
        "inputPorts": [
            {"id": "text1", "required": True},
            {"id": "text2", "required": False},
            {"id": "text3", "required": False},
        ],
        "outputPorts": [{"id": "text"}],
        "envKeyName": [],
    },
    "router": {
        "inputPorts": [{"id": "input", "required": True}],
        "outputPorts": [{"id": "out1"}, {"id": "out2"}, {"id": "out3"}],
        "envKeyName": [],
    },
    "reroute": {
        "inputPorts": [{"id": "input", "required": True}],
        "outputPorts": [{"id": "output"}],
        "envKeyName": [],
    },
    "gpt-4o-chat": {
        "inputPorts": [
            {"id": "messages", "required": True},
            {"id": "images", "required": False},
        ],
        "outputPorts": [{"id": "text"}],
        "envKeyName": "OPENAI_API_KEY",
    },
    "gemini-chat": {
        "inputPorts": [
            {"id": "messages", "required": True},
            {"id": "images", "required": False},
        ],
        "outputPorts": [{"id": "text"}],
        "envKeyName": "GOOGLE_API_KEY",
    },
    "imagen-4-generate": {
        "inputPorts": [{"id": "prompt", "required": True}],
        "outputPorts": [{"id": "image"}],
        "envKeyName": "GOOGLE_API_KEY",
    },
    "kling-v2-1": {
        "inputPorts": [
            {"id": "image", "required": True},
            {"id": "prompt", "required": False},
        ],
        "outputPorts": [{"id": "video"}],
        "envKeyName": "FAL_KEY",
    },
    "sora-2": {
        "inputPorts": [{"id": "prompt", "required": True}],
        "outputPorts": [{"id": "video"}],
        "envKeyName": "FAL_KEY",
    },
    "nano-banana": {
        "inputPorts": [
            {"id": "prompt", "required": True},
            {"id": "images", "required": False},
        ],
        "outputPorts": [{"id": "image"}, {"id": "text"}],
        "envKeyName": "GOOGLE_API_KEY",
    },
    "veo-3": {
        "inputPorts": [{"id": "prompt", "required": True}],
        "outputPorts": [{"id": "video"}],
        "envKeyName": "FAL_KEY",
    },
    "flux-schnell": {
        "inputPorts": [{"id": "prompt", "required": True}],
        "outputPorts": [{"id": "image"}],
        "envKeyName": "FAL_KEY",
    },
    "fast-sdxl": {
        "inputPorts": [{"id": "prompt", "required": True}],
        "outputPorts": [{"id": "image"}],
        "envKeyName": "FAL_KEY",
    },
    "wan-2-6-t2v": {
        "inputPorts": [{"id": "prompt", "required": True}],
        "outputPorts": [{"id": "video"}],
        "envKeyName": "FAL_KEY",
    },
    "luma-ray2-t2v": {
        "inputPorts": [{"id": "prompt", "required": True}],
        "outputPorts": [{"id": "video"}],
        "envKeyName": "FAL_KEY",
    },
    "ltx-video-2": {
        "inputPorts": [
            {"id": "image", "required": True},
            {"id": "prompt", "required": True},
        ],
        "outputPorts": [{"id": "video"}],
        "envKeyName": "FAL_KEY",
    },
}

LOCAL_EXECUTION_NODE_IDS = frozenset(
    {
        "text-input",
        "image-input",
        "video-input",
        "audio-input",
        "sticky-note",
        "frame-extractor",
        "array-builder",
        "array-selector",
        "image-compare",
        "svg-rasterize",
        "mask-painter",
        "iterator-image",
        "iterator-text",
        "preview",
        "combine-text",
        "router",
        "reroute",
    }
)


_REGISTRY_NODE_DEFS: dict[str, dict[str, Any]] | None = None


def _registry_node_defs() -> dict[str, dict[str, Any]]:
    """Load the full exported node registry lazily for metadata not in NODE_DEFS."""
    global _REGISTRY_NODE_DEFS
    if _REGISTRY_NODE_DEFS is None:
        try:
            from services.node_registry import NodeRegistry

            _REGISTRY_NODE_DEFS = NodeRegistry().get_all()
        except Exception:
            _REGISTRY_NODE_DEFS = {}
    return _REGISTRY_NODE_DEFS


def _node_def_for(definition_id: str) -> dict[str, Any] | None:
    return _registry_node_defs().get(definition_id) or NODE_DEFS.get(definition_id)


def _multiple_input_ports(definition_id: str) -> set[str]:
    node_def = _node_def_for(definition_id)
    if not node_def:
        return set()
    return {
        str(port["id"])
        for port in node_def.get("inputPorts", [])
        if isinstance(port, dict) and port.get("id") and port.get("multiple")
    }


def _as_multiple_values(value: Any) -> list[Any]:
    if value is None:
        return []
    if isinstance(value, list):
        return list(value)
    return [value]


def _append_multiple_input(
    existing: PortValueDict | None,
    incoming: PortValueDict,
) -> PortValueDict:
    values = _as_multiple_values(existing.value if existing else None)
    values.extend(_as_multiple_values(incoming.value))

    port_type = incoming.type
    if existing:
        port_type = existing.type if existing.type == incoming.type else "Any"
    return PortValueDict(type=port_type, value=values)


def topological_sort(nodes: list[GraphNode], edges: list[GraphEdge]) -> list[str]:
    graph: dict[str, set[str]] = {node.id: set() for node in nodes}
    for edge in edges:
        if edge.target in graph:
            graph[edge.target].add(edge.source)

    sorter = TopologicalSorter(graph)
    try:
        return list(sorter.static_order())
    except _GraphlibCycleError as exc:
        raise CycleError(str(exc)) from exc


def get_subgraph(
    nodes: list[GraphNode],
    edges: list[GraphEdge],
    target_node_id: str,
) -> tuple[list[GraphNode], list[GraphEdge]]:
    """Return the subgraph of all ancestors of target_node_id (inclusive).

    Traverses edges in reverse (target -> source) to find every node that
    feeds into the target. Returns filtered lists of nodes and edges that
    belong to this subgraph.
    """
    # Build reverse adjacency: node_id -> set of upstream node_ids
    reverse_adj: dict[str, set[str]] = {n.id: set() for n in nodes}
    for edge in edges:
        if edge.target in reverse_adj:
            reverse_adj[edge.target].add(edge.source)

    # BFS from target node to find all ancestors
    needed: set[str] = set()
    queue = [target_node_id]
    while queue:
        nid = queue.pop()
        if nid in needed:
            continue
        needed.add(nid)
        for upstream in reverse_adj.get(nid, set()):
            if upstream not in needed:
                queue.append(upstream)

    node_map = {n.id: n for n in nodes}
    sub_nodes = [node_map[nid] for nid in needed if nid in node_map]
    sub_edges = [e for e in edges if e.source in needed and e.target in needed]
    return sub_nodes, sub_edges


def validate_graph(
    nodes: list[GraphNode],
    edges: list[GraphEdge],
    api_keys: dict[str, str],
) -> list[ValidationErrorDetail]:
    errors: list[ValidationErrorDetail] = []

    connected_ports: set[tuple[str, str]] = set()
    for edge in edges:
        if edge.target_handle:
            connected_ports.add((edge.target, edge.target_handle))

    for node in nodes:
        node_def = _node_def_for(node.definition_id)
        if not node_def:
            # Unknown or future dynamic node — still validate API key for
            # known dynamic shells when the exported registry is unavailable.
            DYNAMIC_ENV_KEYS: dict[str, str] = {
                "openrouter-universal": "OPENROUTER_API_KEY",
                "replicate-universal": "REPLICATE_API_TOKEN",
                "fal-universal": "FAL_KEY",
            }
            env_key = DYNAMIC_ENV_KEYS.get(node.definition_id)
            if env_key and not api_keys.get(env_key):
                errors.append(
                    ValidationErrorDetail(
                        node_id=node.id,
                        port_id="",
                        message=f"Missing API key: {env_key}",
                    )
                )
            continue

        for port in node_def["inputPorts"]:
            if port.get("required", False):
                if (node.id, port["id"]) not in connected_ports:
                    errors.append(
                        ValidationErrorDetail(
                            node_id=node.id,
                            port_id=port["id"],
                            message=f"Required input port '{port['id']}' is not connected",
                        )
                    )

        env_key = node_def.get("envKeyName", [])
        if isinstance(env_key, str) and env_key:
            key_names = [env_key]
        elif isinstance(env_key, list):
            key_names = env_key
        else:
            key_names = []

        if key_names and not any(api_keys.get(k) for k in key_names):
            key_display = " or ".join(key_names)
            errors.append(
                ValidationErrorDetail(
                    node_id=node.id,
                    port_id="",
                    message=f"Missing API key: {key_display}",
                )
            )

    return errors


NodeHandler = Callable[
    [GraphNode, dict[str, PortValueDict], dict[str, str]],
    Awaitable[dict[str, Any]],
]


async def execute_graph(
    nodes: list[GraphNode],
    edges: list[GraphEdge],
    api_keys: dict[str, str],
    handler_registry: dict[str, NodeHandler],
    emit: Callable[[ExecutionEvent], Awaitable[None]],
    cache: ExecutionCache | None = None,
    max_parallel_nodes: int = 4,
) -> None:
    start_time = time.monotonic()
    nodes_executed = 0
    node_map: dict[str, GraphNode] = {n.id: n for n in nodes}
    outputs_cache: dict[str, dict[str, PortValueDict]] = {}
    order = topological_sort(nodes, edges)
    order_index = {nid: index for index, nid in enumerate(order)}
    concurrency = max(1, max_parallel_nodes)

    incoming_edges: dict[str, list[GraphEdge]] = {nid: [] for nid in node_map}
    outgoing_edges: dict[str, list[GraphEdge]] = {nid: [] for nid in node_map}
    remaining_dependencies: dict[str, int] = {nid: 0 for nid in node_map}
    for edge in edges:
        if edge.source not in node_map or edge.target not in node_map:
            continue
        incoming_edges[edge.target].append(edge)
        outgoing_edges[edge.source].append(edge)
        remaining_dependencies[edge.target] += 1

    ready: deque[str] = deque(
        nid for nid in order if nid in node_map and remaining_dependencies[nid] == 0
    )
    queued_nodes: set[str] = set()
    failed_nodes: set[str] = set()

    async def queue_ready_node(nid: str) -> None:
        if nid in queued_nodes:
            return
        queued_nodes.add(nid)
        await emit(QueuedEvent(node_id=nid))

    async def queue_initial_ready_nodes() -> None:
        for nid in list(ready):
            await queue_ready_node(nid)

    def resolve_inputs(nid: str) -> dict[str, PortValueDict]:
        node = node_map[nid]
        resolved_inputs: dict[str, PortValueDict] = {}
        multiple_ports = _multiple_input_ports(node.definition_id)
        for edge in incoming_edges[nid]:
            if edge.source_handle and edge.target_handle:
                upstream_outputs = outputs_cache.get(edge.source, {})
                if edge.source_handle in upstream_outputs:
                    incoming = upstream_outputs[edge.source_handle]
                    if edge.target_handle in multiple_ports:
                        resolved_inputs[edge.target_handle] = _append_multiple_input(
                            resolved_inputs.get(edge.target_handle),
                            incoming,
                        )
                    else:
                        resolved_inputs[edge.target_handle] = incoming
        return resolved_inputs

    async def run_node(nid: str) -> tuple[str, bool, int]:
        node = node_map[nid]
        await emit(ExecutingEvent(node_id=nid))
        resolved_inputs = resolve_inputs(nid)

        try:
            cache_key: str | None = None
            if cache is not None:
                inputs_for_key = {
                    k: {"type": v.type, "value": v.value}
                    for k, v in resolved_inputs.items()
                }
                cache_key = ExecutionCache.get_key(
                    node.definition_id, dict(node.params), inputs_for_key
                )
                cached_outputs = cache.get(cache_key)
                if cached_outputs is not None:
                    outputs_cache[nid] = {
                        k: PortValueDict(type=v.get("type", "Any"), value=v.get("value"))
                        for k, v in cached_outputs.items()
                    }
                    await emit(ExecutedEvent(node_id=nid, outputs=cached_outputs))
                    return nid, True, 1

            handler = handler_registry.get(node.definition_id)
            if handler is None:
                if node.definition_id == "text-input":
                    text_value = node.params.get("value", "")
                    node_outputs = {"text": {"type": "Text", "value": str(text_value)}}
                elif node.definition_id == "image-input":
                    node_outputs = _image_input_output(node.params)
                elif node.definition_id == "video-input":
                    file_path = node.params.get("filePath", "")
                    node_outputs = {"video": {"type": "Video", "value": str(file_path)}}
                elif node.definition_id == "audio-input":
                    file_path = node.params.get("filePath", "")
                    node_outputs = {"audio": {"type": "Audio", "value": str(file_path)}}
                elif node.definition_id == "sticky-note":
                    node_outputs = {}
                elif node.definition_id == "frame-extractor":
                    video_input = resolved_inputs.get("video")
                    if not video_input or not video_input.value:
                        raise ValueError("Video input is required for frame extraction")
                    try:
                        import subprocess
                        from services.output import get_run_dir
                        from uuid import uuid4

                        video_path = str(video_input.value)
                        mode = node.params.get("mode", "first_frame")
                        timestamp = float(node.params.get("timestamp", 0))

                        run_dir = get_run_dir()
                        out_path = run_dir / f"{uuid4().hex[:12]}.png"

                        if mode == "first_frame":
                            ts = "00:00:00.000"
                        elif mode == "last_frame":
                            probe = subprocess.run(
                                ["ffprobe", "-v", "error", "-show_entries", "format=duration",
                                 "-of", "default=noprint_wrappers=1:nokey=1", video_path],
                                capture_output=True, text=True
                            )
                            dur = float(probe.stdout.strip()) if probe.stdout.strip() else 1.0
                            ts_secs = max(0, dur - 0.1)
                            ts = f"{int(ts_secs//3600):02d}:{int((ts_secs%3600)//60):02d}:{ts_secs%60:06.3f}"
                        elif mode == "middle_frame":
                            probe = subprocess.run(
                                ["ffprobe", "-v", "error", "-show_entries", "format=duration",
                                 "-of", "default=noprint_wrappers=1:nokey=1", video_path],
                                capture_output=True, text=True
                            )
                            dur = float(probe.stdout.strip()) if probe.stdout.strip() else 1.0
                            ts_secs = dur / 2
                            ts = f"{int(ts_secs//3600):02d}:{int((ts_secs%3600)//60):02d}:{ts_secs%60:06.3f}"
                        else:  # timestamp mode
                            ts = f"{int(timestamp//3600):02d}:{int((timestamp%3600)//60):02d}:{timestamp%60:06.3f}"

                        subprocess.run(
                            ["ffmpeg", "-y", "-ss", ts, "-i", video_path, "-frames:v", "1",
                             "-q:v", "2", str(out_path)],
                            capture_output=True, check=True
                        )
                        node_outputs = {"image": {"type": "Image", "value": str(out_path)}}
                    except FileNotFoundError:
                        raise ValueError("ffmpeg not found — install ffmpeg to use Frame Extractor")
                elif node.definition_id == "array-builder":
                    items = []
                    for port_id in ("item1", "item2", "item3", "item4", "item5", "item6", "item7", "item8"):
                        if port_id in resolved_inputs and resolved_inputs[port_id].value is not None:
                            items.append(resolved_inputs[port_id].value)
                    node_outputs = {"array": {"type": "Array", "value": items}}
                elif node.definition_id == "array-selector":
                    array_input = resolved_inputs.get("array")
                    if array_input and isinstance(array_input.value, list) and len(array_input.value) > 0:
                        mode = node.params.get("mode", "first")
                        arr = array_input.value
                        if mode == "first":
                            selected = arr[0]
                        elif mode == "last":
                            selected = arr[-1]
                        elif mode == "random":
                            import random
                            selected = random.choice(arr)
                        else:  # index
                            idx = int(node.params.get("index", 0))
                            selected = arr[min(idx, len(arr) - 1)]
                        node_outputs = {"item": {"type": "Any", "value": selected}}
                    else:
                        node_outputs = {}
                elif node.definition_id == "image-compare":
                    node_outputs = {}
                    if "imageA" in resolved_inputs:
                        node_outputs["imageA"] = {"type": resolved_inputs["imageA"].type, "value": resolved_inputs["imageA"].value}
                    if "imageB" in resolved_inputs:
                        node_outputs["imageB"] = {"type": resolved_inputs["imageB"].type, "value": resolved_inputs["imageB"].value}
                elif node.definition_id == "svg-rasterize":
                    svg_input = resolved_inputs.get("svg")
                    if not svg_input or not svg_input.value:
                        raise ValueError("SVG input is required")
                    svg_value = str(svg_input.value)
                    width = int(node.params.get("width", 1024))
                    height = int(node.params.get("height", 1024))
                    bg = node.params.get("background", "transparent")

                    from services.output import get_run_dir
                    from uuid import uuid4

                    run_dir = get_run_dir()
                    out_path = run_dir / f"{uuid4().hex[:12]}.png"

                    try:
                        import cairosvg
                        # Read SVG content — could be a file path or raw SVG string
                        from pathlib import Path as _Path
                        svg_path = _Path(svg_value)
                        if svg_path.exists():
                            svg_data = svg_path.read_text()
                        else:
                            svg_data = svg_value

                        bg_color = None if bg == "transparent" else "#FFFFFF"
                        cairosvg.svg2png(
                            bytestring=svg_data.encode("utf-8"),
                            write_to=str(out_path),
                            output_width=width,
                            output_height=height,
                            background_color=bg_color,
                        )
                        node_outputs = {"image": {"type": "Image", "value": str(out_path)}}
                    except ImportError:
                        raise ValueError("cairosvg not installed — run: pip install cairosvg")
                elif node.definition_id == "mask-painter":
                    # Painted strokes are stored WHITE-on-BLACK in params._maskData
                    # (a PNG data URI written by the paint modal). Export resizes the
                    # mask to the input image's exact dimensions (Ideogram 422s on a
                    # mismatch) and applies the polarity the consumer expects:
                    #   white-edit (FLUX Fill)  -> painted stays white
                    #   black-edit (Ideogram)   -> inverted, painted becomes black
                    mask_data = node.params.get("_maskData")
                    if not mask_data or not str(mask_data).startswith("data:image"):
                        raise ValueError("No mask painted yet — open the Mask Painter node and paint one")
                    image_input = resolved_inputs.get("image")
                    if not image_input or not image_input.value:
                        raise ValueError("Image input is required (the mask must match its dimensions)")

                    import base64 as _b64
                    import io as _io

                    from PIL import Image as _PILImage, ImageOps as _ImageOps
                    from services.output import get_run_dir
                    from uuid import uuid4

                    mask_bytes = _b64.b64decode(str(mask_data).partition(",")[2])
                    mask_img = _PILImage.open(_io.BytesIO(mask_bytes)).convert("L")

                    # Match the source image dimensions exactly. The source may be a
                    # local path or a data URI; remote URLs are downloaded.
                    src_value = str(image_input.value)
                    if src_value.startswith("data:image"):
                        src_bytes = _b64.b64decode(src_value.partition(",")[2])
                        src_img = _PILImage.open(_io.BytesIO(src_bytes))
                    elif src_value.startswith(("http://", "https://")):
                        import httpx as _httpx
                        async with _httpx.AsyncClient(timeout=60.0) as _client:
                            _resp = await _client.get(src_value, follow_redirects=True)
                            _resp.raise_for_status()
                        src_img = _PILImage.open(_io.BytesIO(_resp.content))
                    else:
                        src_img = _PILImage.open(src_value)

                    if mask_img.size != src_img.size:
                        # NEAREST keeps mask edges hard instead of introducing greys.
                        mask_img = mask_img.resize(src_img.size, _PILImage.NEAREST)

                    if node.params.get("polarity", "white-edit") == "black-edit":
                        mask_img = _ImageOps.invert(mask_img)

                    out_path = get_run_dir() / f"{uuid4().hex[:12]}.png"
                    mask_img.save(out_path, format="PNG")
                    node_outputs = {"mask": {"type": "Image", "value": str(out_path)}}
                elif node.definition_id in ("iterator-image", "iterator-text"):
                    array_input = resolved_inputs.get("array")
                    if array_input and isinstance(array_input.value, list) and len(array_input.value) > 0:
                        cap = int(node.params.get("batch_size_cap", 10))
                        items = array_input.value[:cap]
                        out_type = "Image" if node.definition_id == "iterator-image" else "Text"
                        out_key = "image" if node.definition_id == "iterator-image" else "text"
                        # Emit each item as a separate executed event
                        for i, item in enumerate(items):
                            iter_outputs = {out_key: {"type": out_type, "value": item}}
                            await emit(ExecutedEvent(node_id=nid, outputs=iter_outputs))
                        return nid, True, 1
                    else:
                        node_outputs = {}
                elif node.definition_id == "preview":
                    node_outputs = {}
                    if "input" in resolved_inputs:
                        node_outputs["input"] = {
                            "type": resolved_inputs["input"].type,
                            "value": resolved_inputs["input"].value,
                        }
                elif node.definition_id == "combine-text":
                    texts = []
                    for port_id in ("text1", "text2", "text3"):
                        if port_id in resolved_inputs and resolved_inputs[port_id].value:
                            texts.append(str(resolved_inputs[port_id].value))

                    template = node.params.get("template", "")
                    if template:
                        # Template mode: replace {text1}, {text2}, {text3}
                        result = str(template)
                        for port_id in ("text1", "text2", "text3"):
                            val = ""
                            if port_id in resolved_inputs and resolved_inputs[port_id].value:
                                val = str(resolved_inputs[port_id].value)
                            result = result.replace(f"{{{port_id}}}", val)
                        node_outputs = {"text": {"type": "Text", "value": result}}
                    else:
                        # Separator mode: join non-empty texts
                        separator = node.params.get("separator", "\n")
                        separator = str(separator).replace("\\n", "\n").replace("\\t", "\t")
                        node_outputs = {"text": {"type": "Text", "value": separator.join(texts)}}
                elif node.definition_id == "router":
                    input_val = resolved_inputs.get("input")
                    if input_val and input_val.value is not None:
                        port_val = {"type": input_val.type, "value": input_val.value}
                        node_outputs = {
                            "out1": port_val,
                            "out2": port_val,
                            "out3": port_val,
                        }
                    else:
                        node_outputs = {}
                elif node.definition_id == "reroute":
                    input_val = resolved_inputs.get("input")
                    if input_val and input_val.value is not None:
                        node_outputs = {"output": {"type": input_val.type, "value": input_val.value}}
                    else:
                        node_outputs = {}
                else:
                    raise RuntimeError(f"No handler registered for '{node.definition_id}'")
            else:
                node_outputs = await handler(node, resolved_inputs, api_keys)

            outputs_cache[nid] = {
                k: PortValueDict(type=v.get("type", "Any"), value=v.get("value"))
                for k, v in node_outputs.items()
            }

            if cache is not None and cache_key is not None and handler is not None:
                cache.set(cache_key, node_outputs)

            await _maybe_probe_video_output(node, node_outputs)

            await emit(ExecutedEvent(node_id=nid, outputs=node_outputs))
            return nid, True, 1

        except Exception as exc:
            category, friendly, retryable = classify_error(str(exc))
            await emit(
                ErrorEvent(
                    node_id=nid,
                    error=str(exc),
                    retryable=retryable,
                    category=category,
                    friendly=friendly,
                )
            )
            return nid, False, 0

    await queue_initial_ready_nodes()
    running: dict[asyncio.Task[tuple[str, bool, int]], str] = {}

    while ready or running:
        while ready and len(running) < concurrency:
            nid = ready.popleft()
            if nid in failed_nodes:
                continue
            task = asyncio.create_task(run_node(nid))
            running[task] = nid

        if not running:
            break

        done, _pending = await asyncio.wait(
            running.keys(),
            return_when=asyncio.FIRST_COMPLETED,
        )

        for task in done:
            running.pop(task, None)
            nid, succeeded, executed_count = task.result()
            nodes_executed += executed_count

            if not succeeded:
                failed_nodes.add(nid)
                continue

            newly_ready: list[str] = []
            for edge in outgoing_edges[nid]:
                if edge.target in failed_nodes:
                    continue
                remaining_dependencies[edge.target] -= 1
                if remaining_dependencies[edge.target] == 0:
                    newly_ready.append(edge.target)

            for target in sorted(newly_ready, key=lambda node_id: order_index[node_id]):
                await queue_ready_node(target)
                ready.append(target)

    duration = time.monotonic() - start_time
    await emit(GraphCompleteEvent(duration=round(duration, 3), nodes_executed=nodes_executed))
