agent-lab/src/agent_lab/application/benchmark.py

import asyncio
import json
import time
import unicodedata
from collections.abc import AsyncIterator, Callable, Mapping
from enum import StrEnum
from types import MappingProxyType
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

from agent_lab.application.contracts import (
    AgentParams,
    DebugRunRequest,
    EventAgentParams,
)
from agent_lab.application.events import (
    InMemoryCalendarScheduleAdapter,
    InMemoryDeviceVolumeAdapter,
)
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 ChatMessage, StreamItem, TokenUsage
from agent_lab.infrastructure.chat_client import OpenAICompatibleChatClient
from agent_lab.infrastructure.sqlite_store import SQLiteSessionStore


class BenchmarkMode(StrEnum):
    DUAL_AGENT = "dual_agent"
    CHAT_AGENT_TOOLS = "chat_agent_tools"


class BenchmarkCaseId(StrEnum):
    ORDINARY_CHAT = "ordinary_chat"
    DEVICE_VOLUME_SILENT = "device_volume_silent"
    CALENDAR_SCHEDULE_TEMPLATE = "calendar_schedule_template"
    WEB_SEARCH_TWO_ANSWERS = "web_search_two_answers"
    SESSION_TERMINATE = "session_terminate"
    PARALLEL_VOLUME_SCHEDULE = "parallel_volume_schedule"


JsonScalar: TypeAlias = str | int | float | bool | None

BENCHMARK_MODES: tuple[BenchmarkMode, ...] = tuple(BenchmarkMode)
BENCHMARK_CASE_IDS: tuple[BenchmarkCaseId, ...] = tuple(BenchmarkCaseId)

BENCHMARK_SYSTEM_PROMPT = "Exercise the requested business scenario."
BENCHMARK_MAX_TOKENS = 128
BENCHMARK_MAX_EVENT_LOOPS = 1
BENCHMARK_MAX_PARALLEL_EVENTS = 2
BENCHMARK_BATCH_TIMEOUT_SECONDS = 1.0
BENCHMARK_PARALLEL_BARRIER_TIMEOUT_SECONDS = 0.05


class _StrictBenchmarkModel(BaseModel):
    model_config = ConfigDict(
        extra="forbid",
        strict=True,
        hide_input_in_errors=True,
    )


class _FrozenStrictBenchmarkModel(BaseModel):
    model_config = ConfigDict(
        extra="forbid",
        strict=True,
        hide_input_in_errors=True,
        frozen=True,
    )


def _validate_base_url(value: str) -> str:
    if not value or value != value.strip() or any(char.isspace() for char in value):
        raise ValueError("base_url must be a non-empty HTTP(S) URL")
    if any(unicodedata.category(char).startswith("C") for char in value):
        raise ValueError("base_url must not contain Unicode category C characters")
    try:
        parsed = urlsplit(value)
        hostname = parsed.hostname
    except ValueError as exc:
        raise ValueError("base_url must be a valid HTTP(S) URL") from exc
    if parsed.scheme not in {"http", "https"} or not parsed.netloc or not hostname:
        raise ValueError("base_url must use http or https and include a host")
    if parsed.username is not None or parsed.password is not None or "@" in parsed.netloc:
        raise ValueError("base_url must not contain userinfo")
    if "?" in value:
        raise ValueError("base_url must not contain a query")
    if "#" in value:
        raise ValueError("base_url must not contain a fragment")
    try:
        transport_url = httpx.URL(value)
        _ = (transport_url.host, transport_url.port, parsed.port)
    except (httpx.InvalidURL, ValueError) as exc:
        raise ValueError("base_url must be a valid HTTP(S) URL") from exc
    return value


def _validate_model(value: str) -> str:
    stripped = value.strip()
    if not stripped:
        raise ValueError("model must not be empty")
    return stripped


class BenchmarkTarget(_StrictBenchmarkModel):
    base_url: str
    model: str

    _base_url_validator = field_validator("base_url")(_validate_base_url)
    _model_validator = field_validator("model")(_validate_model)


