feat: run isolated agent benchmarks

Problem: The benchmark catalog lacked an isolated runtime runner, typed per-call timing, deterministic zero-network clients, and semantic results sourced from audit and usage ledgers.

Risk: Provider TTFT, visible TTFT, handler latency, and turn wall time can be conflated, while stream or semantic failures could stop later runs or leak session state. Focused and full-suite tests cover timing separation, cleanup, ordering, isolation, continuation, and event/fallback/tool metrics.

src/agent_lab/application/benchmark.py

@@ -1,10 +1,13 @@
+import asyncio
 import json
+import time
 import unicodedata
-from collections.abc import Mapping
+from collections.abc import AsyncIterator, Callable, Mapping
 from enum import StrEnum
 from types import MappingProxyType
-from typing import Annotated, Literal, TypeAlias
+from typing import Annotated, Any, Literal, TypeAlias
 from urllib.parse import urlsplit
+from uuid import uuid4
 
 import httpx
 from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
@@ -14,8 +17,13 @@ from agent_lab.application.contracts import (
     DebugRunRequest,
     EventAgentParams,
 )
+from agent_lab.application.runtime import DebugRuntime
+from agent_lab.application.queues import RuntimeQueues
+from agent_lab.application.tools import build_default_tool_registry
 from agent_lab.domain.events import ToolCallEvent
-from agent_lab.domain.messages import StreamItem
+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
 
 
 class BenchmarkMode(StrEnum):
@@ -132,6 +140,115 @@ class BenchmarkConfig(BenchmarkTarget):
         return values
 
 
+class BenchmarkModelCallTiming(_StrictBenchmarkModel):
+    call_index: int = Field(ge=1)
+    call_kind: Literal["chat_completion", "argument_fallback"]
+    first_item_kind: str | None
+    provider_ttft_ms: int | None = Field(ge=0)
+    visible_ttft_ms: int | None = Field(ge=0)
+    elapsed_ms: int | None = Field(ge=0)
+    usage: TokenUsage | None
+
+
+class BenchmarkRunResult(_StrictBenchmarkModel):
+    case_id: BenchmarkCaseId
+    mode: BenchmarkMode
+    iteration: int = Field(ge=1)
+    status: Literal["passed", "failed"]
+    initial_provider_ttft_ms: int | None = Field(ge=0)
+    visible_ttft_ms: int | None = Field(ge=0)
+    turn_wall_time_ms: int | None = Field(ge=0)
+    prompt_tokens: int | None = Field(ge=0)
+    completion_tokens: int | None = Field(ge=0)
+    total_tokens: int | None = Field(ge=0)
+    cached_tokens: int | None = Field(ge=0)
+    model_call_count: int | None = Field(ge=0)
+    fallback_count: int | None = Field(ge=0)
+    tool_count: int | None = Field(ge=0)
+    event_names: list[str]
+    event_sources: list[str]
+    tool_statuses: list[str]
+    tool_latencies_ms: list[int | None]
+    semantic_failures: list[str]
+    error: str | None
+    model_calls: list[BenchmarkModelCallTiming]
+
+
+class TimingChatClient:
+    def __init__(
+        self,
+        inner: object,
+        *,
+        clock: Callable[[], float] = time.perf_counter,
+    ) -> None:
+        self.inner = inner
+        self.clock = clock
+        self.timings: list[BenchmarkModelCallTiming] = []
+        self._next_call_index = 0
+
+    async def stream_chat(
+        self,
+        messages: list[ChatMessage],
+        tools: list[dict[str, Any]],
+        params: AgentParams,
+        tool_choice: dict[str, Any] | None = None,
+    ) -> AsyncIterator[StreamItem]:
+        self._next_call_index += 1
+        call_index = self._next_call_index
+        started_at = self.clock()
+        first_item_at: float | None = None
+        first_item_kind: str | None = None
+        first_message_delta_at: float | None = None
+        usage: TokenUsage | None = None
+        try:
+            stream_chat = getattr(self.inner, "stream_chat")
+            async for item in stream_chat(
+                messages=messages,
+                tools=tools,
+                params=params,
+                tool_choice=tool_choice,
+            ):
+                item_at = self.clock()
+                if first_item_at is None:
+                    first_item_at = item_at
+                    first_item_kind = item.kind
+                if item.kind == "message_delta" and first_message_delta_at is None:
+                    first_message_delta_at = item_at
+                if item.kind == "usage" and item.usage is not None:
+                    usage = item.usage
+                yield item
+        finally:
+            timing = BenchmarkModelCallTiming(
+                call_index=call_index,
+                call_kind=(
+                    "argument_fallback"
+                    if tool_choice is not None
+                    else "chat_completion"
+                ),
+                first_item_kind=first_item_kind,
+                provider_ttft_ms=self._elapsed_ms(started_at, first_item_at),
+                visible_ttft_ms=self._elapsed_ms(
+                    started_at,
+                    first_message_delta_at,
+                ),
+                elapsed_ms=self._elapsed_ms(started_at, self.clock()),
+                usage=usage,
+            )
+            self.timings.append(timing)
+            self.timings.sort(key=lambda item: item.call_index)
+
+    async def aclose(self) -> None:
+        close = getattr(self.inner, "aclose", None)
+        if close is not None:
+            await close()
+
+    @staticmethod
+    def _elapsed_ms(started_at: float, ended_at: float | None) -> int | None:
+        if ended_at is None:
+            return None
+        return max(0, round((ended_at - started_at) * 1000))
+
+
 class BenchmarkExpectation(_FrozenStrictBenchmarkModel):
     visible_messages: tuple[str, ...]
     event_names: tuple[str, ...]
