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,
        tool_choice: dict | None = None,
    ) -> AsyncIterator[StreamItem]:
        self.calls.append(
            {
                "messages": list(messages),
                "tools": list(tools),
                "params": params,
                "tool_choice": tool_choice,
            }
        )
        tool_name = tools[0]["function"]["name"]
        yield StreamItem.raw_response_chunk(
            {
                "choices": [
                    {
                        "delta": {
                            "tool_calls": [
                                {
                                    "index": 0,
                                    "function": {"name": tool_name},
                                }
                            ]
                        },
                        "finish_reason": None,
                    }
                ]
            }
        )
        yield StreamItem.provider_tool_call(
            ToolCallEvent(
                id="llm_call_1",
                name=tool_name,
                arguments=self.arguments,
                raw_arguments=json.dumps(self.arguments),
            )
        )


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

    async def stream_chat(
        self,
        messages: list[ChatMessage],
        tools: list[dict],
        params: AgentParams,
        tool_choice: dict | None = None,
    ) -> AsyncIterator[StreamItem]:
        self.calls.append(
            {
                "messages": list(messages),
                "tools": list(tools),
                "params": params,
                "tool_choice": tool_choice,
            }
        )
        yield StreamItem.message_delta("I should have called the tool.")


class WrongSourceToolCallChatClient:
    def __init__(self, item: StreamItem) -> None:
        self.item = item

    async def stream_chat(
        self,
        messages: list[ChatMessage],
        tools: list[dict],
        params: AgentParams,
        tool_choice: dict | None = None,
    ) -> AsyncIterator[StreamItem]:
        yield self.item


@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]["tool_choice"] == {
        "type": "function",
        "function": {"name": "handoff_note"},
    }
    assert chat_client.calls[0]["params"].model == "event-model"
    assert agent.raw_model_chunks([event]) == [
        {
            "event_id": "call_1",
            "event_name": "handoff_note",
            "chunks": [
                {
                    "choices": [
                        {
                            "delta": {
                                "tool_calls": [
                                    {
                                        "index": 0,
                                        "function": {"name": "handoff_note"},
                                    }
                                ]
                            },
                            "finish_reason": None,
                        }
                    ]
                }
            ],
        }
    ]


@pytest.mark.asyncio
async def test_event_agent_falls_back_to_context_arguments_when_llm_returns_no_tool_call():
    chat_client = NoToolCallChatClient()
    agent = EventAgent(
        enabled_tools=["mock_search"],
        chat_client=chat_client,
        params=AgentParams(model="event-model", temperature=0, max_tokens=80),
    )
    event = ToolCallEvent(
        id="call_1",
        name="mock_search",
        arguments={},
        raw_arguments="{}",
    )
    history = [
        ChatMessage(role="user", content="Find latency docs"),
        ChatMessage(role="assistant", content="Need a search for latency docs"),
    ]

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

    payload = json.loads(reply.content)
    assert payload["tool"] == "mock_search"
    assert payload["query"] == "Need a search for latency docs"
    assert "event agent did not return arguments" not in reply.content
    assert chat_client.calls[0]["tools"][0]["function"]["name"] == "mock_search"


@pytest.mark.asyncio
@pytest.mark.parametrize(
    "item",
    [
        StreamItem.text_event(
            ToolCallEvent(
                id="text_event_1",
                name="mock_search",
                arguments={"query": "wrong source"},
                raw_arguments='{"query":"wrong source"}',
            )
        ),
        StreamItem.event(
            ToolCallEvent(
                id="legacy_event_1",
                name="mock_search",
                arguments={"query": "legacy arguments"},
                raw_arguments='{"query":"legacy arguments"}',
            )
        ),
    ],
    ids=["text_event", "legacy_event"],
)
async def test_event_agent_ignores_non_provider_tool_call_sources(item: StreamItem):
    event = ToolCallEvent(
        id="call_1",
        name="mock_search",
        arguments={},
        raw_arguments="{}",
    )
    agent = EventAgent(
        enabled_tools=["mock_search"],
        chat_client=WrongSourceToolCallChatClient(item),
    )

    reply = await agent.handle(
        event,
        history=[ChatMessage(role="assistant", content="fallback query")],
    )

    payload = json.loads(reply.content)
    assert payload["tool"] == "mock_search"
    assert payload["query"] == "fallback query"


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


@pytest.mark.asyncio
async def test_event_agent_projects_complete_tool_round_to_visible_history():
    chat_client = ToolCallingChatClient({"message": "projected history"})
    event = ToolCallEvent(
        id="call_current",
        name="handoff_note",
        arguments={},
        raw_arguments="{}",
    )
    prior_call = ToolCallEvent(
        id="call_prior",
        name="mock_search",
        arguments={"query": "latency"},
        raw_arguments='{"query":"latency"}',
    )

    await EventAgent(
        enabled_tools=["handoff_note"],
        chat_client=chat_client,
    ).handle(
        event,
        history=[
            ChatMessage(role="user", content="Find latency docs"),
            ChatMessage(
                role="assistant",
                content="I checked the latency sources.",
                tool_calls=[prior_call],
            ),
            ChatMessage(
                role="tool",
                content='{"results":["doc"]}',
                tool_call_id="call_prior",
            ),
            ChatMessage(role="user", content="Prepare a handoff"),
        ],
    )

    projected_messages = chat_client.calls[0]["messages"]
    visible_assistant = next(
        message
        for message in projected_messages
        if message.content == "I checked the latency sources."
    )
    assert visible_assistant.tool_calls == []
    assert not any(message.role == "tool" for message in projected_messages)