class BenchmarkConfig(BenchmarkTarget):
    schema_version: Literal[1]
    runs_per_case: int = Field(default=1, ge=1)
    modes: list[Annotated[BenchmarkMode, Field(strict=False)]] = Field(
        default_factory=lambda: list(BENCHMARK_MODES),
        min_length=1,
    )
    cases: list[Annotated[BenchmarkCaseId, Field(strict=False)]] = Field(
        default_factory=lambda: list(BENCHMARK_CASE_IDS),
        min_length=1,
    )

    @field_validator("schema_version", mode="before")
    @classmethod
    def require_integer_schema_version(cls, value: object) -> object:
        if type(value) is not int or value != 1:
            raise ValueError("schema_version must be the integer 1")
        return value

    @field_validator("modes", "cases")
    @classmethod
    def reject_duplicates(
        cls,
        values: list[BenchmarkMode | BenchmarkCaseId],
    ) -> list[BenchmarkMode | BenchmarkCaseId]:
        if len(values) != len(set(values)):
            raise ValueError("values must not contain duplicates")
        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]
    batch_event_names: list[str]
    tool_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]
    parallel_events_overlapped: bool | None = None


class BenchmarkParallelEventProbe:
    _EXPECTED_EVENTS = frozenset(
        {"device.volume.adjust", "calendar.schedule.create"}
    )

    def __init__(
        self,
        *,
        barrier_timeout_seconds: float = BENCHMARK_PARALLEL_BARRIER_TIMEOUT_SECONDS,
    ) -> None:
        self.barrier_timeout_seconds = barrier_timeout_seconds
        self._volume = InMemoryDeviceVolumeAdapter()
        self._calendar = InMemoryCalendarScheduleAdapter()
        self._active: set[str] = set()
        self._release = asyncio.Event()
        self._lock = asyncio.Lock()
        self.overlapped = False

    async def adjust(
        self,
        event_id: str,
        *,
        mode: str,
        value: int | None = None,
        delta: int | None = None,
    ) -> dict[str, Any]:
        await self._await_overlap("device.volume.adjust")
        return self._volume.adjust(
            event_id,
            mode=mode,
            value=value,
            delta=delta,
        )

    async def create(
        self,
        event_id: str,
        *,
        title: str,
        start_at: str,
        timezone: str,
        recurrence: str | None = None,
        reminder_minutes: int | None = None,
    ) -> dict[str, Any]:
        await self._await_overlap("calendar.schedule.create")
        return self._calendar.create(
            event_id,
            title=title,
            start_at=start_at,
            timezone=timezone,
            recurrence=recurrence,
            reminder_minutes=reminder_minutes,
        )

    async def _await_overlap(self, event_name: str) -> None:
        async with self._lock:
            self._active.add(event_name)
            if self._EXPECTED_EVENTS.issubset(self._active):
                self.overlapped = True
                self._release.set()
        try:
            await asyncio.wait_for(
                self._release.wait(),
                timeout=self.barrier_timeout_seconds,
            )
        except TimeoutError:
            pass
        finally:
            async with self._lock:
                self._active.discard(event_name)


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
        provider_elapsed_seconds = 0.0
        iterator: AsyncIterator[StreamItem] | None = None
        provider_exception_active = False
        try:
            stream_chat = getattr(self.inner, "stream_chat")
            iterator = stream_chat(
                messages=messages,
                tools=tools,
                params=params,
                tool_choice=tool_choice,
            ).__aiter__()
            while True:
                wait_started_at = self.clock()
                try:
                    item = await iterator.__anext__()
                except StopAsyncIteration:
                    provider_elapsed_seconds += max(
                        0.0,
                        self.clock() - wait_started_at,
                    )
                    break
                except BaseException:
                    provider_exception_active = True
                    provider_elapsed_seconds += max(
                        0.0,
                        self.clock() - wait_started_at,
                    )
                    raise
                provider_elapsed_seconds += max(
                    0.0,
                    self.clock() - wait_started_at,
                )
                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 (item.content or "").strip()
                    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:
            try:
                close = getattr(iterator, "aclose", None)
                if close is not None:
                    try:
                        await close()
                    except BaseException:
                        if not provider_exception_active:
                            raise
            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=max(0, round(provider_elapsed_seconds * 1000)),
                    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, ...]
    answer_count: int = Field(ge=1)
    expected_model_call_count: int = Field(ge=1)
    expected_fallback_count: int = Field(ge=0)
    terminal: bool
    first_reply_before_tool: bool

    @model_validator(mode="after")
    def match_answer_count(self) -> "BenchmarkExpectation":
        if self.answer_count != len(self.visible_messages):
            raise ValueError("answer_count must match visible_messages")
        return self


