qihuaiflow/backend/python/app/core/node_registry.py

"""app/core/node_registry.py — #874「工具即节点」统一注册表（双 Claw 共用 · 单一实现）。

why
---
改造前，平台里「一个能力」有三套互不相识的表达：

  1. function calling —— `_build_system_prompt()` 把 skill 拼成 function_definitions；
  2. MCP 工具       —— `build_mcp_tool_defs()` 另拼一套 `mcp__srv__tool__idx`；
  3. DAG 编排       —— `orchestrator.VALID_NODE_TYPES = (llm, local, external, reduce)`，
                       skill / MCP 根本无法作为图节点出现。

结果：画布里画不出「调某个技能」，YAML 工作流校验和函数调用各说各话，
IM Claw 想跑编排还得再写一份 prompt 拼装。三套表达 = 三份 bug。

本模块把「能力」收敛为唯一抽象 —— **节点(Node)**：

    BaseNode(Protocol): node_type / name / input_schema / output_schema / risk_level
                        + async run(ctx: NodeContext) -> NodeResult

同一个注册表同时供给三个消费方（设计书 §B 验收口径）：

    ① 函数调用   discover() → CapabilitySnapshot.function_definitions
    ② DAG 编排   orchestrator.run_graph → registry.get(node_type).run(ctx)
    ③ YAML 校验  yaml_workflow.VALID_NODE_TYPES → registry.node_types()

单引擎原则
---------
本模块 **不实现任何业务逻辑**，只做「分派 + 事件发射」。真正干活的原语一律复用：

    LLM        -> orchestrator._llm_complete
    local 智能体 -> orchestrator._run_local_agent_turn
    external    -> chat_agents._call_external_agent
    skill       -> chat_skill._execute_skill_tool
    mcp         -> mcp_client.call_mcp_tool

对 orchestrator 的引用全部为函数内延迟导入 —— orchestrator 反向 import 本模块，
模块级互引会死锁在 import 阶段。

双 Claw 对齐
-----------
节点过程通过 `NodeContext.emit_*` 发射 **统一 EngineEvent**，Web Claw(SseAdapter)
与 IM Claw(BufferAdapter) 吃同一份事件流，编排过程在两端天然对齐，IM 侧零特殊逻辑。

线格式取舍（重要）：设计书写的是 `EngineEvent(kind="node")`，但前端 AiProcessBlock
只认 thinking/skill/mcp/agent/kb/media，发未知 type 会静默丢帧。故这里做**语义映射**：
节点按自身性质映射到前端已识别的 step 类型（SkillNode→skill、McpToolNode→mcp、
AgentNode→agent、编排类→thinking），节点身份则通过 `step["node"]={id,type,label}`
附加字段承载 —— 新增键非空才输出，老前端忽略，新前端可据此画节点进度条。
"""
from __future__ import annotations

import asyncio
import inspect
import json
import logging
import time
from dataclasses import dataclass, field
from typing import Any, Awaitable, Callable, Dict, List, Optional, Protocol, runtime_checkable

from app.core.engine_event import (
    CLAW_IM,
    CLAW_WEB,
    PHASE_FAIL,
    PHASE_OK,
    PHASE_START,
    SEV_ERROR,
    STEP_AGENT,
    STEP_MCP,
    STEP_SKILL,
    STEP_THINKING,
    Diagnosis,
    make_step,
)

logger = logging.getLogger("aiflow.node_registry")

__all__ = [
    "NODE_LLM", "NODE_REDUCE", "NODE_LOCAL", "NODE_EXTERNAL",
    "NODE_SKILL", "NODE_MCP", "NODE_AGENT", "NODE_LOOP",
    "RISK_LOW", "RISK_MEDIUM", "RISK_HIGH",
    "BaseNode", "NodeBase", "NodeContext", "NodeResult", "NodeError", "StepCounter",
    "NodeRegistry", "CapabilitySnapshot",
    "registry", "normalize_claw",
]

# ---- 节点类型枚举（设计书 §B）------------------------------------------
NODE_LLM = "llm"            # 纯文本变换 / 路由判断
NODE_REDUCE = "reduce"      # 多路结果汇总
NODE_LOCAL = "local"        # 本地智能体（带工具轮）
NODE_EXTERNAL = "external"  # 外部智能体（Dify / n8n / Coze）
NODE_SKILL = "skill"        # 单个技能直调（新）
NODE_MCP = "mcp"            # 单个 MCP 工具直调（新）
NODE_AGENT = "agent"        # 智能体自动分派 local/external（新）
NODE_LOOP = "loop"          # 对列表批量执行子节点（新）

