feat(workflow #874): NodeRegistry 统一工作流节点抽象 + 双 Claw 事件对齐

#874 P3 工作流引擎 NodeRegistry 统一（加速带「工具即节点」思想落地，零源码拷贝重写）：

- app/core/node_registry.py（新增）：把 skill / MCP / Agent / DAG 四套割裂表达收敛为
  单一 Node 抽象。BaseNode(Protocol) 定义 node_type/name/input_schema/output_schema/
  risk_level + async run(ctx)；NodeContext 携带 db/tenant_id/claw_type/trace_id/黑板/emit；
  NodeResult.to_blackboard() 与既有 orchestrator 黑板结构无缝互转。内建八型节点
  llm/reduce/local/external/agent/skill/mcp/loop，均继承 NodeBase 公共生命周期
  （发 start 步骤 -> 执行 -> ok/fail 覆盖同 index），异常统一归一为 NodeError（带结构化
  diagnosis）。SkillNode 用 asyncio.to_thread 包裹同步执行器避免阻塞事件循环；
  LoopNode 硬限流 20 次迭代 / 并发上限 5，子节点经 ctx.child() 复用同一 StepCounter 与
  emit，保证步骤序号全图唯一。CapabilitySnapshot 统一承载 skills/mcp_servers/agents/
  function_definitions/mcp_tool_map，供 _build_system_prompt、run_graph、yaml 校验三处
  共用同一份能力真相。normalize_claw() 将 im_wechat/im_platform 归一为 "im"，避免
  Grafana 维度碎片化。

- app/core/capability_discovery.py：新增 discover_agents()，与 discover_skills 同构的
  三级降级策略（显式安装 -> published 兜底 -> rank_by_query 裁剪），补齐 Agent 维度的
  统一能力发现，杜绝各处内联第二份发现逻辑（#866 回退事故教训）。

- app/core/orchestrator.py：VALID_NODE_TYPES 与 validate_graph 改为运行期读
  registry.node_types()/requires_agent()，注册表成为节点类型的单一真相源；_run_node 改为
  拼装 NodeContext 后委托 registry.get(ntype).run(ctx)，本体只保留调度与拓扑；run_graph
  新增 on_event/claw_type/tenant_id/snapshot 参数，编排全过程以统一 EngineEvent 下发。

- app/core/yaml_workflow.py：valid_node_types() 实时读注册表；validate_yaml 补 skill/mcp/
  loop 必填校验；yaml_to_orchestration / orchestration_to_yaml 双向透传节点扩展字段
  （label/skill_id/func_name/tool/server_id/args/body/items/split/max_iterations/
  concurrency/join），修复配置往返丢字段。

- app/core/engine_event.py：make_step 新增 node 参数，以 step["node"]={id,type,label}
  承载节点身份。设计书原文的 EngineEvent(kind="node") 与前端仅识别 thinking/skill/mcp/
  agent/kb/media 的现状冲突（未知 kind 会被静默丢帧），故映射为已知 kind + 附加字段，
  线格式对旧前端字节兼容。

- app/api/chat_service.py：Web Claw 编排调用点透传 on_event/claw_type/tenant_id，
  租户与通道标签复用 #875 的 ContextVar（get_route_ctx），不再新增一层参数透传。
  IM Claw 经 EngineCore 自动继承同一事件流（单引擎原则，im_llm.py 零改动）。

- tests/test_node_registry.py（新增）：R-NODE-1~11 确定性哨兵，无网络无外部依赖。守卫
  内建八型齐全、注册表为单一真相源（orchestrator 与 yaml 校验实时同源）、describe 契约、
  双 Claw 事件对齐（SseAdapter 与 BufferAdapter 逐键相等且均带 node 身份）、claw 归一与
  tenant 标签、新节点必填校验、YAML 往返不吞字段、Loop 限流聚合、失败结构化诊断两端可见、
  禁止第二份能力发现、register 拒绝非法节点。

验证：容器内 py_compile 全通过；pytest tests/ 退出码 0（85 项 = 83 passed + 2 skipped，
0 FAILED/ERROR），含 R-NODE 16 项与 R-DUAL 双 Claw 哨兵。

safe-align: 仅复制 7 个 git-tracked 文件，未触碰 secrets/.venv/node_modules/dist/caches。
本地提交不 push。

backend/python/app/api/chat_service.py

@@ -2200,6 +2200,14 @@ async def _send_message_core(
 
 
             from app.core.orchestrator import run_graph
+            # #874 双 Claw 对齐：编排过程以统一 EngineEvent 下发，
+            # 租户/通道标签复用 #875 的 ContextVar，避免再加一层参数透传。
+            try:
+                from app.core.llm_router import get_route_ctx as _get_route_ctx
+                _rctx = _get_route_ctx()
+            except Exception:
+                _rctx = {}
+
 
 
 
@@ -2228,6 +2236,9 @@ async def _send_message_core(
 
 
                 trace_id=get_trace_id(),
+                on_event=on_event,
+                claw_type=(_rctx.get("claw_type") or "web"),
+                tenant_id=(_rctx.get("tenant_id") or ""),
 
 
 

backend/python/app/core/capability_discovery.py

@@ -28,7 +28,9 @@
 --------------------------------
   CAP_SKILL_LIMIT               单轮注入技能数上限，默认 12
   CAP_MCP_LIMIT                 单轮注入 MCP 服务数上限，默认 8
+  CAP_AGENT_LIMIT               单轮可见 Agent 数上限，默认 6
   CAP_SKILL_FALLBACK_PUBLISHED  安装表为空时是否降级为 published，默认 开
+  CAP_AGENT_FALLBACK_PUBLISHED  Agent 同上，默认 开
   CAP_MCP_AUTO                  未显式指定时是否自动挂载已启用 MCP，默认 开
 """
 
@@ -44,6 +46,7 @@ logger = logging.getLogger(__name__)
 __all__ = [
     "discover_skills",
     "discover_mcp_servers",
+    "discover_agents",
     "build_mcp_tool_defs",
     "rank_by_query",
 ]
@@ -226,6 +229,97 @@ def discover_mcp_servers(
     return out
 
 
+# --------------------------------------------------------------------------
+# Agent 发现（#874：画布节点候选 / IM `/flow` 可选清单）
+# --------------------------------------------------------------------------
+def discover_agents(
+    db,
+    user_id: str | None,
+    query: str = "",
+    explicit_ids: Iterable[str] | None = None,
+    limit: int | None = None,
+    claw: str = "web",
+) -> list:
+    """发现当前请求可用的智能体（双 Claw 共用），策略与 discover_skills 同构。
+
+    返回 `_resolve_market_agents` 的结构：
+        [{id, name, role_prompt, instructions, skill_ids, knowledge_base_ids,
+          model_id, tools, description, tool_policy, platform, agent_endpoint,
+          agent_api_key, orchestration}]
+
+    用途定位：给 NodeRegistry 生成 agent 型节点候选、给 IM `/flow` 列可选智能体。
+    若要拿它往 system prompt 里注入，请自行收窄 limit —— 多个 Agent 人格叠加会串味。
+    """
+    limit = limit if limit is not None else _env_int("CAP_AGENT_LIMIT", 6)
+    if limit <= 0:
+        return []
+
+    from app.api.chat_skill import _resolve_market_agents
+    from app.models import AppInstallation, MarketApp
+
+    # 0) 显式指定（Web Claw 勾选 / 画布已引用）—— 不做相关性裁剪，用户意愿最高
+    ids = [i for i in (explicit_ids or []) if i]
+    if ids:
+        out = _resolve_market_agents(db, ids, user_id or "")
+        logger.info("[cap] agents claw=%s source=explicit picked=%d", claw, len(out))
+        return out
+
+    pool: list = []
+    source = "empty"
+
+    # 1) 用户显式安装
+    if user_id:
+        try:
+            app_ids = [
+                i.app_id
+                for i in db.query(AppInstallation)
+                .filter(AppInstallation.user_id == user_id)
+                .all()
+            ]
+            if app_ids:
+                aids = [
+                    a.id
+                    for a in db.query(MarketApp)
+                    .filter(
+                        MarketApp.id.in_(app_ids),
+                        MarketApp.type == "agent",
+                        MarketApp.status == "published",
+                    )
+                    .all()
+                ]
+                if aids:
+                    pool = _resolve_market_agents(db, aids, user_id)
+                    source = "installed"
+        except Exception as e:
+            logger.warning("[cap] 安装 Agent 发现失败(降级继续): %s", e)
+
+    # 2) 降级：平台已发布全集
+    if not pool and _env_flag("CAP_AGENT_FALLBACK_PUBLISHED", True):
+        try:
+            aids = [
+                a.id
+                for a in db.query(MarketApp)
+                .filter(MarketApp.type == "agent", MarketApp.status == "published")
+                .all()
+            ]
+            if aids:
+                pool = _resolve_market_agents(db, aids, user_id or "")
+                source = "published"
+        except Exception as e:
+            logger.warning("[cap] published Agent 发现失败: %s", e)
+
+    # 3) 预算裁剪
+    picked = rank_by_query(pool, query, limit)
+    logger.info(
+        "[cap] agents claw=%s source=%s pool=%d picked=%d",
+        claw,
+        source,
+        len(pool),
+        len(picked),
+    )
+    return picked
+
+
 # --------------------------------------------------------------------------
 # MCP → function definitions
 # --------------------------------------------------------------------------

backend/python/app/core/engine_event.py

@@ -27,7 +27,8 @@ step 对象契约（AiProcessBlock 消费）:
     skill    -> {"type":"skill", "name": str, "args": dict, "result": str}
     mcp      -> {"type":"mcp", "server": str, "name": str, "args": dict, "result": str}
 
-    附加(新增，前端可选消费): "diagnosis": {...}, "phase": "start|ok|fail"
+    附加(新增，前端可选消费): "diagnosis": {...}, "phase": "start|ok|fail",
+                              "node": {"id","type","label"}   # #874 节点身份
 """
 from __future__ import annotations
 
@@ -233,12 +234,18 @@ def make_step(
     phase: str = "",
     index: Optional[int] = None,
     diagnosis: Any = None,
+    node: Optional[Dict[str, Any]] = None,
     **kw,
 ) -> EngineEvent:
     """过程步骤（思考 / 技能 / MCP / Agent / KB / 媒体）。
 
     index 用于「原地更新同一步骤」：先发 phase=start 占位，
     执行完再用同一 index 发 phase=ok|fail 覆盖，前端即得到实时进度。
+
+    node（#874）承载图节点身份 {id, type, label}。设计书原文写的是
+    `EngineEvent(kind="node")`，但前端只认既有 step 类型，发未知 type 会静默丢帧；
+    故 kind 仍取节点映射后的已知类型，节点身份改由本字段附加 —— 非空才输出，
+    老前端忽略，新前端可据此渲染 DAG 节点进度。
     """
     step: Dict[str, Any] = {"type": kind}
     if kind == STEP_THINKING:
@@ -254,6 +261,8 @@ def make_step(
     step["result"] = result or ""
     if phase:
         step["phase"] = phase
+    if isinstance(node, dict) and node:
+        step["node"] = node
 
     diag = Diagnosis.from_dict(diagnosis) if diagnosis is not None else None
     if diag is not None:

backend/python/app/core/orchestrator.py

@@ -7,6 +7,18 @@
 - 节点执行复用既有原语(_build_system_prompt / _execute_skill_tool / _call_external_agent /
   3.1 工具沙箱)，完全向后兼容——agent.orchestration 为 null 时 chat_service 走原逻辑。
 - 并发安全：每个节点用独立 SessionLocal 会话做 DB 读，避免共享请求会话跨协程竞态。
+
+#874 变更
+--------
+节点类型不再硬编码在本模块，改由 `app.core.node_registry` 统一持有：
+
+  * 合法类型      -> registry.node_types()（同时供 yaml_workflow 校验，单一真相源）
+  * 节点执行      -> registry.get(type).run(NodeContext)，本模块只负责拓扑与调度
+  * 双 Claw 对齐  -> run_graph(on_event=...) 把节点过程发成统一 EngineEvent，
+                    Web(SseAdapter) 与 IM(BufferAdapter) 吃同一份事件流
+
+原有 llm/local/external/reduce 语义逐字保留（节点实现内部仍调用本模块的
+_llm_complete / _run_local_agent_turn），新增 skill/mcp/agent/loop 四型。
 """
 from __future__ import annotations
 
@@ -17,9 +29,20 @@ from typing import Any, Optional
 
 import httpx
 
+from app.core.engine_event import STEP_THINKING, make_step
+from app.core.node_registry import (
+    NodeContext,
+    NodeError,
+    StepCounter,
+    normalize_claw,
+    registry,
+)
+
 logger = logging.getLogger("aiflow.orchestrator")
 
-VALID_NODE_TYPES = ("llm", "local", "external", "reduce")
+# 兼容导出：历史调用方直接 import 了这个名字。真正的校验一律走 registry.node_types()，
+# 以便运行期注册的第三方节点立即生效（本常量只是 import 期快照）。
+VALID_NODE_TYPES = registry.node_types()
 
 
 class OrchestrationError(ValueError):
@@ -50,12 +73,25 @@ def validate_graph(graph: Any) -> None:
             raise OrchestrationError(f"节点 id 重复: {nid}")
         ids.add(nid)
         ntype = n.get("type")
-        if ntype not in VALID_NODE_TYPES:
+        valid = registry.node_types()
+        if ntype not in valid:
             raise OrchestrationError(
-                f"节点 {nid} 类型非法: {ntype} (允许 {', '.join(VALID_NODE_TYPES)})"
+                f"节点 {nid} 类型非法: {ntype} (允许 {', '.join(valid)})"
             )
-        if ntype in ("local", "external") and not n.get("agent_id"):
+        if registry.requires_agent(ntype) and not n.get("agent_id"):
             raise OrchestrationError(f"节点 {nid} 类型 {ntype} 必须指定 agent_id")
+        if ntype == "skill" and not (n.get("skill_id") or n.get("func_name")):
+            raise OrchestrationError(f"节点 {nid} 类型 skill 必须指定 skill_id 或 func_name")
+        if ntype == "mcp" and not n.get("tool"):
+            raise OrchestrationError(f"节点 {nid} 类型 mcp 必须指定 tool")
+        if ntype == "loop":
+            body = n.get("body")
+            if not isinstance(body, dict) or not body.get("type"):
+                raise OrchestrationError(f"节点 {nid} 类型 loop 必须提供含 type 的 body")
+            if body["type"] not in valid:
+                raise OrchestrationError(f"节点 {nid} 的 body 类型非法: {body['type']}")
+            if body["type"] == "loop":
+                raise OrchestrationError(f"节点 {nid} 的 body 不允许再嵌套 loop")
 
     for i, e in enumerate(edges):
         if not isinstance(e, dict):
@@ -275,12 +311,33 @@ async def _run_local_agent_turn(agent: dict, node_input: str, db, user_id, conv_
 # ---------------------------------------------------------------------------
 # 单节点执行（含独立 DB 会话，保证并发安全）
 # ---------------------------------------------------------------------------
-async def _run_node(node, graph, blackboard, query, user_id, conv_id, trace_id) -> dict:
+async def _run_node(
-    from app.api.chat_skill import _resolve_market_agents
+    node,
+    graph,
+    blackboard,
+    query,
+    user_id,
+    conv_id,
+    trace_id,
+    *,
+    on_event=None,
+    claw_type: str = "",
+    tenant_id: str = "",
+    counter: Optional[StepCounter] = None,
+    snapshot=None,
+) -> dict:
+    """执行单节点：拼装上游输入 → 交给 registry 里的节点实现 → 回填黑板结构。
+
+    本函数不再包含任何节点类型分支 —— 新增能力只需注册新节点，无需改调度层。
+    """
     from app.database import SessionLocal
 
     nid = node["id"]
     ntype = node["type"]
+    impl = registry.get(ntype)
+    if impl is None:
+        raise OrchestrationError(f"未知节点类型 {ntype}")
+
     preds = _predecessors(graph, nid)
     pred_results = [blackboard[p] for p in preds]
     is_entry = not preds
@@ -298,66 +355,104 @@ async def _run_node(node, graph, blackboard, query, user_id, conv_id, trace_id)
         upstream = ""
 
     with SessionLocal() as db:
-        if ntype in ("llm", "reduce"):
+        ctx = NodeContext(
-            if ntype == "llm":
+            db=db,
-                sys_prompt = node.get("system_prompt") or (
+            node=node,
-                    "你是一个处理节点，请根据上游输入完成指定的文本变换、摘要或路由判断，"
+            graph=graph,
-                    "只输出处理后的结果，不要多余解释。"
+            blackboard=blackboard,
+            query=query,
+            upstream=upstream,
+            is_entry=is_entry,
+            user_id=user_id or "",
+            conv_id=conv_id,
+            trace_id=str(trace_id or ""),
+            tenant_id=str(tenant_id or ""),
+            claw_type=claw_type or "",
+            snapshot=snapshot,
+            on_event=on_event,
+            counter=counter or StepCounter(),
         )
-            else:
+        try:
-                sys_prompt = node.get("system_prompt") or (
+            result = await impl.run(ctx)
-                    "你是一个结果汇总器，请将多个子任务的输出综合成连贯、完整、面向用户的最终回答。"
+        except NodeError as e:
-                )
+            # 归一为 OrchestrationError，run_graph 的 on_error=halt|continue 语义不变
-            uc = _assemble_node_input(query, upstream, is_entry)
+            raise OrchestrationError(str(e)) from e
-            text = await _llm_complete(sys_prompt, uc, node.get("model_id"), db, user_id)
+        return result.to_blackboard()
-            return {"ok": True, "node_id": nid, "node_type": ntype, "output": text}
-
-        if ntype == "local":
-            agents = _resolve_market_agents(db, [node["agent_id"]], user_id)
-            if not agents:
-                raise OrchestrationError(f"local 节点 {nid} 引用的 agent_id={node['agent_id']} 不存在")
-            ni = _assemble_node_input(query, upstream, is_entry)
-            res = await _run_local_agent_turn(agents[0], ni, db, user_id, conv_id)
-            return {
-                "ok": True,
-                "node_id": nid,
-                "node_type": "local",
-                "output": res["text"],
-                "used_tools": res.get("used_tools"),
-            }
-
-        if ntype == "external":
-            from app.api.chat_agents import _call_external_agent
-
-            agents = _resolve_market_agents(db, [node["agent_id"]], user_id)
-            if not agents:
-                raise OrchestrationError(f"external 节点 {nid} 引用的 agent_id={node['agent_id']} 不存在")
-            ni = _assemble_node_input(query, upstream, is_entry)
-            answer, _ = await _call_external_agent(
-                db, agents[0], query=ni, user_id=user_id, ext_conversation_id=None, history=[]
-            )
-            return {"ok": True, "node_id": nid, "node_type": "external", "output": answer or ""}
-
-    raise OrchestrationError(f"未知节点类型 {ntype}")
 
 
 # ---------------------------------------------------------------------------
 # 入口
 # ---------------------------------------------------------------------------
-async def run_graph(graph, query, db, user_id, conv_id, trace_id=None):
+async def _emit(on_event, event) -> None:
+    """图级事件发射（失败安全 —— 观测不得拖垮编排）。"""
+    if on_event is None:
+        return
+    try:
+        res = on_event(event)
+        if asyncio.iscoroutine(res) or isinstance(res, asyncio.Future):
+            await res
+    except Exception as e:
+        logger.debug("[orchestrator] emit 失败(忽略): %s", e)
+
+
+async def run_graph(
+    graph,
+    query,
+    db,
+    user_id,
+    conv_id,
+    trace_id=None,
+    *,
+    on_event=None,
+    claw_type: str = "",
+    tenant_id: str = "",
+    snapshot=None,
+):
     """执行编排图，返回 (answer: str, trace: dict)。
 
     - 拓扑分层后逐层 asyncio.gather 并行执行。
     - 节点异常按 on_error=halt|continue 处理（默认 continue，失败时下游拿到错误作为输入）。
+    - on_event 非空时，图级与节点级过程以统一 EngineEvent 推送：
+      Web Claw 传 SseAdapter（逐帧渲染），IM Claw 传 BufferAdapter（整条装配）。
+      两端共用同一份事件流，IM 侧无需任何编排专属逻辑。
     """
     validate_graph(graph)
     layers = topological_layers(graph)
     blackboard: dict = {}
     trace = {"nodes": [], "trace_id": trace_id, "layers": layers}
 
+    claw = normalize_claw(claw_type)
+    counter = StepCounter()
+    node_count = len(graph.get("nodes") or [])
+
+    await _emit(
+        on_event,
+        make_step(
+            STEP_THINKING,
+            text=f"开始编排执行：{node_count} 个节点 / {len(layers)} 层",
+            trace_id=str(trace_id or ""),
+            tenant_id=str(tenant_id or ""),
+            claw=claw,
+            index=counter.next(),
+        ),
+    )
+
     for layer in layers:
         tasks = [
-            _run_node(_node_by_id(graph, nid), graph, blackboard, query, user_id, conv_id, trace_id)
+            _run_node(
+                _node_by_id(graph, nid),
+                graph,
+                blackboard,
+                query,
+                user_id,
+                conv_id,
+                trace_id,
+                on_event=on_event,
+                claw_type=claw_type,
+                tenant_id=tenant_id,
+                counter=counter,
+                snapshot=snapshot,
+            )
             for nid in layer
         ]
         results = await asyncio.gather(*tasks, return_exceptions=True)
@@ -392,4 +487,17 @@ async def run_graph(graph, query, db, user_id, conv_id, trace_id=None):
     else:
         answer = f"⚠️ 编排输出节点 {output_node} 执行失败: {final.get('error', '未知错误')}"
     trace["output_node"] = output_node
+
+    ok_cnt = sum(1 for n in trace["nodes"] if n.get("ok"))
+    await _emit(
+        on_event,
+        make_step(
+            STEP_THINKING,
+            text=f"编排执行完成：{ok_cnt}/{len(trace['nodes'])} 个节点成功，输出节点 {output_node}",
+            trace_id=str(trace_id or ""),
+            tenant_id=str(tenant_id or ""),
+            claw=claw,
+            index=counter.next(),
+        ),
+    )
     return answer, trace

backend/python/app/core/yaml_workflow.py

@@ -72,10 +72,38 @@ _REVERSE_CONFIG = {
     "agent_endpoint", "api_key", "strategy", "max_tokens",
 }
 
-VALID_NODE_TYPES = {"llm", "local", "external", "reduce"}
 VALID_ON_ERROR = {"halt", "continue"}
 VALID_REDUCE_STRATEGY = {"concat", "first", "last"}
 
+# #874 新节点(skill/mcp/loop)的身份与配置字段：必须留在节点顶层双向透传。
+# 若任其落进 config 子对象，反向转换时 _CONFIG_MAP 取不回来 —— YAML 存一次就丢配置。
+_NODE_FIELDS = (
+    "label", "skill_id", "func_name", "tool", "server_id", "args",
+    "body", "items", "split", "max_iterations", "concurrency", "join",
+)
+
+
+def valid_node_types() -> set:
+    """合法节点类型 —— 唯一真相源是 NodeRegistry（#874）。
+
+    改造前这里硬编码了一份 {llm, local, external, reduce}，与 orchestrator 里的
+    元组各写各的：注册新节点必须改两处，漏一处就出现「画布能跑、YAML 报非法」。
+    现在统一读注册表；注册表不可用时退回内建四型，保证校验器永不因导入失败而瘫痪。
+    """
+    try:
+        from app.core.node_registry import registry
+
+        types = set(registry.node_types())
+        if types:
+            return types
+    except Exception:  # pragma: no cover - 仅在注册表不可用时兜底
+        pass
+    return {"llm", "local", "external", "reduce"}
+
+
+# 兼容导出：历史调用方 `from ... import VALID_NODE_TYPES` 仍可用（import 期快照）
+VALID_NODE_TYPES = valid_node_types()
+
 
 # ── validation ──────────────────────────────────────────────────────────────
 
@@ -112,18 +140,29 @@ def validate_yaml(data: Any) -> dict:
             errors.append(f"节点 id 重复: {nid}")
         node_ids.add(nid)
 
+        allowed = valid_node_types()
         ntype = node.get("type")
-        if ntype not in VALID_NODE_TYPES:
+        if ntype not in allowed:
-            errors.append(f"节点 {nid} 类型非法: {ntype} (允许 {', '.join(sorted(VALID_NODE_TYPES))})")
+            errors.append(f"节点 {nid} 类型非法: {ntype} (允许 {', '.join(sorted(allowed))})")
 
         on_error = node.get("on_error", "continue")
         if on_error and on_error not in VALID_ON_ERROR:
             errors.append(f"节点 {nid} on_error 非法: {on_error}")
 
         # type-specific checks
-        if ntype in ("local", "external"):
+        if ntype in ("local", "external", "agent"):
             if not node.get("agent_id"):
                 errors.append(f"节点 {nid} 类型 {ntype} 必须指定 agent_id")
+        if ntype == "skill" and not (node.get("skill_id") or node.get("func_name")):
+            errors.append(f"节点 {nid} 类型 skill 必须指定 skill_id 或 func_name")
+        if ntype == "mcp" and not node.get("tool"):
+            errors.append(f"节点 {nid} 类型 mcp 必须指定 tool")
+        if ntype == "loop":
+            body = node.get("body")
+            if not isinstance(body, dict) or not body.get("type"):
+                errors.append(f"节点 {nid} 类型 loop 必须提供含 type 的 body")
+            elif body["type"] not in allowed or body["type"] == "loop":
+                errors.append(f"节点 {nid} 的 body 类型非法: {body.get('type')}")
 
         # config validation
         cfg = node.get("config", {})
@@ -205,10 +244,15 @@ def yaml_to_orchestration(data: dict) -> dict:
             "on_error": raw_node.get("on_error", "continue"),
         }
 
-        # agent_id for local/external
+        # agent_id for local/external/agent
         if raw_node.get("agent_id"):
             node["agent_id"] = raw_node["agent_id"]
 
+        # #874 skill/mcp/loop 节点字段原样透传
+        for _f in _NODE_FIELDS:
+            if raw_node.get(_f) is not None:
+                node[_f] = raw_node[_f]
+
         # flatten config into top-level keys
         config = raw_node.get("config", {})
         if isinstance(config, dict):
@@ -218,7 +262,8 @@ def yaml_to_orchestration(data: dict) -> dict:
 
         # also carry over any extra keys from raw_node (preserve flexibility)
         for k, v in raw_node.items():
-            if k not in ("id", "type", "config", "description", "on_error", "agent_id"):
+            if k not in ("id", "type", "config", "description", "on_error", "agent_id") \
+               and k not in _NODE_FIELDS:
                 if k in ("skill_ids", "knowledge_base_ids", "model_id", "system_prompt",
                           "agent_endpoint", "api_key", "strategy", "max_tokens"):
                     node[k] = v
@@ -262,6 +307,9 @@ def orchestration_to_yaml(data: dict) -> dict:
             node["on_error"] = json_node["on_error"]
         if json_node.get("agent_id"):
             node["agent_id"] = json_node["agent_id"]
+        for _f in _NODE_FIELDS:
+            if json_node.get(_f) is not None:
+                node[_f] = json_node[_f]
 
         # nest known config keys into config sub-object
         config: dict = {}
@@ -272,7 +320,7 @@ def orchestration_to_yaml(data: dict) -> dict:
         # also preserve any extra keys not in the map
         for k, v in json_node.items():
             if k not in ("id", "type", "description", "on_error", "agent_id") \
-               and k not in _REVERSE_CONFIG:
+               and k not in _NODE_FIELDS and k not in _REVERSE_CONFIG:
                 if v is not None:
                     config[k] = v
 

backend/python/tests/test_node_registry.py

@@ -0,0 +1,370 @@
+"""NodeRegistry 回归哨兵 —— R-NODE-*（设计书 leagent_accel_belt_design.md §B，任务 #874）。
+
+守护两条硬约束：
+
+  1. **单一真相源**：节点类型只有 NodeRegistry 一份。任何人把
+     orchestrator / yaml_workflow 的类型集合改回硬编码，这里立刻变红 ——
+     历史上「画布能跑、YAML 报非法」的分裂 bug 正是两份清单各写各的。
+
+  2. **双 Claw 对齐**：同一张图在 Web Claw(SseAdapter) 与 IM Claw(BufferAdapter)
+     上必须产出**逐键相同**的过程事件，且事件带 tenant_id + 归一后的 claw
+     （im_wechat → im，否则 Grafana 维度碎成多片）。
+
+全部用例确定性、无网络、无 DB：节点执行被替换为测试节点，SessionLocal 被 mock。
+"""
+import asyncio
+import os
+import sys
+
+import pytest
+
+_HERE = os.path.dirname(os.path.abspath(__file__))
+sys.path.insert(0, os.path.join(_HERE, ".."))
+
+from app.core import orchestrator as orch  # noqa: E402
+from app.core import yaml_workflow as ywf  # noqa: E402
+from app.core.claw_adapters import BufferAdapter, SseAdapter  # noqa: E402
+from app.core.engine_event import SEV_ERROR  # noqa: E402
+from app.core.node_registry import (  # noqa: E402
+    NODE_AGENT,
+    NODE_EXTERNAL,
+    NODE_LLM,
+    NODE_LOCAL,
+    NODE_LOOP,
+    NODE_MCP,
+    NODE_REDUCE,
+    NODE_SKILL,
+    NodeBase,
+    NodeContext,
+    NodeError,
+    NodeResult,
+    StepCounter,
+    normalize_claw,
+    registry,
+)
+
+BUILTIN_TYPES = {
+    NODE_LLM, NODE_REDUCE, NODE_LOCAL, NODE_EXTERNAL,
+    NODE_SKILL, NODE_MCP, NODE_AGENT, NODE_LOOP,
+}
+TENANT = "t-node-874"
+TRACE = "trace-node-874"
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# 测试替身
+# ─────────────────────────────────────────────────────────────────────────────
+class EchoNode(NodeBase):
+    """确定性节点：回显节点输入，不碰 DB / 网络。"""
+
+    node_type = "test_echo"
+    name = "Echo(test)"
+
+    async def _execute(self, ctx: NodeContext) -> NodeResult:
+        return NodeResult(output=f"echo<{ctx.node_id}>:{ctx.node_input()[:20]}")
+
+
+class BoomNode(NodeBase):
+    """确定性失败节点：验证 phase=fail + 结构化诊断在两端都可见。"""
+
+    node_type = "test_boom"
+    name = "Boom(test)"
+
+    async def _execute(self, ctx: NodeContext) -> NodeResult:
+        raise RuntimeError("boom for test")
+
+
+class _DummySession:
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *a):
+        return False
+
+    def close(self):
+        pass
+
+
+@pytest.fixture(autouse=True)
+def _test_nodes(monkeypatch):
+    """注册测试节点 + 屏蔽真实 DB 会话；用完清理，避免污染注册表单例。"""
+    registry.register(EchoNode, override=True)
+    registry.register(BoomNode, override=True)
+    monkeypatch.setattr("app.database.SessionLocal", lambda: _DummySession())
+    yield
+    registry._nodes.pop("test_echo", None)
+    registry._nodes.pop("test_boom", None)
+
+
+def _graph(ntype: str = "test_echo") -> dict:
+    return {
+        "nodes": [
+            {"id": "a", "type": ntype, "label": "节点A"},
+            {"id": "b", "type": ntype, "label": "节点B"},
+        ],
+        "edges": [{"from": "a", "to": "b"}],
+        "output_node": "b",
+    }
+
+
+def _fold_sse(adapter: SseAdapter) -> dict:
+    """把 SSE 帧按 index 折叠成最终态，口径与 BufferAdapter 的原地覆盖一致。"""
+    out: dict = {}
+    while not adapter.queue.empty():
+        wire = adapter.queue.get_nowait()
+        if wire.get("type") != "step":
+            continue
+        out[wire.get("index")] = wire["step"]
+    return out
+
+
+def _fold_buffer(adapter: BufferAdapter) -> dict:
+    return {i: s for i, s in enumerate(adapter.steps)}
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# R-NODE-1：内建节点齐全
+# ─────────────────────────────────────────────────────────────────────────────
+def test_rnode1_builtin_types_registered():
+    types = set(registry.node_types())
+    missing = BUILTIN_TYPES - types
+    assert not missing, f"内建节点缺失: {missing}"
+    for t in BUILTIN_TYPES:
+        impl = registry.get(t)
+        assert impl is not None and callable(getattr(impl, "run", None))
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# R-NODE-2：单一真相源 —— 校验器必须实时读注册表
+# ─────────────────────────────────────────────────────────────────────────────
+def test_rnode2_single_source_of_truth():
+    """orchestrator 与 yaml_workflow 的合法类型集合必须来自 registry。
+
+    测试节点是运行期注册的：若任一侧改回硬编码常量，它就不会出现在集合里。
+    """
+    reg = set(registry.node_types())
+    assert "test_echo" in reg
+
+    assert set(ywf.valid_node_types()) == reg, "yaml_workflow 未实时读注册表"
+
+    # orchestrator 侧：运行期注册的类型必须直接通过图校验
+    orch.validate_graph(_graph())
+
+    # 反向：未注册类型必须被两侧同时拒绝
+    bad = {"nodes": [{"id": "x", "type": "nope_not_registered"}], "edges": []}
+    with pytest.raises(orch.OrchestrationError):
+        orch.validate_graph(bad)
+    res = ywf.validate_yaml({"version": ywf.SPEC_VERSION, "nodes": [{"id": "x", "type": "nope_not_registered"}]})
+    assert not res["ok"]
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# R-NODE-3：节点元信息契约（画布表单依赖）
+# ─────────────────────────────────────────────────────────────────────────────
+def test_rnode3_describe_contract():
+    for d in registry.describe():
+        assert d["node_type"]
+        assert d["name"]
+        assert d["risk_level"] in ("low", "medium", "high"), d
+        assert isinstance(d["input_schema"], dict)
+        assert isinstance(d["output_schema"], dict)
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# R-NODE-4：双 Claw 对齐（核心哨兵）
+# ─────────────────────────────────────────────────────────────────────────────
+def test_rnode4_dual_claw_event_alignment():
+    """同一张图在 Web / IM 两端产出逐键相同的过程事件。
+
+    这是 #874 的验收门禁：Web 画布与 IM `/flow` 共用 run_graph，
+    若谁给某一端加了「专属分支」，两端事件立刻不等，此处变红。
+    """
+    async def _run(adapter, claw_type):
+        return await orch.run_graph(
+            _graph(), "用户问题", None, "u1", None, TRACE,
+            on_event=adapter, claw_type=claw_type, tenant_id=TENANT,
+        )
+
+    sse = SseAdapter(wire_meta=True, trace_id=TRACE, tenant_id=TENANT)
+    buf = BufferAdapter(trace_id=TRACE, tenant_id=TENANT)
+
+    ans_web, trace_web = asyncio.run(_run(sse, "web"))
+    ans_im, trace_im = asyncio.run(_run(buf, "im_wechat"))
+
+    assert ans_web == ans_im, "同一张图两端答案不一致"
+    assert [n["id"] for n in trace_web["nodes"]] == [n["id"] for n in trace_im["nodes"]]
+
+    web_steps, im_steps = _fold_sse(sse), _fold_buffer(buf)
+    assert web_steps, "Web Claw 未收到任何节点事件"
+    assert web_steps == im_steps, (
+        "双 Claw 事件不一致(有人给某一端加了专属逻辑):\n"
+        f"web={web_steps}\nim={im_steps}"
+    )
+
+    # 节点身份必须随事件下发，否则前端画不出 DAG 进度
+    node_steps = [s for s in web_steps.values() if "node" in s]
+    assert node_steps, "节点事件缺少 node 身份字段"
+    assert {s["node"]["id"] for s in node_steps} == {"a", "b"}
+    assert all(s["node"]["type"] == "test_echo" for s in node_steps)
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# R-NODE-5：claw 维度归一 + 租户标签（#876 看板依赖）
+# ─────────────────────────────────────────────────────────────────────────────
+def test_rnode5_claw_normalized_and_tenant_tagged():
+    assert normalize_claw("web") == "web"
+    assert normalize_claw("im") == "im"
+    assert normalize_claw("im_wechat") == "im", "IM 变体未归一 → Grafana 维度会碎片化"
+    assert normalize_claw("im_platform") == "im"
+    assert normalize_claw("") == ""
+
+    seen = []
+
+    async def collector(ev):
+        seen.append(ev)
+
+    asyncio.run(orch.run_graph(
+        _graph(), "q", None, "u1", None, TRACE,
+        on_event=collector, claw_type="im_wechat", tenant_id=TENANT,
+    ))
+
+    assert seen, "未发出任何引擎事件"
+    for ev in seen:
+        assert ev.tenant_id == TENANT, f"事件缺 tenant_id: {ev}"
+        assert ev.claw == "im", f"claw 未归一: {ev.claw}"
+        assert ev.trace_id == TRACE
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# R-NODE-6：新节点类型的必填字段校验（两侧一致）
+# ─────────────────────────────────────────────────────────────────────────────
+@pytest.mark.parametrize("node,keyword", [
+    ({"id": "n", "type": NODE_SKILL}, "skill_id"),
+    ({"id": "n", "type": NODE_MCP}, "tool"),
+    ({"id": "n", "type": NODE_AGENT}, "agent_id"),
+    ({"id": "n", "type": NODE_LOOP}, "body"),
+    ({"id": "n", "type": NODE_LOOP, "body": {"type": NODE_LOOP}}, "body"),
+])
+def test_rnode6_required_fields(node, keyword):
+    with pytest.raises(orch.OrchestrationError) as ei:
+        orch.validate_graph({"nodes": [node], "edges": []})
+    assert keyword in str(ei.value)
+
+    res = ywf.validate_yaml({"version": ywf.SPEC_VERSION, "nodes": [node]})
+    assert not res["ok"] and any(keyword in e for e in res["errors"])
+
+
+def test_rnode6b_valid_new_nodes_pass():
+    ok_nodes = [
+        {"id": "s", "type": NODE_SKILL, "skill_id": "sk-1"},
+        {"id": "m", "type": NODE_MCP, "server_id": "srv-1", "tool": "search"},
+        {"id": "g", "type": NODE_AGENT, "agent_id": "ag-1"},
+        {"id": "l", "type": NODE_LOOP, "body": {"type": NODE_LLM}},
+    ]
+    orch.validate_graph({"nodes": ok_nodes, "edges": []})
+    res = ywf.validate_yaml({"version": ywf.SPEC_VERSION, "nodes": ok_nodes})
+    assert res["ok"], res["errors"]
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# R-NODE-7：YAML ↔ JSON 往返不得吞掉新节点字段
+# ─────────────────────────────────────────────────────────────────────────────
+def test_rnode7_yaml_roundtrip_preserves_fields():
+    src = {
+        "nodes": [
+            {"id": "s", "type": NODE_SKILL, "skill_id": "sk-1", "args": {"q": "x"}},
+            {"id": "m", "type": NODE_MCP, "server_id": "srv-1", "tool": "search"},
+            {"id": "l", "type": NODE_LOOP, "body": {"type": NODE_LLM},
+             "max_iterations": 3, "join": "concat"},
+        ],
+        "edges": [{"from": "s", "to": "m"}],
+        "output_node": "m",
+    }
+    back = ywf.yaml_to_orchestration(ywf.orchestration_to_yaml(src))
+    got = {n["id"]: n for n in back["nodes"]}
+
+    assert got["s"]["skill_id"] == "sk-1"
+    assert got["s"]["args"] == {"q": "x"}
+    assert got["m"]["tool"] == "search" and got["m"]["server_id"] == "srv-1"
+    assert got["l"]["body"] == {"type": NODE_LLM}
+    assert got["l"]["max_iterations"] == 3 and got["l"]["join"] == "concat"
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# R-NODE-8：LoopNode 限流与聚合
+# ─────────────────────────────────────────────────────────────────────────────
+def test_rnode8_loop_respects_limit_and_joins():
+    node = {
+        "id": "loop1", "type": NODE_LOOP,
+        "body": {"type": "test_echo"},
+        "items": [f"item{i}" for i in range(50)],
+        "max_iterations": 3,
+    }
+    ctx = NodeContext(
+        db=None, node=node, query="q", upstream="", is_entry=True,
+        tenant_id=TENANT, claw_type="web", counter=StepCounter(),
+    )
+    res = asyncio.run(registry.get(NODE_LOOP).run(ctx))
+
+    assert res.meta["iterations"] == 3, "max_iterations 未生效 —— 图里写个大数就能打满机器"
+    assert res.meta["failed"] == 0
+    assert res.output.count("echo<") == 3
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# R-NODE-9：失败必须带结构化诊断，且两端都看得见
+# ─────────────────────────────────────────────────────────────────────────────
+def test_rnode9_failure_diagnosis_visible_on_both_claws():
+    graph = {"nodes": [{"id": "a", "type": "test_boom", "label": "会炸的节点"}], "edges": []}
+
+    async def _run(adapter, claw_type):
+        return await orch.run_graph(
+            graph, "q", None, "u1", None, TRACE,
+            on_event=adapter, claw_type=claw_type, tenant_id=TENANT,
+        )
+
+    sse = SseAdapter(wire_meta=True, trace_id=TRACE, tenant_id=TENANT)
+    buf = BufferAdapter(trace_id=TRACE, tenant_id=TENANT)
+    answer_web, _ = asyncio.run(_run(sse, "web"))
+    answer_im, _ = asyncio.run(_run(buf, "im_wechat"))
+
+    assert "失败" in answer_web and answer_web == answer_im
+
+    web_steps, im_steps = _fold_sse(sse), _fold_buffer(buf)
+    assert web_steps == im_steps, "失败路径上双 Claw 事件不一致"
+
+    diags = [s["diagnosis"] for s in web_steps.values() if s.get("diagnosis")]
+    assert diags, "节点失败未产出结构化诊断（前端只能靠字符串猜错误）"
+    d = diags[0]
+    assert d["severity"] == SEV_ERROR and d["code"] and d["summary"]
+    fail_steps = [s for s in web_steps.values() if s.get("phase") == "fail"]
+    assert fail_steps and fail_steps[0]["result"], "失败步骤 result 为空会被前端渲染成「运行中」"
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# R-NODE-10：禁止第二份能力发现实现（#866 事故的精确回归签名）
+# ─────────────────────────────────────────────────────────────────────────────
+def test_rnode10_no_second_capability_discovery():
+    path = os.path.join(_HERE, "..", "app", "core", "node_registry.py")
+    with open(path, "r", encoding="utf-8") as f:
+        src = f.read()
+
+    assert "capability_discovery" in src, "注册表必须走统一能力发现层"
+    for banned in ('MarketApp.status == "published"', "AppInstallation", "_list_enabled_mcp_servers"):
+        assert banned not in src, (
+            f"node_registry 内出现第二份能力发现逻辑({banned})；"
+            "能力发现只允许 capability_discovery 一份实现"
+        )
+
+
+def test_rnode11_registry_rejects_invalid_node():
+    class NoType:
+        node_type = ""
+
+    class NoRun:
+        node_type = "x_no_run"
+
+    with pytest.raises(ValueError):
+        registry.register(NoType)
+    with pytest.raises(ValueError):
+        registry.register(NoRun)