class BenchmarkMockEvent(_FrozenStrictBenchmarkModel):
    id: str
    name: str
    arguments: tuple[tuple[str, JsonScalar], ...]

    @model_validator(mode="after")
    def validate_event(self) -> "BenchmarkMockEvent":
        if not self.id.strip():
            raise ValueError("mock event id must not be empty")
        if not self.name.strip():
            raise ValueError("mock event name must not be empty")
        keys = tuple(key for key, _ in self.arguments)
        if any(not key.strip() for key in keys):
            raise ValueError("mock event argument names must not be empty")
        if len(keys) != len(set(keys)):
            raise ValueError("mock event arguments must not contain duplicate names")
        return self


class BenchmarkCase(_FrozenStrictBenchmarkModel):
    case_id: BenchmarkCaseId
    user_message: str
    enabled_tools: tuple[str, ...]
    expectation: BenchmarkExpectation
    mock_visible_messages: tuple[str, ...]
    mock_events: tuple[BenchmarkMockEvent, ...]

    @model_validator(mode="after")
    def validate_case_semantics(self) -> "BenchmarkCase":
        if not self.user_message.strip():
            raise ValueError("user_message must not be empty")
        if len(self.enabled_tools) != len(set(self.enabled_tools)):
            raise ValueError("enabled_tools must not contain duplicates")
        mock_event_names = tuple(event.name for event in self.mock_events)
        if mock_event_names != self.expectation.event_names:
            raise ValueError("mock event order must match expected event order")
        if self.expectation.first_reply_before_tool:
            if not self.mock_events or not self.mock_visible_messages:
                raise ValueError(
                    "first_reply_before_tool requires mock text and at least one event"
                )
        return self


def _mock_event(
    event_id: str,
    name: str,
    arguments: tuple[tuple[str, JsonScalar], ...] = (),
) -> BenchmarkMockEvent:
    return BenchmarkMockEvent(id=event_id, name=name, arguments=arguments)


_SCHEDULE_ARGUMENTS: tuple[tuple[str, JsonScalar], ...] = (
    ("title", "Design review"),
    ("start_at", "2026-07-14T09:30:00+08:00"),
    ("timezone", "Asia/Shanghai"),
)
_SCHEDULE_CONFIRMATION = (
    "Scheduled Design review for 2026-07-14T09:30:00+08:00 "
    "(Asia/Shanghai)."
)