# ---- 风险分级：供审批策略 / 成本分层复用 --------------------------------
RISK_LOW = "low"
RISK_MEDIUM = "medium"
RISK_HIGH = "high"

_MAX_RESULT_CHARS = 600      # 事件里回传的结果摘要上限，防止撑爆 SSE 帧
_LOOP_HARD_LIMIT = 20        # 循环迭代硬顶，防止图里写个 10000 把机器打满
_LOOP_CONCURRENCY_CAP = 5


class NodeError(RuntimeError):
    """节点执行失败。run_graph 按 on_error=halt|continue 处理。"""

    def __init__(self, message: str, *, diagnosis: Optional[Diagnosis] = None):
        super().__init__(message)
        self.diagnosis = diagnosis


def normalize_claw(claw_type: Any) -> str:
    """claw 维度归一。

    设计书/otel 用 `web|im_wechat|im_platform`，engine_event 用 `web|im`。
    不归一会让 Grafana 维度碎成四片，看板永远对不上。
    """
    raw = str(claw_type or "").strip().lower()
    if not raw:
        return ""
    return CLAW_IM if raw.startswith("im") else CLAW_WEB


def _truncate(text: Any, limit: int = _MAX_RESULT_CHARS) -> str:
    s = text if isinstance(text, str) else str(text or "")
    return s if len(s) <= limit else s[:limit] + f"...(共 {len(s)} 字)"


class StepCounter:
    """全图共享的步骤序号发生器。

    同一 index 先发 phase=start 占位、后发 phase=ok|fail 覆盖，
    前端即得到「原地更新」的实时进度，而非追加一堆重复卡片。
    """

    __slots__ = ("_n",)

    def __init__(self, start: int = 0):
        self._n = start

    def next(self) -> int:
        n = self._n
        self._n += 1
        return n


# =========================================================================
# 上下文 / 结果
# =========================================================================
@dataclass
class NodeContext:
    """节点运行上下文（设计书 §B：tenant_id / claw_type / trace_id / blackboard / emit）。"""

    db: Any
    node: Dict[str, Any]
    graph: Dict[str, Any] = field(default_factory=dict)
    blackboard: Dict[str, Any] = field(default_factory=dict)
    query: str = ""
    upstream: str = ""
    is_entry: bool = True
    user_id: str = ""
    conv_id: Any = None
    trace_id: str = ""
    tenant_id: str = ""
    claw_type: str = ""
    snapshot: Optional["CapabilitySnapshot"] = None
    on_event: Optional[Callable[[Any], Any]] = None
    counter: "StepCounter" = field(default_factory=lambda: StepCounter())

    # -- 便捷取值 -----------------------------------------------------
    @property
    def claw(self) -> str:
        return normalize_claw(self.claw_type)

    @property
    def node_id(self) -> str:
        return str(self.node.get("id") or "")

    def cfg(self, key: str, default: Any = None) -> Any:
        val = self.node.get(key)
        return default if val in (None, "") else val

    def node_input(self) -> str:
        """入口节点吃用户原问题；下游节点综合原问题 + 上游输出。

        与 orchestrator._assemble_node_input 保持同一口径（该函数为其唯一实现，
        此处复用而非复制）。
        """
        from app.core.orchestrator import _assemble_node_input

        return _assemble_node_input(self.query, self.upstream, self.is_entry)

    def child(self, node: Dict[str, Any], *, upstream: str = "") -> "NodeContext":
        """派生子节点上下文（LoopNode 用）。共享 counter/emit，保证事件序不乱。"""
        return NodeContext(
            db=self.db,
            node=node,
            graph=self.graph,
            blackboard=self.blackboard,
            query=self.query,
            upstream=upstream or self.upstream,
            is_entry=False,
            user_id=self.user_id,
            conv_id=self.conv_id,
            trace_id=self.trace_id,
            tenant_id=self.tenant_id,
            claw_type=self.claw_type,
            snapshot=self.snapshot,
            on_event=self.on_event,
            counter=self.counter,
        )

    # -- 事件发射（唯一出口，失败安全）--------------------------------
    async def emit(self, event: Any) -> None:
        if self.on_event is None:
            return
        try:
            res = self.on_event(event)
            if inspect.isawaitable(res):
                await res
        except Exception as e:  # 观测不得影响主流程
            logger.debug("[node] emit 失败(忽略) node=%s: %s", self.node_id, e)

    async def emit_step(
        self,
        wire_kind: str,
        *,
        name: str = "",
        server: str = "",
        text: str = "",
        args: Optional[Dict[str, Any]] = None,
        result: str = "",
        phase: str = "",
        index: Optional[int] = None,
        diagnosis: Any = None,
        node_type: str = "",
    ) -> None:
        """构造并发射一条节点过程事件。

        step["node"] 承载节点身份（设计书 kind="node" 的兼容落地形式）。
        """
        node_meta = {
            "id": self.node_id,
            "type": node_type or str(self.node.get("type") or ""),
        }
        label = self.node.get("label") or self.node.get("name")
        if label:
            node_meta["label"] = str(label)

        evt = make_step(
            wire_kind,
            name=name,
            server=server,
            text=text,
            args=args or {},
            result=result,
            phase=phase,
            index=index,
            diagnosis=diagnosis,
            node=node_meta,
            trace_id=self.trace_id,
            tenant_id=self.tenant_id,
            claw=self.claw,
        )
        await self.emit(evt)


