agent-lab/tests/test_event_agent.py

import json

import pytest

from agent_lab.application.event_agent import EventAgent
from agent_lab.application.tools import ToolDefinition, ToolRegistry
from agent_lab.domain.events import ToolCallEvent
from agent_lab.domain.messages import ChatMessage


@pytest.mark.asyncio
async def test_event_agent_resolves_tool_arguments_from_history():
    agent = EventAgent(enabled_tools=["handoff_note"])
    event = ToolCallEvent(
        id="call_1",
        name="handoff_note",
        arguments={"message": "chat agent argument should be ignored"},
        raw_arguments='{"message":"chat agent argument should be ignored"}',
    )
    history = [
        ChatMessage(role="user", content="debug this event flow"),
        ChatMessage(role="assistant", content="I need the event agent."),
    ]

    reply = await agent.handle(event, history=history)

    assert reply.role == "tool"
    assert reply.tool_call_id == "call_1"
    assert reply.name == "handoff_note"
    assert json.loads(reply.content) == {
        "tool": "handoff_note",
        "message": "I need the event agent.",
    }


@pytest.mark.asyncio
async def test_event_agent_returns_registry_errors_for_disabled_and_unknown_tools():
    registry = ToolRegistry(
        [
            ToolDefinition(
                name="handoff_note",
                description="Send a note to the event agent.",
                parameters={"type": "object"},
                handler=lambda event: {"tool": event.name, "message": "handled"},
            )
        ]
    )

    disabled_reply = await EventAgent(
        enabled_tools=[],
        registry=registry,
    ).handle(
        ToolCallEvent(
            id="call_1",
            name="handoff_note",
            arguments={"message": "inspect this event"},
            raw_arguments='{"message":"inspect this event"}',
        )
    )
    unknown_reply = await EventAgent(
        enabled_tools=["missing_tool"],
        registry=registry,
    ).handle(
        ToolCallEvent(
            id="call_2",
            name="missing_tool",
            arguments={},
            raw_arguments="{}",
        )
    )

    assert json.loads(disabled_reply.content) == {
        "tool": "handoff_note",
        "error": "tool disabled",
    }
    assert json.loads(unknown_reply.content) == {
        "tool": "missing_tool",
        "error": "unknown tool",
    }


@pytest.mark.asyncio
async def test_event_agent_returns_structured_error_when_tool_handler_raises():
    def fail_tool(event: ToolCallEvent) -> dict:
        raise RuntimeError("boom")

    registry = ToolRegistry(
        [
            ToolDefinition(
                name="handoff_note",
                description="Send a note to the event agent.",
                parameters={"type": "object"},
                handler=fail_tool,
            )
        ]
    )
    event = ToolCallEvent(
        id="call_1",
        name="handoff_note",
        arguments={"message": "inspect this event"},
        raw_arguments='{"message":"inspect this event"}',
    )

    reply = await EventAgent(
        enabled_tools=["handoff_note"],
        registry=registry,
    ).handle(event)

    assert reply.role == "tool"
    assert reply.tool_call_id == "call_1"
    assert json.loads(reply.content) == {
        "tool": "handoff_note",
        "error": "tool handler failed: boom",
    }