BENCHMARK_CASE_CATALOG: Mapping[BenchmarkCaseId, BenchmarkCase] = MappingProxyType(
    {
        BenchmarkCaseId.ORDINARY_CHAT: BenchmarkCase(
            case_id=BenchmarkCaseId.ORDINARY_CHAT,
            user_message="Hello.",
            enabled_tools=(
                "session.terminate",
                "device.volume.adjust",
                "calendar.schedule.create",
                "knowledge.web.search",
            ),
            expectation=BenchmarkExpectation(
                visible_messages=("Ordinary answer.",),
                event_names=(),
                answer_count=1,
                expected_model_call_count=1,
                expected_fallback_count=0,
                terminal=False,
                first_reply_before_tool=False,
            ),
            mock_visible_messages=("Ordinary answer.",),
            mock_events=(),
        ),
        BenchmarkCaseId.DEVICE_VOLUME_SILENT: BenchmarkCase(
            case_id=BenchmarkCaseId.DEVICE_VOLUME_SILENT,
            user_message="set volume to 30",
            enabled_tools=("device.volume.adjust",),
            expectation=BenchmarkExpectation(
                visible_messages=("I will set the volume to 30.",),
                event_names=("device.volume.adjust",),
                answer_count=1,
                expected_model_call_count=1,
                expected_fallback_count=0,
                terminal=False,
                first_reply_before_tool=True,
            ),
            mock_visible_messages=("I will set the volume to 30.",),
            mock_events=(
                _mock_event(
                    "volume-1",
                    "device.volume.adjust",
                    (("mode", "absolute"), ("value", 30)),
                ),
            ),
        ),
        BenchmarkCaseId.CALENDAR_SCHEDULE_TEMPLATE: BenchmarkCase(
            case_id=BenchmarkCaseId.CALENDAR_SCHEDULE_TEMPLATE,
            user_message=(
                'schedule "Design review" at 2026-07-14T09:30:00+08:00 '
                "timezone Asia/Shanghai"
            ),
            enabled_tools=("calendar.schedule.create",),
            expectation=BenchmarkExpectation(
                visible_messages=("I will schedule that.", _SCHEDULE_CONFIRMATION),
                event_names=("calendar.schedule.create",),
                answer_count=2,
                expected_model_call_count=1,
                expected_fallback_count=0,
                terminal=False,
                first_reply_before_tool=True,
            ),
            mock_visible_messages=("I will schedule that.",),
            mock_events=(
                _mock_event(
                    "schedule-1",
                    "calendar.schedule.create",
                    _SCHEDULE_ARGUMENTS,
                ),
            ),
        ),
        BenchmarkCaseId.WEB_SEARCH_TWO_ANSWERS: BenchmarkCase(
            case_id=BenchmarkCaseId.WEB_SEARCH_TWO_ANSWERS,
            user_message="event batch safety",
            enabled_tools=("knowledge.web.search",),
            expectation=BenchmarkExpectation(
                visible_messages=("I will check.", "Grounded search update."),
                event_names=("knowledge.web.search",),
                answer_count=2,
                expected_model_call_count=2,
                expected_fallback_count=0,
                terminal=False,
                first_reply_before_tool=True,
            ),
            mock_visible_messages=("I will check.", "Grounded search update."),
            mock_events=(
                _mock_event(
                    "search-1",
                    "knowledge.web.search",
                    (("query", "event batch safety"),),
                ),
            ),
        ),
        BenchmarkCaseId.SESSION_TERMINATE: BenchmarkCase(
            case_id=BenchmarkCaseId.SESSION_TERMINATE,
            user_message="end this session",
            enabled_tools=("session.terminate",),
            expectation=BenchmarkExpectation(
                visible_messages=("Goodbye.",),
                event_names=("session.terminate",),
                answer_count=1,
                expected_model_call_count=1,
                expected_fallback_count=0,
                terminal=True,
                first_reply_before_tool=True,
            ),
            mock_visible_messages=("Goodbye.",),
            mock_events=(_mock_event("terminate-1", "session.terminate"),),
        ),
        BenchmarkCaseId.PARALLEL_VOLUME_SCHEDULE: BenchmarkCase(
            case_id=BenchmarkCaseId.PARALLEL_VOLUME_SCHEDULE,
            user_message=(
                'set volume to 30 and schedule "Design review" at '
                "2026-07-14T09:30:00+08:00 timezone Asia/Shanghai"
            ),
            enabled_tools=(
                "device.volume.adjust",
                "calendar.schedule.create",
            ),
            expectation=BenchmarkExpectation(
                visible_messages=(
                    "I will set the volume to 30 and schedule that.",
                    _SCHEDULE_CONFIRMATION,
                ),
                event_names=(
                    "device.volume.adjust",
                    "calendar.schedule.create",
                ),
                answer_count=2,
                expected_model_call_count=1,
                expected_fallback_count=0,
                terminal=False,
                first_reply_before_tool=True,
            ),
            mock_visible_messages=(
                "I will set the volume to 30 and schedule that.",
            ),
            mock_events=(
                _mock_event(
                    "parallel-volume",
                    "device.volume.adjust",
                    (("mode", "absolute"), ("value", 30)),
                ),
                _mock_event(
                    "parallel-schedule",
                    "calendar.schedule.create",
                    _SCHEDULE_ARGUMENTS,
                ),
            ),
        ),
    }
)


def _normalize_benchmark_mode(mode: BenchmarkMode | str) -> BenchmarkMode:
    try:
        return BenchmarkMode(mode)
    except ValueError as exc:
        raise ValueError(f"unsupported benchmark mode: {mode}") from exc


