agent-lab/tests/test_debug_runtime.py

import asyncio
import importlib
import json
import logging
from collections.abc import AsyncIterator
from typing import Any

import pytest

from agent_lab.application.contracts import AgentParams, DebugRunRequest, EventAgentParams
from agent_lab.application.event_agent import EventAgentRequest
from agent_lab.application.events import EventBatchExecutor, ResultPolicy
from agent_lab.application.runtime import DebugRuntime
from agent_lab.application.tools import ToolDefinition, ToolExecutionContext, ToolRegistry
from agent_lab.domain.events import ToolCallEvent
from agent_lab.domain.messages import ChatMessage, StreamItem, TokenUsage
from agent_lab.infrastructure.chat_client import OpenAICompatibleChatClient
from agent_lab.infrastructure.sqlite_store import SQLiteSessionStore


def _runtime_queues_class():
    module = importlib.import_module("agent_lab.application.queues")
    return module.RuntimeQueues


async def _collect_outputs(stream: AsyncIterator[dict[str, Any]]) -> list[dict[str, Any]]:
    return [message async for message in stream]


def _without_audit(outputs: list[dict[str, Any]]) -> list[dict[str, Any]]:
    return [message for message in outputs if message["type"] != "audit"]


def _message_types(outputs: list[dict[str, Any]]) -> list[str]:
    return [message["type"] for message in _without_audit(outputs)]


def _event_tool_call_from_tools(
    tools: list[dict],
    messages: list[ChatMessage],
) -> StreamItem:
    tool_name = tools[0]["function"]["name"]
    content = ""
    for role in ("assistant", "user"):
        content = next(
            (
                message.content
                for message in reversed(messages)
                if message.role == role and message.content.strip()
            ),
            "",
        )
        if content:
            break
    arguments = {
        "message": content,
        "query": content,
        "title": content,
    }
    return StreamItem.provider_tool_call(
        ToolCallEvent(
            id="event_agent_call_1",
            name=tool_name,
            arguments=arguments,
            raw_arguments=json.dumps(arguments),
        )
    )


async def _next_non_audit(
    stream: AsyncIterator[dict[str, Any]],
) -> dict[str, Any]:
    while True:
        message = await anext(stream)
        if message["type"] != "audit":
            return message


async def _next_non_audit_from_queue(queues: Any) -> dict[str, Any]:
    while True:
        message = await queues.output.get()
        if message["type"] != "audit":
            return message


class RecordingQueue(asyncio.Queue):
    def __init__(self, name: str, log: list[tuple[str, str, str]]) -> None:
        super().__init__()
        self.name = name
        self.log = log

    async def put(self, item: Any) -> None:
        self.log.append((self.name, "put", self._describe(item)))
        await super().put(item)

    async def get(self) -> Any:
        item = await super().get()
        self.log.append((self.name, "get", self._describe(item)))
        return item

    def _describe(self, item: Any) -> str:
        if isinstance(item, ChatMessage):
            if item.role == "tool":
                return f"tool:{item.tool_call_id}"
            return item.role
        if isinstance(item, ToolCallEvent):
            return f"event:{item.name}:{item.id}"
        if isinstance(item, EventAgentRequest):
            events = ",".join(f"{event.name}:{event.id}" for event in item.events)
            return f"event_request:{events}"
        if isinstance(item, dict):
            return f"output:{item.get('type')}"
        return type(item).__name__


class FakeChatClient:
    def __init__(self) -> None:
        self.calls = 0

    async def stream_chat(
        self,
        messages: list[ChatMessage],
        tools: list[dict],
        params: AgentParams,
        tool_choice: dict[str, Any] | None = None,
    ) -> AsyncIterator[StreamItem]:
        if tools:
            yield StreamItem.raw_response_chunk(
                {
                    "choices": [
                        {
                            "delta": {
                                "tool_calls": [
                                    {
                                        "index": 0,
                                        "function": {"name": tools[0]["function"]["name"]},
                                    }
                                ]
                            },
                            "finish_reason": None,
                        }
                    ]
                }
            )
            yield _event_tool_call_from_tools(tools, messages)
            return
        self.calls += 1
        if self.calls == 1:
            yield StreamItem.raw_response_chunk(
                {
                    "choices": [
                        {
                            "delta": {"content": "<agent_events>handoff_note</agent_events>"},
                            "finish_reason": None,
                        }
                    ]
                }
            )
            yield StreamItem.text_event(
                ToolCallEvent(
                    id="call_1",
                    name="handoff_note",
                    arguments={},
                    raw_arguments="{}",
                )
            )
            return

        assert not any(message.role == "tool" for message in messages)
        assert any(message.name == "event_agent" for message in messages)
        yield StreamItem.message_delta("final answer")


class WrongSourceEventChatClient:
    def __init__(self, item: StreamItem) -> None:
        self.item = item
        self.calls = 0

    async def stream_chat(
        self,
        messages: list[ChatMessage],
        tools: list[dict],
        params: AgentParams,
        tool_choice: dict[str, Any] | None = None,
    ) -> AsyncIterator[StreamItem]:
        if tools:
            yield _event_tool_call_from_tools(tools, messages)
            return
        self.calls += 1
        if self.calls == 1:
            yield self.item
            return
        yield StreamItem.message_delta("unexpected continuation")


class IncrementingClock:
    def __init__(self, current: float = 100.0, step: float = 0.01) -> None:
        self.current = current
        self.step = step

    def __call__(self) -> float:
        self.current += self.step
        return self.current


class StrictHistoryChatClient:
    def __init__(self) -> None:
        self.calls = 0
        self.second_call_messages: list[ChatMessage] = []

    async def stream_chat(
        self,
        messages: list[ChatMessage],
        tools: list[dict],
        params: AgentParams,
        tool_choice: dict[str, Any] | None = None,
    ) -> AsyncIterator[StreamItem]:
        if tools:
            yield StreamItem.raw_response_chunk(
                {
                    "choices": [
                        {
                            "delta": {
                                "tool_calls": [
                                    {
                                        "index": 0,
                                        "function": {"name": tools[0]["function"]["name"]},
                                    }
                                ]
                            },
                            "finish_reason": None,
                        }
                    ]
                }
            )
            yield _event_tool_call_from_tools(tools, messages)
            return
        self.calls += 1
        if self.calls == 1:
            yield StreamItem.raw_response_chunk(
                {
                    "choices": [
                        {
                            "delta": {"content": "<agent_events>handoff_note</agent_events>"},
                            "finish_reason": None,
                        }
                    ]
                }
            )
            yield StreamItem.text_event(
                ToolCallEvent(
                    id="call_1",
                    name="handoff_note",
                    arguments={},
                    raw_arguments="{}",
                )
            )
            return

        self.second_call_messages = list(messages)
        yield StreamItem.message_delta("final answer")


class MultiEventChatClient:
    def __init__(self) -> None:
        self.calls = 0
        self.second_call_messages: list[ChatMessage] = []

    async def stream_chat(
        self,
        messages: list[ChatMessage],
        tools: list[dict],
        params: AgentParams,
        tool_choice: dict[str, Any] | None = None,
    ) -> AsyncIterator[StreamItem]:
        if tools:
            yield _event_tool_call_from_tools(tools, messages)
            return
        self.calls += 1
        if self.calls == 1:
            yield StreamItem.message_delta("Checking events.")
            yield StreamItem.text_event(
                ToolCallEvent(
                    id="call_1",
                    name="handoff_note",
                    arguments={"message": "ignored chat argument"},
                    raw_arguments='{"message":"ignored chat argument"}',
                )
            )
            yield StreamItem.text_event(
                ToolCallEvent(
                    id="call_2",
                    name="audit_note",
                    arguments={"message": "ignored chat argument"},
                    raw_arguments='{"message":"ignored chat argument"}',
                )
            )
            return

        self.second_call_messages = list(messages)
        yield StreamItem.message_delta("Final answer.")


class ToolCapturingChatClient:
    def __init__(self) -> None:
        self.tools: list[dict[str, Any]] = []
        self.messages: list[ChatMessage] = []

    async def stream_chat(
        self,
        messages: list[ChatMessage],
        tools: list[dict],
        params: AgentParams,
        tool_choice: dict[str, Any] | None = None,
    ) -> AsyncIterator[StreamItem]:
        self.messages = list(messages)
        self.tools = list(tools)
        yield StreamItem.message_delta("final answer")


class EventLoopLimitChatClient:
    def __init__(self) -> None:
        self.calls = 0
        self.tools_by_call: list[list[dict[str, Any]]] = []
        self.messages_by_call: list[list[ChatMessage]] = []

    async def stream_chat(
        self,
        messages: list[ChatMessage],
        tools: list[dict],
        params: AgentParams,
        tool_choice: dict[str, Any] | None = None,
    ) -> AsyncIterator[StreamItem]:
        if tools:
            yield _event_tool_call_from_tools(tools, messages)
            return
        self.calls += 1
        self.tools_by_call.append(list(tools))
        self.messages_by_call.append(list(messages))
        if self.calls == 1:
            yield StreamItem.text_event(
                ToolCallEvent(
                    id="call_1",
                    name="handoff_note",
                    arguments={},
                    raw_arguments="{}",
                )
            )
            return

        yield StreamItem.message_delta("final after event limit")


class RoundStatsChatClient:
    async def stream_chat(
        self,
        messages: list[ChatMessage],
        tools: list[dict],
        params: AgentParams,
        tool_choice: dict[str, Any] | None = None,
    ) -> AsyncIterator[StreamItem]:
        yield StreamItem.message_delta("hello")
        yield StreamItem.usage_item(
            TokenUsage(
                prompt_tokens=10,
                completion_tokens=20,
                total_tokens=30,
                cached_tokens=5,
            )
        )


class EventRoundStatsChatClient:
    def __init__(self) -> None:
        self.calls = 0

    async def stream_chat(
        self,
        messages: list[ChatMessage],
        tools: list[dict],
        params: AgentParams,
        tool_choice: dict[str, Any] | None = None,
    ) -> AsyncIterator[StreamItem]:
        if tools:
            yield StreamItem.raw_response_chunk(
                {
                    "choices": [
                        {
                            "delta": {
                                "tool_calls": [
                                    {
                                        "index": 0,
                                        "function": {
                                            "name": tools[0]["function"]["name"],
                                        },
                                    }
                                ]
                            },
                            "finish_reason": None,
                        }
                    ]
                }
            )
            yield _event_tool_call_from_tools(tools, messages)
            return
        self.calls += 1
        if self.calls == 1:
            yield StreamItem.raw_response_chunk(
                {
                    "choices": [
                        {
                            "delta": {
                                "content": "<agent_events>handoff_note</agent_events>",
                            },
                            "finish_reason": None,
                        }
                    ]
                }
            )
            yield StreamItem.text_event(
                ToolCallEvent(
                    id="call_1",
                    name="handoff_note",
                    arguments={},
                    raw_arguments="{}",
                )
            )
            yield StreamItem.usage_item(
                TokenUsage(prompt_tokens=3, completion_tokens=0, total_tokens=3)
            )
            return

        yield StreamItem.raw_response_chunk(
            {
                "choices": [
                    {
                        "delta": {"content": "final answer"},
                        "finish_reason": None,
                    }
                ]
            }
        )
        yield StreamItem.message_delta("final answer")
        yield StreamItem.usage_item(
            TokenUsage(prompt_tokens=4, completion_tokens=6, total_tokens=10)
        )