@dataclass
class NodeResult:
    """节点成功产物。失败一律抛 NodeError（保持 run_graph 的 on_error 语义不变）。"""

    output: str = ""
    node_id: str = ""
    node_type: str = ""
    used_tools: List[str] = field(default_factory=list)
    meta: Dict[str, Any] = field(default_factory=dict)

    def to_blackboard(self) -> Dict[str, Any]:
        """转成 orchestrator 黑板结构（字段集与改造前逐键一致）。"""
        out: Dict[str, Any] = {
            "ok": True,
            "node_id": self.node_id,
            "node_type": self.node_type,
            "output": self.output,
        }
        if self.used_tools:
            out["used_tools"] = self.used_tools
        if self.meta:
            out["meta"] = self.meta
        return out


# =========================================================================
# 节点协议 + 公共基类
# =========================================================================
@runtime_checkable
class BaseNode(Protocol):
    """节点协议（设计书 §B）。第三方节点只要满足此协议即可 register()。"""

    node_type: str
    name: str
    input_schema: Dict[str, Any]
    output_schema: Dict[str, Any]
    risk_level: str

    async def run(self, ctx: NodeContext) -> NodeResult: ...


_TEXT_OUT = {
    "type": "object",
    "properties": {"output": {"type": "string", "description": "节点文本输出"}},
}


class NodeBase:
    """公共实现：生命周期事件 + 异常归一。子类只实现 `_execute`。"""

    node_type: str = ""
    name: str = ""
    input_schema: Dict[str, Any] = {"type": "object", "properties": {}}
    output_schema: Dict[str, Any] = _TEXT_OUT
    risk_level: str = RISK_LOW
    wire_kind: str = STEP_THINKING          # 映射到前端已识别的 step 类型
    fail_code: str = "NODE_FAILED"

    # -- 子类钩子 -----------------------------------------------------
    async def _execute(self, ctx: NodeContext) -> NodeResult:
        raise NotImplementedError

    def _start_label(self, ctx: NodeContext) -> str:
        return str(ctx.node.get("label") or ctx.node.get("name") or ctx.node_id or self.name)

    def _start_args(self, ctx: NodeContext) -> Dict[str, Any]:
        return {}

    def _fail_hint(self, ctx: NodeContext) -> str:
        return "检查该节点配置与上游输入，或将 on_error 设为 continue 以跳过"

    # -- 统一生命周期 -------------------------------------------------
    async def run(self, ctx: NodeContext) -> NodeResult:
        idx = ctx.counter.next()
        label = self._start_label(ctx)
        t0 = time.time()

        await ctx.emit_step(
            self.wire_kind,
            name=label,
            text=f"执行节点 {label}" if self.wire_kind == STEP_THINKING else "",
            args=self._start_args(ctx),
            phase=PHASE_START,
            index=idx,
            node_type=self.node_type,
        )

        try:
            res = await self._execute(ctx)
        except NodeError as e:
            diag = e.diagnosis or Diagnosis(
                code=self.fail_code,
                severity=SEV_ERROR,
                summary=f"节点 {label} 执行失败",
                detail=str(e),
                hint=self._fail_hint(ctx),
                retryable=True,
            )
            await ctx.emit_step(
                self.wire_kind, name=label, phase=PHASE_FAIL, index=idx,
                diagnosis=diag, node_type=self.node_type,
            )
            raise
        except Exception as e:
            diag = Diagnosis(
                code=self.fail_code,
                severity=SEV_ERROR,
                summary=f"节点 {label} 执行失败",
                detail=f"{type(e).__name__}: {e}",
                hint=self._fail_hint(ctx),
                retryable=True,
            )
            await ctx.emit_step(
                self.wire_kind, name=label, phase=PHASE_FAIL, index=idx,
                diagnosis=diag, node_type=self.node_type,
            )
            raise NodeError(str(e), diagnosis=diag) from e

        res.node_id = res.node_id or ctx.node_id
        res.node_type = res.node_type or self.node_type
        res.meta.setdefault("elapsed_ms", int((time.time() - t0) * 1000))

        await ctx.emit_step(
            self.wire_kind,
            name=label,
            args=self._start_args(ctx),
            result=_truncate(res.output),
            phase=PHASE_OK,
            index=idx,
            node_type=self.node_type,
        )
        return res