def build_benchmark_request(
    case: BenchmarkCase,
    mode: BenchmarkMode | str,
    model: str,
) -> DebugRunRequest:
    normalized_mode = _normalize_benchmark_mode(mode)
    validated_model = _validate_model(model)
    return DebugRunRequest(
        user_message=case.user_message,
        system_prompts=[BENCHMARK_SYSTEM_PROMPT],
        pre_messages=[],
        chat_agent=AgentParams(
            model=validated_model,
            temperature=0.0,
            max_tokens=BENCHMARK_MAX_TOKENS,
        ),
        event_agent=EventAgentParams(
            model=validated_model,
            temperature=0.0,
            max_tokens=BENCHMARK_MAX_TOKENS,
            enabled_tools=list(case.enabled_tools),
            max_event_loops=BENCHMARK_MAX_EVENT_LOOPS,
            max_parallel_events=BENCHMARK_MAX_PARALLEL_EVENTS,
            batch_timeout_seconds=BENCHMARK_BATCH_TIMEOUT_SECONDS,
        ),
        tool_invocation_mode=normalized_mode.value,
    )


def build_mock_rounds(
    case: BenchmarkCase,
    mode: BenchmarkMode | str,
) -> list[list[StreamItem]]:
    normalized_mode = _normalize_benchmark_mode(mode)

    first_round: list[StreamItem] = []
    if case.mock_visible_messages:
        first_round.append(StreamItem.message_delta(case.mock_visible_messages[0]))
    for mock_event in case.mock_events:
        arguments = dict(mock_event.arguments)
        event = ToolCallEvent(
            id=mock_event.id,
            name=mock_event.name,
            arguments=arguments,
            raw_arguments=json.dumps(
                arguments,
                sort_keys=True,
                separators=(",", ":"),
            ),
        )
        event_item = (
            StreamItem.text_event(event)
            if normalized_mode is BenchmarkMode.DUAL_AGENT
            else StreamItem.provider_tool_call(event)
        )
        first_round.append(event_item)

    rounds = [first_round]
    rounds.extend(
        [StreamItem.message_delta(message)]
        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]