class ContextBoundaryChatClient:
    def __init__(self) -> None:
        self.calls = 0
        self.event_agent_messages: list[list[ChatMessage]] = []

    async def stream_chat(
        self,
        messages: list[ChatMessage],
        tools: list[dict],
        params: AgentParams,
        tool_choice: dict[str, Any] | None = None,
    ) -> AsyncIterator[StreamItem]:
        if tools:
            self.event_agent_messages.append(list(messages))
            yield _event_tool_call_from_tools(tools, messages)
            return
        self.calls += 1
        if self.calls == 1:
            yield StreamItem.message_delta("I will check.")
            yield StreamItem.text_event(
                ToolCallEvent(
                    id="call_1",
                    name="handoff_note",
                    arguments={},
                    raw_arguments="{}",
                )
            )
            return
        yield StreamItem.message_delta("final answer")


class TwoTurnSessionChatClient:
    def __init__(self) -> None:
        self.calls = 0
        self.messages_by_call: list[list[ChatMessage]] = []

    async def stream_chat(
        self,
        messages: list[ChatMessage],
        tools: list[dict],
        params: AgentParams,
        tool_choice: dict[str, Any] | None = None,
    ) -> AsyncIterator[StreamItem]:
        if tools:
            yield _event_tool_call_from_tools(tools, messages)
            return
        self.calls += 1
        self.messages_by_call.append(list(messages))
        if self.calls in {1, 3}:
            yield StreamItem.text_event(
                ToolCallEvent(
                    id=f"call_{self.calls}",
                    name="handoff_note",
                    arguments={},
                    raw_arguments="{}",
                )
            )
            return

        yield StreamItem.message_delta(f"final answer {self.calls}")


class SlowAfterEventChatClient:
    def __init__(self) -> None:
        self.calls = 0
        self.event_seen = asyncio.Event()
        self.release_stream = asyncio.Event()

    async def stream_chat(
        self,
        messages: list[ChatMessage],
        tools: list[dict],
        params: AgentParams,
        tool_choice: dict[str, Any] | None = None,
    ) -> AsyncIterator[StreamItem]:
        if tools:
            yield _event_tool_call_from_tools(tools, messages)
            return
        self.calls += 1
        if self.calls == 1:
            yield StreamItem.message_delta("Need event.")
            yield StreamItem.text_event(
                ToolCallEvent(
                    id="call_1",
                    name="handoff_note",
                    arguments={},
                    raw_arguments="{}",
                )
            )
            self.event_seen.set()
            await self.release_stream.wait()
            return

        yield StreamItem.message_delta("final answer")


def test_runtime_queues_exposes_input_output_and_events_queues():
    RuntimeQueues = _runtime_queues_class()

    queues = RuntimeQueues()

    assert isinstance(queues.input, asyncio.Queue)
    assert isinstance(queues.output, asyncio.Queue)
    assert isinstance(queues.events, asyncio.Queue)
    assert queues.input is not queues.output
    assert queues.input is not queues.events
    assert queues.output is not queues.events


@pytest.mark.asyncio
async def test_runtime_routes_chat_events_through_event_agent_then_continues_chat():
    request = DebugRunRequest(
        user_message="debug this",
        system_prompts=["You are a debugger."],
        pre_messages=[],
        chat_agent=AgentParams(model="fake-model", temperature=0.1, max_tokens=200),
        event_agent=EventAgentParams(enabled_tools=["handoff_note"], max_event_loops=3),
    )
    client = FakeChatClient()
    runtime = DebugRuntime(client)

    outputs = [message async for message in runtime.run(request)]
    business_outputs = _without_audit(outputs)

    assert client.calls == 2
    assert _message_types(outputs) == [
        "session_started",
        "event",
        "tool_result",
        "round_stats",
        "message_delta",
        "round_stats",
        "done",
    ]
    assert business_outputs[1]["event"]["name"] == "handoff_note"
    assert business_outputs[4]["content"] == "final answer"


@pytest.mark.asyncio
@pytest.mark.parametrize(
    "item",
    [
        StreamItem.provider_tool_call(
            ToolCallEvent(
                id="provider_call_1",
                name="handoff_note",
                arguments={"message": "wrong source"},
                raw_arguments='{"message":"wrong source"}',
            )
        ),
        StreamItem.event(
            ToolCallEvent(
                id="legacy_event_1",
                name="handoff_note",
                arguments={},
                raw_arguments="{}",
            )
        ),
    ],
    ids=["provider_tool_call", "legacy_event"],
)
async def test_runtime_ignores_non_text_event_sources(item: StreamItem):
    request = DebugRunRequest(
        user_message="debug this",
        system_prompts=[],
        pre_messages=[],
        chat_agent=AgentParams(model="fake-model"),
        event_agent=EventAgentParams(enabled_tools=["handoff_note"], max_event_loops=2),
    )
    client = WrongSourceEventChatClient(item)

    outputs = [message async for message in DebugRuntime(client).run(request)]

    assert client.calls == 1
    assert "event" not in _message_types(outputs)
    assert "tool_result" not in _message_types(outputs)


@pytest.mark.asyncio
async def test_runtime_emits_audit_events_and_backend_logs(caplog):
    request = DebugRunRequest(
        user_message="debug this",
        system_prompts=[],
        pre_messages=[],
        chat_agent=AgentParams(model="fake-model", temperature=0.1, max_tokens=200),
        event_agent=EventAgentParams(enabled_tools=["handoff_note"], max_event_loops=1),
    )
    runtime = DebugRuntime(FakeChatClient())

    with caplog.at_level(logging.INFO, logger="agent_lab.application.runtime"):
        outputs = [message async for message in runtime.run(request)]

    audit_events = [
        message["event"]
        for message in outputs
        if message["type"] == "audit"
    ]
    assert audit_events == [
        "session_started",
        "chat_round_started",
        "chat_agent_request",
        "chat_event_detected",
        "chat_agent_response",
        "event_batch_started",
        "event_agent_request",
        "event_agent_response",
        "event_agent_completed",
        "event_batch_results",
        "event_policy_decision",
        "chat_round_finished",
        "chat_round_started",
        "chat_agent_request",
        "chat_message_stream_started",
        "chat_message_stream_finished",
        "chat_agent_response",
        "chat_round_finished",
        "session_finished",
    ]
    assert "chat_event_detected" in caplog.text
    assert "event_agent_completed" in caplog.text


@pytest.mark.asyncio
async def test_runtime_audits_chat_message_stream_boundaries_in_output_order():
    request = DebugRunRequest(
        user_message="debug this",
        system_prompts=[],
        pre_messages=[],
        chat_agent=AgentParams(model="fake-model", temperature=0.1, max_tokens=200),
        event_agent=EventAgentParams(enabled_tools=[], max_event_loops=1),
    )
    runtime = DebugRuntime(RoundStatsChatClient())

    outputs = [message async for message in runtime.run(request)]
    ordered_labels = [
        message["event"] if message["type"] == "audit" else message["type"]
        for message in outputs
    ]

    assert ordered_labels.index("chat_message_stream_started") < ordered_labels.index(
        "message_delta"
    )
    assert ordered_labels.index("message_delta") < ordered_labels.index(
        "chat_message_stream_finished"
    )
    assert ordered_labels.index("chat_message_stream_finished") < ordered_labels.index(
        "chat_agent_response"
    )

    stream_started = next(
        message
        for message in outputs
        if message.get("event") == "chat_message_stream_started"
    )
    stream_finished = next(
        message
        for message in outputs
        if message.get("event") == "chat_message_stream_finished"
    )
    assert stream_started["details"]["agent"] == "chat_agent"
    assert stream_started["details"]["round_index"] == 1
    assert stream_finished["details"]["delta_count"] == 1
    assert stream_finished["details"]["content_length"] == len("hello")


@pytest.mark.asyncio
async def test_runtime_audit_events_include_turn_relative_elapsed_time():
    request = DebugRunRequest(
        user_message="debug this",
        system_prompts=[],
        pre_messages=[],
        chat_agent=AgentParams(model="fake-model", temperature=0.1, max_tokens=200),
        event_agent=EventAgentParams(enabled_tools=["handoff_note"], max_event_loops=1),
    )
    runtime = DebugRuntime(FakeChatClient(), clock=IncrementingClock())

    queues = runtime.start_session(request)
    outputs: list[dict[str, Any]] = []
    while not any(message["type"] == "turn_completed" for message in outputs):
        outputs.append(await asyncio.wait_for(queues.output.get(), timeout=1))

    await runtime.aclose()

    turn_audits = [
        message
        for message in outputs
        if message["type"] == "audit"
    ]
    elapsed_values = [
        message["details"].get("turn_elapsed_ms") for message in turn_audits
    ]

    assert turn_audits
    assert all(isinstance(value, int) for value in elapsed_values)
    assert all(value >= 0 for value in elapsed_values)
    assert elapsed_values == sorted(elapsed_values)
    assert next(
        message
        for message in turn_audits
        if message["event"] == "event_agent_request"
    )["details"]["turn_elapsed_ms"] >= 0


@pytest.mark.asyncio
async def test_runtime_audit_includes_model_params_prompts_results_and_usage():
    request = DebugRunRequest(
        user_message="debug this",
        system_prompts=["You are a debugger."],
        pre_messages=[],
        chat_agent=AgentParams(model="chat-model", temperature=0.1, max_tokens=200),
        event_agent=EventAgentParams(
            model="event-model",
            temperature=0.4,
            max_tokens=80,
            enabled_tools=["handoff_note"],
            max_event_loops=1,
        ),
    )
    runtime = DebugRuntime(EventRoundStatsChatClient())

    outputs = [message async for message in runtime.run(request)]
    audits = [message for message in outputs if message["type"] == "audit"]

    chat_request = next(
        message for message in audits if message["event"] == "chat_agent_request"
    )
    assert chat_request["details"]["agent"] == "chat_agent"
    assert chat_request["details"]["params"]["model"] == "chat-model"
    assert chat_request["details"]["params"]["temperature"] == 0.1
    assert chat_request["details"]["params"]["max_tokens"] == 200
    assert chat_request["details"]["tools"] == []
    assert chat_request["details"]["messages"][0] == {
        "role": "system",
        "content": "You are a debugger.",
        "name": None,
        "tool_call_id": None,
        "tool_calls": [],
    }

    chat_response = next(
        message for message in audits if message["event"] == "chat_agent_response"
    )
    assert chat_response["details"]["event_names"] == ["handoff_note"]
    assert chat_response["details"]["raw_chunks"] == [
        {
            "choices": [
                {
                    "delta": {"content": "<agent_events>handoff_note</agent_events>"},
                    "finish_reason": None,
                }
            ]
        }
    ]
    assert chat_response["details"]["usage"] == {
        "prompt_tokens": 3,
        "completion_tokens": 0,
        "total_tokens": 3,
        "cached_tokens": 0,
    }

    event_request = next(
        message for message in audits if message["event"] == "event_agent_request"
    )
    assert event_request["details"]["agent"] == "event_agent"
    assert event_request["details"]["params"]["model"] == "event-model"
    assert event_request["details"]["params"]["temperature"] == 0.4
    assert event_request["details"]["events"][0]["name"] == "handoff_note"
    assert event_request["details"]["tools"][0]["function"]["name"] == "handoff_note"
    assert "Authorization" not in str(event_request["details"])

    event_response = next(
        message for message in audits if message["event"] == "event_agent_response"
    )
    assert event_response["details"]["replies"][0]["role"] == "tool"
    assert '"tool": "handoff_note"' in event_response["details"]["replies"][0]["content"]
    assert event_response["details"]["raw_model_chunks"][0]["event_name"] == "handoff_note"
    assert event_response["details"]["raw_model_chunks"][0]["chunks"] == []


