ai-video-fullstack/backend/services/node_specs.py

"""Workflow v3 node catalog, v2 compatibility normalization, and validation."""

from __future__ import annotations

from collections import defaultdict, deque
from copy import deepcopy
from typing import Any


SPEC_VERSION = "3"
NODE_TYPES = {"start", "agent", "action", "handoff", "end"}
EDGE_MODES = {"llm", "expression", "always"}
AGENT_ENTRY_MODES = {"wait_user", "generate", "fixed_speech"}
AUTOMATIC_NODE_TYPES = {"start", "action", "handoff"}
EXPRESSION_OPERATORS = {
    "eq",
    "neq",
    "gt",
    "gte",
    "lt",
    "lte",
    "contains",
    "in",
    "exists",
}

NODE_SPECS: list[dict[str, Any]] = [
    {
        "name": "start",
        "displayName": "Start",
        "category": "control_node",
        "description": "初始化会话、动态变量和全局观察器，可播放固定开场白。",
        "icon": "Play",
        "accent": "mint",
        "addable": False,
        "constraints": {
            "minIncoming": 0,
            "maxIncoming": 0,
            "minOutgoing": 0,
            "minInstances": 1,
            "maxInstances": 1,
        },
        "fields": [
            {"key": "name", "label": "节点名称", "type": "text", "default": "Start"},
            {"key": "greeting", "label": "固定开场白", "type": "textarea", "default": ""},
        ],
    },
    {
        "name": "agent",
        "displayName": "Agent",
        "category": "conversation_node",
        "description": "阶段智能体：绑定上下文、工具、知识库及 ASR/TTS 资源。",
        "icon": "Bot",
        "accent": "sky",
        "addable": True,
        "constraints": {"minIncoming": 1, "minOutgoing": 0},
        "fields": [
            {"key": "name", "label": "节点名称", "type": "text", "default": "Agent"},
            {
                "key": "prompt",
                "label": "阶段提示词",
                "type": "textarea",
                "required": True,
                "default": "",
            },
        ],
    },
    {
        "name": "action",
        "displayName": "Action",
        "category": "execution_node",
        "description": "确定性执行指定工具，并将结果字段写入会话动态变量。",
        "icon": "Zap",
        "accent": "peach",
        "addable": True,
        "constraints": {"minIncoming": 1, "minOutgoing": 1},
        "fields": [
            {"key": "name", "label": "节点名称", "type": "text", "default": "Action"},
        ],
    },
    {
        "name": "handoff",
        "displayName": "Handoff",
        "category": "execution_node",
        "description": "转交其他 AI、人工、队列或电话；MVP 发送转交事件后继续路由。",
        "icon": "PhoneForwarded",
        "accent": "lavender",
        "addable": True,
        "constraints": {"minIncoming": 1, "minOutgoing": 0},
        "fields": [
            {"key": "name", "label": "节点名称", "type": "text", "default": "Handoff"},
            {"key": "target", "label": "转交目标", "type": "text", "default": ""},
            {"key": "message", "label": "转交提示", "type": "textarea", "default": ""},
        ],
    },
    {
        "name": "end",
        "displayName": "End",
        "category": "control_node",
        "description": "结束 AI 流程或整个音视频会话。",
        "icon": "Flag",
        "accent": "rose",
        "addable": True,
        "constraints": {"minIncoming": 1, "minOutgoing": 0, "maxOutgoing": 0},
        "fields": [
            {"key": "name", "label": "节点名称", "type": "text", "default": "End"},
            {"key": "message", "label": "固定结束语", "type": "textarea", "default": ""},
        ],
    },
]

_SPEC_BY_NAME = {spec["name"]: spec for spec in NODE_SPECS}


def node_types_response() -> dict[str, Any]:
    return {"specVersion": SPEC_VERSION, "nodeTypes": NODE_SPECS}


def _edge_data_v3(edge: dict, source_type: str) -> dict:
    data = deepcopy(edge.get("data") or {})
    if data.get("mode") in EDGE_MODES:
        data.setdefault("priority", 10)
        return data
    condition = str(data.pop("condition", "") or "").strip()
    transition = data.pop("transition_speech", data.get("transitionSpeech", ""))
    data.update(
        {
            "mode": "llm" if condition and source_type == "agent" else "always",
            "priority": 10,
            "condition": condition,
            "transitionSpeech": transition,
        }
    )
    return data


