fix: cross-check benchmark event ledgers

Problem: Benchmark semantics compared detected event names and aggregate counts but discarded event identities from batch-result audits and tool-execution usage rows. Failed pre-ledger runs also erased timing-derived attempts and any usage already returned.

Risk: A successful status or matching count for the wrong event could be reported as a passing benchmark, while failed runs could underreport attempted model calls and token cost. Results now preserve and cross-check all event ledgers, pair batch identities with statuses, and retain timing-derived attempts and usage.

src/agent_lab/application/benchmark.py

@@ -166,6 +166,8 @@ class BenchmarkRunResult(_StrictBenchmarkModel):
     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]
@@ -787,13 +789,25 @@ class BenchmarkRunner:
             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"]
-        model_call_count = sum(
+        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
         )
-        fallback_count = int(session_usage["fallback_count"])
+        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
+        )
         tool_count = int(session_usage["tool_count"])
         semantic_failures = self._semantic_failures(
             case=case,
@@ -801,10 +815,12 @@ class BenchmarkRunner:
             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=model_call_count,
-            fallback_count=fallback_count,
+            model_call_count=ledger_model_call_count,
+            fallback_count=ledger_fallback_count,
             tool_count=tool_count,
             timings=timings,
             check_mock_text=self.mock,
@@ -828,10 +844,12 @@ class BenchmarkRunner:
             completion_tokens=int(session_usage["completion_tokens"]),
             total_tokens=int(session_usage["total_tokens"]),
             cached_tokens=int(session_usage["cached_tokens"]),
-            model_call_count=model_call_count,
-            fallback_count=fallback_count,
+            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],
@@ -848,6 +866,8 @@ class BenchmarkRunner:
         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,
@@ -880,6 +900,16 @@ class BenchmarkRunner:
             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
@@ -891,11 +921,22 @@ class BenchmarkRunner:
                 f"event sources expected {expected_sources!r}, "
                 f"got {tuple(event_sources)!r}"
             )
-        expected_statuses = ("success",) * len(expected.event_names)
-        if tuple(tool_statuses) != expected_statuses:
+        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"tool statuses expected {expected_statuses!r}, "
-                f"got {tuple(tool_statuses)!r}"
+                f"event batch statuses expected {expected_statuses!r}, "
+                f"got {actual_statuses!r}"
             )
         if tool_count != len(expected.event_names):
             failures.append(
@@ -984,6 +1025,23 @@ class BenchmarkRunner:
             (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
+        )
+        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
+        )
         return BenchmarkRunResult(
             case_id=case_id,
             mode=mode,
@@ -994,14 +1052,16 @@ class BenchmarkRunner:
             ),
             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,
+            prompt_tokens=prompt_tokens,
+            completion_tokens=completion_tokens,
+            total_tokens=total_tokens,
+            cached_tokens=cached_tokens,
+            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=[],

tests/test_benchmark_runner.py

@@ -238,6 +238,8 @@ def test_benchmark_result_models_are_strict_and_allow_nullable_metrics():
         fallback_count=None,
         tool_count=None,
         event_names=[],
+        batch_event_names=[],
+        tool_event_names=[],
         event_sources=[],
         tool_statuses=[],
         tool_latencies_ms=[],
@@ -510,6 +512,12 @@ async def test_mock_runner_is_ordered_ignores_factory_and_reports_event_ledgers(
     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
@@ -526,6 +534,8 @@ async def test_mock_runner_is_ordered_ignores_factory_and_reports_event_ledgers(
         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)
 
 
@@ -630,6 +640,12 @@ async def test_runner_continues_after_one_factory_failure():
 
     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
 
@@ -802,6 +818,8 @@ def _semantic_failures_for_counts(
         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,
@@ -851,3 +869,228 @@ def test_semantics_keep_timing_vs_ledger_consistency_failure():
         "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)