@pytest.mark.asyncio
async def test_runtime_round_started_separates_available_events_from_round_budget():
    request = DebugRunRequest(
        user_message="debug this",
        system_prompts=[],
        pre_messages=[],
        chat_agent=AgentParams(model="chat-model", temperature=0.1, max_tokens=200),
        event_agent=EventAgentParams(
            enabled_tools=["handoff_note"],
            max_event_loops=1,
        ),
    )
    runtime = DebugRuntime(EventRoundStatsChatClient())

    outputs = [message async for message in runtime.run(request)]
    round_starts = [
        message
        for message in outputs
        if message.get("event") == "chat_round_started"
    ]

    assert round_starts[0]["details"]["events_enabled"] == ["handoff_note"]
    assert round_starts[0]["details"]["configured_events"] == ["handoff_note"]
    assert round_starts[0]["details"]["event_generation_enabled"] is True
    assert round_starts[0]["details"]["event_prompt_events"] == ["handoff_note"]
    assert round_starts[1]["details"]["events_enabled"] == []
    assert round_starts[1]["details"]["configured_events"] == ["handoff_note"]
    assert round_starts[1]["details"]["event_generation_enabled"] is False
    assert round_starts[1]["details"]["event_prompt_events"] == []


@pytest.mark.asyncio
async def test_runtime_event_agent_history_excludes_chat_agent_system_context():
    request = DebugRunRequest(
        user_message="debug this",
        system_prompts=["ChatAgent root prompt."],
        pre_messages=[
            ChatMessage(role="system", content="ChatAgent pre system."),
            ChatMessage(role="user", content="earlier user"),
            ChatMessage(role="assistant", content="earlier assistant"),
        ],
        chat_agent=AgentParams(model="chat-model", temperature=0.1, max_tokens=200),
        event_agent=EventAgentParams(
            model="event-model",
            enabled_tools=["handoff_note"],
            max_event_loops=1,
        ),
    )
    client = ContextBoundaryChatClient()
    runtime = DebugRuntime(client)

    outputs = [message async for message in runtime.run(request)]
    event_request = next(
        message for message in outputs if message.get("event") == "event_agent_request"
    )

    assert [
        (message["role"], message["content"])
        for message in event_request["details"]["history"]
    ] == [
        ("user", "earlier user"),
        ("assistant", "earlier assistant"),
        ("user", "debug this"),
        ("assistant", "I will check."),
    ]
    assert not any(
        message["role"] == "system"
        for message in event_request["details"]["history"]
    )
    assert client.event_agent_messages == []


@pytest.mark.asyncio
async def test_runtime_session_event_agent_history_excludes_internal_replies():
    request = DebugRunRequest(
        user_message="debug this",
        system_prompts=["ChatAgent root prompt."],
        pre_messages=[ChatMessage(role="assistant", content="prior answer")],
        chat_agent=AgentParams(model="chat-model", temperature=0.1, max_tokens=200),
        event_agent=EventAgentParams(
            model="event-model",
            enabled_tools=["handoff_note"],
            max_event_loops=1,
        ),
    )
    runtime = DebugRuntime(ContextBoundaryChatClient())

    queues = runtime.start_session(request)
    outputs: list[dict[str, Any]] = []
    while not any(message["type"] == "turn_completed" for message in outputs):
        outputs.append(await asyncio.wait_for(queues.output.get(), timeout=1))

    await runtime.aclose()

    event_request = next(
        message for message in outputs if message.get("event") == "event_agent_request"
    )
    assert [
        (message["role"], message["content"], message["name"])
        for message in event_request["details"]["history"]
    ] == [
        ("assistant", "prior answer", None),
        ("user", "debug this", None),
        ("assistant", "I will check.", None),
    ]
    assert not any(
        message["name"] == "event_agent"
        for message in event_request["details"]["history"]
    )


@pytest.mark.asyncio
async def test_runtime_outputs_event_as_soon_as_chat_stream_detects_it():
    RuntimeQueues = _runtime_queues_class()
    queues = RuntimeQueues()
    client = SlowAfterEventChatClient()
    runtime = DebugRuntime(client, queues=queues)
    request = DebugRunRequest(
        user_message="debug this",
        system_prompts=[],
        pre_messages=[],
        chat_agent=AgentParams(model="fake-model", temperature=0.1, max_tokens=200),
        event_agent=EventAgentParams(enabled_tools=["handoff_note"], max_event_loops=1),
    )

    runtime.start(request)

    try:
        assert await _next_non_audit_from_queue(queues) == {"type": "session_started"}
        assert await _next_non_audit_from_queue(queues) == {
            "type": "message_delta",
            "content": "Need event.",
        }
        await asyncio.wait_for(client.event_seen.wait(), timeout=1)
        event_message = await asyncio.wait_for(
            _next_non_audit_from_queue(queues),
            timeout=0.2,
        )

        assert event_message["type"] == "event"
        assert event_message["event"]["name"] == "handoff_note"
    finally:
        client.release_stream.set()
        await runtime.aclose()


@pytest.mark.asyncio
async def test_runtime_batches_round_events_before_continuing_chat_agent():
    def resolve_from_history(
        event: ToolCallEvent,
        context: ToolExecutionContext,
    ) -> dict[str, Any]:
        return {"message": context.history[-1].content, "event": event.name}

    registry = ToolRegistry(
        [
            ToolDefinition(
                name="handoff_note",
                description="Send a handoff note.",
                parameters={
                    "type": "object",
                    "properties": {"message": {"type": "string"}},
                    "required": ["message"],
                },
                handler=lambda event: {
                    "tool": event.name,
                    "message": event.arguments["message"],
                },
                argument_resolver=resolve_from_history,
            ),
            ToolDefinition(
                name="audit_note",
                description="Send an audit note.",
                parameters={
                    "type": "object",
                    "properties": {"message": {"type": "string"}},
                    "required": ["message"],
                },
                handler=lambda event: {
                    "tool": event.name,
                    "message": event.arguments["message"],
                },
                argument_resolver=resolve_from_history,
            ),
        ]
    )
    request = DebugRunRequest(
        user_message="debug this",
        system_prompts=[],
        pre_messages=[],
        chat_agent=AgentParams(model="fake-model", temperature=0.1, max_tokens=200),
        event_agent=EventAgentParams(
            enabled_tools=["handoff_note", "audit_note"],
            max_event_loops=2,
        ),
    )
    client = MultiEventChatClient()
    runtime = DebugRuntime(client, registry=registry)

    outputs = [message async for message in runtime.run(request)]

    assert _message_types(outputs) == [
        "session_started",
        "message_delta",
        "event",
        "event",
        "tool_result",
        "tool_result",
        "round_stats",
        "message_delta",
        "round_stats",
        "done",
    ]
    assert client.calls == 2
    assert [message.role for message in client.second_call_messages] == [
        "system",
        "user",
        "assistant",
        "user",
    ]
    assistant_message = client.second_call_messages[2]
    assert assistant_message.content == "Checking events."
    assert not any(message.role == "tool" for message in client.second_call_messages)
    assert client.second_call_messages[-1].content == (
        "EventAgent results:\n"
        '{"tool": "handoff_note", "message": "Checking events."}\n'
        '{"tool": "audit_note", "message": "Checking events."}'
    )
    assert client.second_call_messages[-1].name == "event_agent"


@pytest.mark.asyncio
async def test_runtime_start_returns_queues_for_downstream_output_consumer():
    request = DebugRunRequest(
        user_message="debug this",
        system_prompts=[],
        pre_messages=[],
        chat_agent=AgentParams(model="fake-model", temperature=0.1, max_tokens=200),
        event_agent=EventAgentParams(enabled_tools=[], max_event_loops=1),
    )
    runtime = DebugRuntime(RoundStatsChatClient())

    queues = runtime.start(request)

    outputs: list[dict[str, Any]] = []
    while True:
        message = await asyncio.wait_for(queues.output.get(), timeout=1)
        outputs.append(message)
        if message["type"] == "done":
            break
    assert _message_types(outputs) == [
        "session_started",
        "message_delta",
        "usage",
        "round_stats",
        "done",
    ]


@pytest.mark.asyncio
async def test_runtime_finalizes_chat_after_reaching_event_loop_limit():
    request = DebugRunRequest(
        user_message="debug this",
        system_prompts=[],
        pre_messages=[],
        chat_agent=AgentParams(model="fake-model", temperature=0.1, max_tokens=200),
        event_agent=EventAgentParams(enabled_tools=["handoff_note"], max_event_loops=1),
    )
    client = EventLoopLimitChatClient()
    runtime = DebugRuntime(client)

    outputs = [message async for message in runtime.run(request)]
    business_outputs = _without_audit(outputs)

    assert client.calls == 2
    assert client.tools_by_call[0] == []
    assert client.tools_by_call[1] == []
    assert "Available events:" in client.messages_by_call[0][0].content
    assert not any(
        "Available events:" in message.content
        for message in client.messages_by_call[1]
        if message.role == "system"
    )
    assert _message_types(outputs) == [
        "session_started",
        "event",
        "tool_result",
        "round_stats",
        "message_delta",
        "round_stats",
        "done",
    ]
    assert business_outputs[4]["content"] == "final after event limit"


@pytest.mark.asyncio
async def test_runtime_buffers_upstream_user_input_until_after_matching_tool_reply():
    RuntimeQueues = _runtime_queues_class()
    queues = RuntimeQueues()
    request = DebugRunRequest(
        user_message="debug this",
        system_prompts=[],
        pre_messages=[],
        chat_agent=AgentParams(model="fake-model", temperature=0.1, max_tokens=200),
        event_agent=EventAgentParams(enabled_tools=["handoff_note"], max_event_loops=2),
    )
    client = StrictHistoryChatClient()
    runtime = DebugRuntime(client, queues=queues)
    stream = runtime.run(request)

    assert await _next_non_audit(stream) == {"type": "session_started"}
    event_message = await _next_non_audit(stream)
    assert event_message["type"] == "event"
    await queues.input.put(ChatMessage(role="user", content="follow-up while tool runs"))
    remaining = [message async for message in stream]

    assert remaining[-1] == {"type": "done"}
    assert [message.role for message in client.second_call_messages] == [
        "system",
        "user",
        "assistant",
        "user",
        "user",
    ]
    assert not any(message.role == "tool" for message in client.second_call_messages)
    assert client.second_call_messages[3].content.startswith("EventAgent results:\n")
    assert client.second_call_messages[3].name == "event_agent"
    assert client.second_call_messages[4].content == "follow-up while tool runs"


@pytest.mark.asyncio
async def test_runtime_continues_when_event_agent_tool_handler_raises():
    def fail_tool(event: ToolCallEvent) -> dict[str, Any]:
        raise RuntimeError("boom")

    registry = ToolRegistry(
        [
            ToolDefinition(
                name="handoff_note",
                description="Broken handoff tool.",
                parameters={"type": "object"},
                handler=fail_tool,
            )
        ]
    )
    request = DebugRunRequest(
        user_message="debug this",
        system_prompts=[],
        pre_messages=[],
        chat_agent=AgentParams(model="fake-model", temperature=0.1, max_tokens=200),
        event_agent=EventAgentParams(enabled_tools=["handoff_note"], max_event_loops=2),
    )
    runtime = DebugRuntime(FakeChatClient(), registry=registry)

    outputs = await asyncio.wait_for(
        _collect_outputs(runtime.run(request)),
        timeout=1,
    )
    business_outputs = _without_audit(outputs)

    assert _message_types(outputs) == [
        "session_started",
        "event",
        "tool_result",
        "round_stats",
        "message_delta",
        "round_stats",
        "done",
    ]
    assert json.loads(business_outputs[2]["message"]["content"]) == {
        "tool": "handoff_note",
        "error": "tool handler failed: boom",
    }


