qihuaiflow/backend/python/tests/test_node_registry.py

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