"""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)