agent-lab/tests/test_benchmark_runner.py

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 (
    InMemoryCalendarScheduleAdapter,
    InMemoryDeviceVolumeAdapter,
    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 WhitespaceTimingClient:
    def __init__(self, clock: ManualClock) -> None:
        self.clock = clock

    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"})
        self.clock.now = 0.019
        yield StreamItem.message_delta("")
        self.clock.now = 0.023
        yield StreamItem.message_delta("   \n")
        self.clock.now = 0.037
        yield StreamItem.message_delta("visible")
        self.clock.now = 0.050
        yield StreamItem.usage_item(TokenUsage(total_tokens=5))

    async def aclose(self) -> None:
        return None


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


class IncrementalTimingClient:
    def __init__(self, clock: ManualClock) -> None:
        self.clock = clock

    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
        for item in (
            StreamItem.raw_response_chunk({"provider": "first"}),
            StreamItem.message_delta("visible"),
            StreamItem.usage_item(TokenUsage(total_tokens=5)),
        ):
            self.clock.now += 0.010
            yield item
        self.clock.now += 0.010

    async def aclose(self) -> None:
        return None


class SelectiveUsageBenchmarkClient:
    def __init__(
        self,
        case: benchmark.BenchmarkCase,
        mode: benchmark.BenchmarkMode,
        *,
        missing_usage_calls: set[int],
    ) -> None:
        self.rounds = list(benchmark.build_mock_rounds(case, mode))
        self.missing_usage_calls = missing_usage_calls
        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
        self.calls += 1
        yield StreamItem.raw_response_chunk({"call": self.calls})
        for item in self.rounds[self.calls - 1]:
            yield item
        if self.calls not in self.missing_usage_calls:
            yield StreamItem.usage_item(
                TokenUsage(
                    prompt_tokens=10,
                    completion_tokens=5,
                    total_tokens=15,
                    cached_tokens=2,
                )
            )

    async def aclose(self) -> None:
        return None


class NonOverlappingParallelProbe:
    def __init__(self) -> None:
        self.overlapped = False
        self.volume = InMemoryDeviceVolumeAdapter()
        self.calendar = InMemoryCalendarScheduleAdapter()

    async def adjust(
        self,
        event_id: str,
        *,
        mode: str,
        value: int | None = None,
        delta: int | None = None,
    ) -> dict[str, Any]:
        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]:
        return self.calendar.create(
            event_id,
            title=title,
            start_at=start_at,
            timezone=timezone,
            recurrence=recurrence,
            reminder_minutes=reminder_minutes,
        )


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=[],
        batch_event_names=[],
        tool_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_excludes_consumer_backpressure_from_model_elapsed():
    clock = ManualClock()
    client = benchmark.TimingChatClient(
        IncrementalTimingClient(clock),
        clock=clock,
    )

    items = []
    async for item in client.stream_chat(
        messages=[],
        tools=[],
        params=AgentParams(model="benchmark-model"),
    ):
        items.append(item)
        clock.now += 0.100

    assert [item.kind for item in items] == ["raw_chunk", "message_delta", "usage"]
    assert client.timings[0].provider_ttft_ms == 10
    assert client.timings[0].visible_ttft_ms == 120
    assert client.timings[0].elapsed_ms == 40


@pytest.mark.asyncio
async def test_timing_client_ignores_blank_deltas_for_visible_ttft():
    clock = ManualClock()
    client = benchmark.TimingChatClient(
        WhitespaceTimingClient(clock),
        clock=clock,
    )

    async for _ in client.stream_chat(
        messages=[],
        tools=[],
        params=AgentParams(model="benchmark-model"),
    ):
        pass

    assert client.timings[0].provider_ttft_ms == 11
    assert client.timings[0].visible_ttft_ms == 37


@pytest.mark.asyncio
async def test_benchmark_output_queue_ignores_blank_deltas_for_visible_ttft():
    clock = ManualClock()
    queue = benchmark._BenchmarkOutputQueue(clock)

    clock.now = 0.019
    await queue.put({"type": "message_delta", "content": ""})
    clock.now = 0.023
    await queue.put({"type": "message_delta", "content": " \n\t "})
    assert queue.first_message_delta_at is None

    clock.now = 0.037
    await queue.put({"type": "message_delta", "content": "visible"})

    assert queue.first_message_delta_at == 0.037


