agent-lab/tests/test_event_agent.py

import json
from collections.abc import AsyncIterator

import pytest

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


class ToolCallingChatClient:
    def __init__(self, arguments: dict) -> None:
        self.arguments = arguments
        self.calls: list[dict] = []

    async def stream_chat(
        self,
        messages: list[ChatMessage],
        tools: list[dict],
        params: AgentParams,
    ) -> AsyncIterator[StreamItem]:
        self.calls.append(
            {
                "messages": list(messages),
                "tools": list(tools),
                "params": params,
            }
        )
        tool_name = tools[0]["function"]["name"]
        yield StreamItem.event(
            ToolCallEvent(
                id="llm_call_1",
                name=tool_name,
                arguments=self.arguments,
                raw_arguments=json.dumps(self.arguments),
            )
        )


@pytest.mark.asyncio
async def test_event_agent_resolves_tool_arguments_with_llm_tool_call():
    chat_client = ToolCallingChatClient({"message": "LLM generated handoff"})
    agent = EventAgent(
        enabled_tools=["handoff_note"],
        chat_client=chat_client,
        params=AgentParams(model="event-model", temperature=0, max_tokens=80),
    )
    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": "LLM generated handoff",
    }
    assert chat_client.calls[0]["tools"] == [
        {
            "type": "function",
            "function": {
                "name": "handoff_note",
                "description": "Send a note to the event agent.",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "message": {"type": "string"},
                    },
                    "required": ["message"],
                },
            },
        }
    ]
    assert chat_client.calls[0]["params"].model == "event-model"


@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",
    }


@pytest.mark.asyncio
async def test_event_agent_llm_receives_history_and_agent_config_context():
    chat_client = ToolCallingChatClient(
        {"message": "Use strict tool parameters.", "thinking": "disabled"}
    )
    registry = ToolRegistry(
        [
            ToolDefinition(
                name="handoff_note",
                description="Send a note to the event agent.",
                parameters={"type": "object"},
                handler=lambda event: {
                    "tool": event.name,
                    "message": event.arguments["message"],
                    "thinking": event.arguments["thinking"],
                },
            )
        ]
    )
    event = ToolCallEvent(
        id="call_1",
        name="handoff_note",
        arguments={},
        raw_arguments="{}",
    )

    reply = await EventAgent(
        enabled_tools=["handoff_note"],
        registry=registry,
        chat_client=chat_client,
        params=AgentParams(model="event-model", temperature=0.4, max_tokens=120),
    ).handle(
        event,
        history=[ChatMessage(role="user", content="debug this")],
        system_prompt="Use strict tool parameters.",
        extra_body={"thinking": {"type": "disabled"}},
    )

    messages = chat_client.calls[0]["messages"]
    assert any(message.content == "debug this" for message in messages)
    assert any(message.content == "Use strict tool parameters." for message in messages)
    assert chat_client.calls[0]["params"].extra_body == {"thinking": {"type": "disabled"}}
    assert json.loads(reply.content) == {
        "tool": "handoff_note",
        "message": "Use strict tool parameters.",
        "thinking": "disabled",
    }