@pytest.mark.asyncio
async def test_runtime_uses_event_and_input_queues_for_event_agent_handoff():
    RuntimeQueues = _runtime_queues_class()
    queue_log: list[tuple[str, str, str]] = []
    queues = RuntimeQueues(
        input=RecordingQueue("input", queue_log),
        output=RecordingQueue("output", queue_log),
        events=RecordingQueue("events", queue_log),
    )
    request = DebugRunRequest(
        user_message="debug this",
        system_prompts=["You are a debugger."],
        pre_messages=[],
        chat_agent=AgentParams(model="fake-model", temperature=0.1, max_tokens=200),
        event_agent=EventAgentParams(enabled_tools=["handoff_note"], max_event_loops=3),
    )
    runtime = DebugRuntime(FakeChatClient(), queues=queues)

    outputs = [message async for message in runtime.run(request)]

    assert _message_types(outputs) == [
        "session_started",
        "event",
        "tool_result",
        "round_stats",
        "message_delta",
        "round_stats",
        "done",
    ]
    assert queue_log.index(("input", "put", "user")) < queue_log.index(
        ("input", "get", "user")
    )
    assert queue_log.index(
        ("events", "put", "event_request:handoff_note:call_1")
    ) < queue_log.index(("events", "get", "event_request:handoff_note:call_1"))
    assert queue_log.index(
        ("events", "get", "event_request:handoff_note:call_1")
    ) < queue_log.index(("input", "put", "tool:call_1"))
    assert queue_log.index(("input", "put", "tool:call_1")) < queue_log.index(
        ("input", "get", "tool:call_1")
    )


@pytest.mark.asyncio
async def test_runtime_run_consumes_output_queue_in_stream_order():
    RuntimeQueues = _runtime_queues_class()
    queue_log: list[tuple[str, str, str]] = []
    queues = RuntimeQueues(
        input=RecordingQueue("input", queue_log),
        output=RecordingQueue("output", queue_log),
        events=RecordingQueue("events", queue_log),
    )
    request = DebugRunRequest(
        user_message="debug this",
        system_prompts=["You are a debugger."],
        pre_messages=[],
        chat_agent=AgentParams(model="fake-model", temperature=0.1, max_tokens=200),
        event_agent=EventAgentParams(enabled_tools=["handoff_note"], max_event_loops=3),
    )
    runtime = DebugRuntime(FakeChatClient(), queues=queues)

    outputs = [message async for message in runtime.run(request)]

    assert _message_types(outputs) == [
        "session_started",
        "event",
        "tool_result",
        "round_stats",
        "message_delta",
        "round_stats",
        "done",
    ]
    output_puts = [
        entry[2] for entry in queue_log if entry[0] == "output" and entry[1] == "put"
    ]
    output_gets = [
        entry[2] for entry in queue_log if entry[0] == "output" and entry[1] == "get"
    ]
    assert [message for message in output_puts if message != "output:audit"] == [
        "output:session_started",
        "output:event",
        "output:tool_result",
        "output:round_stats",
        "output:message_delta",
        "output:round_stats",
        "output:done",
    ]
    assert output_gets == output_puts


@pytest.mark.asyncio
async def test_runtime_continues_with_event_summary_without_tool_call_history():
    request = DebugRunRequest(
        user_message="debug this",
        system_prompts=["You are a debugger."],
        pre_messages=[],
        chat_agent=AgentParams(model="fake-model", temperature=0.1, max_tokens=200),
        event_agent=EventAgentParams(enabled_tools=["handoff_note"], max_event_loops=2),
    )
    client = StrictHistoryChatClient()
    runtime = DebugRuntime(client)

    outputs = [message async for message in runtime.run(request)]

    assert client.calls == 2
    assert [message.role for message in client.second_call_messages] == [
        "system",
        "system",
        "user",
        "assistant",
        "user",
    ]
    assistant_message = client.second_call_messages[3]
    assert assistant_message.content == ""
    assert "handoff_note" in client.second_call_messages[1].content
    assert not any(message.role == "tool" for message in client.second_call_messages)
    assert client.second_call_messages[4].content.startswith("EventAgent results:\n")
    assert client.second_call_messages[4].name == "event_agent"
    assert outputs[-1] == {"type": "done"}


@pytest.mark.asyncio
async def test_runtime_passes_event_catalog_system_message_without_chat_tools():
    registry = ToolRegistry(
        [
            ToolDefinition(
                name="handoff_note",
                description="Registry-owned handoff tool.",
                parameters={
                    "type": "object",
                    "properties": {
                        "message": {"type": "string"},
                        "priority": {"type": "number"},
                    },
                    "required": ["message"],
                },
                handler=lambda event: {"tool": event.name, "message": "handled"},
            )
        ]
    )
    request = DebugRunRequest(
        user_message="debug this",
        system_prompts=[],
        pre_messages=[],
        chat_agent=AgentParams(model="fake-model", temperature=0.1, max_tokens=200),
        event_agent=EventAgentParams(enabled_tools=["handoff_note"], max_event_loops=2),
    )
    client = ToolCapturingChatClient()
    runtime = DebugRuntime(client, registry=registry)

    outputs = [message async for message in runtime.run(request)]

    assert outputs[-1] == {"type": "done"}
    assert client.tools == []
    assert client.messages[0].role == "system"
    assert "Available events:" in client.messages[0].content
    assert "- handoff_note: Registry-owned handoff tool." in client.messages[0].content
    assert "message" not in client.messages[0].content
    assert "priority" not in client.messages[0].content


@pytest.mark.asyncio
async def test_runtime_emits_round_stats_with_clock_and_usage_after_model_turn():
    request = DebugRunRequest(
        user_message="debug this",
        system_prompts=[],
        pre_messages=[],
        chat_agent=AgentParams(model="fake-model", temperature=0.1, max_tokens=200),
        event_agent=EventAgentParams(enabled_tools=[], max_event_loops=1),
    )
    ticks = iter(
        [
            0.9,
            0.91,
            1.0,
            1.01,
            1.02,
            1.123,
            1.2,
            1.25,
            1.3,
            1.456,
            1.7,
            1.8,
        ]
    )
    runtime = DebugRuntime(RoundStatsChatClient(), clock=lambda: next(ticks))

    outputs = [message async for message in runtime.run(request)]
    business_outputs = _without_audit(outputs)

    assert _message_types(outputs) == [
        "session_started",
        "message_delta",
        "usage",
        "round_stats",
        "done",
    ]
    assert business_outputs[3] == {
        "type": "round_stats",
        "round_index": 1,
        "ttft_ms": 123,
        "elapsed_ms": 456,
        "prompt_tokens": 10,
        "completion_tokens": 20,
        "total_tokens": 30,
        "cached_tokens": 5,
        "had_event": False,
    }


@pytest.mark.asyncio
async def test_runtime_emits_round_stats_for_each_chat_call_in_event_handoff():
    request = DebugRunRequest(
        user_message="debug this",
        system_prompts=[],
        pre_messages=[],
        chat_agent=AgentParams(model="fake-model", temperature=0.1, max_tokens=200),
        event_agent=EventAgentParams(enabled_tools=["handoff_note"], max_event_loops=2),
    )
    ticks = iter(
        [
            1.9,
            1.91,
            2.0,
            2.01,
            2.02,
            2.03,
            2.04,
            2.05,
            2.06,
            2.07,
            2.08,
            2.09,
            2.1,
            2.25,
            2.26,
            3.0,
            3.01,
            3.02,
            3.05,
            3.1,
            3.15,
            3.18,
            3.2,
            3.23,
            3.24,
            3.25,
            3.26,
        ]
    )
    client = EventRoundStatsChatClient()
    runtime = DebugRuntime(client, clock=lambda: next(ticks))

    outputs = [message async for message in runtime.run(request)]
    stats = [message for message in outputs if message["type"] == "round_stats"]

    assert client.calls == 2
    assert stats == [
        {
            "type": "round_stats",
            "round_index": 1,
            "ttft_ms": None,
            "elapsed_ms": 250,
            "prompt_tokens": 3,
            "completion_tokens": 0,
            "total_tokens": 3,
            "cached_tokens": 0,
            "had_event": True,
        },
        {
            "type": "round_stats",
            "round_index": 2,
            "ttft_ms": 50,
            "elapsed_ms": 200,
            "prompt_tokens": 4,
            "completion_tokens": 6,
            "total_tokens": 10,
            "cached_tokens": 0,
            "had_event": False,
        },
    ]


@pytest.mark.asyncio
async def test_runtime_session_resets_event_budget_for_each_user_turn():
    request = DebugRunRequest(
        user_message="first turn",
        system_prompts=[],
        pre_messages=[],
        chat_agent=AgentParams(model="fake-model", temperature=0.1, max_tokens=200),
        event_agent=EventAgentParams(enabled_tools=["handoff_note"], max_event_loops=1),
    )
    client = TwoTurnSessionChatClient()
    runtime = DebugRuntime(client)

    queues = runtime.start_session(request)
    outputs: list[dict[str, Any]] = []
    while len([message for message in outputs if message["type"] == "turn_completed"]) < 1:
        outputs.append(await asyncio.wait_for(queues.output.get(), timeout=1))

    await queues.input.put(ChatMessage(role="user", content="second turn"))
    while len([message for message in outputs if message["type"] == "turn_completed"]) < 2:
        outputs.append(await asyncio.wait_for(queues.output.get(), timeout=1))

    await runtime.aclose()

    business_types = [message["type"] for message in _without_audit(outputs)]
    assert business_types.count("turn_started") == 2
    assert business_types.count("turn_completed") == 2
    assert client.calls == 4
    assert "Available events:" in client.messages_by_call[0][0].content
    assert "Available events:" in client.messages_by_call[2][0].content


@pytest.mark.asyncio
async def test_runtime_persists_session_turn_messages_audit_and_usage(tmp_path):
    store = SQLiteSessionStore(tmp_path / "agent_lab.sqlite3")
    request = DebugRunRequest(
        session_id="session-1",
        user_message="persist this",
        system_prompts=["You are a debugger."],
        pre_messages=[],
        chat_agent=AgentParams(model="chat-model", temperature=0.1, max_tokens=200),
        event_agent=EventAgentParams(enabled_tools=[], max_event_loops=1),
    )
    runtime = DebugRuntime(RoundStatsChatClient(), session_store=store)

    queues = runtime.start_session(request)
    outputs: list[dict[str, Any]] = []
    while not any(message["type"] == "turn_completed" for message in outputs):
        outputs.append(await asyncio.wait_for(queues.output.get(), timeout=1))
    await runtime.aclose()

    assert _without_audit(outputs)[0] == {
        "type": "session_started",
        "session_id": "session-1",
    }
    assert store.get_session("session-1")["config"]["chat_agent"]["model"] == "chat-model"
    assert [
        (message["turn_index"], message["role"], message["content"])
        for message in store.list_messages("session-1")
    ] == [
        (1, "user", "persist this"),
        (1, "assistant", "hello"),
    ]
    audit_events = [
        audit["event"]
        for audit in store.list_audit_logs("session-1")
    ]
    assert "session_started" in audit_events
    assert "chat_agent_request" in audit_events
    assert "turn_completed" in audit_events
    usage = store.usage_summary("session-1")
    assert usage["calls"][0]["total_tokens"] == 30
    assert usage["turns"][0]["turn_index"] == 1
    assert usage["session"]["total_tokens"] == 30