@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.batch_event_names) for item in search_results} == {
        ("knowledge.web.search",)
    }
    assert {tuple(item.tool_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.batch_event_names == [] for item in ordinary_results)
    assert all(item.tool_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)
    parallel_results = [
        result
        for result in results
        if result.case_id is benchmark.BenchmarkCaseId.PARALLEL_VOLUME_SCHEDULE
    ]
    assert {result.parallel_events_overlapped for result in parallel_results} == {True}
    assert all(
        result.parallel_events_overlapped is None
        for result in results
        if result.case_id is not benchmark.BenchmarkCaseId.PARALLEL_VOLUME_SCHEDULE
    )


@pytest.mark.asyncio
async def test_runner_fails_parallel_case_when_successful_handlers_do_not_overlap():
    probe = NonOverlappingParallelProbe()
    config = benchmark.BenchmarkConfig(
        schema_version=1,
        base_url="https://provider.example/v1",
        model="benchmark-model",
        cases=[benchmark.BenchmarkCaseId.PARALLEL_VOLUME_SCHEDULE],
        modes=[benchmark.BenchmarkMode.DUAL_AGENT],
    )

    result = (
        await benchmark.BenchmarkRunner(
            config,
            api_key=None,
            mock=True,
            parallel_probe_factory=lambda: probe,
        ).run()
    )[0]

    assert result.event_names == [
        "device.volume.adjust",
        "calendar.schedule.create",
    ]
    assert result.tool_statuses == ["success", "success"]
    assert result.tool_count == 2
    assert result.parallel_events_overlapped is False
    assert result.status == "failed"
    assert "parallel event handlers did not overlap" in result.semantic_failures


@pytest.mark.asyncio
@pytest.mark.parametrize(
    ("case_id", "missing_usage_calls", "expected_missing"),
    [
        (benchmark.BenchmarkCaseId.ORDINARY_CHAT, {1}, 1),
        (benchmark.BenchmarkCaseId.WEB_SEARCH_TWO_ANSWERS, {2}, 1),
    ],
)
async def test_runner_rejects_missing_or_partial_model_usage(
    case_id: benchmark.BenchmarkCaseId,
    missing_usage_calls: set[int],
    expected_missing: int,
):
    case = benchmark.BENCHMARK_CASE_CATALOG[case_id]
    client = SelectiveUsageBenchmarkClient(
        case,
        benchmark.BenchmarkMode.DUAL_AGENT,
        missing_usage_calls=missing_usage_calls,
    )
    config = benchmark.BenchmarkConfig(
        schema_version=1,
        base_url="https://provider.example/v1",
        model="benchmark-model",
        cases=[case_id],
        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 == "failed"
    assert (
        result.prompt_tokens,
        result.completion_tokens,
        result.total_tokens,
        result.cached_tokens,
    ) == (None, None, None, None)
    assert f"model usage missing for {expected_missing} call(s)" in (
        result.semantic_failures
    )


@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[0].model_call_count == 0
    assert results[0].fallback_count == 0
    assert results[0].prompt_tokens is None
    assert results[0].completion_tokens is None
    assert results[0].total_tokens is None
    assert results[0].cached_tokens is None
    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 == "failed"
    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 result.semantic_failures == [
        "fallback_count expected 0, ledger got 1",
        "fallback_count expected 0, timing got 1",
    ]
    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


def _semantic_failures_for_counts(
    *,
    ledger_model_count: int,
    timing_model_count: int,
    ledger_fallback_count: int = 0,
    timing_fallback_count: int = 0,
) -> list[str]:
    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],
    )
    runner = benchmark.BenchmarkRunner(config, api_key=None, mock=True)
    timings = [
        benchmark.BenchmarkModelCallTiming(
            call_index=index,
            call_kind=call_kind,
            first_item_kind="raw_chunk",
            provider_ttft_ms=0,
            visible_ttft_ms=0 if call_kind == "chat_completion" else None,
            elapsed_ms=0,
            usage=TokenUsage(),
        )
        for index, call_kind in enumerate(
            ["chat_completion"] * timing_model_count
            + ["argument_fallback"] * timing_fallback_count,
            start=1,
        )
    ]
    return runner._semantic_failures(
        case=benchmark.BENCHMARK_CASE_CATALOG[
            benchmark.BenchmarkCaseId.ORDINARY_CHAT
        ],
        mode=benchmark.BenchmarkMode.DUAL_AGENT,
        outputs=[{"type": "message_delta", "content": "Ordinary answer."}],
        audits=[],
        event_names=[],
        batch_event_names=[],
        tool_event_names=[],
        event_sources=[],
        tool_statuses=[],
        model_call_count=ledger_model_count,
        fallback_count=ledger_fallback_count,
        tool_count=0,
        timings=timings,
        check_mock_text=True,
    )


