tests/test_agent_loop.py

"""Drive the agent loop against the fakes and print every event.

    python tests/test_agent_loop.py

Exits non-zero if the loop does not reach a final answer, so it doubles as a
smoke test before a deploy.
"""

from __future__ import annotations

import asyncio
import os
import sys
import time
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

from tests.fake_servers import LLM_PORT, MCP_PORT, start_background  # noqa: E402

os.environ.update(
    {
        "LLM_BASE_URL": f"http://127.0.0.1:{LLM_PORT}/v1",
        "LLM_MODEL": "fake-model",
        "LLM_API_KEY": "test-key",
        "VFRPLAN_MCP_URL": f"http://127.0.0.1:{MCP_PORT}/mcp",
        "VFRPLAN_MCP_TOKEN": "",
        "APP_ADMIN_TOKEN": "admin-secret",
    }
)

from app.agent import run_agent  # noqa: E402
from app.config import AppConfig  # noqa: E402


async def main() -> int:
    start_background()
    time.sleep(2.0)  # let both servers bind

    config = AppConfig()
    print(f"LLM  : {config.llm.chat_url}  model={config.llm.model}")
    print(f"MCP  : {config.mcp.url}\n")

    saw_final = False
    saw_tool_call = False
    saw_tool_result = False
    saw_reasoning = False

    async for event in run_agent(config, "What is the current METAR at LEBL?"):
        kind = event["type"]
        if kind == "status":
            print(f"  · {event['message']}")
        elif kind == "tools_loaded":
            print(f"  ✓ MCP tools ({event['count']}): {', '.join(event['names'])}")
        elif kind == "reasoning":
            saw_reasoning = True
            print(f"  🧠 reasoning[{event['step']}]: {event['content']}")
        elif kind == "thinking":
            print(f"  💭 thinking[{event['step']}]: {event['content']}")
        elif kind == "tool_call":
            saw_tool_call = True
            print(f"  🔧 call[{event['step']}]: {event['name']}({event['arguments']})")
        elif kind == "tool_result":
            saw_tool_result = True
            print(f"  📄 result: ok={event['ok']} {event['content'][:120]}")
        elif kind == "final":
            saw_final = True
            print(f"\n  ✅ FINAL:\n{event['content']}\n")
        elif kind == "error":
            print(f"\n  ❌ ERROR: {event['message']}\n")

    checks = {
        "reasoning surfaced": saw_reasoning,
        "tool called": saw_tool_call,
        "tool result fed back": saw_tool_result,
        "final answer produced": saw_final,
    }
    print("-" * 60)
    for label, passed in checks.items():
        print(f"  [{'PASS' if passed else 'FAIL'}] {label}")

    return 0 if all(checks.values()) else 1


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