@pytest.mark.asyncio
async def test_runtime_continues_persisted_turn_indexes_for_existing_session(tmp_path):
    store = SQLiteSessionStore(tmp_path / "agent_lab.sqlite3")
    session_id = store.create_session(title="existing", config={})
    store.start_turn(session_id, turn_index=1, user_message="old")
    request = DebugRunRequest(
        session_id=session_id,
        user_message="new",
        system_prompts=[],
        pre_messages=[],
        chat_agent=AgentParams(model="chat-model", temperature=0.1, max_tokens=200),
        event_agent=EventAgentParams(enabled_tools=[], max_event_loops=1),
    )
    runtime = DebugRuntime(RoundStatsChatClient(), session_store=store)

    queues = runtime.start_session(request)
    outputs: list[dict[str, Any]] = []
    while not any(message["type"] == "turn_completed" for message in outputs):
        outputs.append(await asyncio.wait_for(queues.output.get(), timeout=1))
    await runtime.aclose()

    assert [
        (message["turn_index"], message["content"])
        for message in store.list_messages(session_id)
    ] == [
        (2, "new"),
        (2, "hello"),
    ]


class ScriptedChatClient:
    def __init__(self, rounds: list[list[StreamItem]]) -> None:
        self.rounds = rounds
        self.calls = 0
        self.messages_by_call: list[list[ChatMessage]] = []
        self.tools_by_call: list[list[dict[str, Any]]] = []

    async def stream_chat(
        self,
        messages: list[ChatMessage],
        tools: list[dict],
        params: AgentParams,
        tool_choice: dict[str, Any] | None = None,
    ) -> AsyncIterator[StreamItem]:
        self.messages_by_call.append(list(messages))
        self.tools_by_call.append(list(tools))
        round_items = self.rounds[self.calls]
        self.calls += 1
        for item in round_items:
            yield item


class TranscriptValidatingScriptedChatClient(ScriptedChatClient):
    async def stream_chat(
        self,
        messages: list[ChatMessage],
        tools: list[dict],
        params: AgentParams,
        tool_choice: dict[str, Any] | None = None,
    ) -> AsyncIterator[StreamItem]:
        OpenAICompatibleChatClient._validate_tool_transcript(self, messages)
        async for item in super().stream_chat(
            messages,
            tools,
            params,
            tool_choice,
        ):
            yield item


def _direct_request(
    *,
    enabled_tools: list[str],
    max_event_loops: int = 1,
    session_id: str | None = None,
) -> DebugRunRequest:
    return DebugRunRequest(
        session_id=session_id,
        user_message="debug direct tools",
        system_prompts=["You are a debugger."],
        pre_messages=[],
        chat_agent=AgentParams(model="chat-model"),
        event_agent=EventAgentParams(
            enabled_tools=enabled_tools,
            max_event_loops=max_event_loops,
        ),
        tool_invocation_mode="chat_agent_tools",
    )


@pytest.mark.asyncio
async def test_direct_mode_uses_enabled_schemas_and_valid_ordered_provider_transcript():
    executed: list[tuple[str, dict[str, Any], str]] = []

    def handle(event: ToolCallEvent) -> dict[str, Any]:
        executed.append((event.name, event.arguments, event.raw_arguments))
        return {"tool": event.name, "value": event.arguments["value"]}

    registry = ToolRegistry(
        [
            ToolDefinition(
                name="first_tool",
                description="First direct tool.",
                parameters={
                    "type": "object",
                    "properties": {"value": {"type": "string"}},
                    "required": ["value"],
                },
                handler=handle,
            ),
            ToolDefinition(
                name="second_tool",
                description="Second direct tool.",
                parameters={
                    "type": "object",
                    "properties": {"value": {"type": "string"}},
                    "required": ["value"],
                },
                handler=handle,
            ),
        ]
    )
    calls = [
        ToolCallEvent(
            id="provider-1",
            name="first_tool",
            arguments={"value": "one"},
            raw_arguments='{"value":"one"}',
        ),
        ToolCallEvent(
            id="provider-2",
            name="second_tool",
            arguments={"value": "two"},
            raw_arguments='{"value":"two"}',
        ),
    ]
    ignored_text_event = ToolCallEvent(
        id="text-ignored",
        name="first_tool",
        arguments={},
        raw_arguments="{}",
    )
    client = ScriptedChatClient(
        [
            [
                StreamItem.message_delta("Visible before tools."),
                StreamItem.provider_tool_call(calls[0]),
                StreamItem.text_event(ignored_text_event),
                StreamItem.provider_tool_call(calls[1]),
            ],
            [StreamItem.message_delta("Final answer.")],
        ]
    )

    outputs = await _collect_outputs(
        DebugRuntime(client, registry=registry).run(
            _direct_request(enabled_tools=["first_tool", "second_tool"])
        )
    )

    assert [tool["function"]["name"] for tool in client.tools_by_call[0]] == [
        "first_tool",
        "second_tool",
    ]
    assert client.tools_by_call[1] == []
    assert not any(
        "Available events:" in message.content
        for message in client.messages_by_call[0]
        if message.role == "system"
    )
    transcript = client.messages_by_call[1]
    assert [message.role for message in transcript[-4:]] == [
        "user",
        "assistant",
        "tool",
        "tool",
    ]
    assert transcript[-3].content == "Visible before tools."
    assert transcript[-3].tool_calls == calls
    assert [message.tool_call_id for message in transcript[-2:]] == [
        "provider-1",
        "provider-2",
    ]
    assert [message.name for message in transcript[-2:]] == [
        "first_tool",
        "second_tool",
    ]
    assert executed == [
        ("first_tool", {"value": "one"}, '{"value":"one"}'),
        ("second_tool", {"value": "two"}, '{"value":"two"}'),
    ]
    emitted_events = [
        message["event"] for message in outputs if message["type"] == "event"
    ]
    assert emitted_events == [call.model_dump() for call in calls]
    assert "event_agent_request" not in [
        message.get("event") for message in outputs if message["type"] == "audit"
    ]
    assert not any(
        message.get("event", {}).get("id") == "text-ignored"
        for message in outputs
        if message["type"] == "event"
    )
    assert all(
        message["details"]["tool_invocation_mode"] == "chat_agent_tools"
        for message in outputs
        if message["type"] == "audit"
        and message["event"] in {"chat_round_started", "chat_agent_request"}
    )


@pytest.mark.asyncio
async def test_direct_mode_does_not_execute_unknown_disabled_or_failed_handlers():
    executed: list[str] = []

    def disabled_handler(event: ToolCallEvent) -> dict[str, Any]:
        executed.append(event.name)
        return {"tool": event.name}

    def failing_handler(event: ToolCallEvent) -> dict[str, Any]:
        executed.append(event.name)
        raise RuntimeError("direct boom")

    registry = ToolRegistry(
        [
            ToolDefinition(
                name="enabled_tool",
                description="Enabled.",
                parameters={"type": "object"},
                handler=lambda event: {"tool": event.name},
            ),
            ToolDefinition(
                name="disabled_tool",
                description="Disabled.",
                parameters={"type": "object"},
                handler=disabled_handler,
            ),
            ToolDefinition(
                name="failing_tool",
                description="Fails.",
                parameters={"type": "object"},
                handler=failing_handler,
            ),
        ]
    )
    client = ScriptedChatClient(
        [
            [
                StreamItem.provider_tool_call(
                    ToolCallEvent(
                        id="unknown-1",
                        name="unknown_tool",
                        arguments={"kept": True},
                        raw_arguments='{"kept":true}',
                    )
                ),
                StreamItem.provider_tool_call(
                    ToolCallEvent(
                        id="disabled-1",
                        name="disabled_tool",
                        arguments={},
                        raw_arguments="{}",
                    )
                ),
                StreamItem.provider_tool_call(
                    ToolCallEvent(
                        id="failed-1",
                        name="failing_tool",
                        arguments={},
                        raw_arguments="{}",
                    )
                ),
            ],
            [StreamItem.message_delta("continued")],
        ]
    )

    outputs = await _collect_outputs(
        DebugRuntime(client, registry=registry).run(
            _direct_request(enabled_tools=["enabled_tool", "failing_tool"])
        )
    )

    assert [tool["function"]["name"] for tool in client.tools_by_call[0]] == [
        "enabled_tool",
        "failing_tool",
    ]
    assert executed == ["failing_tool"]
    results = [
        json.loads(message["message"]["content"])
        for message in outputs
        if message["type"] == "tool_result"
    ]
    assert results == [
        {"tool": "unknown_tool", "error": "unknown tool"},
        {"tool": "disabled_tool", "error": "tool disabled"},
        {"tool": "failing_tool", "error": "tool handler failed: direct boom"},
    ]


@pytest.mark.asyncio
async def test_direct_mode_ignores_provider_calls_after_event_budget_exhaustion():
    executed: list[str] = []
    registry = ToolRegistry(
        [
            ToolDefinition(
                name="once_tool",
                description="Run once.",
                parameters={"type": "object"},
                handler=lambda event: executed.append(event.id) or {"tool": event.name},
            )
        ]
    )
    client = ScriptedChatClient(
        [
            [
                StreamItem.provider_tool_call(
                    ToolCallEvent(
                        id="accepted",
                        name="once_tool",
                        arguments={},
                        raw_arguments="{}",
                    )
                )
            ],
            [
                StreamItem.provider_tool_call(
                    ToolCallEvent(
                        id="ignored",
                        name="once_tool",
                        arguments={},
                        raw_arguments="{}",
                    )
                ),
                StreamItem.message_delta("budget exhausted"),
            ],
        ]
    )

    outputs = await _collect_outputs(
        DebugRuntime(client, registry=registry).run(
            _direct_request(enabled_tools=["once_tool"], max_event_loops=1)
        )
    )

    assert client.tools_by_call == [
        [registry.tool_schema("once_tool")],
        [],
    ]
    assert executed == ["accepted"]
    assert [
        message["event"]["id"]
        for message in outputs
        if message["type"] == "event"
    ] == ["accepted"]
    assert outputs[-1] == {"type": "done"}