def _normalize_agent_data(data: dict[str, Any]) -> None:
    """Add v3 Agent defaults without changing existing node-level behavior."""
    data.setdefault("contextPolicy", "inherit")
    data.setdefault("entryMode", "wait_user")
    data.setdefault("entrySpeech", "")
    if "inheritGlobalConfig" not in data:
        has_node_overrides = any(
            (
                data.get("llmResourceId"),
                data.get("asrResourceId"),
                data.get("ttsResourceId"),
                data.get("knowledgeBaseId"),
                data.get("toolIds"),
            )
        )
        data["inheritGlobalConfig"] = not has_node_overrides


def _normalize_settings(settings: dict[str, Any], *, global_prompt: str = "") -> None:
    settings.setdefault("globalPrompt", global_prompt)
    settings.setdefault("defaultLlmResourceId", "")
    settings.setdefault("defaultAsrResourceId", "")
    settings.setdefault("defaultTtsResourceId", "")
    settings.setdefault("toolIds", [])
    settings.setdefault("knowledgeBaseId", "")
    settings.setdefault("knowledgeMode", "automatic")
    settings.setdefault("knowledgeTopN", 5)
    settings.setdefault("knowledgeScoreThreshold", 0.0)
    settings.setdefault("enableInterrupt", True)
    settings.setdefault("turnConfig", {})


def normalize_graph(graph: dict[str, Any] | None) -> dict[str, Any]:
    """Return a deep-copied v3 graph; preserve v3 IDs and migrate v2 semantics."""
    source = deepcopy(graph or {})
    if str(source.get("specVersion") or "") == SPEC_VERSION:
        settings = source.setdefault("settings", {})
        _normalize_settings(settings)
        source.setdefault("nodes", [])
        source.setdefault("edges", [])
        for node in source["nodes"]:
            if node.get("type") != "agent":
                continue
            data = node.setdefault("data", {})
            _normalize_agent_data(data)
        return source

    nodes = source.get("nodes") or []
    edges = source.get("edges") or []
    global_prompt = ""
    mapped_nodes: list[dict] = []
    type_by_id: dict[str, str] = {}
    start_prompt_nodes: dict[str, str] = {}

    type_map = {
        "startCall": "start",
        "agentNode": "agent",
        "endCall": "end",
        "start": "start",
        "agent": "agent",
        "action": "action",
        "handoff": "handoff",
        "end": "end",
    }
    for node in nodes:
        old_type = str(node.get("type") or "")
        data = deepcopy(node.get("data") or {})
        if old_type == "globalNode":
            global_prompt = str(data.get("prompt") or "")
            continue
        new_type = type_map.get(old_type, old_type)
        migrated = deepcopy(node)
        migrated["type"] = new_type
        if new_type == "end":
            data["message"] = data.pop("message", data.pop("prompt", ""))
            data.setdefault("scope", "session")
        elif new_type == "agent":
            _normalize_agent_data(data)
        elif new_type == "start":
            prompt = str(data.pop("prompt", "") or "").strip()
            if prompt:
                start_prompt_nodes[str(node.get("id"))] = prompt
            for key in ("allowInterrupt", "addGlobalPrompt"):
                data.pop(key, None)
        migrated["data"] = data
        mapped_nodes.append(migrated)
        if migrated.get("id"):
            type_by_id[str(migrated["id"])] = new_type

    mapped_edges: list[dict] = []
    for edge in edges:
        migrated = deepcopy(edge)
        migrated["data"] = _edge_data_v3(
            migrated, type_by_id.get(str(migrated.get("source")), "")
        )
        mapped_edges.append(migrated)

    # A v2 Start was conversational. Insert a synthetic Agent so its prompt remains active.
    for start_id, prompt in start_prompt_nodes.items():
        synthetic_id = f"{start_id}-migrated-agent"
        start_node = next((n for n in mapped_nodes if n.get("id") == start_id), None)
        position = (start_node or {}).get("position") or {"x": 100, "y": 120}
        mapped_nodes.append(
            {
                "id": synthetic_id,
                "type": "agent",
                "position": {"x": position.get("x", 100) + 300, "y": position.get("y", 120)},
                "data": {
                    "name": "迁移的开场 Agent",
                    "prompt": prompt,
                    "contextPolicy": "inherit",
                    "inheritGlobalConfig": True,
                    "entryMode": "wait_user",
                    "entrySpeech": "",
                },
            }
        )
        for edge in mapped_edges:
            if edge.get("source") == start_id:
                edge["source"] = synthetic_id
                edge["data"] = _edge_data_v3(edge, "agent")
                if str(edge["data"].get("condition") or "").strip():
                    edge["data"]["mode"] = "llm"
        mapped_edges.append(
            {
                "id": f"e-{start_id}-{synthetic_id}",
                "source": start_id,
                "target": synthetic_id,
                "data": {"mode": "always", "priority": 0, "transitionSpeech": ""},
            }
        )

    settings = deepcopy(source.get("settings") or {})
    _normalize_settings(settings, global_prompt=global_prompt)
    return {
        "specVersion": 3,
        "settings": settings,
        "nodes": mapped_nodes,
        "edges": mapped_edges,
        **({"viewport": deepcopy(source["viewport"])} if source.get("viewport") else {}),
    }


