vent_agent_2.0/scripts/selfcheck.py

"""结构自检：导入 → 建表 → checkpointer → agent 编译 → 工具注册 → 权限。

比 pytest 更「从头到尾」，比 http_e2e 更轻——改完结构（加模块、换依赖、
动 agent 组装）先跑这个，能在几秒内定位「哪一层断了」，不必起服务。

不需要模型网络：agent 编译只构造图，不发请求。

用法：python scripts/selfcheck.py
退出码 0 表示全部通过。
"""

from __future__ import annotations

import asyncio
import importlib
import json
import sys
import traceback
from pathlib import Path

# 让脚本能从任意目录运行
ROOT = Path(__file__).resolve().parent.parent
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))

FAILED: list[str] = []


def step(name: str, fn) -> None:
    """跑一步检查；异常记为失败但不中断后续步骤（一次看全所有问题）。"""
    try:
        detail = fn()
        print(f"  [OK]   {name}" + (f"  — {detail}" if detail else ""))
    except Exception as exc:
        FAILED.append(name)
        print(f"  [FAIL] {name}  — {type(exc).__name__}: {exc}")
        traceback.print_exc(limit=4)


# 需要走一遍导入链的模块：漏掉一个都可能让「测试过了但服务起不来」
MODULES = [
    "config", "obs",
    "db.db_conn", "db.chat_store", "db.schedule_store", "db.billing_store",
    "billing.pricing", "billing.guard", "billing.meter",
    "tools.registry", "tools.demo_tools",
    "agents.model_factory", "agents.prompts", "agents.middleware",
    "agents.agent_memory", "agents.agent_factory", "agents.agent_registry",
    "api.sse_events", "api.subagent_tracker", "api.context_meter",
    "api.source_tracker", "api.concurrency", "api.sse_core", "api.models",
    "api.chat_routes", "api.session_routes", "api.model_routes",
    "api.usage_routes", "api.schedule_routes", "api",
    "scheduler.runner", "scheduler.scheduler", "main",
]


def check_imports() -> str:
    for name in MODULES:
        importlib.import_module(name)
    return f"{len(MODULES)} 个模块"


def check_syntax() -> str:
    """静态语法检查：抓出导入不到的孤儿模块（比如新写但没被 import 的文件）。"""
    import ast

    files = [
        p for p in ROOT.rglob("*.py")
        if not any(part in {".venv", "venv", "__pycache__", "data", ".tmp"} for part in p.parts)
    ]
    bad: list[str] = []
    for path in files:
        try:
            ast.parse(path.read_text(encoding="utf-8"))
        except SyntaxError as exc:
            bad.append(f"{path.relative_to(ROOT)}:{exc.lineno}")
    if bad:
        raise SyntaxError("语法错误：" + ", ".join(bad))
    return f"{len(files)} 个文件"


def check_schema() -> str:
    from db import db_conn

    db_conn.init_schema()
    health = db_conn.health()
    if health.get("status") != "ok":
        raise RuntimeError(f"存储不可用：{health}")
    return f"{health['dialect']} → {health['target']}"


def check_pricing() -> str:
    from billing import pricing

    created = pricing.seed_default_pricing()
    from db import billing_store

    missing = [m for m in {pricing.config.MODEL, pricing.config.SUMMARY_MODEL}
               if m and billing_store.get_pricing(m) is None]
    if missing:
        raise RuntimeError(f"模型未定价（会在请求开始时 fail-closed）：{missing}")
    return f"新增 {created} 条，当前模型已覆盖"


def check_permissions() -> str:
    """权限规则顺序：具体的 allow 必须在宽泛的 deny 之前，否则被短路。"""
    from agents.agent_factory import build_permissions

    rules = build_permissions()
    modes = [r.mode for r in rules]
    if "deny" in modes and modes.index("deny") != len(modes) - 1:
        raise RuntimeError(
            f"deny 规则必须排在最后（首条命中即返回，提前会短路后面的 allow）：{modes}"
        )
    return " · ".join(f"{'/'.join(r.operations)}={'+'.join(r.paths[:1])}:{r.mode}" for r in rules)


