From 33d0f9c32e562a68a12ebb63f209c9478668b21d Mon Sep 17 00:00:00 2001 From: QihuAI Design Date: Tue, 4 Aug 2026 15:16:00 +0800 Subject: [PATCH] =?UTF-8?q?feat(workflow=20#874):=20NodeRegistry=20?= =?UTF-8?q?=E7=BB=9F=E4=B8=80=E5=B7=A5=E4=BD=9C=E6=B5=81=E8=8A=82=E7=82=B9?= =?UTF-8?q?=E6=8A=BD=E8=B1=A1=20+=20=E5=8F=8C=20Claw=20=E4=BA=8B=E4=BB=B6?= =?UTF-8?q?=E5=AF=B9=E9=BD=90?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit #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 | 11 + .../python/app/core/capability_discovery.py | 94 ++ backend/python/app/core/engine_event.py | 11 +- backend/python/app/core/node_registry.py | 1006 +++++++++++++++++ backend/python/app/core/orchestrator.py | 206 +++- backend/python/app/core/yaml_workflow.py | 62 +- backend/python/tests/test_node_registry.py | 370 ++++++ 7 files changed, 1703 insertions(+), 57 deletions(-) create mode 100644 backend/python/app/core/node_registry.py create mode 100644 backend/python/tests/test_node_registry.py diff --git a/backend/python/app/api/chat_service.py b/backend/python/app/api/chat_service.py index 0827df5..d6879d0 100644 --- a/backend/python/app/api/chat_service.py +++ b/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 ""), diff --git a/backend/python/app/core/capability_discovery.py b/backend/python/app/core/capability_discovery.py index 2de3939..c1e1227 100644 --- a/backend/python/app/core/capability_discovery.py +++ b/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 # -------------------------------------------------------------------------- diff --git a/backend/python/app/core/engine_event.py b/backend/python/app/core/engine_event.py index cd89fb6..2080e53 100644 --- a/backend/python/app/core/engine_event.py +++ b/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: diff --git a/backend/python/app/core/node_registry.py b/backend/python/app/core/node_registry.py new file mode 100644 index 0000000..934a038 --- /dev/null +++ b/backend/python/app/core/node_registry.py @@ -0,0 +1,1006 @@ +"""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() diff --git a/backend/python/app/core/orchestrator.py b/backend/python/app/core/orchestrator.py index e229c36..decd513 100644 --- a/backend/python/app/core/orchestrator.py +++ b/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: - from app.api.chat_skill import _resolve_market_agents +async def _run_node( + 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"): - if ntype == "llm": - sys_prompt = node.get("system_prompt") or ( - "你是一个处理节点，请根据上游输入完成指定的文本变换、摘要或路由判断，" - "只输出处理后的结果，不要多余解释。" - ) - else: - sys_prompt = node.get("system_prompt") or ( - "你是一个结果汇总器，请将多个子任务的输出综合成连贯、完整、面向用户的最终回答。" - ) - uc = _assemble_node_input(query, upstream, is_entry) - text = await _llm_complete(sys_prompt, uc, node.get("model_id"), db, user_id) - 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}") + ctx = NodeContext( + db=db, + 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(), + ) + try: + result = await impl.run(ctx) + except NodeError as e: + # 归一为 OrchestrationError，run_graph 的 on_error=halt|continue 语义不变 + raise OrchestrationError(str(e)) from e + return result.to_blackboard() # --------------------------------------------------------------------------- # 入口 # --------------------------------------------------------------------------- -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 diff --git a/backend/python/app/core/yaml_workflow.py b/backend/python/app/core/yaml_workflow.py index f15c88b..f66a45d 100644 --- a/backend/python/app/core/yaml_workflow.py +++ b/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: - errors.append(f"节点 {nid} 类型非法: {ntype} (允许 {', '.join(sorted(VALID_NODE_TYPES))})") + if ntype not in allowed: + 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 diff --git a/backend/python/tests/test_node_registry.py b/backend/python/tests/test_node_registry.py new file mode 100644 index 0000000..573e4f3 --- /dev/null +++ b/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)