PlantAI-rag-new/tooling/cap/cap_selfcheck.py

#!/usr/bin/env python3
"""tooling/cap/cap_selfcheck.py —— 机制端到端自检（单进程热加载证明）。

流程（单一子进程内完成，与生产同语义——registry 在启动时加载一次，
之后的 yaml 变化必须靠 reload() 热加载，diff 才有意义）：
  1. 初始工具表（空 yaml → 无 fake_echo）
  2. install_manager.register_manifest 落盘 → reload → diff.added 含 fake_echo
  3. 实际调用 echo（MCP stdio 子进程）→ 返回正确
  4. unregister → reload → diff.removed 含 fake_echo、工具表为空
  5. 恢复原 capabilities.yaml

用法：
  QA_PYTHON=<venv python> python3 cap_selfcheck.py <qa-config-dir>

<qa-config-dir> = 编译树根（含 configs/capabilities.yaml），即 GENE_FAMILY_ROOT。
"""
from __future__ import annotations

import json
import os
import subprocess
import sys
import textwrap
from pathlib import Path

TOOL_DIR = Path(__file__).parent
FAKE_MCP = TOOL_DIR / "fake_mcp_server.py"

INNER = textwrap.dedent(
    """
    import json, sys, textwrap
    from pathlib import Path

    root = Path(sys.argv[1])
    sys.path.insert(0, str(root))
    sys.path.insert(0, str(root / "src"))

    from src.rag.qa_v1_1_73.capabilities.registry import get_registry
    from src.rag.qa_v1_1_73.capabilities.install_manager import InstallManager, get_install_manager
    from src.rag.qa_v1_1_73.capabilities.providers import get_dispatcher

    caps_yaml = root / "configs" / "capabilities.yaml"
    original = caps_yaml.read_text(encoding="utf-8")

    manifest = (
        "capabilities:\\n"
        "  - name: fake_echo\\n"
        "    provider: mcp\\n"
        "    permission: read_only\\n"
        "    description: 自检用 echo 工具\\n"
        "    session_default: true\\n"
        "    schema:\\n"
        "      text: {type: string, required: true, description: 要回显的文本}\\n"
        "    mcp:\\n"
        "      command: " + sys.argv[2] + "\\n"
        "      args: [\\"" + sys.argv[3] + "\\"]\\n"
        "      tool: echo\\n"
    )

    results = {}
    try:
        # 1. 初始工具表
        r = get_registry()
        results["initial_tools"] = [t.name for t in r.list_all()]
        assert "fake_echo" not in results["initial_tools"], "预置状态异常"

        # 2. 安装 + 热加载（同进程，diff 有意义）
        manager = InstallManager(caps_yaml)
        manifests, warnings = manager.register_manifest(manifest)
        assert warnings == [] and [m.name for m in manifests] == ["fake_echo"]
        w, diff = manager.reload(registry=r)
        assert "fake_echo" in diff["added"], f"热加载未新增: {diff}"
        results["hot_reload_added"] = diff["added"]
        results["registry_after_add"] = [t.name for t in r.list_all()]

        # 3. MCP 实际调用
        m = r.get("fake_echo").manifest
        out = get_dispatcher().execute(m, {"text": "你好机制"})
        assert "你好机制" in out, f"echo 返回异常: {out}"
        results["mcp_call_ok"] = True

        # 4. 卸载 + 热卸载（同进程）
        assert manager.unregister("fake_echo")
        w, diff = manager.reload(registry=r)
        assert "fake_echo" in diff["removed"], f"热卸载未移除: {diff}"
        get_dispatcher().shutdown_connections(diff["removed"])
        results["hot_reload_removed"] = diff["removed"]
        results["registry_after_remove"] = [t.name for t in r.list_all()]

        # 5. 恢复
        caps_yaml.write_text(original, encoding="utf-8")
        r.reload()
        results["restored"] = True

        print("SELFCHECK PASS:", json.dumps(results, ensure_ascii=False, indent=2))
    except Exception as exc:
        print(f"SELFCHECK FAIL: {exc}")
        caps_yaml.write_text(original, encoding="utf-8")
        sys.exit(1)
    """
)


def _python() -> str:
    return os.environ.get(
        "QA_PYTHON", "/home/xzq/5090/teamWorkspace/zh/plantai-rag-new-venv/bin/python"
    )


def main() -> int:
    root = Path(sys.argv[1]).resolve() if len(sys.argv) > 1 else None
    if root is None or not (root / "configs").is_dir():
        print(__doc__)
        return 2
    env = dict(os.environ)
    env["GENE_FAMILY_ROOT"] = str(root)
    proc = subprocess.run(
        [_python(), "-c", INNER, str(root), _python(), str(FAKE_MCP)],
        capture_output=True,
        text=True,
        timeout=180,
        env=env,
        cwd=str(root),
    )
    sys.stdout.write(proc.stdout)
    sys.stderr.write(proc.stderr[-3000:])
    return proc.returncode


if __name__ == "__main__":
    raise SystemExit(main())