def _validate_expression(expression: Any) -> list[str]:
    if not isinstance(expression, dict):
        return ["表达式条件不能为空"]
    combinator = expression.get("combinator", "and")
    if combinator not in {"and", "or"}:
        return ["表达式组合方式必须是 and 或 or"]
    rules = expression.get("rules")
    if not isinstance(rules, list) or not rules:
        return ["表达式至少需要一条规则"]
    errors = []
    for rule in rules:
        if not isinstance(rule, dict) or not rule.get("variable"):
            errors.append("表达式规则缺少变量")
        elif rule.get("operator") not in EXPRESSION_OPERATORS:
            errors.append(f"不支持的表达式运算符:{rule.get('operator')}")
    return errors


def validate_graph(graph: dict[str, Any]) -> list[str]:
    graph = normalize_graph(graph)
    nodes = graph.get("nodes") or []
    edges = graph.get("edges") or []
    if not nodes:
        return []

    errors: list[str] = []
    node_by_id: dict[str, dict] = {}
    counts: dict[str, int] = defaultdict(int)
    for node in nodes:
        node_id = str(node.get("id") or "")
        node_type = str(node.get("type") or "")
        if not node_id:
            errors.append("存在缺少 id 的节点")
            continue
        if node_id in node_by_id:
            errors.append(f"节点 id 重复:{node_id}")
        if node_type not in NODE_TYPES:
            errors.append(f"未知节点类型:{node_type}(节点 {node_id})")
        node_by_id[node_id] = node
        counts[node_type] += 1

        if node_type == "agent":
            data = node.get("data") or {}
            entry_mode = data.get("entryMode", "wait_user")
            if entry_mode not in AGENT_ENTRY_MODES:
                errors.append(f"Agent 节点 {node_id} 的进入模式无效:{entry_mode}")
            elif entry_mode == "fixed_speech" and not str(
                data.get("entrySpeech") or ""
            ).strip():
                errors.append(f"Agent 节点 {node_id} 的固定进入语不能为空")

    if counts["start"] != 1:
        errors.append("工作流必须有且仅有一个 Start 节点")
    incoming: dict[str, int] = defaultdict(int)
    outgoing: dict[str, int] = defaultdict(int)
    adj: dict[str, list[str]] = defaultdict(list)
    auto_adj: dict[str, list[str]] = defaultdict(list)
    priorities: dict[str, set[int]] = defaultdict(set)
    always_counts: dict[str, int] = defaultdict(int)
    for edge in edges:
        edge_id = str(edge.get("id") or "")
        source_id = str(edge.get("source") or "")
        target_id = str(edge.get("target") or "")
        if source_id not in node_by_id:
            errors.append(f"边 {edge_id} 指向不存在的源节点:{source_id}")
            continue
        if target_id not in node_by_id:
            errors.append(f"边 {edge_id} 指向不存在的目标节点:{target_id}")
            continue
        if source_id == target_id and node_by_id[source_id].get("type") != "agent":
            errors.append(f"自动节点不能自连:{source_id}")
        data = edge.get("data") or {}
        mode = data.get("mode")
        if mode not in EDGE_MODES:
            errors.append(f"边 {edge_id} 的判断模式无效:{mode}")
        if mode == "llm" and node_by_id[source_id].get("type") != "agent":
            errors.append(f"LLM 判断边只能从 Agent 发出:{edge_id}")
        if mode == "llm" and not str(data.get("condition") or "").strip():
            errors.append(f"LLM 判断边缺少自然语言条件:{edge_id}")
        if mode == "expression":
            expression_errors = _validate_expression(data.get("expression"))
            errors.extend(f"边 {edge_id}:{item}" for item in expression_errors)
        try:
            priority = int(data.get("priority", 10))
        except (TypeError, ValueError):
            errors.append(f"边 {edge_id} 的优先级必须是整数")
            priority = 10
        if priority in priorities[source_id]:
            errors.append(f"节点 {source_id} 的出边优先级不能重复:{priority}")
        priorities[source_id].add(priority)
        if mode == "always":
            always_counts[source_id] += 1
            if always_counts[source_id] > 1:
                errors.append(f"节点 {source_id} 最多只能有一条默认边")
        incoming[target_id] += 1
        outgoing[source_id] += 1
        adj[source_id].append(target_id)
        source_is_automatic = node_by_id[source_id].get("type") != "agent"
        target_is_automatic = node_by_id[target_id].get("type") != "agent"
        if source_is_automatic and target_is_automatic:
            auto_adj[source_id].append(target_id)

    for node_id, node in node_by_id.items():
        spec = _SPEC_BY_NAME.get(str(node.get("type")))
        if not spec:
            continue
        constraints = spec["constraints"]
        for actual, suffix, label in (
            (incoming[node_id], "Incoming", "入边"),
            (outgoing[node_id], "Outgoing", "出边"),
        ):
            lo = constraints.get(f"min{suffix}")
            hi = constraints.get(f"max{suffix}")
            if lo is not None and actual < lo:
                errors.append(f"节点 {node_id} 的{label}不能少于 {lo}")
            if hi is not None and actual > hi:
                errors.append(f"节点 {node_id} 的{label}不能多于 {hi}")
        node_type = node.get("type")
        if (
            node_type in AUTOMATIC_NODE_TYPES
            and outgoing[node_id] > 0
            and always_counts[node_id] != 1
        ):
            errors.append(f"自动节点 {node_id} 存在出边时必须有且仅有一条默认边")

    start_id = next(
        (node_id for node_id, node in node_by_id.items() if node.get("type") == "start"),
        None,
    )
    if start_id:
        reached = {start_id}
        queue = deque([start_id])
        while queue:
            current = queue.popleft()
            for target in adj[current]:
                if target not in reached:
                    reached.add(target)
                    queue.append(target)
        for node_id in node_by_id.keys() - reached:
            errors.append(f"节点不可从 Start 到达:{node_id}")

    # Reject cycles made only of instantaneous nodes; Agent cycles are valid waits.
    visiting: set[str] = set()
    visited: set[str] = set()

    def visit(node_id: str) -> bool:
        if node_id in visiting:
            return True
        if node_id in visited:
            return False
        visiting.add(node_id)
        for target in auto_adj[node_id]:
            if visit(target):
                return True
        visiting.remove(node_id)
        visited.add(node_id)
        return False

    automatic_node_ids = (
        node_id
        for node_id, node in node_by_id.items()
        if node.get("type") != "agent"
    )
    if any(visit(node_id) for node_id in automatic_node_ids):
        errors.append("Start/Action/Handoff/End 之间不能形成无等待循环")
    return list(dict.fromkeys(errors))