@@ -419,3 +536,446 @@ def build_mock_rounds(
         for message in case.mock_visible_messages[1:]
     )
     return rounds
+
+
+class MockBenchmarkChatClient:
+    def __init__(
+        self,
+        case: BenchmarkCase,
+        mode: BenchmarkMode | str,
+    ) -> None:
+        self.case = case
+        self.mode = _normalize_benchmark_mode(mode)
+        self.rounds = build_mock_rounds(case, self.mode)
+        self.calls = 0
+
+    async def stream_chat(
+        self,
+        messages: list[ChatMessage],
+        tools: list[dict[str, Any]],
+        params: AgentParams,
+        tool_choice: dict[str, Any] | None = None,
+    ) -> AsyncIterator[StreamItem]:
+        del messages, tools, params, tool_choice
+        call_index = self.calls
+        if call_index >= len(self.rounds):
+            raise RuntimeError(
+                f"mock benchmark stream exhausted: {self.case.case_id}"
+            )
+        self.calls += 1
+        yield StreamItem.raw_response_chunk(
+            {
+                "mock": True,
+                "case_id": self.case.case_id.value,
+                "mode": self.mode.value,
+                "call_index": self.calls,
+            }
+        )
+        for item in self.rounds[call_index]:
+            yield item
+        yield StreamItem.usage_item(
+            TokenUsage(
+                prompt_tokens=10,
+                completion_tokens=5,
+                total_tokens=15,
+                cached_tokens=2,
+            )
+        )
+
+    async def aclose(self) -> None:
+        return None
+
+
+BenchmarkClientFactory: TypeAlias = Callable[[BenchmarkConfig, str], object]
+
+
+class _BenchmarkOutputQueue(asyncio.Queue[dict[str, Any]]):
+    def __init__(self, clock: Callable[[], float]) -> None:
+        super().__init__()
+        self.clock = clock
+        self.first_message_delta_at: float | None = None
+
+    async def put(self, item: dict[str, Any]) -> None:
+        if (
+            item.get("type") == "message_delta"
+            and self.first_message_delta_at is None
+        ):
+            self.first_message_delta_at = self.clock()
+        await super().put(item)
+
+
+class BenchmarkRunner:
+    def __init__(
+        self,
+        config: BenchmarkConfig,
+        api_key: str | None,
+        mock: bool = False,
+        client_factory: BenchmarkClientFactory | None = None,
+        clock: Callable[[], float] = time.perf_counter,
+    ) -> None:
+        if not mock and not (api_key or "").strip():
+            raise ValueError("api_key is required for live benchmark runs")
+        self.config = config
+        self.api_key = (api_key or "").strip()
+        self.mock = mock
+        self.client_factory = client_factory or self._default_client_factory
+        self.clock = clock
+
+    async def run(self) -> list[BenchmarkRunResult]:
+        results: list[BenchmarkRunResult] = []
+        for case_id in self.config.cases:
+            case = BENCHMARK_CASE_CATALOG[case_id]
+            for mode in self.config.modes:
+                for iteration in range(1, self.config.runs_per_case + 1):
+                    results.append(await self._run_one(case, mode, iteration))
+        return results
+
+    async def _run_one(
+        self,
+        case: BenchmarkCase,
+        mode: BenchmarkMode,
+        iteration: int,
+    ) -> BenchmarkRunResult:
+        store = SQLiteSessionStore(":memory:")
+        runtime: DebugRuntime | None = None
+        timing_client: TimingChatClient | None = None
+        result: BenchmarkRunResult | None = None
+        close_error: Exception | None = None
+        try:
+            inner_client = (
+                MockBenchmarkChatClient(case, mode)
+                if self.mock
+                else self.client_factory(self.config, self.api_key)
+            )
+            timing_client = TimingChatClient(inner_client, clock=self.clock)
+            session_id = f"benchmark-{uuid4().hex}"
+            request = build_benchmark_request(
+                case,
+                mode,
+                self.config.model,
+            ).model_copy(update={"session_id": session_id})
+            output_queue = _BenchmarkOutputQueue(self.clock)
+            runtime = DebugRuntime(
+                timing_client,
+                queues=RuntimeQueues(output=output_queue),
+                registry=build_default_tool_registry(),
+                session_store=store,
+                clock=self.clock,
+            )
+            turn_started_at = self.clock()
+            outputs, visible_ttft_ms, runtime_error = await self._consume_turn(
+                runtime,
+                request,
+                turn_started_at,
+                output_queue,
+            )
+            audits = store.list_audit_logs(session_id)
+            usage = store.usage_summary(session_id)
+            result = self._build_result(
+                case=case,
+                mode=mode,
+                iteration=iteration,
+                outputs=outputs,
+                audits=audits,
+                usage=usage,
+                timings=timing_client.timings,
+                visible_ttft_ms=visible_ttft_ms,
+                runtime_error=runtime_error,
+            )
+        except Exception as exc:
+            result = self._failed_result(
+                case.case_id,
+                mode,
+                iteration,
+                error=str(exc),
+                timings=timing_client.timings if timing_client is not None else [],
+            )
+        finally:
+            if runtime is not None:
+                try:
+                    await runtime.aclose()
+                except Exception as exc:
+                    close_error = exc
+            if timing_client is not None:
+                try:
+                    await timing_client.aclose()
+                except Exception as exc:
+                    close_error = close_error or exc
+
+        assert result is not None
+        if close_error is not None:
+            error = str(close_error)
+            if result.error:
+                error = f"{result.error}; close failed: {error}"
+            result = result.model_copy(update={"status": "failed", "error": error})
+        return result
+
+    async def _consume_turn(
+        self,
+        runtime: DebugRuntime,
+        request: DebugRunRequest,
+        turn_started_at: float,
+        output_queue: _BenchmarkOutputQueue,
+    ) -> tuple[list[dict[str, Any]], int | None, str | None]:
+        queues = runtime.start_session(request)
+        outputs: list[dict[str, Any]] = []
+        visible_ttft_ms: int | None = None
+        runtime_error: str | None = None
+        while True:
+            message = await queues.output.get()
+            outputs.append(message)
+            message_type = message.get("type")
+            if message_type == "error":
+                runtime_error = str(message.get("message") or "runtime error")
+                break
+            if message_type == "turn_completed":
+                break
+        if output_queue.first_message_delta_at is not None:
+            visible_ttft_ms = max(
+                0,
+                round(
+                    (output_queue.first_message_delta_at - turn_started_at) * 1000
+                ),
+            )
+        return outputs, visible_ttft_ms, runtime_error
+
+    def _build_result(
+        self,
+        *,
+        case: BenchmarkCase,
+        mode: BenchmarkMode,
+        iteration: int,
+        outputs: list[dict[str, Any]],
+        audits: list[dict[str, Any]],
+        usage: dict[str, Any],
+        timings: list[BenchmarkModelCallTiming],
+        visible_ttft_ms: int | None,
+        runtime_error: str | None,
+    ) -> BenchmarkRunResult:
+        session_usage = usage["session"]
+        calls = usage["calls"]
+        event_audits = [
+            audit for audit in audits if audit["event"] == "chat_event_detected"
+        ]
+        batch_audits = [
+            audit for audit in audits if audit["event"] == "event_batch_results"
+        ]
+        batch_results = [
+            item
+            for audit in batch_audits
+            for item in audit["details"].get("results", [])
+        ]
+        event_names = [audit["details"]["event_name"] for audit in event_audits]
+        event_sources = [audit["details"]["event_source"] for audit in event_audits]
+        tool_statuses = [str(item["status"]) for item in batch_results]
+        tool_calls = [call for call in calls if call["call_kind"] == "tool_execution"]
+        model_call_count = sum(
+            call["call_kind"] == "chat_completion" for call in calls
+        )
+        fallback_count = int(session_usage["fallback_count"])
+        tool_count = int(session_usage["tool_count"])
+        semantic_failures = self._semantic_failures(
+            case=case,
+            mode=mode,
+            outputs=outputs,
+            audits=audits,
+            event_names=event_names,
+            event_sources=event_sources,
+            tool_statuses=tool_statuses,
+            model_call_count=model_call_count,
+            fallback_count=fallback_count,
+            tool_count=tool_count,
+            timings=timings,
+            check_mock_text=self.mock,
+        )
+        initial_call = next(
+            (timing for timing in timings if timing.call_kind == "chat_completion"),
+            None,
+        )
+        status = "failed" if runtime_error or semantic_failures else "passed"
+        return BenchmarkRunResult(
+            case_id=case.case_id,
+            mode=mode,
+            iteration=iteration,
+            status=status,
+            initial_provider_ttft_ms=(
+                initial_call.provider_ttft_ms if initial_call is not None else None
+            ),
+            visible_ttft_ms=visible_ttft_ms,
+            turn_wall_time_ms=session_usage["turn_wall_time_ms"],
+            prompt_tokens=int(session_usage["prompt_tokens"]),
+            completion_tokens=int(session_usage["completion_tokens"]),
+            total_tokens=int(session_usage["total_tokens"]),
+            cached_tokens=int(session_usage["cached_tokens"]),
+            model_call_count=model_call_count,
+            fallback_count=fallback_count,
+            tool_count=tool_count,
+            event_names=event_names,
+            event_sources=event_sources,
+            tool_statuses=tool_statuses,
+            tool_latencies_ms=[call["tool_latency_ms"] for call in tool_calls],
+            semantic_failures=semantic_failures,
+            error=runtime_error,
+            model_calls=list(timings),
+        )
+
+    def _semantic_failures(
+        self,
+        *,
+        case: BenchmarkCase,
+        mode: BenchmarkMode,
+        outputs: list[dict[str, Any]],
+        audits: list[dict[str, Any]],
+        event_names: list[str],
+        event_sources: list[str],
+        tool_statuses: list[str],
+        model_call_count: int,
+        fallback_count: int,
+        tool_count: int,
+        timings: list[BenchmarkModelCallTiming],
+        check_mock_text: bool,
+    ) -> list[str]:
+        failures: list[str] = []
+        answers, first_answer_index, first_tool_index = self._visible_answers(outputs)
+        expected = case.expectation
+        if len(answers) != expected.answer_count:
+            failures.append(
+                f"answer_count expected {expected.answer_count}, got {len(answers)}"
+            )
+        if any(not answer.strip() for answer in answers):
+            failures.append("visible answers must be non-empty")
+        if check_mock_text and tuple(answers) != expected.visible_messages:
+            failures.append(
+                f"mock visible messages expected {expected.visible_messages!r}, "
+                f"got {tuple(answers)!r}"
+            )
+        if expected.first_reply_before_tool and not (
+            first_answer_index is not None
+            and first_tool_index is not None
+            and first_answer_index < first_tool_index
+        ):
+            failures.append("first visible answer must precede tool results")
+        if tuple(event_names) != expected.event_names:
+            failures.append(
+                f"event names expected {expected.event_names!r}, got {tuple(event_names)!r}"
+            )
+        expected_source = (
+            "text_event"
+            if mode is BenchmarkMode.DUAL_AGENT
+            else "provider_resolved"
+        )
+        expected_sources = (expected_source,) * len(expected.event_names)
+        if tuple(event_sources) != expected_sources:
+            failures.append(
+                f"event sources expected {expected_sources!r}, "
+                f"got {tuple(event_sources)!r}"
+            )
+        expected_statuses = ("success",) * len(expected.event_names)
+        if tuple(tool_statuses) != expected_statuses:
+            failures.append(
+                f"tool statuses expected {expected_statuses!r}, "
+                f"got {tuple(tool_statuses)!r}"
+            )
+        if tool_count != len(expected.event_names):
+            failures.append(
+                f"tool_count expected {len(expected.event_names)}, got {tool_count}"
+            )
+        timing_chat_count = sum(
+            timing.call_kind == "chat_completion" for timing in timings
+        )
+        timing_fallback_count = sum(
+            timing.call_kind == "argument_fallback" for timing in timings
+        )
+        if model_call_count != timing_chat_count:
+            failures.append(
+                f"model_call_count ledger={model_call_count}, timing={timing_chat_count}"
+            )
+        if fallback_count != timing_fallback_count:
+            failures.append(
+                f"fallback_count ledger={fallback_count}, timing={timing_fallback_count}"
+            )
+        terminal = any(audit["event"] == "terminal_completed" for audit in audits)
+        if terminal is not expected.terminal:
+            failures.append(
+                f"terminal expected {expected.terminal}, got {terminal}"
+            )
+        return failures
+
+    @staticmethod
+    def _visible_answers(
+        outputs: list[dict[str, Any]],
+    ) -> tuple[list[str], int | None, int | None]:
+        answers: list[str] = []
+        current: list[str] = []
+        first_answer_index: int | None = None
+        first_tool_index: int | None = None
+        for index, message in enumerate(outputs):
+            message_type = message.get("type")
+            if message_type == "message_delta":
+                content = str(message.get("content") or "")
+                if first_answer_index is None and content.strip():
+                    first_answer_index = index
+                current.append(content)
+                continue
+            if message_type == "tool_result":
+                if first_tool_index is None:
+                    first_tool_index = index
+                answer = "".join(current).strip()
+                if answer:
+                    answers.append(answer)
+                current = []
+        answer = "".join(current).strip()
+        if answer:
+            answers.append(answer)
+        return answers, first_answer_index, first_tool_index
+
+    @staticmethod
+    def _failed_result(
+        case_id: BenchmarkCaseId,
+        mode: BenchmarkMode,
+        iteration: int,
+        *,
+        error: str,
+        timings: list[BenchmarkModelCallTiming],
+    ) -> BenchmarkRunResult:
+        initial_call = next(
+            (timing for timing in timings if timing.call_kind == "chat_completion"),
+            None,
+        )
+        return BenchmarkRunResult(
+            case_id=case_id,
+            mode=mode,
+            iteration=iteration,
+            status="failed",
+            initial_provider_ttft_ms=(
+                initial_call.provider_ttft_ms if initial_call is not None else None
+            ),
+            visible_ttft_ms=None,
+            turn_wall_time_ms=None,
+            prompt_tokens=None,
+            completion_tokens=None,
+            total_tokens=None,
+            cached_tokens=None,
+            model_call_count=None,
+            fallback_count=None,
+            tool_count=None,
+            event_names=[],
+            event_sources=[],
+            tool_statuses=[],
+            tool_latencies_ms=[],
+            semantic_failures=[],
+            error=error,
+            model_calls=list(timings),
+        )
+
+    @staticmethod
+    def _default_client_factory(
+        config: BenchmarkConfig,
+        api_key: str,
+    ) -> OpenAICompatibleChatClient:
+        return OpenAICompatibleChatClient(
+            api_key=api_key,
+            base_url=config.base_url,
+            default_model=config.model,
+            request_timeout_seconds=60.0,
+        )