BenchmarkParallelProbeFactory: TypeAlias = Callable[[], BenchmarkParallelEventProbe]


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:
        content = item.get("content")
        if (
            item.get("type") == "message_delta"
            and isinstance(content, str)
            and content.strip()
            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,
        parallel_probe_factory: BenchmarkParallelProbeFactory | 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.parallel_probe_factory = (
            parallel_probe_factory or BenchmarkParallelEventProbe
        )
        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
        parallel_probe: BenchmarkParallelEventProbe | 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)
            if case.case_id is BenchmarkCaseId.PARALLEL_VOLUME_SCHEDULE:
                parallel_probe = self.parallel_probe_factory()
                registry = build_default_tool_registry(
                    device_volume_port=parallel_probe,
                    calendar_schedule_port=parallel_probe,
                )
            else:
                registry = build_default_tool_registry()
            runtime = DebugRuntime(
                timing_client,
                queues=RuntimeQueues(output=output_queue),
                registry=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,
                parallel_events_overlapped=(
                    parallel_probe.overlapped
                    if parallel_probe is not None
                    else None
                ),
            )
        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,
        parallel_events_overlapped: bool | None = 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]
        batch_event_names = [
            str(item.get("event_name") or "") for item in batch_results
        ]
        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"]
        tool_event_names = [
            str(call.get("event_name") or "") for call in tool_calls
        ]
        ledger_model_call_count = sum(
            call["call_kind"] == "chat_completion" for call in calls
        )
        ledger_fallback_count = int(session_usage["fallback_count"])
        timing_model_call_count = sum(
            timing.call_kind == "chat_completion" for timing in timings
        )
        timing_fallback_count = sum(
            timing.call_kind == "argument_fallback" for timing in timings
        )
        token_totals, missing_usage_count = self._token_totals(timings)
        tool_count = int(session_usage["tool_count"])
        semantic_failures = self._semantic_failures(
            case=case,
            mode=mode,
            outputs=outputs,
            audits=audits,
            event_names=event_names,
            batch_event_names=batch_event_names,
            tool_event_names=tool_event_names,
            event_sources=event_sources,
            tool_statuses=tool_statuses,
            model_call_count=ledger_model_call_count,
            fallback_count=ledger_fallback_count,
            tool_count=tool_count,
            timings=timings,
            check_mock_text=self.mock,
            missing_usage_count=missing_usage_count,
            parallel_events_overlapped=parallel_events_overlapped,
        )
        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=token_totals[0],
            completion_tokens=token_totals[1],
            total_tokens=token_totals[2],
            cached_tokens=token_totals[3],
            model_call_count=timing_model_call_count,
            fallback_count=timing_fallback_count,
            tool_count=tool_count,
            event_names=event_names,
            batch_event_names=batch_event_names,
            tool_event_names=tool_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),
            parallel_events_overlapped=parallel_events_overlapped,
        )

    def _semantic_failures(
        self,
        *,
        case: BenchmarkCase,
        mode: BenchmarkMode,
        outputs: list[dict[str, Any]],
        audits: list[dict[str, Any]],
        event_names: list[str],
        batch_event_names: list[str],
        tool_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,
        missing_usage_count: int = 0,
        parallel_events_overlapped: bool | None = None,
    ) -> 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}"
            )
        if tuple(batch_event_names) != expected.event_names:
            failures.append(
                "batch event names expected "
                f"{expected.event_names!r}, got {tuple(batch_event_names)!r}"
            )
        if tuple(tool_event_names) != expected.event_names:
            failures.append(
                "tool event names expected "
                f"{expected.event_names!r}, got {tuple(tool_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 = tuple(
            (event_name, "success") for event_name in expected.event_names
        )
        actual_statuses = tuple(
            (
                batch_event_names[index]
                if index < len(batch_event_names)
                else None,
                tool_statuses[index] if index < len(tool_statuses) else None,
            )
            for index in range(max(len(batch_event_names), len(tool_statuses)))
        )
        if actual_statuses != expected_statuses:
            failures.append(
                f"event batch statuses expected {expected_statuses!r}, "
                f"got {actual_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 != expected.expected_model_call_count:
            failures.append(
                "model_call_count expected "
                f"{expected.expected_model_call_count}, ledger got {model_call_count}"
            )
        if timing_chat_count != expected.expected_model_call_count:
            failures.append(
                "model_call_count expected "
                f"{expected.expected_model_call_count}, timing got {timing_chat_count}"
            )
        if fallback_count != expected.expected_fallback_count:
            failures.append(
                "fallback_count expected "
                f"{expected.expected_fallback_count}, ledger got {fallback_count}"
            )
        if timing_fallback_count != expected.expected_fallback_count:
            failures.append(
                "fallback_count expected "
                f"{expected.expected_fallback_count}, timing got "
                f"{timing_fallback_count}"
            )
        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}"
            )
        if missing_usage_count:
            failures.append(
                f"model usage missing for {missing_usage_count} call(s)"
            )
        if (
            case.case_id is BenchmarkCaseId.PARALLEL_VOLUME_SCHEDULE
            and parallel_events_overlapped is not True
        ):
            failures.append("parallel event handlers did not overlap")
        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,
        )
        model_call_count = sum(
            timing.call_kind == "chat_completion" for timing in timings
        )
        fallback_count = sum(
            timing.call_kind == "argument_fallback" for timing in timings
        )
        token_totals, missing_usage_count = BenchmarkRunner._token_totals(timings)
        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=token_totals[0],
            completion_tokens=token_totals[1],
            total_tokens=token_totals[2],
            cached_tokens=token_totals[3],
            model_call_count=model_call_count,
            fallback_count=fallback_count,
            tool_count=None,
            event_names=[],
            batch_event_names=[],
            tool_event_names=[],
            event_sources=[],
            tool_statuses=[],
            tool_latencies_ms=[],
            semantic_failures=(
                [f"model usage missing for {missing_usage_count} call(s)"]
                if missing_usage_count
                else []
            ),
            error=error,
            model_calls=list(timings),
            parallel_events_overlapped=None,
        )

    @staticmethod
    def _token_totals(
        timings: list[BenchmarkModelCallTiming],
    ) -> tuple[tuple[int | None, int | None, int | None, int | None], int]:
        if not timings:
            return (None, None, None, None), 0
        missing_usage_count = sum(timing.usage is None for timing in timings)
        if missing_usage_count:
            return (None, None, None, None), missing_usage_count
        usages = [timing.usage for timing in timings]
        assert all(usage is not None for usage in usages)
        complete_usages = [usage for usage in usages if usage is not None]
        return (
            sum(usage.prompt_tokens for usage in complete_usages),
            sum(usage.completion_tokens for usage in complete_usages),
            sum(usage.total_tokens for usage in complete_usages),
            sum(usage.cached_tokens for usage in complete_usages),
        ), 0

    @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,
        )