fix: preserve benchmark metric integrity

Problem: model elapsed included consumer backpressure, missing provider usage produced false zero or partial token totals, and the parallel benchmark only checked successful event cardinality instead of handler overlap.

Risk: providers that omit usage now produce failed benchmark samples with null token metrics; a non-overlapping parallel case may wait up to the 50 ms probe barrier timeout before completing.

README.md

@@ -161,7 +161,7 @@ The six built-in cases are:
 | `calendar_schedule_template` | One schedule event followed by the deterministic confirmation template. |
 | `web_search_two_answers` | A fast first answer, one search event, and exactly one grounded second answer. |
 | `session_terminate` | One farewell, one terminate event, and a terminal session. |
-| `parallel_volume_schedule` | Volume and schedule events execute in one accepted batch with stable results. |
+| `parallel_volume_schedule` | Volume and schedule events both succeed and their async handlers are proven to overlap. |
 
 Each run records provider TTFT (first provider stream item), visible TTFT (first
 non-whitespace user-visible delta), turn wall time, per-call model latency, tool
@@ -170,6 +170,15 @@ counts, and event names, sources, statuses, and semantic correctness. Provider
 TTFT and visible TTFT are intentionally separate; model and tool spans may
 overlap and are not added to manufacture a total latency.
 
+Per-call model elapsed time accumulates only time awaiting the provider stream
+iterator. It excludes downstream runtime work and consumer backpressure after a
+yield, while provider TTFT and visible TTFT remain wall-clock arrival metrics.
+Token totals are reported only when every model and argument-fallback call
+returns usage; otherwise all token metrics are `null` and the run fails with an
+explicit missing-usage semantic error. The parallel volume/schedule case uses a
+benchmark-owned barrier probe and passes only when both event handlers are
+active at the same time, not merely when two successful event results exist.
+
 By default, reports are published as one complete bundle:
 
 ```text

src/agent_lab/application/benchmark.py

@@ -17,6 +17,10 @@ from agent_lab.application.contracts import (
     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
@@ -50,6 +54,7 @@ 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):
@@ -174,6 +179,79 @@ class BenchmarkRunResult(_StrictBenchmarkModel):
     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:
@@ -202,14 +280,36 @@ class TimingChatClient:
         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
         try:
             stream_chat = getattr(self.inner, "stream_chat")
-            async for item in 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_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
@@ -224,24 +324,29 @@ class TimingChatClient:
                     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)
+            try:
+                close = getattr(iterator, "aclose", None)
+                if close is not None:
+                    await close()
+            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)
@@ -607,6 +712,7 @@ class MockBenchmarkChatClient:
 
 
 BenchmarkClientFactory: TypeAlias = Callable[[BenchmarkConfig, str], object]
+BenchmarkParallelProbeFactory: TypeAlias = Callable[[], BenchmarkParallelEventProbe]
 
 
 class _BenchmarkOutputQueue(asyncio.Queue[dict[str, Any]]):
@@ -634,6 +740,7 @@ class BenchmarkRunner:
         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():
@@ -642,6 +749,9 @@ class BenchmarkRunner:
         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]:
@@ -662,6 +772,7 @@ class BenchmarkRunner:
         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:
@@ -678,10 +789,18 @@ class BenchmarkRunner:
                 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=build_default_tool_registry(),
+                registry=registry,
                 session_store=store,
                 clock=self.clock,
             )
@@ -704,6 +823,11 @@ class BenchmarkRunner:
                 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(
@@ -774,6 +898,7 @@ class BenchmarkRunner:
         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"]
@@ -808,6 +933,7 @@ class BenchmarkRunner:
         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,
@@ -824,6 +950,8 @@ class BenchmarkRunner:
             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"),
@@ -840,10 +968,10 @@ class BenchmarkRunner:
             ),
             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"]),
+            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,
@@ -856,6 +984,7 @@ class BenchmarkRunner:
             semantic_failures=semantic_failures,
             error=runtime_error,
             model_calls=list(timings),
+            parallel_events_overlapped=parallel_events_overlapped,
         )
 
     def _semantic_failures(
@@ -875,6 +1004,8 @@ class BenchmarkRunner:
         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)
@@ -977,6 +1108,15 @@ class BenchmarkRunner:
             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(
@@ -1031,17 +1171,7 @@ class BenchmarkRunner:
         fallback_count = sum(
             timing.call_kind == "argument_fallback" for timing in timings
         )
-        usages = [timing.usage for timing in timings if timing.usage is not None]
-        prompt_tokens = (
-            sum(usage.prompt_tokens for usage in usages) if usages else None
-        )
-        completion_tokens = (
-            sum(usage.completion_tokens for usage in usages) if usages else None
-        )
-        total_tokens = sum(usage.total_tokens for usage in usages) if usages else None
-        cached_tokens = (
-            sum(usage.cached_tokens for usage in usages) if usages else None
-        )
+        token_totals, missing_usage_count = BenchmarkRunner._token_totals(timings)
         return BenchmarkRunResult(
             case_id=case_id,
             mode=mode,
@@ -1052,10 +1182,10 @@ class BenchmarkRunner:
             ),
             visible_ttft_ms=None,
             turn_wall_time_ms=None,
-            prompt_tokens=prompt_tokens,
-            completion_tokens=completion_tokens,
-            total_tokens=total_tokens,
-            cached_tokens=cached_tokens,
+            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,
@@ -1065,11 +1195,35 @@ class BenchmarkRunner:
             event_sources=[],
             tool_statuses=[],
             tool_latencies_ms=[],
-            semantic_failures=[],
+            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,

src/agent_lab/infrastructure/benchmark_reporting.py

@@ -460,8 +460,8 @@ def render_markdown(report: BenchmarkReport) -> str:
             "",
             "## Runs",
             "",
-            "| Case | Mode | Iteration | Status | Error |",
-            "| --- | --- | ---: | --- | --- |",
+            "| Case | Mode | Iteration | Status | Parallel overlap | Error |",
+            "| --- | --- | ---: | --- | --- | --- |",
         ]
     )
     for run in payload["runs"]:
@@ -480,6 +480,11 @@ def render_markdown(report: BenchmarkReport) -> str:
                     _escape_markdown(str(run["mode"])),
                     str(run["iteration"]),
                     str(run["status"]),
+                    (
+                        str(run["parallel_events_overlapped"]).lower()
+                        if run["parallel_events_overlapped"] is not None
+                        else "-"
+                    ),
                     _escape_markdown(str(error_text)),
                 ]
             )

tests/test_benchmark_reporting.py

@@ -62,6 +62,7 @@ def _run(
     semantic_failures: list[str] | None = None,
     error: str | None = None,
     model_calls: list[BenchmarkModelCallTiming] | None = None,
+    parallel_events_overlapped: bool | None = None,
 ) -> BenchmarkRunResult:
     return BenchmarkRunResult(
         case_id=case_id,
@@ -87,6 +88,7 @@ def _run(
         semantic_failures=semantic_failures or [],
         error=error,
         model_calls=model_calls or [],
+        parallel_events_overlapped=parallel_events_overlapped,
     )
 
 
@@ -625,6 +627,29 @@ def test_markdown_preserves_runtime_error_and_semantic_failures_together():
     assert "expected two visible answers" in markdown
 
 
+def test_markdown_runs_table_shows_parallel_event_overlap():
+    reporting = _reporting_module()
+    report = reporting.build_benchmark_report(
+        _config(cases=[BenchmarkCaseId.PARALLEL_VOLUME_SCHEDULE]),
+        [
+            _run(
+                case_id=BenchmarkCaseId.PARALLEL_VOLUME_SCHEDULE,
+                parallel_events_overlapped=True,
+            )
+        ],
+        mock=True,
+    )
+
+    payload = json.loads(reporting.report_to_json(report))
+    markdown = reporting.render_markdown(report)
+
+    assert payload["runs"][0]["parallel_events_overlapped"] is True
+    assert "| Parallel overlap |" in markdown
+    assert "| parallel_volume_schedule | dual_agent | 1 | passed | true | - |" in (
+        markdown
+    )
+
+
 def test_atomic_write_publishes_complete_bundle_through_relative_symlink(
     tmp_path: Path,
 ):

tests/test_benchmark_runner.py

@@ -7,7 +7,11 @@ 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.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
@@ -204,6 +208,110 @@ class ConcurrentTimingClient:
         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")
@@ -298,6 +406,29 @@ async def test_timing_client_keeps_provider_visible_and_elapsed_timings_separate
     ]
 
 
+@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()
@@ -552,6 +683,95 @@ async def test_mock_runner_passes_all_catalog_cases_in_both_modes():
     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
@@ -1094,3 +1314,48 @@ def test_failed_result_keeps_tokens_nullable_when_no_timing_usage_returned():
         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)"]