@pytest.mark.asyncio
async def test_direct_mode_session_turns_reset_budget_and_snapshot_mode(tmp_path):
    store = SQLiteSessionStore(tmp_path / "agent_lab.sqlite3")
    registry = ToolRegistry(
        [
            ToolDefinition(
                name="turn_tool",
                description="Per-turn tool.",
                parameters={"type": "object"},
                handler=lambda event: {"tool": event.name, "call_id": event.id},
            )
        ]
    )
    client = ScriptedChatClient(
        [
            [
                StreamItem.provider_tool_call(
                    ToolCallEvent(
                        id="turn-1-call",
                        name="turn_tool",
                        arguments={},
                        raw_arguments="{}",
                    )
                )
            ],
            [StreamItem.message_delta("turn one done")],
            [
                StreamItem.provider_tool_call(
                    ToolCallEvent(
                        id="turn-2-call",
                        name="turn_tool",
                        arguments={},
                        raw_arguments="{}",
                    )
                )
            ],
            [StreamItem.message_delta("turn two done")],
        ]
    )
    runtime = DebugRuntime(client, registry=registry, session_store=store)
    request = _direct_request(
        enabled_tools=["turn_tool"],
        max_event_loops=1,
        session_id="direct-session",
    )

    queues = runtime.start_session(request)
    outputs: list[dict[str, Any]] = []
    while sum(message["type"] == "turn_completed" for message in outputs) < 1:
        outputs.append(await asyncio.wait_for(queues.output.get(), timeout=1))
    await queues.input.put(ChatMessage(role="user", content="second direct turn"))
    while sum(message["type"] == "turn_completed" for message in outputs) < 2:
        outputs.append(await asyncio.wait_for(queues.output.get(), timeout=1))
    await runtime.aclose()

    assert [bool(tools) for tools in client.tools_by_call] == [True, False, True, False]
    assert [message.role for message in client.messages_by_call[1][-3:]] == [
        "user",
        "assistant",
        "tool",
    ]
    second_turn_transcript = client.messages_by_call[3]
    assert not any(message.name == "event_agent" for message in second_turn_transcript)
    assert [
        message.tool_call_id
        for message in second_turn_transcript
        if message.role == "tool"
    ] == ["turn-1-call", "turn-2-call"]
    session = store.get_session("direct-session")
    assert session is not None
    assert session["config"]["tool_invocation_mode"] == "chat_agent_tools"
    persisted_messages = store.list_messages("direct-session")
    assert [message["role"] for message in persisted_messages] == [
        "user",
        "assistant",
        "tool",
        "assistant",
        "user",
        "assistant",
        "tool",
        "assistant",
    ]
    assert [
        call["id"]
        for message in persisted_messages
        for call in message["tool_calls"]
    ] == ["turn-1-call", "turn-2-call"]
    assert [
        message["tool_call_id"]
        for message in persisted_messages
        if message["role"] == "tool"
    ] == ["turn-1-call", "turn-2-call"]
    request_audits = [
        audit
        for audit in store.list_audit_logs("direct-session")
        if audit["event"] == "chat_agent_request"
    ]
    assert request_audits
    assert all(
        audit["details"]["tool_invocation_mode"] == "chat_agent_tools"
        for audit in request_audits
    )


@pytest.mark.asyncio
async def test_direct_mode_reconnect_restores_provider_valid_session_transcript(tmp_path):
    store = SQLiteSessionStore(tmp_path / "agent_lab.sqlite3")
    session_id = store.create_session(
        title="restored direct",
        config={"tool_invocation_mode": "chat_agent_tools"},
        session_id="restored-direct",
    )
    store.start_turn(session_id, turn_index=1, user_message="past user")
    previous_call = ToolCallEvent(
        id="past-call",
        name="safe_tool",
        arguments={"value": 1},
        raw_arguments='{"value":1}',
    )
    for message in [
        ChatMessage(role="user", content="past user"),
        ChatMessage(role="assistant", content="", tool_calls=[previous_call]),
        ChatMessage(
            role="tool",
            content='{"ok":true}',
            name="safe_tool",
            tool_call_id="past-call",
        ),
        ChatMessage(role="assistant", content="past answer"),
    ]:
        store.append_message(session_id, turn_index=1, message=message)
    store.complete_turn(session_id, turn_index=1)
    client = TranscriptValidatingScriptedChatClient(
        [[StreamItem.message_delta("current answer")]]
    )
    runtime = DebugRuntime(
        client,
        registry=_single_tool_registry(lambda event: {"ok": True}),
        session_store=store,
    )
    request = _direct_request(
        enabled_tools=["safe_tool"],
        session_id=session_id,
    ).model_copy(
        update={
            "user_message": "current user",
            "pre_messages": [ChatMessage(role="user", content="configured pre")],
        }
    )

    queues = runtime.start_session(request)
    outputs: list[dict[str, Any]] = []
    while not any(message["type"] == "turn_completed" for message in outputs):
        outputs.append(await asyncio.wait_for(queues.output.get(), timeout=1))
    await runtime.aclose()

    transcript = client.messages_by_call[0]
    assert [(message.role, message.content) for message in transcript] == [
        ("system", "You are a debugger."),
        ("user", "configured pre"),
        ("user", "past user"),
        ("assistant", ""),
        ("tool", '{"ok":true}'),
        ("assistant", "past answer"),
        ("user", "current user"),
    ]
    assert transcript[3].tool_calls == [previous_call]
    assert sum(
        message.role == "user" and message.content == "current user"
        for message in transcript
    ) == 1


@pytest.mark.asyncio
async def test_cancelled_direct_batch_persists_no_partial_protocol_and_reconnects(tmp_path):
    store = SQLiteSessionStore(tmp_path / "agent_lab.sqlite3")
    handler_started = asyncio.Event()
    release_handler = asyncio.Event()

    async def blocking_handler(event: ToolCallEvent) -> dict[str, Any]:
        handler_started.set()
        await release_handler.wait()
        return {"ok": True, "call_id": event.id}

    registry = _single_tool_registry(blocking_handler)
    first_client = ScriptedChatClient(
        [[StreamItem.provider_tool_call(_tool_call("cancelled-call"))]]
    )
    first_runtime = DebugRuntime(
        first_client,
        registry=registry,
        session_store=store,
    )
    request = _direct_request(
        enabled_tools=["safe_tool"],
        session_id="cancelled-direct",
    )

    first_runtime.start_session(request)
    await asyncio.wait_for(handler_started.wait(), timeout=1)
    await first_runtime.aclose()

    persisted_after_cancel = store.list_messages("cancelled-direct")
    assert [message["role"] for message in persisted_after_cancel] == ["user"]
    assert not any(message["tool_calls"] for message in persisted_after_cancel)

    reconnect_client = TranscriptValidatingScriptedChatClient(
        [[StreamItem.message_delta("reconnected")]]
    )
    reconnect_runtime = DebugRuntime(
        reconnect_client,
        registry=registry,
        session_store=store,
    )
    reconnect_request = request.model_copy(
        update={"user_message": "after cancellation"}
    )
    queues = reconnect_runtime.start_session(reconnect_request)
    outputs: list[dict[str, Any]] = []
    while not any(
        message["type"] in {"turn_completed", "error"} for message in outputs
    ):
        outputs.append(await asyncio.wait_for(queues.output.get(), timeout=1))
    await reconnect_runtime.aclose()

    assert not any(message["type"] == "error" for message in outputs)
    assert [
        (message.role, message.content)
        for message in reconnect_client.messages_by_call[0]
        if message.role != "system"
    ] == [
        ("user", "debug direct tools"),
        ("user", "after cancellation"),
    ]


@pytest.mark.asyncio
async def test_direct_restore_drops_invalid_protocol_fragments_and_keeps_visible_history(
    tmp_path,
):
    store = SQLiteSessionStore(tmp_path / "agent_lab.sqlite3")
    session_id = store.create_session(
        title="repair transcript",
        config={"tool_invocation_mode": "chat_agent_tools"},
        session_id="repair-transcript",
    )
    store.start_turn(session_id, turn_index=1, user_message="visible user")
    complete_calls = [_tool_call("complete-1"), _tool_call("complete-2")]
    store.append_messages(
        session_id,
        turn_index=1,
        messages=[
            ChatMessage(role="user", content="visible user"),
            ChatMessage(
                role="assistant",
                content="using tools",
                tool_calls=complete_calls,
            ),
            ChatMessage(
                role="tool",
                content='{"id":1}',
                name="safe_tool",
                tool_call_id="complete-1",
            ),
            ChatMessage(
                role="tool",
                content='{"id":2}',
                name="safe_tool",
                tool_call_id="complete-2",
            ),
            ChatMessage(role="assistant", content="visible answer"),
            ChatMessage(
                role="tool",
                content='{"orphan":true}',
                name="safe_tool",
                tool_call_id="orphan-call",
            ),
            ChatMessage(role="assistant", content="visible after orphan"),
            ChatMessage(
                role="assistant",
                content="",
                tool_calls=[_tool_call("incomplete-call")],
            ),
        ],
    )
    store.complete_turn(session_id, turn_index=1)
    client = TranscriptValidatingScriptedChatClient(
        [[StreamItem.message_delta("repaired")]]
    )
    runtime = DebugRuntime(
        client,
        registry=_single_tool_registry(lambda event: {"ok": True}),
        session_store=store,
    )
    request = _direct_request(
        enabled_tools=["safe_tool"],
        session_id=session_id,
    ).model_copy(update={"user_message": "current user"})

    queues = runtime.start_session(request)
    outputs: list[dict[str, Any]] = []
    while not any(
        message["type"] in {"turn_completed", "error"} for message in outputs
    ):
        outputs.append(await asyncio.wait_for(queues.output.get(), timeout=1))
    await runtime.aclose()

    assert not any(message["type"] == "error" for message in outputs)
    transcript = client.messages_by_call[0]
    assert [(message.role, message.content) for message in transcript] == [
        ("system", "You are a debugger."),
        ("user", "visible user"),
        ("assistant", "using tools"),
        ("tool", '{"id":1}'),
        ("tool", '{"id":2}'),
        ("assistant", "visible answer"),
        ("assistant", "visible after orphan"),
        ("user", "current user"),
    ]
    assert transcript[2].tool_calls == complete_calls


@pytest.mark.asyncio
async def test_dual_mode_reconnect_restores_visible_history_without_tool_protocol(tmp_path):
    store = SQLiteSessionStore(tmp_path / "agent_lab.sqlite3")
    session_id = store.create_session(
        title="restored dual",
        config={"tool_invocation_mode": "dual_agent"},
        session_id="restored-dual",
    )
    store.start_turn(session_id, turn_index=1, user_message="past user")
    store.append_message(
        session_id,
        turn_index=1,
        message=ChatMessage(role="user", content="past user"),
    )
    store.append_message(
        session_id,
        turn_index=1,
        message=ChatMessage(role="assistant", content="past answer"),
    )
    store.complete_turn(session_id, turn_index=1)
    registry = _single_tool_registry(lambda event: {"ok": True})
    client = ScriptedChatClient(
        [
            [StreamItem.text_event(_tool_call("dual-call"))],
            [StreamItem.message_delta("current answer")],
        ]
    )
    runtime = DebugRuntime(client, registry=registry, session_store=store)
    request = DebugRunRequest(
        session_id=session_id,
        user_message="current user",
        system_prompts=["system context"],
        pre_messages=[ChatMessage(role="assistant", content="configured pre")],
        chat_agent=AgentParams(model="chat-model"),
        event_agent=EventAgentParams(
            enabled_tools=["safe_tool"],
            max_event_loops=1,
        ),
        tool_invocation_mode="dual_agent",
    )

    queues = runtime.start_session(request)
    outputs: list[dict[str, Any]] = []
    while not any(message["type"] == "turn_completed" for message in outputs):
        outputs.append(await asyncio.wait_for(queues.output.get(), timeout=1))
    await runtime.aclose()

    first_request = client.messages_by_call[0]
    assert [(message.role, message.content) for message in first_request if message.role != "system"] == [
        ("assistant", "configured pre"),
        ("user", "past user"),
        ("assistant", "past answer"),
        ("user", "current user"),
    ]
    assert sum(
        message.role == "user" and message.content == "current user"
        for message in first_request
    ) == 1
    persisted_messages = store.list_messages(session_id)
    assert not any(message["role"] == "tool" for message in persisted_messages)
    assert not any(message["tool_calls"] for message in persisted_messages)
    assert not any(message["content"] == "" for message in persisted_messages)