# =========================================================================
# 内建节点 —— 全部委托既有原语，零业务重复
# =========================================================================
class LlmNode(NodeBase):
    node_type = NODE_LLM
    name = "LLM 处理"
    risk_level = RISK_LOW
    fail_code = "LLM_NODE_FAILED"
    input_schema = {
        "type": "object",
        "properties": {
            "system_prompt": {"type": "string", "description": "节点系统提示词"},
            "model_id": {"type": "string", "description": "指定模型；留空走 auto 路由"},
        },
    }
    _DEFAULT_PROMPT = (
        "你是一个处理节点，请根据上游输入完成指定的文本变换、摘要或路由判断，"
        "只输出处理后的结果，不要多余解释。"
    )

    async def _execute(self, ctx: NodeContext) -> NodeResult:
        from app.core.orchestrator import _llm_complete

        text = await _llm_complete(
            ctx.cfg("system_prompt", self._DEFAULT_PROMPT),
            ctx.node_input(),
            ctx.node.get("model_id"),
            ctx.db,
            ctx.user_id,
        )
        return NodeResult(output=text)


class ReduceNode(LlmNode):
    node_type = NODE_REDUCE
    name = "结果汇总"
    fail_code = "REDUCE_NODE_FAILED"
    _DEFAULT_PROMPT = (
        "你是一个结果汇总器，请将多个子任务的输出综合成连贯、完整、面向用户的最终回答。"
    )


class LocalAgentNode(NodeBase):
    node_type = NODE_LOCAL
    name = "本地智能体"
    risk_level = RISK_MEDIUM
    wire_kind = STEP_AGENT
    fail_code = "LOCAL_AGENT_FAILED"
    input_schema = {
        "type": "object",
        "required": ["agent_id"],
        "properties": {"agent_id": {"type": "string", "description": "MarketApp(agent) ID"}},
    }

    def _resolve(self, ctx: NodeContext) -> Dict[str, Any]:
        from app.api.chat_skill import _resolve_market_agents

        aid = ctx.node.get("agent_id")
        if not aid:
            raise NodeError(f"{self.node_type} 节点 {ctx.node_id} 缺少 agent_id")
        agents = _resolve_market_agents(ctx.db, [aid], ctx.user_id)
        if not agents:
            raise NodeError(f"节点 {ctx.node_id} 引用的 agent_id={aid} 不存在")
        return agents[0]

    def _start_args(self, ctx: NodeContext) -> Dict[str, Any]:
        return {"agent_id": ctx.node.get("agent_id")}

    async def _execute(self, ctx: NodeContext) -> NodeResult:
        from app.core.orchestrator import _run_local_agent_turn

        agent = self._resolve(ctx)
        res = await _run_local_agent_turn(agent, ctx.node_input(), ctx.db, ctx.user_id, ctx.conv_id)
        return NodeResult(
            output=res.get("text") or "",
            used_tools=list(res.get("used_tools") or []),
            meta={"agent_name": agent.get("name")},
        )


class ExternalAgentNode(LocalAgentNode):
    node_type = NODE_EXTERNAL
    name = "外部智能体"
    risk_level = RISK_HIGH
    fail_code = "EXTERNAL_AGENT_FAILED"

    async def _execute(self, ctx: NodeContext) -> NodeResult:
        from app.api.chat_agents import _call_external_agent

        agent = self._resolve(ctx)
        answer, _ = await _call_external_agent(
            ctx.db, agent, query=ctx.node_input(), user_id=ctx.user_id,
            ext_conversation_id=None, history=[],
        )
        return NodeResult(
            output=answer or "",
            meta={"agent_name": agent.get("name"), "platform": agent.get("platform")},
        )

    def _fail_hint(self, ctx: NodeContext) -> str:
        return "确认该 Agent 的 platform / agent_endpoint / 凭据配置有效且远端可达"