tests/test_benchmark_runner.py

@@ -0,0 +1,696 @@
+import asyncio
+from collections.abc import AsyncIterator
+from typing import Any
+
+import pytest
+from pydantic import ValidationError
+
+from agent_lab.application import benchmark
+from agent_lab.application.contracts import AgentParams
+from agent_lab.application.events import ResultPolicy
+from agent_lab.application.tools import ToolDefinition, ToolRegistry
+from agent_lab.domain.events import ToolCallEvent
+from agent_lab.domain.messages import ChatMessage, StreamItem
+from agent_lab.domain.messages import TokenUsage
+from agent_lab.infrastructure.sqlite_store import SQLiteSessionStore
+
+
+class ManualClock:
+    def __init__(self) -> None:
+        self.now = 0.0
+
+    def __call__(self) -> float:
+        return self.now
+
+
+class ScriptedTimingClient:
+    def __init__(self, clock: ManualClock, *, fail: bool = False) -> None:
+        self.clock = clock
+        self.fail = fail
+        self.closed = False
+
+    async def stream_chat(
+        self,
+        messages: list[ChatMessage],
+        tools: list[dict[str, Any]],
+        params: AgentParams,
+        tool_choice: dict[str, Any] | None = None,
+    ) -> AsyncIterator[StreamItem]:
+        del messages, tools, params, tool_choice
+        self.clock.now = 0.011
+        yield StreamItem.raw_response_chunk({"provider": "first"})
+        if self.fail:
+            self.clock.now = 0.023
+            raise RuntimeError("stream failed")
+        self.clock.now = 0.037
+        yield StreamItem.message_delta("visible")
+        self.clock.now = 0.041
+        yield StreamItem.usage_item(TokenUsage(total_tokens=3))
+        self.clock.now = 0.059
+        yield StreamItem.usage_item(TokenUsage(total_tokens=5))
+        self.clock.now = 0.071
+
+    async def aclose(self) -> None:
+        self.closed = True
+
+
+class FallbackBenchmarkClient:
+    def __init__(self) -> None:
+        self.calls = 0
+        self.closed = False
+
+    async def stream_chat(
+        self,
+        messages: list[ChatMessage],
+        tools: list[dict[str, Any]],
+        params: AgentParams,
+        tool_choice: dict[str, Any] | None = None,
+    ) -> AsyncIterator[StreamItem]:
+        del messages, tools, params
+        self.calls += 1
+        if self.calls == 1:
+            assert tool_choice is None
+            yield StreamItem.raw_response_chunk({"chat": 1})
+            yield StreamItem.message_delta("I will set the volume to 30.")
+            yield StreamItem.text_event(
+                ToolCallEvent(
+                    id="volume-1",
+                    name="device.volume.adjust",
+                    arguments={"mode": "absolute"},
+                    raw_arguments='{"mode":"absolute"}',
+                )
+            )
+            yield StreamItem.usage_item(TokenUsage(total_tokens=5))
+            return
+        if self.calls == 2:
+            assert tool_choice is not None
+            yield StreamItem.raw_response_chunk({"fallback": 1})
+            yield StreamItem.provider_tool_call(
+                ToolCallEvent(
+                    id="fallback-volume",
+                    name="device.volume.adjust",
+                    arguments={"mode": "absolute", "value": 30},
+                    raw_arguments='{"mode":"absolute","value":30}',
+                )
+            )
+            yield StreamItem.usage_item(TokenUsage(total_tokens=7))
+            return
+        raise RuntimeError("unexpected fallback client call")
+
+    async def aclose(self) -> None:
+        self.closed = True
+
+
+class RecordingSQLiteSessionStore(SQLiteSessionStore):
+    instances: list["RecordingSQLiteSessionStore"] = []
+
+    def __init__(self, database_path: str) -> None:
+        super().__init__(database_path)
+        self.instances.append(self)
+
+
+class BlockingTimingClient:
+    def __init__(self, clock: ManualClock) -> None:
+        self.clock = clock
+        self.blocked = asyncio.Event()
+        self.closed = False
+
+    async def stream_chat(
+        self,
+        messages: list[ChatMessage],
+        tools: list[dict[str, Any]],
+        params: AgentParams,
+        tool_choice: dict[str, Any] | None = None,
+    ) -> AsyncIterator[StreamItem]:
+        del messages, tools, params, tool_choice
+        self.clock.now = 0.010
+        yield StreamItem.raw_response_chunk({"first": True})
+        self.blocked.set()
+        await asyncio.Event().wait()
+
+    async def aclose(self) -> None:
+        self.closed = True
+
+
+class NoAnswerClient:
+    def __init__(self) -> None:
+        self.closed = False
+
+    async def stream_chat(
+        self,
+        messages: list[ChatMessage],
+        tools: list[dict[str, Any]],
+        params: AgentParams,
+        tool_choice: dict[str, Any] | None = None,
+    ) -> AsyncIterator[StreamItem]:
+        del messages, tools, params, tool_choice
+        yield StreamItem.raw_response_chunk({"no_answer": True})
+        yield StreamItem.usage_item(TokenUsage(total_tokens=1))
+
+    async def aclose(self) -> None:
+        self.closed = True
+
+
+class ConcurrentTimingClient:
+    def __init__(self) -> None:
+        self.calls = 0
+        self.first_started = asyncio.Event()
+        self.release_first = asyncio.Event()
+
+    async def stream_chat(
+        self,
+        messages: list[ChatMessage],
+        tools: list[dict[str, Any]],
+        params: AgentParams,
+        tool_choice: dict[str, Any] | None = None,
+    ) -> AsyncIterator[StreamItem]:
+        del messages, tools, params, tool_choice
+        self.calls += 1
+        call_number = self.calls
+        yield StreamItem.raw_response_chunk({"call": call_number})
+        if call_number == 1:
+            self.first_started.set()
+            await self.release_first.wait()
+        yield StreamItem.usage_item(TokenUsage(total_tokens=call_number))
+
+    async def aclose(self) -> None:
+        return None
+
+
+def test_benchmark_result_models_are_strict_and_allow_nullable_metrics():
+    timing_type = getattr(benchmark, "BenchmarkModelCallTiming")
+    result_type = getattr(benchmark, "BenchmarkRunResult")
+
+    timing = timing_type(
+        call_index=1,
+        call_kind="chat_completion",
+        first_item_kind="raw_chunk",
+        provider_ttft_ms=11,
+        visible_ttft_ms=37,
+        elapsed_ms=59,
+        usage=TokenUsage(
+            prompt_tokens=2,
+            completion_tokens=3,
+            total_tokens=5,
+            cached_tokens=1,
+        ),
+    )
+    result = result_type(
+        case_id=benchmark.BenchmarkCaseId.ORDINARY_CHAT,
+        mode=benchmark.BenchmarkMode.DUAL_AGENT,
+        iteration=1,
+        status="failed",
+        initial_provider_ttft_ms=None,
+        visible_ttft_ms=None,
+        turn_wall_time_ms=None,
+        prompt_tokens=None,
+        completion_tokens=None,
+        total_tokens=None,
+        cached_tokens=None,
+        model_call_count=None,
+        fallback_count=None,
+        tool_count=None,
+        event_names=[],
+        event_sources=[],
+        tool_statuses=[],
+        tool_latencies_ms=[],
+        semantic_failures=["missing visible answer"],
+        error=None,
+        model_calls=[timing],
+    )
+
+    assert result.model_calls == [timing]
+    assert result.initial_provider_ttft_ms is None
+    with pytest.raises(ValidationError):
+        timing_type(
+            call_index="1",
+            call_kind="chat_completion",
+            first_item_kind=None,
+            provider_ttft_ms=None,
+            visible_ttft_ms=None,
+            elapsed_ms=None,
+            usage=None,
+        )
+    with pytest.raises(ValidationError):
+        result_type.model_validate(result.model_dump() | {"unexpected": True})
+
+
+@pytest.mark.asyncio
+async def test_timing_client_keeps_provider_visible_and_elapsed_timings_separate():
+    clock = ManualClock()
+    inner = ScriptedTimingClient(clock)
+    client = benchmark.TimingChatClient(inner, clock=clock)
+
+    items = [
+        item
+        async for item in client.stream_chat(
+            messages=[],
+            tools=[],
+            params=AgentParams(model="benchmark-model"),
+        )
+    ]
+
+    assert [item.kind for item in items] == [
+        "raw_chunk",
+        "message_delta",
+        "usage",
+        "usage",
+    ]
+    assert client.timings == [
+        benchmark.BenchmarkModelCallTiming(
+            call_index=1,
+            call_kind="chat_completion",
+            first_item_kind="raw_chunk",
+            provider_ttft_ms=11,
+            visible_ttft_ms=37,
+            elapsed_ms=71,
+            usage=TokenUsage(total_tokens=5),
+        )
+    ]
+
+
+@pytest.mark.asyncio
+async def test_timing_client_records_failed_fallback_and_closes_inner_client():
+    clock = ManualClock()
+    inner = ScriptedTimingClient(clock, fail=True)
+    client = benchmark.TimingChatClient(inner, clock=clock)
+
+    with pytest.raises(RuntimeError, match="stream failed"):
+        async for _ in client.stream_chat(
+            messages=[],
+            tools=[{"type": "function", "function": {"name": "mock"}}],
+            params=AgentParams(model="benchmark-model"),
+            tool_choice={"type": "function", "function": {"name": "mock"}},
+        ):
+            pass
+    await client.aclose()
+
+    assert inner.closed is True
+    assert client.timings == [
+        benchmark.BenchmarkModelCallTiming(
+            call_index=1,
+            call_kind="argument_fallback",
+            first_item_kind="raw_chunk",
+            provider_ttft_ms=11,
+            visible_ttft_ms=None,
+            elapsed_ms=23,
+            usage=None,
+        )
+    ]
+
+
+@pytest.mark.asyncio
+async def test_timing_client_finishes_timing_when_stream_is_cancelled():
+    clock = ManualClock()
+    inner = BlockingTimingClient(clock)
+    client = benchmark.TimingChatClient(inner, clock=clock)
+
+    async def consume() -> None:
+        async for _ in client.stream_chat(
+            messages=[],
+            tools=[],
+            params=AgentParams(model="benchmark-model"),
+        ):
+            pass
+
+    task = asyncio.create_task(consume())
+    await inner.blocked.wait()
+    clock.now = 0.025
+    task.cancel()
+    await asyncio.gather(task, return_exceptions=True)
+    await client.aclose()
+
+    assert client.timings[0].first_item_kind == "raw_chunk"
+    assert client.timings[0].provider_ttft_ms == 10
+    assert client.timings[0].visible_ttft_ms is None
+    assert client.timings[0].elapsed_ms == 25
+    assert inner.closed is True
+
+
+@pytest.mark.asyncio
+async def test_timing_client_indexes_concurrent_calls_by_start_order():
+    inner = ConcurrentTimingClient()
+    client = benchmark.TimingChatClient(inner)
+
+    async def collect() -> None:
+        async for _ in client.stream_chat(
+            messages=[],
+            tools=[],
+            params=AgentParams(model="benchmark-model"),
+        ):
+            pass
+
+    first = asyncio.create_task(collect())
+    await inner.first_started.wait()
+    await collect()
+    inner.release_first.set()
+    await first
+
+    assert [timing.call_index for timing in client.timings] == [1, 2]
+    assert [timing.usage.total_tokens for timing in client.timings] == [1, 2]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("mode", benchmark.BENCHMARK_MODES)
+@pytest.mark.parametrize("case_id", benchmark.BENCHMARK_CASE_IDS)
+async def test_mock_benchmark_client_emits_raw_semantics_and_fixed_usage(
+    case_id: benchmark.BenchmarkCaseId,
+    mode: benchmark.BenchmarkMode,
+):
+    case = benchmark.BENCHMARK_CASE_CATALOG[case_id]
+    client = benchmark.MockBenchmarkChatClient(case, mode)
+    expected_rounds = benchmark.build_mock_rounds(case, mode)
+
+    emitted_rounds = []
+    for _ in expected_rounds:
+        emitted_rounds.append(
+            [
+                item
+                async for item in client.stream_chat(
+                    messages=[],
+                    tools=[],
+                    params=AgentParams(model="benchmark-model"),
+                )
+            ]
+        )
+    await client.aclose()
+
+    assert [items[0].kind for items in emitted_rounds] == [
+        "raw_chunk"
+    ] * len(expected_rounds)
+    assert [items[1:-1] for items in emitted_rounds] == expected_rounds
+    assert [items[-1].usage for items in emitted_rounds] == [
+        TokenUsage(
+            prompt_tokens=10,
+            completion_tokens=5,
+            total_tokens=15,
+            cached_tokens=2,
+        )
+    ] * len(expected_rounds)
+
+    with pytest.raises(RuntimeError, match=f"mock benchmark stream exhausted: {case_id}"):
+        async for _ in client.stream_chat(
+            messages=[],
+            tools=[],
+            params=AgentParams(model="benchmark-model"),
+        ):
+            pass
+
+
+@pytest.mark.asyncio
+async def test_mock_runner_is_ordered_ignores_factory_and_reports_event_ledgers():
+    config = benchmark.BenchmarkConfig(
+        schema_version=1,
+        base_url="https://provider.example/v1",
+        model="benchmark-model",
+        runs_per_case=2,
+        cases=[
+            benchmark.BenchmarkCaseId.WEB_SEARCH_TWO_ANSWERS,
+            benchmark.BenchmarkCaseId.ORDINARY_CHAT,
+        ],
+        modes=[
+            benchmark.BenchmarkMode.CHAT_AGENT_TOOLS,
+            benchmark.BenchmarkMode.DUAL_AGENT,
+        ],
+    )
+    factory_calls = 0
+
+    def forbidden_factory(config: benchmark.BenchmarkConfig, api_key: str):
+        del config, api_key
+        nonlocal factory_calls
+        factory_calls += 1
+        raise AssertionError("mock mode must not call client_factory")
+
+    results = await benchmark.BenchmarkRunner(
+        config,
+        api_key=None,
+        mock=True,
+        client_factory=forbidden_factory,
+    ).run()
+
+    assert factory_calls == 0
+    assert [(item.case_id, item.mode, item.iteration) for item in results] == [
+        (case_id, mode, iteration)
+        for case_id in config.cases
+        for mode in config.modes
+        for iteration in range(1, 3)
+    ]
+    assert {item.status for item in results} == {"passed"}
+    search_results = [
+        item
+        for item in results
+        if item.case_id is benchmark.BenchmarkCaseId.WEB_SEARCH_TWO_ANSWERS
+    ]
+    assert {item.model_call_count for item in search_results} == {2}
+    assert {item.fallback_count for item in search_results} == {0}
+    assert {item.tool_count for item in search_results} == {1}
+    assert {tuple(item.event_names) for item in search_results} == {
+        ("knowledge.web.search",)
+    }
+    assert {tuple(item.tool_statuses) for item in search_results} == {("success",)}
+    assert {
+        tuple(item.event_sources) for item in search_results
+    } == {("provider_resolved",), ("text_event",)}
+    assert all(len(item.tool_latencies_ms) == 1 for item in search_results)
+    assert all(
+        latency is not None and latency >= 0
+        for item in search_results
+        for latency in item.tool_latencies_ms
+    )
+    ordinary_results = [
+        item
+        for item in results
+        if item.case_id is benchmark.BenchmarkCaseId.ORDINARY_CHAT
+    ]
+    assert all(item.event_names == [] for item in ordinary_results)
+    assert all(item.tool_count == 0 for item in ordinary_results)
+
+
+@pytest.mark.asyncio
+async def test_mock_runner_passes_all_catalog_cases_in_both_modes():
+    config = benchmark.BenchmarkConfig(
+        schema_version=1,
+        base_url="https://provider.example/v1",
+        model="benchmark-model",
+    )
+
+    results = await benchmark.BenchmarkRunner(config, api_key=None, mock=True).run()
+
+    assert len(results) == 12
+    assert all(result.status == "passed" for result in results)
+    assert all(result.semantic_failures == [] for result in results)
+
+
+@pytest.mark.asyncio
+async def test_runner_uses_a_fresh_in_memory_store_and_unique_session_per_run(monkeypatch):
+    RecordingSQLiteSessionStore.instances = []
+    monkeypatch.setattr(benchmark, "SQLiteSessionStore", RecordingSQLiteSessionStore)
+    config = benchmark.BenchmarkConfig(
+        schema_version=1,
+        base_url="https://provider.example/v1",
+        model="benchmark-model",
+        runs_per_case=2,
+        cases=[benchmark.BenchmarkCaseId.ORDINARY_CHAT],
+        modes=[benchmark.BenchmarkMode.DUAL_AGENT],
+    )
+
+    results = await benchmark.BenchmarkRunner(config, api_key=None, mock=True).run()
+
+    assert len(results) == 2
+    assert len(RecordingSQLiteSessionStore.instances) == 2
+    sessions = [store.list_sessions() for store in RecordingSQLiteSessionStore.instances]
+    assert all(store.database_path == ":memory:" for store in RecordingSQLiteSessionStore.instances)
+    assert [len(records) for records in sessions] == [1, 1]
+    assert len({records[0]["id"] for records in sessions}) == 2
+
+
+@pytest.mark.asyncio
+async def test_runner_keeps_provider_visible_and_turn_wall_time_distinct():
+    clock = ManualClock()
+    client = ScriptedTimingClient(clock)
+    config = benchmark.BenchmarkConfig(
+        schema_version=1,
+        base_url="https://provider.example/v1",
+        model="benchmark-model",
+        cases=[benchmark.BenchmarkCaseId.ORDINARY_CHAT],
+        modes=[benchmark.BenchmarkMode.DUAL_AGENT],
+    )
+
+    result = (
+        await benchmark.BenchmarkRunner(
+            config,
+            api_key="test-key",
+            client_factory=lambda config, api_key: client,
+            clock=clock,
+        ).run()
+    )[0]
+
+    assert result.status == "passed"
+    assert result.initial_provider_ttft_ms == 11
+    assert result.visible_ttft_ms == 37
+    assert result.turn_wall_time_ms == 71
+    assert result.model_calls[0].elapsed_ms == 71
+    assert client.closed is True
+
+
+@pytest.mark.asyncio
+async def test_runner_continues_after_one_factory_failure():
+    config = benchmark.BenchmarkConfig(
+        schema_version=1,
+        base_url="https://provider.example/v1",
+        model="benchmark-model",
+        cases=[benchmark.BenchmarkCaseId.ORDINARY_CHAT],
+        modes=[
+            benchmark.BenchmarkMode.DUAL_AGENT,
+            benchmark.BenchmarkMode.CHAT_AGENT_TOOLS,
+        ],
+    )
+    clients: list[ScriptedTimingClient] = []
+    calls = 0
+
+    def factory(config: benchmark.BenchmarkConfig, api_key: str):
+        del config, api_key
+        nonlocal calls
+        calls += 1
+        if calls == 1:
+            raise RuntimeError("first factory failed")
+        clock = ManualClock()
+        client = ScriptedTimingClient(clock)
+        clients.append(client)
+        return client
+
+    results = await benchmark.BenchmarkRunner(
+        config,
+        api_key="test-key",
+        client_factory=factory,
+    ).run()
+
+    assert [result.status for result in results] == ["failed", "passed"]
+    assert results[0].error == "first factory failed"
+    assert results[1].error is None
+    assert clients[0].closed is True
+
+
+@pytest.mark.asyncio
+async def test_runner_marks_semantic_mismatch_failed_and_continues():
+    config = benchmark.BenchmarkConfig(
+        schema_version=1,
+        base_url="https://provider.example/v1",
+        model="benchmark-model",
+        cases=[benchmark.BenchmarkCaseId.ORDINARY_CHAT],
+        modes=[
+            benchmark.BenchmarkMode.DUAL_AGENT,
+            benchmark.BenchmarkMode.CHAT_AGENT_TOOLS,
+        ],
+    )
+    failed_client = NoAnswerClient()
+    calls = 0
+
+    def factory(config: benchmark.BenchmarkConfig, api_key: str):
+        del config, api_key
+        nonlocal calls
+        calls += 1
+        if calls == 1:
+            return failed_client
+        return benchmark.MockBenchmarkChatClient(
+            benchmark.BENCHMARK_CASE_CATALOG[benchmark.BenchmarkCaseId.ORDINARY_CHAT],
+            benchmark.BenchmarkMode.CHAT_AGENT_TOOLS,
+        )
+
+    results = await benchmark.BenchmarkRunner(
+        config,
+        api_key="test-key",
+        client_factory=factory,
+    ).run()
+
+    assert [result.status for result in results] == ["failed", "passed"]
+    assert results[0].error is None
+    assert results[0].semantic_failures == ["answer_count expected 1, got 0"]
+    assert failed_client.closed is True
+
+
+@pytest.mark.asyncio
+async def test_runner_reports_argument_fallback_model_and_tool_ledgers(monkeypatch):
+    client = FallbackBenchmarkClient()
+    registry = ToolRegistry(
+        [
+            ToolDefinition(
+                name="device.volume.adjust",
+                description="Force fallback for benchmark ledger coverage.",
+                parameters={
+                    "type": "object",
+                    "properties": {
+                        "mode": {"type": "string"},
+                        "value": {"type": "integer"},
+                    },
+                    "required": ["mode", "value"],
+                },
+                handler=lambda event: {
+                    "tool": event.name,
+                    "status": "payload-status-is-not-kernel-status",
+                },
+                argument_resolver=lambda event, context: {},
+                result_policy=ResultPolicy.SILENT_SUCCESS,
+            )
+        ]
+    )
+    monkeypatch.setattr(benchmark, "build_default_tool_registry", lambda: registry)
+    config = benchmark.BenchmarkConfig(
+        schema_version=1,
+        base_url="https://provider.example/v1",
+        model="benchmark-model",
+        cases=[benchmark.BenchmarkCaseId.DEVICE_VOLUME_SILENT],
+        modes=[benchmark.BenchmarkMode.DUAL_AGENT],
+    )
+
+    result = (
+        await benchmark.BenchmarkRunner(
+            config,
+            api_key="test-key",
+            client_factory=lambda config, api_key: client,
+        ).run()
+    )[0]
+
+    assert result.status == "passed"
+    assert result.model_call_count == 1
+    assert result.fallback_count == 1
+    assert result.tool_count == 1
+    assert result.total_tokens == 12
+    assert [timing.call_kind for timing in result.model_calls] == [
+        "chat_completion",
+        "argument_fallback",
+    ]
+    assert result.event_names == ["device.volume.adjust"]
+    assert result.event_sources == ["text_event"]
+    assert result.tool_statuses == ["success"]
+    assert client.calls == 2
+    assert client.closed is True
+
+
+def test_live_runner_requires_an_api_key():
+    config = benchmark.BenchmarkConfig(
+        schema_version=1,
+        base_url="https://provider.example/v1",
+        model="benchmark-model",
+        cases=[benchmark.BenchmarkCaseId.ORDINARY_CHAT],
+        modes=[benchmark.BenchmarkMode.DUAL_AGENT],
+    )
+
+    with pytest.raises(ValueError, match="api_key"):
+        benchmark.BenchmarkRunner(config, api_key=None)
+
+
+def test_visible_answer_order_ignores_empty_deltas_before_a_tool_result():
+    answers, first_answer_index, first_tool_index = (
+        benchmark.BenchmarkRunner._visible_answers(
+            [
+                {"type": "message_delta", "content": "  "},
+                {"type": "tool_result", "message": {}},
+                {"type": "message_delta", "content": "late answer"},
+            ]
+        )
+    )
+
+    assert answers == ["late answer"]
+    assert first_answer_index == 2
+    assert first_tool_index == 1