def _single_tool_registry(
    handler: Any,
    *,
    name: str = "safe_tool",
) -> ToolRegistry:
    return ToolRegistry(
        [
            ToolDefinition(
                name=name,
                description="Tool used by execution-safety tests.",
                parameters={"type": "object"},
                handler=handler,
            )
        ]
    )


def _tool_call(call_id: str, *, name: str = "safe_tool") -> ToolCallEvent:
    return ToolCallEvent(
        id=call_id,
        name=name,
        arguments={},
        raw_arguments="{}",
    )


@pytest.mark.asyncio
async def test_direct_mode_rejects_provider_call_id_reused_across_one_shot_rounds():
    executed: list[str] = []
    registry = _single_tool_registry(
        lambda event: executed.append(event.id) or {"tool": event.name}
    )
    client = ScriptedChatClient(
        [
            [StreamItem.provider_tool_call(_tool_call("duplicate-call"))],
            [StreamItem.provider_tool_call(_tool_call("duplicate-call"))],
        ]
    )

    outputs = await _collect_outputs(
        DebugRuntime(client, registry=registry).run(
            _direct_request(enabled_tools=["safe_tool"], max_event_loops=2)
        )
    )

    assert executed == ["duplicate-call"]
    assert [
        message["message"]["tool_call_id"]
        for message in outputs
        if message["type"] == "tool_result"
    ] == ["duplicate-call"]
    assert outputs[-1]["type"] == "error"
    assert "duplicate provider tool-call ID: duplicate-call" in outputs[-1]["message"]


@pytest.mark.asyncio
async def test_direct_mode_rejects_provider_call_id_reused_in_session_rounds():
    executed: list[str] = []
    registry = _single_tool_registry(
        lambda event: executed.append(event.id) or {"tool": event.name}
    )
    client = ScriptedChatClient(
        [
            [StreamItem.provider_tool_call(_tool_call("session-duplicate"))],
            [StreamItem.provider_tool_call(_tool_call("session-duplicate"))],
        ]
    )
    runtime = DebugRuntime(client, registry=registry)

    queues = runtime.start_session(
        _direct_request(enabled_tools=["safe_tool"], max_event_loops=2)
    )
    outputs: list[dict[str, Any]] = []
    while not any(message["type"] == "error" for message in outputs):
        outputs.append(await asyncio.wait_for(queues.output.get(), timeout=1))
    await runtime.aclose()

    assert executed == ["session-duplicate"]
    assert [
        message["message"]["tool_call_id"]
        for message in outputs
        if message["type"] == "tool_result"
    ] == ["session-duplicate"]
    assert "duplicate provider tool-call ID: session-duplicate" in outputs[-1][
        "message"
    ]


@pytest.mark.asyncio
async def test_direct_mode_rejects_duplicate_provider_call_ids_within_batch():
    executed: list[str] = []
    registry = _single_tool_registry(
        lambda event: executed.append(event.id) or {"tool": event.name}
    )
    client = ScriptedChatClient(
        [
            [
                StreamItem.provider_tool_call(_tool_call("same-batch")),
                StreamItem.provider_tool_call(_tool_call("same-batch")),
            ]
        ]
    )

    outputs = await _collect_outputs(
        DebugRuntime(client, registry=registry).run(
            _direct_request(enabled_tools=["safe_tool"])
        )
    )

    assert executed == []
    assert outputs[-1]["type"] == "error"
    assert "duplicate provider tool-call ID: same-batch" in outputs[-1]["message"]


@pytest.mark.asyncio
async def test_direct_mode_executes_batch_concurrently_but_replies_in_provider_order():
    both_started = asyncio.Event()
    first_release = asyncio.Event()
    second_release = asyncio.Event()
    second_finished = asyncio.Event()
    started: list[str] = []

    async def handler(event: ToolCallEvent) -> dict[str, Any]:
        started.append(event.id)
        if len(started) == 2:
            both_started.set()
        if event.id == "first-call":
            await first_release.wait()
        else:
            await second_release.wait()
            second_finished.set()
        return {"tool": event.name, "call_id": event.id}

    registry = _single_tool_registry(handler)
    client = ScriptedChatClient(
        [
            [
                StreamItem.provider_tool_call(
                    _tool_call("first-call").model_copy(
                        update={"arguments": {"order": 1}, "raw_arguments": '{"order":1}'}
                    )
                ),
                StreamItem.provider_tool_call(
                    _tool_call("second-call").model_copy(
                        update={"arguments": {"order": 2}, "raw_arguments": '{"order":2}'}
                    )
                ),
            ],
            [StreamItem.message_delta("finished")],
        ]
    )
    runtime = DebugRuntime(client, registry=registry)
    run_task = asyncio.create_task(
        _collect_outputs(
            runtime.run(_direct_request(enabled_tools=["safe_tool"]))
        )
    )
    overlapped = False
    try:
        try:
            await asyncio.wait_for(both_started.wait(), timeout=1)
        except TimeoutError:
            pass
        else:
            overlapped = True
            second_release.set()
            await asyncio.wait_for(second_finished.wait(), timeout=1)
            assert not run_task.done()
    finally:
        first_release.set()
        second_release.set()
    outputs = await asyncio.wait_for(run_task, timeout=1)

    assert overlapped is True
    assert started == ["first-call", "second-call"]
    assert [
        message["message"]["tool_call_id"]
        for message in outputs
        if message["type"] == "tool_result"
    ] == ["first-call", "second-call"]


def _dual_budget_request() -> DebugRunRequest:
    return DebugRunRequest(
        user_message="dual budget",
        system_prompts=[],
        pre_messages=[],
        chat_agent=AgentParams(model="chat-model"),
        event_agent=EventAgentParams(
            enabled_tools=["safe_tool"],
            max_event_loops=1,
        ),
    )


@pytest.mark.asyncio
async def test_dual_mode_ignores_text_event_after_one_shot_budget_exhaustion():
    executed: list[str] = []
    registry = _single_tool_registry(
        lambda event: executed.append(event.id) or {"tool": event.name}
    )
    client = ScriptedChatClient(
        [
            [StreamItem.text_event(_tool_call("accepted-text-event"))],
            [
                StreamItem.text_event(_tool_call("ignored-text-event")),
                StreamItem.message_delta("final dual answer"),
            ],
        ]
    )

    outputs = await _collect_outputs(
        DebugRuntime(client, registry=registry).run(_dual_budget_request())
    )

    assert executed == ["accepted-text-event"]
    assert [
        message["event"]["id"]
        for message in outputs
        if message["type"] == "event"
    ] == ["accepted-text-event"]
    assert outputs[-1] == {"type": "done"}


@pytest.mark.asyncio
async def test_dual_mode_session_main_path_ignores_text_event_after_budget_exhaustion():
    executed: list[str] = []
    registry = _single_tool_registry(
        lambda event: executed.append(event.id) or {"tool": event.name}
    )
    client = ScriptedChatClient(
        [
            [StreamItem.text_event(_tool_call("session-accepted"))],
            [
                StreamItem.text_event(_tool_call("session-ignored")),
                StreamItem.message_delta("session final"),
            ],
        ]
    )
    runtime = DebugRuntime(client, registry=registry)

    queues = runtime.start_session(_dual_budget_request())
    outputs: list[dict[str, Any]] = []
    while not any(
        message["type"] in {"turn_completed", "error"} for message in outputs
    ):
        outputs.append(await asyncio.wait_for(queues.output.get(), timeout=1))
    await runtime.aclose()

    assert executed == ["session-accepted"]
    assert [
        message["event"]["id"]
        for message in outputs
        if message["type"] == "event"
    ] == ["session-accepted"]
    assert outputs[-1]["type"] == "turn_completed"


def _policy_request(
    *,
    enabled_tools: list[str],
    mode: str = "dual_agent",
    session_id: str | None = None,
) -> DebugRunRequest:
    return DebugRunRequest(
        session_id=session_id,
        user_message="exercise event result policy",
        system_prompts=[],
        pre_messages=[],
        chat_agent=AgentParams(model="chat-model"),
        event_agent=EventAgentParams(
            enabled_tools=enabled_tools,
            max_event_loops=1,
        ),
        tool_invocation_mode=mode,
    )


@pytest.mark.asyncio
async def test_dual_silent_success_does_not_enqueue_legacy_summary_or_call_chat_again():
    registry = ToolRegistry(
        [
            ToolDefinition(
                name="example.silent",
                description="Complete silently.",
                parameters={"type": "object"},
                handler=lambda event: {"tool": event.name, "status": "applied"},
                result_policy=ResultPolicy.SILENT_SUCCESS,
            )
        ]
    )
    client = ScriptedChatClient(
        [
            [
                StreamItem.message_delta("Applying it now."),
                StreamItem.text_event(_tool_call("silent-1", name="example.silent")),
            ]
        ]
    )

    outputs = await _collect_outputs(
        DebugRuntime(client, registry=registry).run(
            _policy_request(enabled_tools=["example.silent"])
        )
    )

    assert client.calls == 1
    assert outputs[-1] == {"type": "done"}
    assert not any(
        message.get("details", {}).get("result_summary")
        for message in outputs
        if message.get("event") == "event_agent_completed"
    )


@pytest.mark.asyncio
async def test_direct_template_success_emits_plugin_confirmation_without_model_call():
    schedule = ToolCallEvent(
        id="schedule-1",
        name="calendar.schedule.create",
        arguments={
            "title": "Design review",
            "start_at": "2026-07-14T09:30:00+08:00",
            "timezone": "Asia/Shanghai",
        },
        raw_arguments=(
            '{"title":"Design review","start_at":"2026-07-14T09:30:00+08:00",'
            '"timezone":"Asia/Shanghai"}'
        ),
    )
    client = ScriptedChatClient(
        [
            [
                StreamItem.message_delta("I will add it."),
                StreamItem.provider_tool_call(schedule),
            ]
        ]
    )

    outputs = await _collect_outputs(
        DebugRuntime(client).run(
            _policy_request(
                enabled_tools=["calendar.schedule.create"],
                mode="chat_agent_tools",
            )
        )
    )

    assert client.calls == 1
    assert [
        message["content"]
        for message in outputs
        if message["type"] == "message_delta"
    ] == [
        "I will add it.",
        "Scheduled Design review for 2026-07-14T09:30:00+08:00 (Asia/Shanghai).",
    ]
    assert outputs[-1] == {"type": "done"}


@pytest.mark.asyncio
async def test_direct_silent_failure_requests_exactly_one_corrective_follow_up():
    def fail(event: ToolCallEvent) -> dict[str, Any]:
        raise RuntimeError("device offline")

    registry = ToolRegistry(
        [
            ToolDefinition(
                name="example.silent",
                description="Complete silently on success.",
                parameters={"type": "object"},
                handler=fail,
                result_policy=ResultPolicy.SILENT_SUCCESS,
            )
        ]
    )
    client = ScriptedChatClient(
        [
            [
                StreamItem.message_delta("Trying it."),
                StreamItem.provider_tool_call(
                    _tool_call("silent-failure", name="example.silent")
                ),
            ],
            [StreamItem.message_delta("I could not apply that change.")],
        ]
    )

    outputs = await _collect_outputs(
        DebugRuntime(client, registry=registry).run(
            _policy_request(
                enabled_tools=["example.silent"],
                mode="chat_agent_tools",
            )
        )
    )

    assert client.calls == 2
    assert [
        message["content"]
        for message in outputs
        if message["type"] == "message_delta"
    ] == ["Trying it.", "I could not apply that change."]