class AgentNode(LocalAgentNode):
    """智能体节点：按配置自动分派本地 / 远程，画布上不用再区分两种节点。"""

    node_type = NODE_AGENT
    name = "智能体"
    risk_level = RISK_MEDIUM
    fail_code = "AGENT_NODE_FAILED"

    @staticmethod
    def is_external(agent: Dict[str, Any]) -> bool:
        return bool((agent.get("platform") or "").strip() and (agent.get("agent_endpoint") or "").strip())

    async def _execute(self, ctx: NodeContext) -> NodeResult:
        agent = self._resolve(ctx)
        if self.is_external(agent):
            from app.api.chat_agents import _call_external_agent

            answer, _ = await _call_external_agent(
                ctx.db, agent, query=ctx.node_input(), user_id=ctx.user_id,
                ext_conversation_id=None, history=[],
            )
            return NodeResult(
                output=answer or "",
                meta={"agent_name": agent.get("name"), "mode": "external",
                      "platform": agent.get("platform")},
            )

        from app.core.orchestrator import _run_local_agent_turn

        res = await _run_local_agent_turn(agent, ctx.node_input(), ctx.db, ctx.user_id, ctx.conv_id)
        return NodeResult(
            output=res.get("text") or "",
            used_tools=list(res.get("used_tools") or []),
            meta={"agent_name": agent.get("name"), "mode": "local"},
        )


class SkillNode(NodeBase):
    """技能直调节点 —— 把 function calling 里的技能搬到画布上。"""

    node_type = NODE_SKILL
    name = "技能"
    risk_level = RISK_MEDIUM
    wire_kind = STEP_SKILL
    fail_code = "SKILL_NODE_FAILED"
    input_schema = {
        "type": "object",
        "properties": {
            "skill_id": {"type": "string", "description": "MarketApp(skill) ID"},
            "func_name": {"type": "string", "description": "技能函数名；与 skill_id 二选一"},
            "args": {"type": "object", "description": "技能入参；留空则以上游文本作为 query"},
        },
    }

    @staticmethod
    def func_name_of(skill: Dict[str, Any]) -> str:
        """与 _build_system_prompt 同源的函数名规则，禁止另立一套。"""
        fn = skill.get("func_name")
        if fn:
            return str(fn)
        return str(skill.get("name") or "").replace(" ", "_").replace("-", "_").lower()

    def _resolve(self, ctx: NodeContext) -> tuple:
        from app.api.chat_skill import _resolve_market_skills

        sid = ctx.node.get("skill_id")
        want_fn = str(ctx.node.get("func_name") or "").strip().lower()
        skills: List[Dict[str, Any]] = []

        if sid:
            skills = _resolve_market_skills(ctx.db, [sid], ctx.user_id) or []
        elif ctx.snapshot and ctx.snapshot.skills:
            skills = [s for s in ctx.snapshot.skills if self.func_name_of(s) == want_fn]

        if not skills:
            raise NodeError(
                f"技能节点 {ctx.node_id} 未解析到技能"
                f"(skill_id={sid or '-'} func_name={want_fn or '-'})"
            )
        return skills, want_fn or self.func_name_of(skills[0])

    def _start_args(self, ctx: NodeContext) -> Dict[str, Any]:
        return dict(ctx.node.get("args") or {})

    async def _execute(self, ctx: NodeContext) -> NodeResult:
        from app.api.chat_skill import _execute_skill_tool

        skills, func_name = self._resolve(ctx)
        args = dict(ctx.node.get("args") or {})
        node_input = ctx.node_input()
        if not args:
            args = {"query": node_input}

        # _execute_skill_tool 为同步实现，放线程池避免阻塞事件循环里的同层并行节点
        result = await asyncio.to_thread(
            _execute_skill_tool, ctx.db, func_name, args, skills, node_input
        )
        text = result if isinstance(result, str) else json.dumps(result, ensure_ascii=False)
        return NodeResult(output=text, used_tools=[func_name], meta={"skill": func_name})

    def _fail_hint(self, ctx: NodeContext) -> str:
        return "确认技能已发布且入参符合其 parameters 定义"