@pytest.mark.parametrize("actual_count", [0, 2])
def test_semantics_reject_missing_or_extra_model_calls_even_when_ledgers_agree(
    actual_count: int,
):
    failures = _semantic_failures_for_counts(
        ledger_model_count=actual_count,
        timing_model_count=actual_count,
    )

    assert failures == [
        f"model_call_count expected 1, ledger got {actual_count}",
        f"model_call_count expected 1, timing got {actual_count}",
    ]


def test_semantics_reject_extra_fallback_calls_even_when_ledgers_agree():
    failures = _semantic_failures_for_counts(
        ledger_model_count=1,
        timing_model_count=1,
        ledger_fallback_count=1,
        timing_fallback_count=1,
    )

    assert failures == [
        "fallback_count expected 0, ledger got 1",
        "fallback_count expected 0, timing got 1",
    ]


def test_semantics_keep_timing_vs_ledger_consistency_failure():
    failures = _semantic_failures_for_counts(
        ledger_model_count=1,
        timing_model_count=2,
    )

    assert failures == [
        "model_call_count expected 1, timing got 2",
        "model_call_count ledger=1, timing=2",
    ]


def test_build_result_rejects_wrong_event_identity_in_all_ledgers_and_pairs_status():
    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],
    )
    runner = benchmark.BenchmarkRunner(config, api_key=None, mock=True)
    timing = benchmark.BenchmarkModelCallTiming(
        call_index=1,
        call_kind="chat_completion",
        first_item_kind="raw_chunk",
        provider_ttft_ms=1,
        visible_ttft_ms=2,
        elapsed_ms=3,
        usage=TokenUsage(total_tokens=5),
    )

    result = runner._build_result(
        case=benchmark.BENCHMARK_CASE_CATALOG[
            benchmark.BenchmarkCaseId.DEVICE_VOLUME_SILENT
        ],
        mode=benchmark.BenchmarkMode.DUAL_AGENT,
        iteration=1,
        outputs=[
            {
                "type": "message_delta",
                "content": "I will set the volume to 30.",
            },
            {"type": "tool_result", "message": {}},
        ],
        audits=[
            {
                "event": "chat_event_detected",
                "details": {
                    "event_name": "wrong.detected",
                    "event_source": "text_event",
                },
            },
            {
                "event": "event_batch_results",
                "details": {
                    "results": [
                        {"event_name": "wrong.batch", "status": "success"}
                    ]
                },
            },
        ],
        usage={
            "calls": [
                {
                    "call_kind": "chat_completion",
                    "event_name": None,
                    "tool_latency_ms": None,
                },
                {
                    "call_kind": "tool_execution",
                    "event_name": "wrong.tool",
                    "tool_latency_ms": 4,
                },
            ],
            "session": {
                "prompt_tokens": 2,
                "completion_tokens": 3,
                "total_tokens": 5,
                "cached_tokens": 0,
                "fallback_count": 0,
                "tool_count": 1,
                "turn_wall_time_ms": 10,
            },
        },
        timings=[timing],
        visible_ttft_ms=2,
        runtime_error=None,
    )

    assert result.status == "failed"
    assert result.event_names == ["wrong.detected"]
    assert result.batch_event_names == ["wrong.batch"]
    assert result.tool_event_names == ["wrong.tool"]
    assert result.tool_statuses == ["success"]
    assert result.semantic_failures == [
        "event names expected ('device.volume.adjust',), got ('wrong.detected',)",
        "batch event names expected ('device.volume.adjust',), got ('wrong.batch',)",
        "tool event names expected ('device.volume.adjust',), got ('wrong.tool',)",
        "event batch statuses expected "
        "(('device.volume.adjust', 'success'),), got (('wrong.batch', 'success'),)",
    ]


