diff --git a/backend/python/tests/test_dual_claw.py b/backend/python/tests/test_dual_claw.py new file mode 100644 index 0000000..eb2dd12 --- /dev/null +++ b/backend/python/tests/test_dual_claw.py @@ -0,0 +1,195 @@ +"""Dual-Claw 回归哨兵 —— R-DUAL-*(设计书 leagent_accel_belt_design.md §6,任务 #873)。 + +硬约束:Web Claw 与 IM Bot Claw 必须消费**同一份**能力清单与**同一套**工具命名。 +本文件是「活体哨兵」:任何把 IM Claw 能力改回退(如 #866 覆盖 #764/#765 导致 +mcp=[]/agents=[] 写死)的提交,会在 pre-push 阶段立即变红。 + +历史教训(见设计书 §2):双端对齐若靠「在第二份实现里再写一遍」必然被下一次覆盖 +回退。**唯一根治法是消灭第二份实现**——即 im_llm 必须复用统一的 capability_discovery。 + +测试分两层: + A. 始终运行(确定性、无网络):源码契约守卫 + DB 级能力一致性 + 命名同源守卫。 + B. 可选运行(需联网 MCP/LLM):置 RUN_DUAL_INTEGRATION=1 时执行真实工具调用, + 作为持续哨兵;默认 skip,避免 CI 网关因外部网络抖动而误杀正常提交。 +""" +import os +import re +import glob +import sys + +import pytest + +_HERE = os.path.dirname(os.path.abspath(__file__)) +APP_DIR = os.path.abspath(os.path.join(_HERE, "..", "app")) +API_DIR = os.path.join(APP_DIR, "api") + +sys.path.insert(0, os.path.join(_HERE, "..")) + +from app.database import SessionLocal # noqa: E402 +from app.core.capability_discovery import ( # noqa: E402 + discover_skills, + discover_mcp_servers, + _safe_name, +) + + +def _read_api_sources(): + parts = [] + for p in sorted(glob.glob(os.path.join(API_DIR, "chat*.py"))) + \ + [os.path.join(API_DIR, "im_llm.py")]: + if os.path.exists(p): + with open(p, "r", encoding="utf-8") as f: + parts.append(f.read()) + return "\n".join(parts) + + +@pytest.fixture(scope="module") +def db(): + session = SessionLocal() + yield session + session.close() + + +# ───────────────────────────────────────────────────────────────────────────── +# R-DUAL-2 (源码守卫):IM Claw 必须复用统一能力发现,禁止写死 mcp=[] +# ───────────────────────────────────────────────────────────────────────────── +def test_rdual2_im_reuses_unified_discovery(): + """IM Bot 引擎必须从 capability_discovery 取能力,且不得残留 mcp=[] 写死。 + + 这是 #866 覆盖 #764/#765 的精确回归签名:当时 im_llm 把 mcp=[] 硬编码, + 导致 IM 永远拿不到 MCP 工具。本测试捕获该字面量复现。 + """ + im_src = "" + im_path = os.path.join(API_DIR, "im_llm.py") + with open(im_path, "r", encoding="utf-8") as f: + im_src = f.read() + + # 1) 必须导入统一发现层 + assert "from app.core.capability_discovery import" in im_src, \ + "im_llm 未从 capability_discovery 导入——IM 可能又退化成第二份实现" + assert "discover_skills" in im_src, "im_llm 未使用 discover_skills" + assert "discover_mcp_servers" in im_src, "im_llm 未使用 discover_mcp_servers" + + # 2) 禁止写死 mcp=[](agents=[] 是允许的:当前无已发布 agent) + assert "mcp=[]" not in im_src, \ + "检测到 im_llm 写死 mcp=[]——IM Claw 的 MCP 能力已回退(#866 类回归)" + + # 3) _build_system_prompt 调用必须传入真实 mcp(而非空列表) + assert "mcp=mcp_servers" in im_src or "mcp=mcp_servers," in im_src, \ + "im_llm 未把发现的 MCP 服务器注入 system prompt" + + +# ───────────────────────────────────────────────────────────────────────────── +# R-DUAL-1 (DB 级一致性):双 Claw 返回同一份能力集合 +# ───────────────────────────────────────────────────────────────────────────── +def test_rdual1_skill_set_parity(db): + """同一用户,Web 与 IM 必须发现相同的技能集合(统一发现层 cliff 保证)。""" + web = discover_skills(db, "dual-claw-sentinel", claw="web") + im = discover_skills(db, "dual-claw-sentinel", claw="im") + + assert len(web) > 0, "Web Claw 未发现任何技能(published fallback 可能失效)" + web_names = {s.get("name") for s in web} + im_names = {s.get("name") for s in im} + assert web_names == im_names, ( + f"Web 与 IM 技能集合不一致:\n web={sorted(web_names)}\n im={sorted(im_names)}" + ) + + +def test_rdual1_mcp_server_parity(db): + """双 Claw 必须挂载相同的 MCP 服务器集合(IM 不再丢 MCP)。""" + web = discover_mcp_servers(db, None, claw="web") + im = discover_mcp_servers(db, None, claw="im") + + assert len(web) > 0, "未发现有启用的 MCP 服务器" + web_names = sorted(s.get("name") for s in web) + im_names = sorted(s.get("name") for s in im) + assert web_names == im_names, ( + f"Web 与 IM 的 MCP 服务器不一致:\n web={web_names}\n im={im_names}" + ) + + +# ───────────────────────────────────────────────────────────────────────────── +# R-DUAL-4 (命名同源):双 Claw 工具命名必须同源,杜绝第二份实现发散 +# ───────────────────────────────────────────────────────────────────────────── +def test_rdual4_tool_naming_single_source(): + """capability_discovery._safe_name 必须与 Web Claw 的命名公式完全一致。 + + Web Claw (chat_service.py:3428) 用: + _sname = re.sub(r'[^A-Za-z0-9_-]', '_', name[:30]) + _tname = re.sub(r'[^A-Za-z0-9_-]', '_', name[:50]) + _fname = f"mcp__{_sname}__{_tname}__{mcp_idx}" + 统一层必须用同一个公式,否则两端工具名对不上 → 工具调用在 IM 端失效。 + """ + # 统一层的公式 + for raw, maxlen in [("My Server!", 30), ("some tool.name", 50), ("中文-工具", 50)]: + expected = re.sub(r"[^A-Za-z0-9_-]", "_", raw[:maxlen]) + assert _safe_name(raw, maxlen) == expected, \ + f"_safe_name 公式与 Web Claw 不一致: {_safe_name(raw, maxlen)!r} != {expected!r}" + + # Web Claw 源码必须仍使用同一公式([:30] 给 server,[:50] 给 tool) + cs_path = os.path.join(API_DIR, "chat_service.py") + with open(cs_path, "r", encoding="utf-8") as f: + cs_src = f.read() + assert "re.sub(r'[^A-Za-z0-9_-]', '_', " in cs_src, \ + "chat_service.py 工具命名公式缺失/被改写" + assert "[:30]" in cs_src and "[:50]" in cs_src, \ + "chat_service.py 的 server/tool 截断长度与统一层不匹配(应为 30/50)" + + +# ───────────────────────────────────────────────────────────────────────────── +# R-DUAL-5 (媒体结构契约):IM 必须保留 __media__ 结构化返回 +# ───────────────────────────────────────────────────────────────────────────── +def test_rdual5_im_preserves_media_struct(): + """微信内联图片依赖 __media__ 结构化返回,禁止在 IM 端退化成纯文本。""" + im_path = os.path.join(API_DIR, "im_llm.py") + with open(im_path, "r", encoding="utf-8") as f: + im_src = f.read() + assert "__media__" in im_src, \ + "im_llm 缺失 __media__ 结构化返回——微信内联图片将失效(R-DUAL-5)" + + +# ───────────────────────────────────────────────────────────────────────────── +# R-DUAL-2 / R-DUAL-3(真实集成,需联网,默认 skip) +# ───────────────────────────────────────────────────────────────────────────── +_RUN_LIVE = os.environ.get("RUN_DUAL_INTEGRATION") == "1" + + +@pytest.mark.skipif(not _RUN_LIVE, reason="set RUN_DUAL_INTEGRATION=1 to run live MCP/LLM integration") +@pytest.mark.asyncio +async def test_rdual2_live_im_mcp_tools_reachable(db): + """真实拉取启用 MCP 服务器的工具列表,证明 IM 端代理绕过已生效(对照 #P0 修复)。""" + from app.mcp_client import list_mcp_tools + + servers = discover_mcp_servers(db, None, claw="im") + assert servers, "没有启用 MCP 服务器可供实时验证" + + total = 0 + for srv in servers: + try: + tools = await list_mcp_tools(srv, timeout=30, raise_on_error=False) + except Exception: + tools = [] + total += len(tools or []) + assert total > 0, "实时拉取 MCP 工具为 0——IM 端 MCP 代理可能再次被代理劫持" + + +@pytest.mark.skipif(not _RUN_LIVE, reason="set RUN_DUAL_INTEGRATION=1 to run live agent integration") +def test_rdual3_im_agent_capability(db): + """若已发布 agent,IM 端必须能纳入;当前无已发布 agent 则 skip(待 #873 薄壳化)。""" + from app.models import MarketApp + + agent_count = db.query(MarketApp).filter( + MarketApp.type == "agent", MarketApp.status == "published" + ).count() + if agent_count == 0: + pytest.skip("尚无已发布 agent(market_apps.type='agent'=0),IM agent 能力待 #873 薄壳化后激活") + # 待 agent 接入后补充:断言 im_llm 经 EngineCore 获得 agent 编排能力 + + +# ───────────────────────────────────────────────────────────────────────────── +# R-DUAL-6(OTel 标签,待 #876):双 Claw span 须携带 tenant_id + claw_type +# ───────────────────────────────────────────────────────────────────────────── +@pytest.mark.skip(reason="pending #876 OTel enhancement: spans must carry tenant_id + claw_type") +def test_rdual6_spans_carry_tenant_and_claw(): + """双 Claw 链路的 span 必须带有 tenant_id 与 claw_type 标签(双端横向对比看板依赖)。""" + pytest.skip("pending #876")