Agent/examples/integration_test_6/quick_test.py

#!/usr/bin/env python3
"""
快速验证脚本 - 测试信号机制基础功能

不运行完整的 Agent，只测试信号机制的基本功能
"""

import sys
from pathlib import Path

# 添加项目根目录到 Python 路径
project_root = Path(__file__).parent.parent.parent
sys.path.insert(0, str(project_root))

from agent.services.subagent.signals import SignalBus, Signal


def test_signal_bus():
    """测试 SignalBus 基本功能"""
    print("=" * 60)
    print("测试 SignalBus 基本功能")
    print("=" * 60)
    print()

    # 1. 创建 SignalBus
    print("1. 创建 SignalBus...")
    bus = SignalBus()
    print("   ✅ SignalBus 创建成功")
    print()

    # 2. 发送信号
    print("2. 发送信号...")
    signal1 = Signal(
        type="subagent.start",
        trace_id="sub-trace-001",
        data={
            "parent_trace_id": "main-trace-001",
            "mode": "evaluate",
            "task": "测试任务"
        }
    )
    bus.emit(signal1)
    print(f"   ✅ 发送信号: {signal1.type}")
    print()

    signal2 = Signal(
        type="subagent.complete",
        trace_id="sub-trace-001",
        data={
            "parent_trace_id": "main-trace-001",
            "result": {"passed": True},
            "status": "completed"
        }
    )
    bus.emit(signal2)
    print(f"   ✅ 发送信号: {signal2.type}")
    print()

    # 3. 检查信号
    print("3. 检查信号...")
    signals = bus.check_buffer("main-trace-001")
    print(f"   ✅ 收到 {len(signals)} 个信号")
    for i, sig in enumerate(signals, 1):
        print(f"      {i}. {sig.type} (trace: {sig.trace_id})")
    print()

    # 4. 验证缓冲池已清空
    print("4. 验证缓冲池已清空...")
    signals2 = bus.check_buffer("main-trace-001")
    if len(signals2) == 0:
        print("   ✅ 缓冲池已清空")
    else:
        print(f"   ❌ 缓冲池未清空，还有 {len(signals2)} 个信号")
    print()

    # 5. 测试多个 trace
    print("5. 测试多个 trace...")
    signal3 = Signal(
        type="subagent.start",
        trace_id="sub-trace-002",
        data={"parent_trace_id": "main-trace-002"}
    )
    bus.emit(signal3)

    signal4 = Signal(
        type="subagent.start",
        trace_id="sub-trace-003",
        data={"parent_trace_id": "main-trace-003"}
    )
    bus.emit(signal4)

    signals_trace2 = bus.check_buffer("main-trace-002")
    signals_trace3 = bus.check_buffer("main-trace-003")

    print(f"   ✅ trace-002 收到 {len(signals_trace2)} 个信号")
    print(f"   ✅ trace-003 收到 {len(signals_trace3)} 个信号")
    print()

    print("=" * 60)
    print("✅ 所有测试通过！SignalBus 工作正常。")
    print("=" * 60)


def test_signal_import():
    """测试信号模块导入"""
    print()
    print("=" * 60)
    print("测试模块导入")
    print("=" * 60)
    print()

    try:
        from agent.core.runner import AgentRunner
        print("✅ AgentRunner 导入成功")

        # 检查是否有 signal_bus 属性
        import inspect
        init_source = inspect.getsource(AgentRunner.__init__)
        if "signal_bus" in init_source:
            print("✅ AgentRunner.__init__ 包含 signal_bus")
        else:
            print("❌ AgentRunner.__init__ 不包含 signal_bus")

    except Exception as e:
        print(f"❌ 导入失败: {e}")

    try:
        from agent.services.subagent.manager import SubAgentManager
        print("✅ SubAgentManager 导入成功")

        # 检查是否导入了 Signal
        import inspect
        source = inspect.getsource(SubAgentManager)
        if "Signal" in source:
            print("✅ SubAgentManager 使用了 Signal")
        else:
            print("❌ SubAgentManager 未使用 Signal")

    except Exception as e:
        print(f"❌ 导入失败: {e}")

    print()
    print("=" * 60)


if __name__ == "__main__":
    test_signal_bus()
    test_signal_import()