@pytest.mark.asyncio
async def test_mixed_builtin_batch_emits_template_then_terminates_without_llm():
    calls = [
        ToolCallEvent(
            id="terminate-1",
            name="session.terminate",
            arguments={},
            raw_arguments="{}",
        ),
        ToolCallEvent(
            id="schedule-1",
            name="calendar.schedule.create",
            arguments={
                "title": "Design review",
                "start_at": "2026-07-14T09:30:00+08:00",
                "timezone": "Asia/Shanghai",
            },
            raw_arguments=(
                '{"title":"Design review","start_at":"2026-07-14T09:30:00+08:00",'
                '"timezone":"Asia/Shanghai"}'
            ),
        ),
        ToolCallEvent(
            id="search-1",
            name="knowledge.web.search",
            arguments={"query": "event batch executors"},
            raw_arguments='{"query":"event batch executors"}',
        ),
        ToolCallEvent(
            id="volume-1",
            name="device.volume.adjust",
            arguments={"mode": "absolute", "value": 30},
            raw_arguments='{"mode":"absolute","value":30}',
        ),
    ]
    client = ScriptedChatClient(
        [
            [
                StreamItem.message_delta("Goodbye, I will finish those first."),
                *[StreamItem.provider_tool_call(call) for call in calls],
            ]
        ]
    )

    outputs = await _collect_outputs(
        DebugRuntime(client).run(
            _policy_request(
                enabled_tools=[call.name for call in calls],
                mode="chat_agent_tools",
            )
        )
    )

    assert client.calls == 1
    assert [
        message["message"]["tool_call_id"]
        for message in outputs
        if message["type"] == "tool_result"
    ] == [call.id for call in calls]
    assert [
        message["content"]
        for message in outputs
        if message["type"] == "message_delta"
    ][-1] == (
        "Scheduled Design review for 2026-07-14T09:30:00+08:00 "
        "(Asia/Shanghai)."
    )
    decision = next(
        message
        for message in outputs
        if message.get("event") == "event_policy_decision"
    )
    assert decision["details"]["terminate"] is True
    assert decision["details"]["llm_follow_up"] is False
    assert any(
        message.get("event") == "terminal_completed" for message in outputs
    )
    assert outputs[-1] == {"type": "done"}


@pytest.mark.asyncio
async def test_reusable_session_terminate_completes_turn_and_session_task():
    client = ScriptedChatClient(
        [
            [
                StreamItem.message_delta("Goodbye."),
                StreamItem.provider_tool_call(
                    ToolCallEvent(
                        id="terminate-session",
                        name="session.terminate",
                        arguments={},
                        raw_arguments="{}",
                    )
                ),
            ]
        ]
    )
    runtime = DebugRuntime(client)
    queues = runtime.start_session(
        _policy_request(
            enabled_tools=["session.terminate"],
            mode="chat_agent_tools",
            session_id="reusable-session",
        )
    )
    outputs: list[dict[str, Any]] = []

    while not any(message["type"] == "done" for message in outputs):
        outputs.append(await asyncio.wait_for(queues.output.get(), timeout=1))

    await asyncio.wait_for(runtime._wait_for_tasks(), timeout=1)

    business_types = [message["type"] for message in _without_audit(outputs)]
    assert business_types[-2:] == ["turn_completed", "done"]


@pytest.mark.asyncio
async def test_dual_text_event_scope_uses_round_and_does_not_replay_stale_history():
    handled_messages: list[str] = []

    def resolve_from_history(
        event: ToolCallEvent,
        context: ToolExecutionContext,
    ) -> dict[str, Any]:
        del event
        return {
            "message": next(
                message.content
                for message in reversed(context.history)
                if message.role == "assistant" and message.content
            )
        }

    registry = ToolRegistry(
        [
            ToolDefinition(
                name="example.history",
                description="Capture the current assistant history.",
                parameters={
                    "type": "object",
                    "properties": {"message": {"type": "string"}},
                    "required": ["message"],
                },
                handler=lambda event: handled_messages.append(
                    event.arguments["message"]
                )
                or {"message": event.arguments["message"]},
                argument_resolver=resolve_from_history,
            )
        ]
    )
    repeated = _tool_call("event_1", name="example.history")
    client = ScriptedChatClient(
        [
            [StreamItem.message_delta("first history"), StreamItem.text_event(repeated)],
            [StreamItem.message_delta("second history"), StreamItem.text_event(repeated)],
            [StreamItem.message_delta("final answer")],
        ]
    )
    request = DebugRunRequest(
        user_message="capture each round",
        system_prompts=[],
        pre_messages=[],
        chat_agent=AgentParams(model="chat-model"),
        event_agent=EventAgentParams(
            enabled_tools=["example.history"],
            max_event_loops=2,
        ),
    )

    outputs = await _collect_outputs(
        DebugRuntime(client, registry=registry).run(request)
    )

    assert handled_messages == ["first history", "second history"]
    scopes = [
        message["details"]["scope"]
        for message in outputs
        if message.get("event") == "event_batch_started"
    ]
    assert scopes[0].endswith(":round-1")
    assert scopes[1].endswith(":round-2")


@pytest.mark.asyncio
@pytest.mark.parametrize("mode", ["dual_agent", "chat_agent_tools"])
async def test_runtime_releases_batch_scope_after_results(monkeypatch, mode: str):
    released: list[Any] = []
    original_release = EventBatchExecutor.release_scope

    def record_release(self: EventBatchExecutor, scope: Any) -> None:
        released.append(scope)
        original_release(self, scope)

    monkeypatch.setattr(EventBatchExecutor, "release_scope", record_release)
    registry = ToolRegistry(
        [
            ToolDefinition(
                name="example.silent",
                description="Complete without another model call.",
                parameters={"type": "object"},
                handler=lambda event: {"event_id": event.id},
                result_policy=ResultPolicy.SILENT_SUCCESS,
            )
        ]
    )
    event = _tool_call("release-1", name="example.silent")
    item = (
        StreamItem.text_event(event)
        if mode == "dual_agent"
        else StreamItem.provider_tool_call(event)
    )
    client = ScriptedChatClient([[item]])

    await _collect_outputs(
        DebugRuntime(client, registry=registry).run(
            _policy_request(enabled_tools=["example.silent"], mode=mode)
        )
    )

    assert len(released) == 1
    assert str(released[0]).endswith(":round-1")


@pytest.mark.asyncio
@pytest.mark.parametrize("mode", ["dual_agent", "chat_agent_tools"])
async def test_tool_only_terminate_emits_plugin_farewell_once(mode: str):
    event = ToolCallEvent(
        id="terminate-only",
        name="session.terminate",
        arguments={},
        raw_arguments="{}",
    )
    item = (
        StreamItem.text_event(event)
        if mode == "dual_agent"
        else StreamItem.provider_tool_call(event)
    )
    client = ScriptedChatClient([[item]])

    outputs = await _collect_outputs(
        DebugRuntime(client).run(
            _policy_request(enabled_tools=["session.terminate"], mode=mode)
        )
    )

    assert client.calls == 1
    assert [
        message["content"]
        for message in outputs
        if message["type"] == "message_delta"
    ] == ["Goodbye."]
    assert outputs[-1] == {"type": "done"}


@pytest.mark.asyncio
@pytest.mark.parametrize("mode", ["dual_agent", "chat_agent_tools"])
async def test_coalesced_terminate_has_one_farewell_and_one_audit_result(mode: str):
    events = [
        ToolCallEvent(
            id=event_id,
            name="session.terminate",
            arguments={},
            raw_arguments="{}",
        )
        for event_id in ("terminate-primary", "terminate-duplicate")
    ]
    items = [
        (
            StreamItem.text_event(event)
            if mode == "dual_agent"
            else StreamItem.provider_tool_call(event)
        )
        for event in events
    ]
    client = ScriptedChatClient([items])

    outputs = await _collect_outputs(
        DebugRuntime(client).run(
            _policy_request(enabled_tools=["session.terminate"], mode=mode)
        )
    )

    assert [
        message["content"]
        for message in outputs
        if message["type"] == "message_delta"
    ] == ["Goodbye."]
    assert [
        message["message"]["tool_call_id"]
        for message in outputs
        if message["type"] == "tool_result"
    ] == [event.id for event in events]
    batch_audit = next(
        message
        for message in outputs
        if message.get("event") == "event_batch_results"
    )
    terminal_audit = next(
        message
        for message in outputs
        if message.get("event") == "terminal_completed"
    )
    completion_audit = next(
        message
        for message in outputs
        if message.get("event")
        == (
            "provider_tools_completed"
            if mode == "chat_agent_tools"
            else "event_agent_completed"
        )
    )
    assert [
        result["event_id"] for result in batch_audit["details"]["results"]
    ] == ["terminate-primary"]
    assert completion_audit["details"]["result_count"] == 1
    assert terminal_audit["details"]["event_ids"] == ["terminate-primary"]


@pytest.mark.asyncio
async def test_tool_only_terminate_ends_reusable_session_after_plugin_farewell():
    client = ScriptedChatClient(
        [
            [
                StreamItem.provider_tool_call(
                    ToolCallEvent(
                        id="terminate-session-only",
                        name="session.terminate",
                        arguments={},
                        raw_arguments="{}",
                    )
                )
            ]
        ]
    )
    runtime = DebugRuntime(client)
    queues = runtime.start_session(
        _policy_request(
            enabled_tools=["session.terminate"],
            mode="chat_agent_tools",
            session_id="terminal-only-session",
        )
    )
    outputs: list[dict[str, Any]] = []

    while not any(message["type"] == "done" for message in outputs):
        outputs.append(await asyncio.wait_for(queues.output.get(), timeout=1))
    await asyncio.wait_for(runtime._wait_for_tasks(), timeout=1)

    assert [
        message["content"]
        for message in outputs
        if message["type"] == "message_delta"
    ] == ["Goodbye."]
    assert [message["type"] for message in _without_audit(outputs)][-2:] == [
        "turn_completed",
        "done",
    ]


@pytest.mark.asyncio
@pytest.mark.parametrize("mode", ["dual_agent", "chat_agent_tools"])
async def test_web_search_has_first_answer_tool_result_and_one_second_answer(mode: str):
    search = ToolCallEvent(
        id="search-two-answer",
        name="knowledge.web.search",
        arguments={"query": "event batch safety"},
        raw_arguments='{"query":"event batch safety"}',
    )
    item = (
        StreamItem.text_event(search)
        if mode == "dual_agent"
        else StreamItem.provider_tool_call(search)
    )
    client = ScriptedChatClient(
        [
            [StreamItem.message_delta("I will check."), item],
            [StreamItem.message_delta("Grounded search update.")],
        ]
    )

    outputs = await _collect_outputs(
        DebugRuntime(client).run(
            _policy_request(enabled_tools=["knowledge.web.search"], mode=mode)
        )
    )

    business = _without_audit(outputs)
    visible = [
        (index, message["content"])
        for index, message in enumerate(business)
        if message["type"] == "message_delta"
    ]
    tool_index = next(
        index for index, message in enumerate(business) if message["type"] == "tool_result"
    )
    assert client.calls == 2
    assert [content for _, content in visible] == [
        "I will check.",
        "Grounded search update.",
    ]
    assert visible[0][0] < tool_index < visible[1][0]
    assert "sources" in business[tool_index]["message"]["content"]