def graph_references(graph: dict[str, Any]) -> dict[str, set[str]]:
    """Collect externally referenced IDs for save/runtime validation."""
    normalized = normalize_graph(graph)
    settings = normalized.get("settings") or {}
    resources = {
        str(value)
        for value in (
            settings.get("defaultLlmResourceId"),
            settings.get("defaultAsrResourceId"),
            settings.get("defaultTtsResourceId"),
        )
        if value
    }
    tools: set[str] = {str(tool_id) for tool_id in settings.get("toolIds") or []}
    knowledge: set[str] = (
        {str(settings["knowledgeBaseId"])}
        if settings.get("knowledgeBaseId")
        else set()
    )
    for node in normalized.get("nodes") or []:
        data = node.get("data") or {}
        inherits_global = (
            node.get("type") == "agent" and data.get("inheritGlobalConfig", True)
        )
        if not inherits_global:
            for resource_id in (
                data.get("llmResourceId"),
                data.get("asrResourceId"),
                data.get("ttsResourceId"),
            ):
                if resource_id:
                    resources.add(str(resource_id))
            for tool_id in data.get("toolIds") or []:
                tools.add(str(tool_id))
            if data.get("knowledgeBaseId"):
                knowledge.add(str(data["knowledgeBaseId"]))
        if data.get("toolId"):
            tools.add(str(data["toolId"]))
    return {"model_resources": resources, "tools": tools, "knowledge_bases": knowledge}