def check_tools() -> str:
    from tools import registry

    count = registry.load_builtin_tools()
    if count == 0:
        raise RuntimeError(
            "工具注册表为空——检查 tools/registry._BUILTIN_MODULES 与模块里的 @register"
        )
    for tool in registry.get_all_tools():
        if not hasattr(tool, "name"):
            raise RuntimeError(f"工具 {tool!r} 不是 BaseTool（子智能体的 tools 需要 .name）")
    groups = registry.get_group_map()
    return f"{count} 个 / {len(groups)} 组：{', '.join(groups)}"


def check_import_boundaries() -> str:
    """确认工具层没有反向依赖 agents（分层纪律，见 AGENTS.md §三）。"""
    import pathlib

    offenders: list[str] = []
    for path in (ROOT / "tools").rglob("*.py"):
        text = path.read_text(encoding="utf-8")
        for line in text.splitlines():
            stripped = line.strip()
            if stripped.startswith(("import agents", "from agents")):
                offenders.append(f"{path.name}: {stripped}")
    if offenders:
        raise RuntimeError(f"tools/ 不得依赖 agents/：{offenders}")
    return "tools/ 与 billing/ 未反向依赖上层"


async def check_agent() -> str:
    from agents import agent_factory, agent_memory, agent_registry

    await agent_memory.init_checkpointer()
    agent = await asyncio.to_thread(agent_registry.get_agent)
    if agent is None:
        raise RuntimeError("agent 编译返回 None")
    info = agent_factory.describe()
    if not info["tools"]["count"]:
        raise RuntimeError("agent 没有挂任何工具")
    if not info["interrupt_on"]:
        raise RuntimeError("interrupt_on 为空：中断/恢复协议会失效")
    await agent_memory.close_checkpointer()
    return (
        f"{info['name']} · 工具 {info['tools']['count']} · "
        f"子智能体 {info['subagents']} · 审批 {info['interrupt_on']} · "
        f"压缩 {info['compaction']['enabled']}"
    )


def check_contract() -> str:
    """SSE 契约自检：事件类型数量与帧格式。"""
    from api.sse_events import ALL_EVENT_TYPES, EventWriter

    if len(ALL_EVENT_TYPES) != 15:
        raise RuntimeError(f"事件类型应为 15 种，实际 {len(ALL_EVENT_TYPES)}")
    frame = EventWriter("s", "t").token("x")
    if not frame.startswith("data: ") or not frame.endswith("\n\n"):
        raise RuntimeError(f"帧格式不对：{frame!r}")
    payload = json.loads(frame[6:])
    if payload["session_id"] != "s" or payload["trace_id"] != "t":
        raise RuntimeError(f"帧缺少 session_id/trace_id：{payload}")
    return f"{len(ALL_EVENT_TYPES)} 种事件"


def main() -> int:
    print("deep_agent 2.0 结构自检")
    print(f"项目根：{ROOT}")
    print(f"方言：{__import__('os').getenv('DB_DIALECT') or '(默认 mysql)'}")
    print()

    step("Python 源码语法", check_syntax)
    step("模块导入链", check_imports)
    step("建表与存储连通", check_schema)
    step("计费单价覆盖", check_pricing)
    step("SSE 契约", check_contract)
    step("权限规则顺序", check_permissions)
    step("工具注册", check_tools)
    step("分层依赖", check_import_boundaries)
    step("agent 编译", lambda: asyncio.run(check_agent()))

    print()
    if FAILED:
        print(f"失败 {len(FAILED)} 项：{', '.join(FAILED)}")
        return 1
    print("全部通过")
    return 0


if __name__ == "__main__":
    sys.exit(main())