Agent/examples/integration_test_6/run.py

#!/usr/bin/env python3
"""
集成测试 6: 信号驱动机制测试

测试目标：
- 验证信号的发送和接收机制
- 验证 wait=True 模式（同步等待信号）
- 验证后台任务执行
- 验证信号轮询机制
- 验证错误信号传播
"""

import asyncio
import sys
import os
from pathlib import Path

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

from dotenv import load_dotenv
load_dotenv()

from agent.llm.prompts import SimplePrompt
from agent.core.runner import AgentRunner
from agent.execution import FileSystemTraceStore, Trace, Message
from agent.llm import create_openrouter_llm_call


async def main():
    """运行测试"""
    # 路径配置
    base_dir = Path(__file__).parent
    prompt_path = base_dir / "task.prompt"
    output_dir = base_dir / "output"

    print("=" * 80)
    print("集成测试 6: 信号驱动机制测试")
    print("=" * 80)
    print()

    # 1. 加载 prompt
    print("1. 加载任务...")
    prompt = SimplePrompt(prompt_path)
    system_prompt = prompt._messages.get("system", "")
    user_prompt = prompt._messages.get("user", "")

    print(f"   ✓ 任务类型: 数据验证模块实现")
    print(f"   ✓ 测试重点: 信号机制")
    print(f"   ✓ 监控内容: 信号发送、接收、轮询")
    print()

    # 2. 创建 Agent Runner
    print("2. 创建 Agent Runner...")
    print(f"   - 模型: Claude Sonnet 4.5")
    print(f"   - 信号机制: 已启用")
    print()

    runner = AgentRunner(
        trace_store=FileSystemTraceStore(base_path=".trace"),
        llm_call=create_openrouter_llm_call(model="anthropic/claude-sonnet-4.5"),
        skills_dir=str(project_root / "agent" / "skills"),
        debug=False
    )

    # 验证 SignalBus 已创建
    if hasattr(runner, 'signal_bus'):
        print("   ✅ SignalBus 已创建")
    else:
        print("   ❌ SignalBus 未创建")
        return

    # 3. 运行 Agent
    print()
    print("3. 启动 Agent...")
    print("=" * 80)
    print()

    # 创建输出目录
    output_dir.mkdir(exist_ok=True)

    # 监控变量
    current_trace_id = None
    goal_used = False
    subagent_used = False
    evaluate_used = False

    iteration_count = 0
    tool_calls_count = {}
    evaluation_count = 0
    evaluation_results = []

    # 信号监控
    signals_emitted = []
    signals_received = []
    signal_types = set()

    # 钩子：监控信号发送
    original_emit = runner.signal_bus.emit
    def monitored_emit(signal):
        signals_emitted.append({
            "type": signal.type,
            "trace_id": signal.trace_id,
            "data_keys": list(signal.data.keys())
        })
        signal_types.add(signal.type)
        print(f"  [信号发送] {signal.type} (trace: {signal.trace_id[:8]}...)")
        return original_emit(signal)

    runner.signal_bus.emit = monitored_emit

    # 钩子：监控信号接收
    original_check_buffer = runner.signal_bus.check_buffer
    def monitored_check_buffer(trace_id):
        signals = original_check_buffer(trace_id)
        if signals:
            for signal in signals:
                signals_received.append({
                    "type": signal.type,
                    "trace_id": signal.trace_id
                })
                print(f"  [信号接收] {signal.type} (trace: {signal.trace_id[:8]}...)")
        return signals

    runner.signal_bus.check_buffer = monitored_check_buffer

    async for item in runner.run(
        task=user_prompt,
        system_prompt=system_prompt,
        model="anthropic/claude-sonnet-4.5",
        temperature=0.5,
        max_iterations=30,
    ):
        # 处理 Trace 对象
        if isinstance(item, Trace):
            current_trace_id = item.trace_id
            if item.status == "running":
                print(f"[Trace] 开始: {item.trace_id[:8]}...")
            elif item.status == "completed":
                print()
                print("=" * 80)
                print(f"[Trace] 完成")
                print(f"  - 总消息数: {item.total_messages}")
                print(f"  - 总 Token 数: {item.total_tokens}")
                print(f"  - 总成本: ${item.total_cost:.4f}")
                print("=" * 80)
            elif item.status == "failed":
                print()
                print(f"[Trace] 失败: {item.error_message}")

        # 处理 Message 对象
        elif isinstance(item, Message):
            if item.role == "assistant":
                iteration_count += 1

                content = item.content
                if isinstance(content, dict):
                    text = content.get("text", "")
                    tool_calls = content.get("tool_calls")

                    # 显示 Agent 的思考
                    if text and not tool_calls:
                        print(f"\n[{iteration_count}] Agent 回复:")
                        print(f"  {text[:200]}{'...' if len(text) > 200 else ''}")
                    elif text:
                        print(f"\n[{iteration_count}] Agent 思考:")
                        print(f"  {text[:150]}{'...' if len(text) > 150 else ''}")

                    # 显示工具调用
                    if tool_calls:
                        for tc in tool_calls:
                            tool_name = tc.get("function", {}).get("name", "unknown")
                            args = tc.get("function", {}).get("arguments", {})

                            # 如果 args 是字符串，尝试解析为 JSON
                            if isinstance(args, str):
                                import json
                                try:
                                    args = json.loads(args)
                                except:
                                    args = {}

                            # 统计工具使用
                            tool_calls_count[tool_name] = tool_calls_count.get(tool_name, 0) + 1

                            # 检测关键工具使用
                            if tool_name == "goal":
                                goal_used = True
                                if isinstance(args, dict):
                                    if args.get("add"):
                                        print(f"  → goal(add): {args['add'][:80]}...")
                                    elif args.get("done"):
                                        print(f"  → goal(done): {args['done'][:80]}...")
                                    elif args.get("focus"):
                                        print(f"  → goal(focus): {args['focus']}")

                            elif tool_name == "subagent":
                                subagent_used = True
                                if isinstance(args, dict):
                                    mode = args.get("mode", "unknown")
                                    wait = args.get("wait", True)
                                    if mode == "evaluate":
                                        evaluate_used = True
                                        evaluation_count += 1
                                        target = args.get("target_goal_id", "?")
                                        wait_str = f"wait={wait}"
                                        print(f"  → subagent(evaluate, {wait_str}): 评估目标 {target} [评估 #{evaluation_count}]")

                            else:
                                # 其他工具简化显示
                                if tool_name in ["read_file", "write_file", "edit_file"]:
                                    if isinstance(args, dict):
                                        file_path = args.get("file_path", "")
                                        if file_path:
                                            file_name = Path(file_path).name
                                            print(f"  → {tool_name}: {file_name}")

            elif item.role == "tool":
                # 检查是否是评估结果
                content = item.content
                if isinstance(content, str):
                    import json
                    try:
                        result = json.loads(content)
                        if isinstance(result, dict) and "passed" in result:
                            passed = result.get("passed", False)
                            reason = result.get("reason", "")[:100]
                            evaluation_results.append({
                                "passed": passed,
                                "reason": reason
                            })
                            status = "✅ 通过" if passed else "❌ 不通过"
                            print(f"  [评估结果] {status}")
                    except:
                        pass

    # 4. 测试结果总结
    print()
    print("=" * 80)
    print("测试结果总结")
    print("=" * 80)
    print()

    print("功能使用情况:")
    print(f"  - goal 工具: {'✅ 使用' if goal_used else '❌ 未使用'}")
    print(f"  - subagent 工具: {'✅ 使用' if subagent_used else '❌ 未使用'}")
    print(f"    - evaluate 模式: {'✅ 使用' if evaluate_used else '❌ 未使用'} ({evaluation_count} 次)")
    print()

    print("工具调用统计:")
    for tool_name, count in sorted(tool_calls_count.items(), key=lambda x: x[1], reverse=True):
        print(f"  - {tool_name}: {count} 次")
    print()

    # 信号机制测试结果
    print("=" * 80)
    print("信号机制测试结果")
    print("=" * 80)
    print()

    print(f"信号统计:")
    print(f"  - 发送信号数: {len(signals_emitted)}")
    print(f"  - 接收信号数: {len(signals_received)}")
    print(f"  - 信号类型: {', '.join(sorted(signal_types))}")
    print()

    if signals_emitted:
        print("发送的信号:")
        for i, sig in enumerate(signals_emitted, 1):
            print(f"  {i}. {sig['type']} (trace: {sig['trace_id'][:8]}...)")
        print()

    if signals_received:
        print("接收的信号:")
        for i, sig in enumerate(signals_received, 1):
            print(f"  {i}. {sig['type']} (trace: {sig['trace_id'][:8]}...)")
        print()

    # 检查输出文件
    print("输出文件:")
    validator_file = output_dir / "validator.py"
    report_file = output_dir / "REPORT.md"

    if validator_file.exists():
        size = validator_file.stat().st_size
        print(f"  ✅ validator.py ({size} bytes)")
    else:
        print(f"  ❌ validator.py (未生成)")

    if report_file.exists():
        size = report_file.stat().st_size
        print(f"  ✅ REPORT.md ({size} bytes)")
    else:
        print(f"  ❌ REPORT.md (未生成)")
    print()

    # 验证测试目标
    print("=" * 80)
    print("测试目标验证")
    print("=" * 80)
    print()

    success = True

    # 1. 验证 SignalBus 创建
    if hasattr(runner, 'signal_bus'):
        print(f"  ✅ SignalBus 已创建")
    else:
        print(f"  ❌ SignalBus 未创建")
        success = False

    # 2. 验证信号发送
    if len(signals_emitted) > 0:
        print(f"  ✅ 信号已发送 ({len(signals_emitted)} 个)")
    else:
        print(f"  ❌ 未发送信号")
        success = False

    # 3. 验证信号接收
    if len(signals_received) > 0:
        print(f"  ✅ 信号已接收 ({len(signals_received)} 个)")
    else:
        print(f"  ❌ 未接收信号")
        success = False

    # 4. 验证信号类型
    expected_types = {"subagent.start", "subagent.complete"}
    if expected_types.issubset(signal_types):
        print(f"  ✅ 包含预期的信号类型")
    else:
        missing = expected_types - signal_types
        print(f"  ⚠️  缺少信号类型: {', '.join(missing)}")

    # 5. 验证 subagent 使用
    if evaluate_used:
        print(f"  ✅ 使用了 subagent(evaluate) ({evaluation_count} 次)")
    else:
        print(f"  ❌ 未使用 subagent(evaluate)")
        success = False

    # 6. 验证评估结果
    if evaluation_results:
        print(f"  ✅ 获得了评估结果 ({len(evaluation_results)} 次)")
    else:
        print(f"  ❌ 未获得评估结果")

    # 7. 验证文件生成
    if validator_file.exists():
        print(f"  ✅ 生成了代码文件")
    else:
        print(f"  ❌ 未生成代码文件")
        success = False

    print()

    if success:
        print("🎉 测试成功！信号驱动机制工作正常。")
    else:
        print("⚠️  测试未完全通过，请检查实现。")

    print()
    if current_trace_id:
        print(f"详细日志: .trace/{current_trace_id}/")
    print("=" * 80)


if __name__ == "__main__":
    asyncio.run(main())