class McpToolNode(NodeBase):
    """MCP 工具直调节点 —— 外部副作用，风险等级最高。"""

    node_type = NODE_MCP
    name = "MCP 工具"
    risk_level = RISK_HIGH
    wire_kind = STEP_MCP
    fail_code = "MCP_NODE_FAILED"
    input_schema = {
        "type": "object",
        "required": ["tool"],
        "properties": {
            "server_id": {"type": "string", "description": "MCP 服务 ID"},
            "tool": {"type": "string", "description": "MCP 工具名"},
            "args": {"type": "object", "description": "工具入参"},
        },
    }

    def _resolve_server(self, ctx: NodeContext) -> Dict[str, Any]:
        sid = ctx.node.get("server_id")
        if ctx.snapshot and sid:
            for s in ctx.snapshot.mcp_servers:
                if s.get("id") == sid:
                    return s
        from app.api.chat_mcp import _resolve_active_mcp_servers

        if not sid:
            raise NodeError(f"MCP 节点 {ctx.node_id} 缺少 server_id")
        servers = _resolve_active_mcp_servers(ctx.db, [sid]) or []
        if not servers:
            raise NodeError(f"MCP 节点 {ctx.node_id} 引用的服务 {sid} 不存在或未启用")
        return servers[0]

    def _start_args(self, ctx: NodeContext) -> Dict[str, Any]:
        return dict(ctx.node.get("args") or {})

    def _start_label(self, ctx: NodeContext) -> str:
        return str(ctx.node.get("tool") or super()._start_label(ctx))

    async def _execute(self, ctx: NodeContext) -> NodeResult:
        from app.mcp_client import call_mcp_tool

        tool = str(ctx.node.get("tool") or "").strip()
        if not tool:
            raise NodeError(f"MCP 节点 {ctx.node_id} 缺少 tool")
        srv = self._resolve_server(ctx)

        args = dict(ctx.node.get("args") or {})
        if not args:
            args = {"query": ctx.node_input()}

        result = await call_mcp_tool(srv, tool, args)
        text = result if isinstance(result, str) else json.dumps(result, ensure_ascii=False)
        return NodeResult(
            output=text,
            used_tools=[f"mcp:{srv.get('name')}:{tool}"],
            meta={"server": srv.get("name"), "tool": tool},
        )

    async def run(self, ctx: NodeContext) -> NodeResult:
        # MCP 事件需带 server 字段（前端按 server 分组渲染），故覆写生命周期头帧
        idx = ctx.counter.next()
        tool = str(ctx.node.get("tool") or "?")
        srv_name = ""
        try:
            srv_name = str(self._resolve_server(ctx).get("name") or "")
        except Exception:
            srv_name = str(ctx.node.get("server_id") or "")

        await ctx.emit_step(
            STEP_MCP, name=tool, server=srv_name, args=self._start_args(ctx),
            phase=PHASE_START, index=idx, node_type=self.node_type,
        )
        t0 = time.time()
        try:
            res = await self._execute(ctx)
        except Exception as e:
            diag = Diagnosis(
                code=self.fail_code, severity=SEV_ERROR,
                summary=f"MCP 工具 {tool} 调用失败",
                detail=f"{type(e).__name__}: {e}",
                hint="确认 MCP 服务在线、凭据有效，且未被出网代理拦截",
                retryable=True,
            )
            await ctx.emit_step(
                STEP_MCP, name=tool, server=srv_name, phase=PHASE_FAIL,
                index=idx, diagnosis=diag, node_type=self.node_type,
            )
            raise NodeError(str(e), diagnosis=diag) from e

        res.node_id = res.node_id or ctx.node_id
        res.node_type = res.node_type or self.node_type
        res.meta.setdefault("elapsed_ms", int((time.time() - t0) * 1000))
        await ctx.emit_step(
            STEP_MCP, name=tool, server=srv_name, args=self._start_args(ctx),
            result=_truncate(res.output), phase=PHASE_OK, index=idx,
            node_type=self.node_type,
        )
        return res


class LoopNode(NodeBase):
    """批量节点：把上游输出拆成条目，对每条跑一次子节点(body)，再按策略聚合。"""

    node_type = NODE_LOOP
    name = "循环"
    risk_level = RISK_MEDIUM
    fail_code = "LOOP_NODE_FAILED"
    input_schema = {
        "type": "object",
        "required": ["body"],
        "properties": {
            "body": {"type": "object", "description": "子节点定义(含 type 及其配置)"},
            "items": {"type": "array", "description": "静态条目；留空则拆分上游输出"},
            "split": {"type": "string", "enum": ["lines", "json"], "default": "lines"},
            "max_iterations": {"type": "integer", "default": 5, "maximum": _LOOP_HARD_LIMIT},
            "concurrency": {"type": "integer", "default": 1, "maximum": _LOOP_CONCURRENCY_CAP},
            "join": {"type": "string", "enum": ["concat", "first", "last"], "default": "concat"},
        },
    }

    def _items(self, ctx: NodeContext) -> List[str]:
        raw = ctx.node.get("items")
        if isinstance(raw, list) and raw:
            items = [str(x) for x in raw]
        else:
            src = ctx.upstream or ctx.query or ""
            if str(ctx.node.get("split") or "lines") == "json":
                try:
                    parsed = json.loads(src)
                    items = [json.dumps(x, ensure_ascii=False) if not isinstance(x, str) else x
                             for x in (parsed if isinstance(parsed, list) else [parsed])]
                except Exception:
                    items = [ln.strip() for ln in src.splitlines() if ln.strip()]
            else:
                items = [ln.strip() for ln in src.splitlines() if ln.strip()]

        limit = ctx.cfg("max_iterations", 5)
        try:
            limit = max(1, min(int(limit), _LOOP_HARD_LIMIT))
        except Exception:
            limit = 5
        return items[:limit]

    async def _execute(self, ctx: NodeContext) -> NodeResult:
        body = ctx.node.get("body")
        if not isinstance(body, dict) or not body.get("type"):
            raise NodeError(f"循环节点 {ctx.node_id} 缺少合法 body(需含 type)")

        impl = registry.get(body["type"])
        if impl is None:
            raise NodeError(f"循环节点 {ctx.node_id} 的 body 类型非法: {body['type']}")

        items = self._items(ctx)
        if not items:
            return NodeResult(output="", meta={"iterations": 0})

        try:
            conc = max(1, min(int(ctx.cfg("concurrency", 1)), _LOOP_CONCURRENCY_CAP))
        except Exception:
            conc = 1
        sem = asyncio.Semaphore(conc)

        async def _one(i: int, item: str) -> str:
            child_def = dict(body)
            child_def.setdefault("id", f"{ctx.node_id}#{i}")
            async with sem:
                res = await impl.run(ctx.child(child_def, upstream=item))
            return res.output or ""

        outs = await asyncio.gather(
            *[_one(i, it) for i, it in enumerate(items)], return_exceptions=True
        )

        texts, used, failed = [], [], 0
        for i, o in enumerate(outs):
            if isinstance(o, Exception):
                failed += 1
                texts.append(f"[第 {i + 1} 项失败] {o}")
            else:
                texts.append(o)
        used.append(f"loop:{body['type']}x{len(items)}")

        join = str(ctx.cfg("join", "concat"))
        if join == "first":
            output = texts[0] if texts else ""
        elif join == "last":
            output = texts[-1] if texts else ""
        else:
            output = "\n\n".join(t for t in texts if t)

        return NodeResult(
            output=output, used_tools=used,
            meta={"iterations": len(items), "failed": failed, "body_type": body["type"]},
        )


