fix: enforce benchmark visible ttft and call counts

Problem: Empty or whitespace-only message deltas could establish visible TTFT, and semantic validation accepted extra or missing model/fallback calls whenever the usage ledger and timing wrapper agreed with each other.

Risk: Benchmark reports could show artificially early visible latency or hide unexpected model cost. Per-case expected call counts are now explicit and checked independently against both persisted ledger data and timing records while retaining cross-source consistency checks.

src/agent_lab/application/benchmark.py

@@ -212,7 +212,11 @@ class TimingChatClient:
                 if first_item_at is None:
                     first_item_at = item_at
                     first_item_kind = item.kind
-                if item.kind == "message_delta" and first_message_delta_at is None:
+                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
@@ -253,6 +257,8 @@ 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
 
@@ -340,6 +346,8 @@ BENCHMARK_CASE_CATALOG: Mapping[BenchmarkCaseId, BenchmarkCase] = MappingProxyTy
                 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,
             ),
@@ -354,6 +362,8 @@ BENCHMARK_CASE_CATALOG: Mapping[BenchmarkCaseId, BenchmarkCase] = MappingProxyTy
                 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,
             ),
@@ -377,6 +387,8 @@ BENCHMARK_CASE_CATALOG: Mapping[BenchmarkCaseId, BenchmarkCase] = MappingProxyTy
                 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,
             ),
@@ -397,6 +409,8 @@ BENCHMARK_CASE_CATALOG: Mapping[BenchmarkCaseId, BenchmarkCase] = MappingProxyTy
                 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,
             ),
@@ -417,6 +431,8 @@ BENCHMARK_CASE_CATALOG: Mapping[BenchmarkCaseId, BenchmarkCase] = MappingProxyTy
                 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,
             ),
@@ -443,6 +459,8 @@ BENCHMARK_CASE_CATALOG: Mapping[BenchmarkCaseId, BenchmarkCase] = MappingProxyTy
                     "calendar.schedule.create",
                 ),
                 answer_count=2,
+                expected_model_call_count=1,
+                expected_fallback_count=0,
                 terminal=False,
                 first_reply_before_tool=True,
             ),
@@ -596,8 +614,11 @@ class _BenchmarkOutputQueue(asyncio.Queue[dict[str, Any]]):
         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()
@@ -886,6 +907,27 @@ class BenchmarkRunner:
         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}"

tests/test_benchmark_config.py

@@ -10,6 +10,7 @@ from agent_lab.application.benchmark import (
     BENCHMARK_MODES,
     BenchmarkCaseId,
     BenchmarkConfig,
+    BenchmarkExpectation,
     BenchmarkMode,
     BenchmarkTarget,
     build_benchmark_request,
@@ -239,6 +240,43 @@ def test_builtin_case_catalog_is_complete_and_immutable():
         first_event.arguments[0] = ("mode", "relative")
 
 
+def test_builtin_case_catalog_declares_expected_model_and_fallback_counts():
+    assert {
+        case_id: (
+            case.expectation.expected_model_call_count,
+            case.expectation.expected_fallback_count,
+        )
+        for case_id, case in BENCHMARK_CASE_CATALOG.items()
+    } == {
+        BenchmarkCaseId.ORDINARY_CHAT: (1, 0),
+        BenchmarkCaseId.DEVICE_VOLUME_SILENT: (1, 0),
+        BenchmarkCaseId.CALENDAR_SCHEDULE_TEMPLATE: (1, 0),
+        BenchmarkCaseId.WEB_SEARCH_TWO_ANSWERS: (2, 0),
+        BenchmarkCaseId.SESSION_TERMINATE: (1, 0),
+        BenchmarkCaseId.PARALLEL_VOLUME_SCHEDULE: (1, 0),
+    }
+
+
+@pytest.mark.parametrize(
+    ("expected_model_call_count", "expected_fallback_count"),
+    [(0, 0), (1, -1)],
+)
+def test_benchmark_expectation_rejects_invalid_expected_call_counts(
+    expected_model_call_count: int,
+    expected_fallback_count: int,
+):
+    with pytest.raises(ValidationError):
+        BenchmarkExpectation(
+            visible_messages=("answer",),
+            event_names=(),
+            answer_count=1,
+            terminal=False,
+            first_reply_before_tool=False,
+            expected_model_call_count=expected_model_call_count,
+            expected_fallback_count=expected_fallback_count,
+        )
+
+
 def test_builtin_cases_own_semantics_and_deterministic_mock_data():
     actual = {
         case_id: {

tests/test_benchmark_runner.py

@@ -54,6 +54,33 @@ class ScriptedTimingClient:
         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
@@ -269,6 +296,42 @@ async def test_timing_client_keeps_provider_visible_and_elapsed_timings_separate
     ]
 
 
+@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()
@@ -651,7 +714,7 @@ async def test_runner_reports_argument_fallback_model_and_tool_ledgers(monkeypat
         ).run()
     )[0]
 
-    assert result.status == "passed"
+    assert result.status == "failed"
     assert result.model_call_count == 1
     assert result.fallback_count == 1
     assert result.tool_count == 1
@@ -663,6 +726,10 @@ async def test_runner_reports_argument_fallback_model_and_tool_ledgers(monkeypat
     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
 
@@ -694,3 +761,93 @@ def test_visible_answer_order_ignores_empty_deltas_before_a_tool_result():
     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=[],
+        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",
+    ]