test: enforce first reply and protocol parity

Problem:
The comparison matrix did not require a visible first reply before event execution or prove that dual and direct modes used only their intended protocol surfaces.

Risk:
Regressions could delay user-visible responses, duplicate terminal farewells, change model-call counts, or leak the wrong event/tool protocol without being caught.

tests/test_tool_invocation_comparison.py

@@ -47,15 +47,22 @@ class ExpectedSemantics:
     terminal_count: int = 0
 
 
+@dataclass(frozen=True)
+class RecordedChatRequest:
+    messages: tuple[ChatMessage, ...]
+    tools: tuple[dict[str, Any], ...]
+    tool_choice: dict[str, Any] | None
+
+
 EXPECTED = {
     "ordinary_chat": ExpectedSemantics(("Ordinary answer.",), (), 1),
     "device_volume_silent": ExpectedSemantics(
-        (),
+        ("I will set the volume to 30.",),
         ("device.volume.adjust",),
         1,
     ),
     "calendar_schedule_template": ExpectedSemantics(
-        (SCHEDULE_CONFIRMATION,),
+        ("I will schedule that.", SCHEDULE_CONFIRMATION),
         ("calendar.schedule.create",),
         1,
     ),
@@ -71,7 +78,10 @@ EXPECTED = {
         terminal_count=1,
     ),
     "parallel_volume_schedule": ExpectedSemantics(
-        (SCHEDULE_CONFIRMATION,),
+        (
+            "I will set the volume to 30 and schedule that.",
+            SCHEDULE_CONFIRMATION,
+        ),
         ("device.volume.adjust", "calendar.schedule.create"),
         1,
     ),
@@ -82,6 +92,7 @@ class ScriptedChatClient:
     def __init__(self, rounds: list[list[StreamItem]]) -> None:
         self.rounds = rounds
         self.calls = 0
+        self.requests: list[RecordedChatRequest] = []
 
     async def stream_chat(
         self,
@@ -90,7 +101,14 @@ class ScriptedChatClient:
         params: AgentParams,
         tool_choice: dict[str, Any] | None = None,
     ) -> AsyncIterator[StreamItem]:
-        del messages, tools, params, tool_choice
+        del params
+        self.requests.append(
+            RecordedChatRequest(
+                messages=tuple(messages),
+                tools=tuple(tools),
+                tool_choice=dict(tool_choice) if tool_choice is not None else None,
+            )
+        )
         items = self.rounds[self.calls]
         self.calls += 1
         for item in items:
@@ -246,6 +264,7 @@ def _scenario_setup(
         enabled_tools = ["device.volume.adjust"]
         rounds = [
             [
+                StreamItem.message_delta("I will set the volume to 30."),
                 _event_item(
                     mode,
                     _event(
@@ -264,6 +283,7 @@ def _scenario_setup(
         enabled_tools = ["calendar.schedule.create"]
         rounds = [
             [
+                StreamItem.message_delta("I will schedule that."),
                 _event_item(
                     mode,
                     _event(
@@ -296,6 +316,7 @@ def _scenario_setup(
         enabled_tools = ["session.terminate"]
         rounds = [
             [
+                StreamItem.message_delta("Goodbye."),
                 _event_item(
                     mode,
                     _event("terminate-1", "session.terminate", {}),
@@ -310,6 +331,9 @@ def _scenario_setup(
         enabled_tools = ["device.volume.adjust", "calendar.schedule.create"]
         rounds = [
             [
+                StreamItem.message_delta(
+                    "I will set the volume to 30 and schedule that."
+                ),
                 _event_item(
                     mode,
                     _event(
@@ -368,6 +392,32 @@ def _tool_payloads(outputs: list[dict[str, Any]]) -> list[dict[str, Any]]:
     ]
 
 
+def _generated_event_catalog_names(
+    messages: tuple[ChatMessage, ...],
+) -> tuple[str, ...]:
+    catalogs = [
+        message.content
+        for message in messages
+        if message.role == "system"
+        and "Available events:" in message.content
+        and "<agent_events>" in message.content
+    ]
+    if not catalogs:
+        return ()
+    assert len(catalogs) == 1
+    return tuple(
+        line.removeprefix("- ").split(":", 1)[0]
+        for line in catalogs[0].splitlines()
+        if line.startswith("- ")
+    )
+
+
+def _provider_tool_names(
+    tools: tuple[dict[str, Any], ...],
+) -> tuple[str, ...]:
+    return tuple(tool["function"]["name"] for tool in tools)
+
+
 def _assert_port_calls(scenario: str, ports: RecordingPorts) -> None:
     expected_session: list[tuple[str, str | None]] = []
     expected_volume: list[tuple[str, str, int | None, int | None]] = []
@@ -462,13 +512,48 @@ async def test_invocation_modes_share_business_semantics(
     assert visible_messages == expected.visible_messages
     assert tuple(payload["tool"] for payload in tool_payloads) == expected.tool_names
     assert client.calls == expected.model_calls
+    assert len(client.requests) == expected.model_calls
     assert terminal_count == expected.terminal_count
     assert outputs[-1] == {"type": "done"}
     _assert_port_calls(scenario, ports)
 
+    initial_request = client.requests[0]
+    enabled_names = tuple(request.event_agent.enabled_tools)
+    assert initial_request.tool_choice is None
+    if mode == "dual_agent":
+        assert initial_request.tools == ()
+        assert _generated_event_catalog_names(initial_request.messages) == enabled_names
+    else:
+        assert _provider_tool_names(initial_request.tools) == enabled_names
+        assert _generated_event_catalog_names(initial_request.messages) == ()
+
+    if expected.tool_names:
+        business = [message for message in outputs if message["type"] != "audit"]
+        first_message_index = next(
+            index
+            for index, message in enumerate(business)
+            if message["type"] == "message_delta"
+        )
+        first_tool_result_index = next(
+            index
+            for index, message in enumerate(business)
+            if message["type"] == "tool_result"
+        )
+        assert first_message_index < first_tool_result_index
+        detected_events = [
+            message
+            for message in outputs
+            if message.get("event") == "chat_event_detected"
+        ]
+        expected_source = (
+            "text_event" if mode == "dual_agent" else "provider_resolved"
+        )
+        assert tuple(
+            message["details"]["event_source"] for message in detected_events
+        ) == (expected_source,) * len(expected.tool_names)
+
     if scenario == "web_search_two_answers":
         assert tool_payloads[0]["sources"][0]["title"] == "Matrix source"
-        business = [message for message in outputs if message["type"] != "audit"]
         visible_indexes = [
             index
             for index, message in enumerate(business)