# =========================================================================
# 能力快照 —— 把「发现到的能力」物化成节点候选 + function definitions
# =========================================================================
@dataclass
class CapabilitySnapshot:
    """一次请求可用能力的只读快照（双 Claw 共用同一份）。"""

    skills: List[Dict[str, Any]] = field(default_factory=list)
    mcp_servers: List[Dict[str, Any]] = field(default_factory=list)
    agents: List[Dict[str, Any]] = field(default_factory=list)
    function_definitions: List[Dict[str, Any]] = field(default_factory=list)
    mcp_tool_map: Dict[str, Any] = field(default_factory=dict)
    claw: str = ""

    def node_candidates(self) -> List[Dict[str, Any]]:
        """画布左侧「可拖拽能力」列表 / IM `/flow` 的可选节点清单。"""
        out: List[Dict[str, Any]] = []
        for s in self.skills:
            out.append({
                "type": NODE_SKILL,
                "label": s.get("name"),
                "skill_id": s.get("id"),
                "func_name": SkillNode.func_name_of(s),
                "description": s.get("description") or "",
                "risk_level": RISK_MEDIUM,
            })
        for fname, info in self.mcp_tool_map.items():
            out.append({
                "type": NODE_MCP,
                "label": f"{info.get('server_name')} / {info.get('name')}",
                "server_id": info.get("server_id"),
                "tool": info.get("name"),
                "func_name": fname,
                "risk_level": RISK_HIGH,
            })
        for a in self.agents:
            out.append({
                "type": NODE_AGENT,
                "label": a.get("name"),
                "agent_id": a.get("id"),
                "description": a.get("description") or "",
                "risk_level": RISK_HIGH if AgentNode.is_external(a) else RISK_MEDIUM,
            })
        return out

    def summary(self) -> Dict[str, int]:
        return {
            "skills": len(self.skills),
            "mcp_servers": len(self.mcp_servers),
            "mcp_tools": len(self.mcp_tool_map),
            "agents": len(self.agents),
            "functions": len(self.function_definitions),
        }