def test_build_result_reports_timing_attempt_counts_and_flags_ledger_mismatch():
    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],
    )
    runner = benchmark.BenchmarkRunner(config, api_key=None, mock=True)
    timing = benchmark.BenchmarkModelCallTiming(
        call_index=1,
        call_kind="chat_completion",
        first_item_kind="raw_chunk",
        provider_ttft_ms=1,
        visible_ttft_ms=2,
        elapsed_ms=3,
        usage=TokenUsage(total_tokens=5),
    )

    result = runner._build_result(
        case=benchmark.BENCHMARK_CASE_CATALOG[
            benchmark.BenchmarkCaseId.ORDINARY_CHAT
        ],
        mode=benchmark.BenchmarkMode.DUAL_AGENT,
        iteration=1,
        outputs=[{"type": "message_delta", "content": "Ordinary answer."}],
        audits=[],
        usage={
            "calls": [],
            "session": {
                "prompt_tokens": 0,
                "completion_tokens": 0,
                "total_tokens": 0,
                "cached_tokens": 0,
                "fallback_count": 0,
                "tool_count": 0,
                "turn_wall_time_ms": 3,
            },
        },
        timings=[timing],
        visible_ttft_ms=2,
        runtime_error=None,
    )

    assert result.status == "failed"
    assert result.model_call_count == 1
    assert result.fallback_count == 0
    assert result.semantic_failures == [
        "model_call_count expected 1, ledger got 0",
        "model_call_count ledger=0, timing=1",
    ]


def test_failed_result_uses_timing_attempt_counts_and_returned_usage():
    timings = [
        benchmark.BenchmarkModelCallTiming(
            call_index=1,
            call_kind="chat_completion",
            first_item_kind="raw_chunk",
            provider_ttft_ms=1,
            visible_ttft_ms=2,
            elapsed_ms=3,
            usage=TokenUsage(
                prompt_tokens=2,
                completion_tokens=3,
                total_tokens=5,
                cached_tokens=1,
            ),
        ),
        benchmark.BenchmarkModelCallTiming(
            call_index=2,
            call_kind="argument_fallback",
            first_item_kind="raw_chunk",
            provider_ttft_ms=4,
            visible_ttft_ms=None,
            elapsed_ms=6,
            usage=TokenUsage(
                prompt_tokens=7,
                completion_tokens=11,
                total_tokens=18,
                cached_tokens=2,
            ),
        ),
    ]

    result = benchmark.BenchmarkRunner._failed_result(
        benchmark.BenchmarkCaseId.ORDINARY_CHAT,
        benchmark.BenchmarkMode.DUAL_AGENT,
        1,
        error="failed before persistence",
        timings=timings,
    )

    assert result.model_call_count == 1
    assert result.fallback_count == 1
    assert (
        result.prompt_tokens,
        result.completion_tokens,
        result.total_tokens,
        result.cached_tokens,
    ) == (9, 14, 23, 3)
    assert result.batch_event_names == []
    assert result.tool_event_names == []


def test_failed_result_keeps_tokens_nullable_when_no_timing_usage_returned():
    timing = benchmark.BenchmarkModelCallTiming(
        call_index=1,
        call_kind="chat_completion",
        first_item_kind="raw_chunk",
        provider_ttft_ms=1,
        visible_ttft_ms=None,
        elapsed_ms=2,
        usage=None,
    )

    result = benchmark.BenchmarkRunner._failed_result(
        benchmark.BenchmarkCaseId.ORDINARY_CHAT,
        benchmark.BenchmarkMode.DUAL_AGENT,
        1,
        error="failed without usage",
        timings=[timing],
    )

    assert result.model_call_count == 1
    assert result.fallback_count == 0
    assert (
        result.prompt_tokens,
        result.completion_tokens,
        result.total_tokens,
        result.cached_tokens,
    ) == (None, None, None, None)
    assert result.semantic_failures == ["model usage missing for 1 call(s)"]


def test_failed_result_does_not_sum_partial_timing_usage():
    timings = [
        benchmark.BenchmarkModelCallTiming(
            call_index=1,
            call_kind="chat_completion",
            first_item_kind="raw_chunk",
            provider_ttft_ms=1,
            visible_ttft_ms=2,
            elapsed_ms=3,
            usage=TokenUsage(
                prompt_tokens=2,
                completion_tokens=3,
                total_tokens=5,
                cached_tokens=1,
            ),
        ),
        benchmark.BenchmarkModelCallTiming(
            call_index=2,
            call_kind="argument_fallback",
            first_item_kind="raw_chunk",
            provider_ttft_ms=4,
            visible_ttft_ms=None,
            elapsed_ms=6,
            usage=None,
        ),
    ]

    result = benchmark.BenchmarkRunner._failed_result(
        benchmark.BenchmarkCaseId.ORDINARY_CHAT,
        benchmark.BenchmarkMode.DUAL_AGENT,
        1,
        error="failed with partial usage",
        timings=timings,
    )

    assert (
        result.prompt_tokens,
        result.completion_tokens,
        result.total_tokens,
        result.cached_tokens,
    ) == (None, None, None, None)
    assert result.semantic_failures == ["model usage missing for 1 call(s)"]