# =========================================================================
# 注册表
# =========================================================================
class NodeRegistry:
    """进程级单例：节点类型 → 实现。

    - 静态部分：内建节点类型，启动即可用，`node_types()` 供 YAML/图校验；
    - 动态部分：`discover()` 按请求扫描已装 skill / 已启用 MCP / 已配置 Agent，
      产出 CapabilitySnapshot（function_definitions + 节点候选），
      让「函数调用」与「画布节点」共用同一份能力清单。
    """

    _inst: Optional["NodeRegistry"] = None

    def __init__(self) -> None:
        self._nodes: Dict[str, Any] = {}

    # -- 单例 ---------------------------------------------------------
    @classmethod
    def instance(cls) -> "NodeRegistry":
        if cls._inst is None:
            cls._inst = cls()
            cls._inst.register_builtins()
        return cls._inst

    # -- 注册 ---------------------------------------------------------
    def register(self, node: Any, *, override: bool = False) -> Any:
        """注册节点实现（类或实例均可）。返回注册后的实例。"""
        impl = node() if inspect.isclass(node) else node
        ntype = getattr(impl, "node_type", "")
        if not ntype:
            raise ValueError("节点必须声明非空 node_type")
        if not callable(getattr(impl, "run", None)):
            raise ValueError(f"节点 {ntype} 缺少 run() 实现")
        if ntype in self._nodes and not override:
            logger.debug("[registry] 节点类型 %s 已存在，跳过重复注册", ntype)
            return self._nodes[ntype]
        self._nodes[ntype] = impl
        return impl

    def register_builtins(self) -> None:
        for cls in (LlmNode, ReduceNode, LocalAgentNode, ExternalAgentNode,
                    AgentNode, SkillNode, McpToolNode, LoopNode):
            self.register(cls)
        logger.info("[registry] 内建节点注册完成: %s", ",".join(self.node_types()))

    # -- 查询 ---------------------------------------------------------
    def get(self, node_type: str) -> Optional[Any]:
        return self._nodes.get(str(node_type or ""))

    def has(self, node_type: str) -> bool:
        return str(node_type or "") in self._nodes

    def node_types(self) -> tuple:
        """供 orchestrator.validate_graph / yaml_workflow 校验（唯一真相源）。"""
        return tuple(sorted(self._nodes.keys()))

    def describe(self) -> List[Dict[str, Any]]:
        """节点类型元信息，供前端画布渲染配置表单。"""
        out = []
        for ntype in self.node_types():
            impl = self._nodes[ntype]
            out.append({
                "node_type": ntype,
                "name": getattr(impl, "name", ntype),
                "risk_level": getattr(impl, "risk_level", RISK_LOW),
                "input_schema": getattr(impl, "input_schema", {}),
                "output_schema": getattr(impl, "output_schema", _TEXT_OUT),
            })
        return out

    def requires_agent(self, node_type: str) -> bool:
        return node_type in (NODE_LOCAL, NODE_EXTERNAL, NODE_AGENT)

    # -- 动态能力发现 --------------------------------------------------
    async def discover(
        self,
        db,
        user_id: str = "",
        query: str = "",
        *,
        claw: str = "web",
        explicit_mcp_ids: Optional[List[str]] = None,
        agent_ids: Optional[List[str]] = None,
        with_functions: bool = True,
    ) -> CapabilitySnapshot:
        """扫描当前可用能力，产出快照。

        能力发现只走 capability_discovery（双 Claw 唯一实现），
        本方法不得内联任何第二套发现逻辑 —— 这是 #866 回退事故的教训。
        """
        from app.core import capability_discovery as cap

        snap = CapabilitySnapshot(claw=normalize_claw(claw) or claw)

        try:
            snap.skills = cap.discover_skills(db, user_id, query=query, claw=claw) or []
        except Exception as e:
            logger.warning("[registry] 技能发现失败(降级空): %s", e)

        try:
            snap.mcp_servers = cap.discover_mcp_servers(
                db, explicit_ids=explicit_mcp_ids, claw=claw
            ) or []
        except Exception as e:
            logger.warning("[registry] MCP 发现失败(降级空): %s", e)

        discover_agents = getattr(cap, "discover_agents", None)
        if callable(discover_agents):
            try:
                snap.agents = discover_agents(
                    db, user_id, query=query, explicit_ids=agent_ids, claw=claw
                ) or []
            except Exception as e:
                logger.warning("[registry] Agent 发现失败(降级空): %s", e)

        if with_functions:
            try:
                from app.api.chat_context import _build_system_prompt

                _, skill_defs = _build_system_prompt(
                    skills=snap.skills, agents=[], kb_context="",
                    user_content=query, auto_discovered=True,
                )
                snap.function_definitions = list(skill_defs or [])
            except Exception as e:
                logger.warning("[registry] 技能 function 定义生成失败: %s", e)

            if snap.mcp_servers:
                try:
                    mcp_defs, tool_map = await cap.build_mcp_tool_defs(
                        snap.mcp_servers, idx_start=len(snap.function_definitions)
                    )
                    snap.function_definitions.extend(mcp_defs or [])
                    snap.mcp_tool_map = tool_map or {}
                except Exception as e:
                    logger.warning("[registry] MCP function 定义生成失败: %s", e)

        logger.info("[registry] discover claw=%s %s", claw, snap.summary())
        return snap


# 模块级单例（导入即完成内建注册，供 yaml_workflow 在 import 期取 node_types）
registry: NodeRegistry = NodeRegistry.instance()