feat: add comparable invocation metrics

Problem: ChatAgent-only round usage hid EventAgent fallback cost, logical tool spans, and true turn wall time, preventing fair dual-agent versus direct-tool comparison.

Risk: Legacy experimental databases containing duplicate non-null metric_key rows still require a follow-up deduplication migration before the unique index can be created.

src/agent_lab/application/event_agent.py

@@ -1,8 +1,9 @@
 import asyncio
 import json
+import time
 from collections.abc import AsyncIterator, Hashable, Iterable, Sequence
 from dataclasses import dataclass
-from typing import Any, Protocol
+from typing import Any, Callable, Protocol
 
 from agent_lab.application.contracts import AgentParams, EventAgentParams
 from agent_lab.application.events import (
@@ -20,7 +21,7 @@ from agent_lab.application.tools import (
     build_default_tool_registry,
 )
 from agent_lab.domain.events import ToolCallEvent
-from agent_lab.domain.messages import ChatMessage, StreamItem
+from agent_lab.domain.messages import ChatMessage, ModelCallMetric, StreamItem, TokenUsage
 
 
 class EventAgentChatClient(Protocol):
@@ -55,16 +56,19 @@ class EventAgent:
         registry: ToolRegistry | None = None,
         chat_client: EventAgentChatClient | None = None,
         params: AgentParams | None = None,
+        monotonic_clock: Callable[[], float] | None = None,
     ) -> None:
         self.enabled_tools = set(enabled_tools)
         self.registry = registry or build_default_tool_registry()
         self.chat_client = chat_client
         self.params = params or EventAgentParams()
+        self._clock = monotonic_clock or time.monotonic
         self.kernel = EventKernel(
             self.registry.event_registry,
             argument_fallback=(
                 self._resolve_arguments_with_llm if chat_client is not None else None
             ),
+            monotonic_clock=self._clock,
         )
         self.executor = EventBatchExecutor(
             self.kernel,
@@ -76,6 +80,7 @@ class EventAgent:
             ),
         )
         self._raw_chunks_by_event: dict[str, list[dict[str, Any]]] = {}
+        self._fallback_calls_by_event: dict[str, list[ModelCallMetric]] = {}
 
     async def handle(
         self,
@@ -103,6 +108,7 @@ class EventAgent:
     ) -> EventBatchResult:
         for event in events:
             self._raw_chunks_by_event[event.id] = []
+            self._fallback_calls_by_event[event.id] = []
         resolved_scope = scope if scope is not None else object()
         try:
             return await self.executor.execute(
@@ -134,6 +140,16 @@ class EventAgent:
             for event in events
         ]
 
+    def fallback_model_calls(
+        self,
+        events: Sequence[ToolCallEvent],
+    ) -> list[dict[str, Any]]:
+        return [
+            metric.as_dict()
+            for event in events
+            for metric in self._fallback_calls_by_event.get(event.id, [])
+        ]
+
     async def _resolve_arguments_with_llm(
         self,
         definition: EventDefinition,
@@ -161,29 +177,65 @@ class EventAgent:
             }
         )
         raw_chunks: list[dict[str, Any]] = []
-        async for item in self.chat_client.stream_chat(
-            messages=messages,
-            tools=[tool],
-            params=params,
-            tool_choice={
-                "type": "function",
-                "function": {"name": event.name},
-            },
-        ):
-            if item.kind == "raw_chunk" and item.raw_chunk is not None:
-                raw_chunks.append(item.raw_chunk)
-                continue
-
-            if item.kind == "provider_tool_call" and item.event is not None:
-                self._raw_chunks_by_event[event.id] = raw_chunks
-                return ResolvedEventArguments(
-                    event_name=item.event.name,
-                    arguments=dict(item.event.arguments),
-                    raw_arguments=item.event.raw_arguments,
+        usage = TokenUsage()
+        resolved: ResolvedEventArguments | None = None
+        started_at = self._clock()
+        first_item_at: float | None = None
+        try:
+            async for item in self.chat_client.stream_chat(
+                messages=messages,
+                tools=[tool],
+                params=params,
+                tool_choice={
+                    "type": "function",
+                    "function": {"name": event.name},
+                },
+            ):
+                if first_item_at is None:
+                    first_item_at = self._clock()
+                if item.kind == "raw_chunk" and item.raw_chunk is not None:
+                    raw_chunks.append(item.raw_chunk)
+                    continue
+                if item.kind == "usage" and item.usage is not None:
+                    usage = item.usage
+                    continue
+                if (
+                    resolved is None
+                    and item.kind == "provider_tool_call"
+                    and item.event is not None
+                ):
+                    resolved = ResolvedEventArguments(
+                        event_name=item.event.name,
+                        arguments=dict(item.event.arguments),
+                        raw_arguments=item.event.raw_arguments,
+                    )
+        finally:
+            self._raw_chunks_by_event[event.id] = raw_chunks
+            self._fallback_calls_by_event.setdefault(event.id, []).append(
+                ModelCallMetric(
+                    event_id=event.id,
+                    event_name=event.name,
+                    usage=usage,
+                    ttft_ms=(
+                        None
+                        if first_item_at is None
+                        else self._elapsed_ms(started_at, ended_at=first_item_at)
+                    ),
+                    elapsed_ms=self._elapsed_ms(started_at),
                 )
+            )
+        return resolved
 
-        self._raw_chunks_by_event[event.id] = raw_chunks
-        return None
+    def _elapsed_ms(
+        self,
+        started_at: float,
+        *,
+        ended_at: float | None = None,
+    ) -> int:
+        return max(
+            0,
+            round(((self._clock() if ended_at is None else ended_at) - started_at) * 1000),
+        )
 
     def _build_argument_messages(
         self,

src/agent_lab/application/events/kernel.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 import asyncio
 import inspect
 import json
+import time
 from collections.abc import Awaitable, Callable, Iterable
 from dataclasses import dataclass, replace
 from typing import Any
@@ -39,9 +40,11 @@ class EventKernel:
         self,
         registry: EventRegistry,
         argument_fallback: ArgumentFallback | None = None,
+        monotonic_clock: Callable[[], float] | None = None,
     ) -> None:
         self.registry = registry
         self.argument_fallback = argument_fallback
+        self._clock = monotonic_clock or time.monotonic
 
     async def execute(
         self,
@@ -152,6 +155,7 @@ class EventKernel:
             arguments=self._strict_json_copy(resolved.arguments),
             raw_arguments=resolved.raw_arguments,
         )
+        handler_started_at = self._clock()
         try:
             if inspect.iscoroutinefunction(definition.handler):
                 payload = definition.handler(resolved_request)
@@ -163,6 +167,7 @@ class EventKernel:
             if inspect.isawaitable(payload):
                 payload = await payload
         except Exception as exc:
+            tool_latency_ms = self._elapsed_ms(handler_started_at)
             return self._result(
                 definition,
                 request,
@@ -170,7 +175,9 @@ class EventKernel:
                 resolved=resolved,
                 error=f"event handler failed: {exc}",
                 used_fallback=used_fallback,
+                tool_latency_ms=tool_latency_ms,
             )
+        tool_latency_ms = self._elapsed_ms(handler_started_at)
         if not isinstance(payload, dict):
             return self._result(
                 definition,
@@ -179,6 +186,7 @@ class EventKernel:
                 resolved=resolved,
                 error="event handler returned non-object payload",
                 used_fallback=used_fallback,
+                tool_latency_ms=tool_latency_ms,
             )
         try:
             payload = self._strict_json_copy(payload)
@@ -190,6 +198,7 @@ class EventKernel:
                 resolved=resolved,
                 error="event handler returned non-JSON payload",
                 used_fallback=used_fallback,
+                tool_latency_ms=tool_latency_ms,
             )
         return self._result(
             definition,
@@ -198,6 +207,7 @@ class EventKernel:
             resolved=resolved,
             payload=dict(payload),
             used_fallback=used_fallback,
+            tool_latency_ms=tool_latency_ms,
         )
 
     def execute_sync(
@@ -254,16 +264,20 @@ class EventKernel:
             arguments=self._strict_json_copy(resolved.arguments),
             raw_arguments=resolved.raw_arguments,
         )
+        handler_started_at = self._clock()
         try:
             payload = definition.handler(resolved_request)
         except Exception as exc:
+            tool_latency_ms = self._elapsed_ms(handler_started_at)
             return self._result(
                 definition,
                 request,
                 EventStatus.HANDLER_ERROR,
                 resolved=resolved,
                 error=f"event handler failed: {exc}",
+                tool_latency_ms=tool_latency_ms,
             )
+        tool_latency_ms = self._elapsed_ms(handler_started_at)
         if inspect.isawaitable(payload):
             if inspect.iscoroutine(payload):
                 payload.close()
@@ -273,6 +287,7 @@ class EventKernel:
                 EventStatus.HANDLER_ERROR,
                 resolved=resolved,
                 error="event handler failed: async event handlers require EventKernel.execute",
+                tool_latency_ms=tool_latency_ms,
             )
         if not isinstance(payload, dict):
             return self._result(
@@ -281,6 +296,7 @@ class EventKernel:
                 EventStatus.HANDLER_ERROR,
                 resolved=resolved,
                 error="event handler returned non-object payload",
+                tool_latency_ms=tool_latency_ms,
             )
         try:
             payload = self._strict_json_copy(payload)
@@ -291,6 +307,7 @@ class EventKernel:
                 EventStatus.HANDLER_ERROR,
                 resolved=resolved,
                 error="event handler returned non-JSON payload",
+                tool_latency_ms=tool_latency_ms,
             )
         return self._result(
             definition,
@@ -298,6 +315,7 @@ class EventKernel:
             EventStatus.SUCCESS,
             resolved=resolved,
             payload=dict(payload),
+            tool_latency_ms=tool_latency_ms,
         )
 
     def _lookup(
@@ -607,6 +625,7 @@ class EventKernel:
         payload: dict[str, Any] | None = None,
         error: str | None = None,
         used_fallback: bool = False,
+        tool_latency_ms: int | None = None,
     ) -> EventResult:
         return EventResult(
             event_id=request.id,
@@ -618,6 +637,7 @@ class EventKernel:
             payload=self._safe_json_object_snapshot(payload or {}),
             error=error,
             used_fallback=used_fallback,
+            tool_latency_ms=tool_latency_ms,
             result_policy=definition.result_policy,
             confirmation_policy=definition.confirmation_policy,
             risk_level=definition.risk_level,
@@ -628,6 +648,9 @@ class EventKernel:
             terminal=definition.terminal,
         )
 
+    def _elapsed_ms(self, started_at: float) -> int:
+        return max(0, round((self._clock() - started_at) * 1000))
+
     def _json_arguments(self, arguments: dict[str, Any]) -> str:
         return json.dumps(
             arguments,

src/agent_lab/application/events/models.py

@@ -113,6 +113,7 @@ class EventResult:
     payload: dict[str, Any] = field(default_factory=dict)
     error: str | None = None
     used_fallback: bool = False
+    tool_latency_ms: int | None = None
     result_policy: ResultPolicy = ResultPolicy.LLM_FOLLOW_UP
     confirmation_policy: ConfirmationPolicy = ConfirmationPolicy.NONE
     risk_level: RiskLevel = RiskLevel.LOW

src/agent_lab/application/runtime.py

@@ -138,6 +138,7 @@ class DebugRuntime:
             registry=self.registry,
             chat_client=self.chat_client,
             params=request.event_agent,
+            monotonic_clock=self.clock,
         )
         return asyncio.create_task(self._consume_events(queues, event_agent))
 
@@ -520,7 +521,8 @@ class DebugRuntime:
         session_id: str | None = None,
         turn_started_at: float | None = None,
     ) -> bool:
-        turn_started_at = turn_started_at or self.clock()
+        if turn_started_at is None:
+            turn_started_at = self.clock()
         messages.append(user_message)
         self._start_persisted_turn(session_id, turn_index, user_message)
         await queues.output.put({"type": "turn_started", "turn_index": turn_index})
@@ -546,6 +548,7 @@ class DebugRuntime:
             round_index += 1
             started_at = self.clock()
             ttft_ms: int | None = None
+            provider_ttft_ms: int | None = None
             prompt_tokens = 0
             completion_tokens = 0
             total_tokens = 0
@@ -603,13 +606,20 @@ class DebugRuntime:
                 tools=chat_tools,
                 params=request.chat_agent,
             ):
+                first_provider_item = provider_ttft_ms is None
+                if first_provider_item:
+                    provider_ttft_ms = self._elapsed_ms(started_at)
                 if item.kind == "raw_chunk" and item.raw_chunk is not None:
                     raw_chunks.append(item.raw_chunk)
                     continue
 
                 if item.kind == "message_delta":
                     if ttft_ms is None:
-                        ttft_ms = self._elapsed_ms(started_at)
+                        ttft_ms = (
+                            provider_ttft_ms
+                            if first_provider_item
+                            else self._elapsed_ms(started_at)
+                        )
                     if not message_stream_started:
                         message_stream_started = True
                         await self._audit(
@@ -787,6 +797,12 @@ class DebugRuntime:
                         {"type": "tool_result", "message": reply.model_dump()}
                     )
                 assert batch_result is not None
+                self._append_persisted_tool_metrics(
+                    session_id,
+                    turn_index,
+                    round_index,
+                    batch_result,
+                )
                 logical_result_count = len(
                     self._logical_batch_results(batch_result)
                 )
@@ -853,8 +869,13 @@ class DebugRuntime:
                     total_tokens=total_tokens,
                     cached_tokens=cached_tokens,
                 ),
-                ttft_ms=ttft_ms,
+                ttft_ms=provider_ttft_ms,
                 elapsed_ms=elapsed_ms,
+                metadata={
+                    "agent": "chat_agent",
+                    "call_kind": "chat_completion",
+                    "metric_key": f"chat:{turn_index}:{round_index}",
+                },
             )
             await queues.output.put(
                 {
@@ -894,6 +915,7 @@ class DebugRuntime:
 
         for deferred_message in deferred_user_messages:
             await queues.input.put(deferred_message)
+        turn_wall_time_ms = self._elapsed_ms(turn_started_at)
         await self._audit(
             queues,
             "turn_completed",
@@ -901,8 +923,13 @@ class DebugRuntime:
             session_id=session_id,
             turn_index=turn_index,
             round_count=round_index,
+            wall_time_ms=turn_wall_time_ms,
+        )
+        self._complete_persisted_turn(
+            session_id,
+            turn_index,
+            wall_time_ms=turn_wall_time_ms,
         )
-        self._complete_persisted_turn(session_id, turn_index)
         await queues.output.put(
             {
                 "type": "turn_completed",
@@ -951,6 +978,26 @@ class DebugRuntime:
                     extra_body=request.extra_body,
                     scope=request.scope,
                 )
+                for metric in event_agent.fallback_model_calls(request.events):
+                    self._append_persisted_usage(
+                        request.session_id,
+                        request.turn_index or 0,
+                        request.round_index or 0,
+                        usage=metric["usage"],
+                        ttft_ms=metric["ttft_ms"],
+                        elapsed_ms=metric["elapsed_ms"],
+                        metadata={
+                            "agent": "event_agent",
+                            "call_kind": "argument_fallback",
+                            "event_id": metric["event_id"],
+                            "event_name": metric["event_name"],
+                            "used_fallback": True,
+                            "metric_key": (
+                                f"fallback:{request.turn_index}:"
+                                f"{request.round_index}:{metric['event_id']}"
+                            ),
+                        },
+                    )
                 replies = event_agent.registry.tool_replies(request.events, batch)
                 await self._audit(
                     queues,
@@ -1092,7 +1139,10 @@ class DebugRuntime:
         executor = self._batch_executors.get(key)
         if executor is None:
             executor = EventBatchExecutor(
-                EventKernel(self.registry.event_registry),
+                EventKernel(
+                    self.registry.event_registry,
+                    monotonic_clock=self.clock,
+                ),
                 max_parallel_events=params.max_parallel_events,
                 batch_timeout_seconds=params.batch_timeout_seconds,
             )
@@ -1353,10 +1403,16 @@ class DebugRuntime:
         self,
         session_id: str | None,
         turn_index: int,
+        *,
+        wall_time_ms: int | None = None,
     ) -> None:
         if self.session_store is None or session_id is None:
             return
-        self.session_store.complete_turn(session_id, turn_index=turn_index)
+        self.session_store.complete_turn(
+            session_id,
+            turn_index=turn_index,
+            wall_time_ms=wall_time_ms,
+        )
 
     def _append_persisted_message(
         self,
@@ -1395,6 +1451,7 @@ class DebugRuntime:
         usage: TokenUsage,
         ttft_ms: int | None,
         elapsed_ms: int,
+        metadata: dict[str, Any] | None = None,
     ) -> None:
         if self.session_store is None or session_id is None:
             return
@@ -1405,8 +1462,38 @@ class DebugRuntime:
             usage=usage,
             ttft_ms=ttft_ms,
             elapsed_ms=elapsed_ms,
+            metadata=metadata,
         )
 
+    def _append_persisted_tool_metrics(
+        self,
+        session_id: str | None,
+        turn_index: int,
+        round_index: int,
+        batch: EventBatchResult,
+    ) -> None:
+        for result in self._logical_batch_results(batch):
+            tool_latency_ms = result.tool_latency_ms or 0
+            self._append_persisted_usage(
+                session_id,
+                turn_index,
+                round_index,
+                usage=TokenUsage(),
+                ttft_ms=None,
+                elapsed_ms=tool_latency_ms,
+                metadata={
+                    "agent": "event_kernel",
+                    "call_kind": "tool_execution",
+                    "event_id": result.event_id,
+                    "event_name": result.event_name,
+                    "used_fallback": result.used_fallback,
+                    "tool_latency_ms": result.tool_latency_ms,
+                    "metric_key": (
+                        f"tool:{turn_index}:{round_index}:{result.event_id}"
+                    ),
+                },
+            )
+
     def _session_title(self, user_message: str) -> str:
         title = " ".join(user_message.split())
         if not title:

src/agent_lab/application/session_store.py

@@ -40,7 +40,13 @@ class SessionStore(Protocol):
     ) -> None:
         ...
 
-    def complete_turn(self, session_id: str, *, turn_index: int) -> None:
+    def complete_turn(
+        self,
+        session_id: str,
+        *,
+        turn_index: int,
+        wall_time_ms: int | None = None,
+    ) -> None:
         ...
 
     def append_message(
@@ -81,6 +87,7 @@ class SessionStore(Protocol):
         usage: TokenUsage,
         ttft_ms: int | None,
         elapsed_ms: int,
+        metadata: dict[str, Any] | None = None,
     ) -> None:
         ...
 

src/agent_lab/domain/messages.py

@@ -43,6 +43,24 @@ class TokenUsage(BaseModel):
     cached_tokens: int = 0
 
 
+@dataclass(frozen=True)
+class ModelCallMetric:
+    event_id: str
+    event_name: str
+    usage: TokenUsage
+    ttft_ms: int | None
+    elapsed_ms: int
+
+    def as_dict(self) -> dict[str, Any]:
+        return {
+            "event_id": self.event_id,
+            "event_name": self.event_name,
+            "usage": self.usage,
+            "ttft_ms": self.ttft_ms,
+            "elapsed_ms": self.elapsed_ms,
+        }
+
+
 @dataclass
 class StreamItem:
     kind: str

src/agent_lab/infrastructure/sqlite_store.py

@@ -159,16 +159,22 @@ class SQLiteSessionStore:
             self._touch_session_locked(session_id, now)
             self._connect().commit()
 
-    def complete_turn(self, session_id: str, *, turn_index: int) -> None:
+    def complete_turn(
+        self,
+        session_id: str,
+        *,
+        turn_index: int,
+        wall_time_ms: int | None = None,
+    ) -> None:
         now = self._now()
         with self._lock:
             self._connect().execute(
                 """
                 UPDATE turns
-                SET completed_at = ?
+                SET completed_at = ?, wall_time_ms = ?
                 WHERE session_id = ? AND turn_index = ?
                 """,
-                (now, session_id, turn_index),
+                (now, wall_time_ms, session_id, turn_index),
             )
             self._touch_session_locked(session_id, now)
             self._connect().commit()
@@ -274,29 +280,42 @@ class SQLiteSessionStore:
         usage: TokenUsage,
         ttft_ms: int | None,
         elapsed_ms: int,
+        metadata: dict[str, Any] | None = None,
     ) -> None:
+        metadata = metadata or {}
+        call_kind = metadata.get("call_kind", "chat_completion")
+        persisted_usage = TokenUsage() if call_kind == "tool_execution" else usage
         now = self._now()
         with self._lock:
             self._connect().execute(
                 """
-                INSERT INTO usage_stats
+                INSERT OR IGNORE INTO usage_stats
                     (
                         session_id, turn_index, round_index, prompt_tokens,
                         completion_tokens, total_tokens, cached_tokens,
-                        ttft_ms, elapsed_ms, created_at
+                        ttft_ms, elapsed_ms, agent, call_kind, event_id,
+                        event_name, used_fallback, tool_latency_ms,
+                        metric_key, created_at
                     )
-                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                 """,
                 (
                     session_id,
                     turn_index,
                     round_index,
-                    usage.prompt_tokens,
-                    usage.completion_tokens,
-                    usage.total_tokens,
-                    usage.cached_tokens,
+                    persisted_usage.prompt_tokens,
+                    persisted_usage.completion_tokens,
+                    persisted_usage.total_tokens,
+                    persisted_usage.cached_tokens,
                     ttft_ms,
                     elapsed_ms,
+                    metadata.get("agent", "chat_agent"),
+                    call_kind,
+                    metadata.get("event_id"),
+                    metadata.get("event_name"),
+                    bool(metadata.get("used_fallback", False)),
+                    metadata.get("tool_latency_ms"),
+                    metadata.get("metric_key"),
                     now,
                 ),
             )
@@ -352,31 +371,52 @@ class SQLiteSessionStore:
         ]
 
     def usage_summary(self, session_id: str) -> dict[str, Any]:
+        session_record = self.get_session(session_id)
+        mode = (
+            session_record["config"].get("tool_invocation_mode", "dual_agent")
+            if session_record is not None
+            else "dual_agent"
+        )
         call_rows = self._fetchall(
             """
             SELECT
                 id, turn_index, round_index, prompt_tokens, completion_tokens,
-                total_tokens, cached_tokens, ttft_ms, elapsed_ms, created_at
+                total_tokens, cached_tokens, ttft_ms, elapsed_ms, agent,
+                call_kind, event_id, event_name, used_fallback,
+                tool_latency_ms, metric_key, created_at
             FROM usage_stats
             WHERE session_id = ?
             ORDER BY id ASC
             """,
             (session_id,),
         )
-        calls = [self._usage_row(row) for row in call_rows]
+        calls = [self._usage_row(row, mode=mode) for row in call_rows]
         turn_rows = self._fetchall(
             """
             SELECT
-                turn_index,
-                COALESCE(SUM(prompt_tokens), 0) AS prompt_tokens,
-                COALESCE(SUM(completion_tokens), 0) AS completion_tokens,
-                COALESCE(SUM(total_tokens), 0) AS total_tokens,
-                COALESCE(SUM(cached_tokens), 0) AS cached_tokens,
-                COALESCE(SUM(elapsed_ms), 0) AS elapsed_ms
-            FROM usage_stats
-            WHERE session_id = ?
-            GROUP BY turn_index
-            ORDER BY turn_index ASC
+                turns.turn_index,
+                COALESCE(SUM(CASE WHEN usage_stats.call_kind != 'tool_execution'
+                    THEN usage_stats.prompt_tokens ELSE 0 END), 0) AS prompt_tokens,
+                COALESCE(SUM(CASE WHEN usage_stats.call_kind != 'tool_execution'
+                    THEN usage_stats.completion_tokens ELSE 0 END), 0) AS completion_tokens,
+                COALESCE(SUM(CASE WHEN usage_stats.call_kind != 'tool_execution'
+                    THEN usage_stats.total_tokens ELSE 0 END), 0) AS total_tokens,
+                COALESCE(SUM(CASE WHEN usage_stats.call_kind != 'tool_execution'
+                    THEN usage_stats.cached_tokens ELSE 0 END), 0) AS cached_tokens,
+                COALESCE(SUM(usage_stats.elapsed_ms), 0) AS elapsed_ms,
+                COUNT(usage_stats.id) AS call_count,
+                COALESCE(SUM(CASE WHEN usage_stats.call_kind = 'argument_fallback'
+                    THEN 1 ELSE 0 END), 0) AS fallback_count,
+                COALESCE(SUM(CASE WHEN usage_stats.call_kind = 'tool_execution'
+                    THEN 1 ELSE 0 END), 0) AS tool_count,
+                turns.wall_time_ms AS turn_wall_time_ms
+            FROM turns
+            JOIN usage_stats
+                ON usage_stats.session_id = turns.session_id
+                AND usage_stats.turn_index = turns.turn_index
+            WHERE turns.session_id = ?
+            GROUP BY turns.turn_index, turns.wall_time_ms
+            ORDER BY turns.turn_index ASC
             """,
             (session_id,),
         )
@@ -388,22 +428,43 @@ class SQLiteSessionStore:
                 "total_tokens": row["total_tokens"],
                 "cached_tokens": row["cached_tokens"],
                 "elapsed_ms": row["elapsed_ms"],
+                "call_count": row["call_count"],
+                "fallback_count": row["fallback_count"],
+                "tool_count": row["tool_count"],
+                "turn_wall_time_ms": row["turn_wall_time_ms"],
             }
             for row in turn_rows
         ]
         session = self._fetchone(
             """
             SELECT
-                COALESCE(SUM(prompt_tokens), 0) AS prompt_tokens,
-                COALESCE(SUM(completion_tokens), 0) AS completion_tokens,
-                COALESCE(SUM(total_tokens), 0) AS total_tokens,
-                COALESCE(SUM(cached_tokens), 0) AS cached_tokens,
-                COALESCE(SUM(elapsed_ms), 0) AS elapsed_ms
+                COALESCE(SUM(CASE WHEN call_kind != 'tool_execution'
+                    THEN prompt_tokens ELSE 0 END), 0) AS prompt_tokens,
+                COALESCE(SUM(CASE WHEN call_kind != 'tool_execution'
+                    THEN completion_tokens ELSE 0 END), 0) AS completion_tokens,
+                COALESCE(SUM(CASE WHEN call_kind != 'tool_execution'
+                    THEN total_tokens ELSE 0 END), 0) AS total_tokens,
+                COALESCE(SUM(CASE WHEN call_kind != 'tool_execution'
+                    THEN cached_tokens ELSE 0 END), 0) AS cached_tokens,
+                COALESCE(SUM(elapsed_ms), 0) AS elapsed_ms,
+                COUNT(*) AS call_count,
+                COALESCE(SUM(CASE WHEN call_kind = 'argument_fallback'
+                    THEN 1 ELSE 0 END), 0) AS fallback_count,
+                COALESCE(SUM(CASE WHEN call_kind = 'tool_execution'
+                    THEN 1 ELSE 0 END), 0) AS tool_count
             FROM usage_stats
             WHERE session_id = ?
             """,
             (session_id,),
         )
+        wall_time = self._fetchone(
+            """
+            SELECT SUM(wall_time_ms) AS turn_wall_time_ms
+            FROM turns
+            WHERE session_id = ?
+            """,
+            (session_id,),
+        )
         return {
             "calls": calls,
             "turns": turns,
@@ -413,10 +474,14 @@ class SQLiteSessionStore:
                 "total_tokens": session["total_tokens"],
                 "cached_tokens": session["cached_tokens"],
                 "elapsed_ms": session["elapsed_ms"],
+                "call_count": session["call_count"],
+                "fallback_count": session["fallback_count"],
+                "tool_count": session["tool_count"],
+                "turn_wall_time_ms": wall_time["turn_wall_time_ms"],
             },
         }
 
-    def _usage_row(self, row: sqlite3.Row) -> dict[str, Any]:
+    def _usage_row(self, row: sqlite3.Row, *, mode: str) -> dict[str, Any]:
         return {
             "id": row["id"],
             "turn_index": row["turn_index"],
@@ -427,6 +492,14 @@ class SQLiteSessionStore:
             "cached_tokens": row["cached_tokens"],
             "ttft_ms": row["ttft_ms"],
             "elapsed_ms": row["elapsed_ms"],
+            "mode": mode,
+            "agent": row["agent"],
+            "call_kind": row["call_kind"],
+            "event_id": row["event_id"],
+            "event_name": row["event_name"],
+            "used_fallback": bool(row["used_fallback"]),
+            "tool_latency_ms": row["tool_latency_ms"],
+            "metric_key": row["metric_key"],
             "created_at": row["created_at"],
         }
 
@@ -476,6 +549,7 @@ class SQLiteSessionStore:
                 user_message TEXT NOT NULL,
                 started_at TEXT NOT NULL,
                 completed_at TEXT,
+                wall_time_ms INTEGER,
                 PRIMARY KEY (session_id, turn_index)
             );
 
@@ -512,6 +586,13 @@ class SQLiteSessionStore:
                 cached_tokens INTEGER NOT NULL,
                 ttft_ms INTEGER,
                 elapsed_ms INTEGER NOT NULL,
+                agent TEXT NOT NULL DEFAULT 'chat_agent',
+                call_kind TEXT NOT NULL DEFAULT 'chat_completion',
+                event_id TEXT,
+                event_name TEXT,
+                used_fallback INTEGER NOT NULL DEFAULT 0,
+                tool_latency_ms INTEGER,
+                metric_key TEXT,
                 created_at TEXT NOT NULL
             );
             """
@@ -532,6 +613,35 @@ class SQLiteSessionStore:
                         or "tool_calls_json" not in self._message_columns()
                     ):
                         raise
+            self._add_column_if_missing(
+                "turns",
+                "wall_time_ms",
+                "INTEGER",
+            )
+            for column_name, declaration in (
+                ("agent", "TEXT NOT NULL DEFAULT 'chat_agent'"),
+                ("call_kind", "TEXT NOT NULL DEFAULT 'chat_completion'"),
+                ("event_id", "TEXT"),
+                ("event_name", "TEXT"),
+                ("used_fallback", "INTEGER NOT NULL DEFAULT 0"),
+                ("tool_latency_ms", "INTEGER"),
+                ("metric_key", "TEXT"),
+            ):
+                self._add_column_if_missing(
+                    "usage_stats",
+                    column_name,
+                    declaration,
+                )
+            self._connection.execute(
+                "DROP INDEX IF EXISTS usage_stats_metric_key_uq"
+            )
+            self._connection.execute(
+                """
+                CREATE UNIQUE INDEX IF NOT EXISTS usage_stats_metric_key_uq
+                ON usage_stats(session_id, metric_key)
+                WHERE metric_key IS NOT NULL
+                """
+            )
             self._connection.commit()
         except BaseException:
             self._connection.rollback()
@@ -544,6 +654,26 @@ class SQLiteSessionStore:
             for row in self._connection.execute("PRAGMA table_info(messages)")
         }
 
+    def _table_columns(self, table_name: str) -> set[str]:
+        assert self._connection is not None
+        return {
+            row[1]
+            for row in self._connection.execute(f"PRAGMA table_info({table_name})")
+        }
+
+    def _add_column_if_missing(
+        self,
+        table_name: str,
+        column_name: str,
+        declaration: str,
+    ) -> None:
+        assert self._connection is not None
+        if column_name in self._table_columns(table_name):
+            return
+        self._connection.execute(
+            f"ALTER TABLE {table_name} ADD COLUMN {column_name} {declaration}"
+        )
+
     def _touch_session_locked(self, session_id: str, now: str) -> None:
         self._connect().execute(
             "UPDATE sessions SET updated_at = ? WHERE id = ?",

src/agent_lab/presentation/static/app.js

@@ -1499,12 +1499,16 @@ function formatTime(value) {
 
 function renderSessionUsage(usage) {
   const session = usage.session || {};
+  const sessionElapsed = session.turn_wall_time_ms ?? session.elapsed_ms ?? 0;
   sessionUsageSummary.textContent = "";
   [
     ["Tokens", session.total_tokens || 0],
     ["Cached", session.cached_tokens || 0],
     ["Prompt", session.prompt_tokens || 0],
-    ["Elapsed", formatMs(session.elapsed_ms || 0)],
+    ["Calls", session.call_count || 0],
+    ["Fallbacks", session.fallback_count || 0],
+    ["Tools", session.tool_count || 0],
+    ["Elapsed", formatMs(sessionElapsed)],
   ].forEach(([label, value]) => {
     const item = document.createElement("div");
     const labelEl = document.createElement("span");
@@ -1518,12 +1522,21 @@ function renderSessionUsage(usage) {
   renderUsageRows(
     sessionUsageTurns,
     usage.turns || [],
-    (turn) => `Turn ${turn.turn_index}: ${turn.total_tokens} tokens, ${turn.elapsed_ms}ms`,
+    (turn) => {
+      const elapsed = turn.turn_wall_time_ms ?? turn.elapsed_ms;
+      return `Turn ${turn.turn_index}: ${turn.total_tokens} tokens, ${formatMs(elapsed)}, ${turn.tool_count || 0} tools, ${turn.fallback_count || 0} fallbacks`;
+    },
   );
   renderUsageRows(
     sessionUsageCalls,
     usage.calls || [],
-    (call) => `T${call.turn_index} R${call.round_index}: ${call.total_tokens} tokens, TTFT ${formatMs(call.ttft_ms)}, elapsed ${formatMs(call.elapsed_ms)}`,
+    (call) => {
+      const fallback = call.used_fallback ? " · fallback" : "";
+      const toolLatency = call.tool_latency_ms === null || call.tool_latency_ms === undefined
+        ? ""
+        : ` · tool ${formatMs(call.tool_latency_ms)}`;
+      return `T${call.turn_index} R${call.round_index} · ${call.mode} · ${call.agent}/${call.call_kind}${fallback}: ${call.total_tokens} tokens, TTFT ${formatMs(call.ttft_ms)}, elapsed ${formatMs(call.elapsed_ms)}${toolLatency}`;
+    },
   );
 }
 

tests/test_debug_runtime.py

@@ -2,6 +2,7 @@ import asyncio
 import importlib
 import json
 import logging
+from dataclasses import replace
 from collections.abc import AsyncIterator
 from typing import Any
 
@@ -9,7 +10,14 @@ import pytest
 
 from agent_lab.application.contracts import AgentParams, DebugRunRequest, EventAgentParams
 from agent_lab.application.event_agent import EventAgentRequest
-from agent_lab.application.events import EventBatchExecutor, ResultPolicy
+from agent_lab.application.events import (
+    EventBatchExecutor,
+    EventBatchResult,
+    EventResult,
+    EventSource,
+    EventStatus,
+    ResultPolicy,
+)
 from agent_lab.application.runtime import DebugRuntime
 from agent_lab.application.tools import ToolDefinition, ToolExecutionContext, ToolRegistry
 from agent_lab.domain.events import ToolCallEvent
@@ -1435,6 +1443,8 @@ async def test_runtime_emits_round_stats_for_each_chat_call_in_event_handoff():
             2.08,
             2.09,
             2.1,
+            2.11,
+            2.12,
             2.25,
             2.26,
             3.0,
@@ -2345,6 +2355,338 @@ def _tool_call(call_id: str, *, name: str = "safe_tool") -> ToolCallEvent:
     )
 
 
+class ManualClock:
+    def __init__(self) -> None:
+        self.now = 0.0
+
+    def __call__(self) -> float:
+        return self.now
+
+
+class ProviderTimingChatClient:
+    def __init__(self, clock: ManualClock) -> None:
+        self.clock = clock
+
+    async def stream_chat(
+        self,
+        messages: list[ChatMessage],
+        tools: list[dict],
+        params: AgentParams,
+        tool_choice: dict[str, Any] | None = None,
+    ) -> AsyncIterator[StreamItem]:
+        self.clock.now = 0.010
+        yield StreamItem.raw_response_chunk({"provider": "first"})
+        self.clock.now = 0.050
+        yield StreamItem.message_delta("visible")
+        yield StreamItem.usage_item(
+            TokenUsage(prompt_tokens=2, completion_tokens=3, total_tokens=5)
+        )
+        self.clock.now = 0.060
+
+
+class ComparableMetricsChatClient:
+    def __init__(self) -> None:
+        self.chat_calls = 0
+        self.fallback_calls = 0
+
+    async def stream_chat(
+        self,
+        messages: list[ChatMessage],
+        tools: list[dict],
+        params: AgentParams,
+        tool_choice: dict[str, Any] | None = None,
+    ) -> AsyncIterator[StreamItem]:
+        if tool_choice is not None:
+            self.fallback_calls += 1
+            tool_name = tools[0]["function"]["name"]
+            yield StreamItem.raw_response_chunk({"fallback": "first"})
+            yield StreamItem.provider_tool_call(
+                ToolCallEvent(
+                    id="provider-fallback",
+                    name=tool_name,
+                    arguments={"query": "logical query"},
+                    raw_arguments='{"query":"logical query"}',
+                )
+            )
+            yield StreamItem.usage_item(
+                TokenUsage(prompt_tokens=3, completion_tokens=4, total_tokens=7)
+            )
+            return
+
+        self.chat_calls += 1
+        if self.chat_calls == 1:
+            event = ToolCallEvent(
+                id="event-1",
+                name="comparable.search",
+                arguments={"query": "logical query"},
+                raw_arguments='{"query":"logical query"}',
+            )
+            if tools:
+                yield StreamItem.provider_tool_call(event)
+            else:
+                yield StreamItem.text_event(event)
+            yield StreamItem.usage_item(
+                TokenUsage(prompt_tokens=2, completion_tokens=3, total_tokens=5)
+            )
+            return
+        yield StreamItem.message_delta("final")
+        yield StreamItem.usage_item(
+            TokenUsage(prompt_tokens=4, completion_tokens=6, total_tokens=10)
+        )
+
+
+async def _run_persisted_turn(
+    runtime: DebugRuntime,
+    request: DebugRunRequest,
+) -> list[dict[str, Any]]:
+    queues = runtime.start_session(request)
+    outputs: list[dict[str, Any]] = []
+    while not any(message["type"] == "turn_completed" for message in outputs):
+        outputs.append(await asyncio.wait_for(queues.output.get(), timeout=1))
+    await runtime.aclose()
+    return outputs
+
+
+def _comparable_registry(*, fallback_required: bool) -> ToolRegistry:
+    return ToolRegistry(
+        [
+            ToolDefinition(
+                name="comparable.search",
+                description="Comparable search.",
+                parameters={
+                    "type": "object",
+                    "properties": {"query": {"type": "string"}},
+                    "required": ["query"],
+                },
+                handler=lambda event: {"query": event.arguments["query"]},
+                argument_resolver=(
+                    (lambda event, context: {})
+                    if fallback_required
+                    else (lambda event, context: {"query": "logical query"})
+                ),
+            )
+        ]
+    )
+
+
+def _comparable_request(*, mode: str, session_id: str) -> DebugRunRequest:
+    return DebugRunRequest(
+        session_id=session_id,
+        user_message="compare invocation",
+        system_prompts=[],
+        pre_messages=[],
+        chat_agent=AgentParams(model="chat-model"),
+        event_agent=EventAgentParams(
+            model="event-model",
+            enabled_tools=["comparable.search"],
+            max_event_loops=1,
+            max_parallel_events=2,
+        ),
+        tool_invocation_mode=mode,
+    )
+
+
+@pytest.mark.asyncio
+async def test_runtime_persists_provider_ttft_but_keeps_visible_round_ttft(tmp_path):
+    clock = ManualClock()
+    store = SQLiteSessionStore(tmp_path / "provider-ttft.sqlite3")
+    runtime = DebugRuntime(
+        ProviderTimingChatClient(clock),
+        session_store=store,
+        clock=clock,
+    )
+
+    outputs = await _run_persisted_turn(
+        runtime,
+        DebugRunRequest(
+            session_id="provider-ttft",
+            user_message="timing",
+            system_prompts=[],
+            pre_messages=[],
+            chat_agent=AgentParams(model="chat-model"),
+            event_agent=EventAgentParams(enabled_tools=[], max_event_loops=1),
+        ),
+    )
+
+    round_stats = next(message for message in outputs if message["type"] == "round_stats")
+    usage = store.usage_summary("provider-ttft")
+    assert round_stats["ttft_ms"] == 50
+    assert usage["calls"][0]["ttft_ms"] == 10
+    assert usage["calls"][0]["metric_key"] == "chat:1:1"
+    assert usage["turns"][0]["turn_wall_time_ms"] == 60
+
+
+@pytest.mark.asyncio
+async def test_runtime_direct_and_dual_persist_comparable_logical_tool_metrics(tmp_path):
+    summaries: dict[str, dict[str, Any]] = {}
+    for mode in ("chat_agent_tools", "dual_agent"):
+        store = SQLiteSessionStore(tmp_path / f"{mode}.sqlite3")
+        client = ComparableMetricsChatClient()
+        runtime = DebugRuntime(
+            client,
+            registry=_comparable_registry(fallback_required=False),
+            session_store=store,
+        )
+        session_id = f"session-{mode}"
+
+        await _run_persisted_turn(
+            runtime,
+            _comparable_request(mode=mode, session_id=session_id),
+        )
+
+        summaries[mode] = store.usage_summary(session_id)
+        assert client.fallback_calls == 0
+
+    for mode, summary in summaries.items():
+        assert summary["session"]["total_tokens"] == 15
+        assert summary["session"]["tool_count"] == 1
+        assert summary["session"]["fallback_count"] == 0
+        assert {call["mode"] for call in summary["calls"]} == {mode}
+        assert {call["metric_key"] for call in summary["calls"]} == {
+            "chat:1:1",
+            "chat:1:2",
+            "tool:1:1:event-1",
+        }
+        assert not any(
+            call["call_kind"] == "argument_fallback" for call in summary["calls"]
+        )
+
+
+@pytest.mark.asyncio
+async def test_runtime_dual_persists_fallback_model_cost_once(tmp_path):
+    store = SQLiteSessionStore(tmp_path / "dual-fallback.sqlite3")
+    client = ComparableMetricsChatClient()
+    runtime = DebugRuntime(
+        client,
+        registry=_comparable_registry(fallback_required=True),
+        session_store=store,
+    )
+
+    await _run_persisted_turn(
+        runtime,
+        _comparable_request(mode="dual_agent", session_id="dual-fallback"),
+    )
+
+    usage = store.usage_summary("dual-fallback")
+    fallback_calls = [
+        call for call in usage["calls"] if call["call_kind"] == "argument_fallback"
+    ]
+    assert client.fallback_calls == 1
+    assert len(fallback_calls) == 1
+    assert fallback_calls[0]["agent"] == "event_agent"
+    assert fallback_calls[0]["event_id"] == "event-1"
+    assert fallback_calls[0]["event_name"] == "comparable.search"
+    assert fallback_calls[0]["total_tokens"] == 7
+    assert fallback_calls[0]["ttft_ms"] is not None
+    assert fallback_calls[0]["metric_key"] == "fallback:1:1:event-1"
+    assert usage["session"]["total_tokens"] == 22
+    assert usage["session"]["fallback_count"] == 1
+    assert usage["session"]["tool_count"] == 1
+
+
+def test_runtime_persists_only_logical_tool_results(tmp_path):
+    store = SQLiteSessionStore(tmp_path / "logical-results.sqlite3")
+    store.create_session(title="logical", config={}, session_id="logical")
+    store.start_turn("logical", turn_index=1, user_message="logical")
+    primary = EventResult(
+        event_id="event-1",
+        event_name="comparable.search",
+        status=EventStatus.SUCCESS,
+        source=EventSource.TEXT_EVENT,
+        tool_latency_ms=20,
+    )
+    duplicate = replace(
+        primary,
+        event_id="event-2",
+        deduplicated_from="event-1",
+        tool_latency_ms=20,
+    )
+    runtime = DebugRuntime(
+        ComparableMetricsChatClient(),
+        session_store=store,
+    )
+
+    runtime._append_persisted_tool_metrics(
+        "logical",
+        1,
+        1,
+        EventBatchResult(results=(primary, duplicate)),
+    )
+
+    usage = store.usage_summary("logical")
+    assert usage["session"]["tool_count"] == 1
+    assert usage["calls"][0]["metric_key"] == "tool:1:1:event-1"
+
+
+@pytest.mark.asyncio
+async def test_runtime_parallel_tool_latencies_are_not_summed_as_turn_wall_time(tmp_path):
+    both_started = asyncio.Event()
+    started = 0
+
+    async def handler(event: ToolCallEvent) -> dict[str, Any]:
+        nonlocal started
+        started += 1
+        if started == 2:
+            both_started.set()
+        await asyncio.wait_for(both_started.wait(), timeout=1)
+        await asyncio.sleep(0.03)
+        return {"event_id": event.id}
+
+    registry = ToolRegistry(
+        [
+            ToolDefinition(
+                name="parallel.tool",
+                description="Parallel tool.",
+                parameters={"type": "object"},
+                handler=handler,
+            )
+        ]
+    )
+    events = [
+        ToolCallEvent(
+            id=f"parallel-{index}",
+            name="parallel.tool",
+            arguments={"index": index},
+            raw_arguments=f'{{"index":{index}}}',
+        )
+        for index in (1, 2)
+    ]
+    client = ScriptedChatClient(
+        [
+            [
+                StreamItem.provider_tool_call(events[0]),
+                StreamItem.provider_tool_call(events[1]),
+            ],
+            [StreamItem.message_delta("done")],
+        ]
+    )
+    store = SQLiteSessionStore(tmp_path / "parallel.sqlite3")
+    runtime = DebugRuntime(client, registry=registry, session_store=store)
+
+    await _run_persisted_turn(
+        runtime,
+        _direct_request(
+            enabled_tools=["parallel.tool"],
+            session_id="parallel",
+        ).model_copy(
+            update={
+                "event_agent": EventAgentParams(
+                    enabled_tools=["parallel.tool"],
+                    max_event_loops=1,
+                    max_parallel_events=2,
+                )
+            }
+        ),
+    )
+
+    usage = store.usage_summary("parallel")
+    tool_calls = [call for call in usage["calls"] if call["call_kind"] == "tool_execution"]
+    tool_latency_sum = sum(call["tool_latency_ms"] for call in tool_calls)
+    assert len(tool_calls) == 2
+    assert tool_latency_sum >= 40
+    assert usage["session"]["turn_wall_time_ms"] < tool_latency_sum
+
+
 @pytest.mark.asyncio
 async def test_direct_mode_rejects_provider_call_id_reused_across_one_shot_rounds():
     executed: list[str] = []

tests/test_event_agent.py

@@ -8,7 +8,7 @@ from agent_lab.application.contracts import AgentParams, EventAgentParams
 from agent_lab.application.event_agent import EventAgent
 from agent_lab.application.tools import ToolDefinition, ToolExecutionContext, ToolRegistry
 from agent_lab.domain.events import ToolCallEvent
-from agent_lab.domain.messages import ChatMessage, StreamItem
+from agent_lab.domain.messages import ChatMessage, StreamItem, TokenUsage
 
 
 class ToolCallingChatClient:
@@ -114,6 +114,48 @@ class WrongNameToolCallingChatClient(ToolCallingChatClient):
         )
 
 
+class TrailingUsageChatClient:
+    async def stream_chat(
+        self,
+        messages: list[ChatMessage],
+        tools: list[dict],
+        params: AgentParams,
+        tool_choice: dict | None = None,
+    ) -> AsyncIterator[StreamItem]:
+        tool_name = tools[0]["function"]["name"]
+        yield StreamItem.raw_response_chunk({"first": tool_name})
+        yield StreamItem.provider_tool_call(
+            ToolCallEvent(
+                id=f"provider-{tool_name}",
+                name=tool_name,
+                arguments={"query": tool_name},
+                raw_arguments=json.dumps({"query": tool_name}),
+            )
+        )
+        yield StreamItem.usage_item(
+            TokenUsage(
+                prompt_tokens=3,
+                completion_tokens=4,
+                total_tokens=7,
+                cached_tokens=1,
+            )
+        )
+
+
+class FailedResolutionUsageChatClient:
+    async def stream_chat(
+        self,
+        messages: list[ChatMessage],
+        tools: list[dict],
+        params: AgentParams,
+        tool_choice: dict | None = None,
+    ) -> AsyncIterator[StreamItem]:
+        yield StreamItem.message_delta("invalid fallback response")
+        yield StreamItem.usage_item(
+            TokenUsage(prompt_tokens=5, completion_tokens=2, total_tokens=7)
+        )
+
+
 @pytest.mark.asyncio
 async def test_event_agent_uses_deterministic_resolver_before_llm_fallback():
     chat_client = ToolCallingChatClient({"message": "LLM generated handoff"})
@@ -180,6 +222,35 @@ async def test_event_agent_deterministic_mock_resolver_skips_llm():
     assert chat_client.calls == []
 
 
+@pytest.mark.asyncio
+async def test_event_agent_injects_clock_into_event_kernel_handler_metrics():
+    clock_values = iter([5.0, 5.030])
+    registry = ToolRegistry(
+        [
+            ToolDefinition(
+                name="timed.tool",
+                description="Timed tool.",
+                parameters={"type": "object"},
+                handler=lambda event: {"tool": event.name},
+            )
+        ]
+    )
+    event = ToolCallEvent(
+        id="timed-event",
+        name="timed.tool",
+        arguments={},
+        raw_arguments="{}",
+    )
+
+    batch = await EventAgent(
+        enabled_tools=[event.name],
+        registry=registry,
+        monotonic_clock=lambda: next(clock_values),
+    ).handle_many([event], history=[])
+
+    assert batch.results[0].tool_latency_ms == 30
+
+
 @pytest.mark.asyncio
 async def test_event_agent_calls_llm_once_for_incomplete_deterministic_arguments():
     chat_client = ToolCallingChatClient({"message": "LLM generated handoff"})
@@ -225,6 +296,141 @@ async def test_event_agent_calls_llm_once_for_incomplete_deterministic_arguments
     }
 
 
+@pytest.mark.asyncio
+async def test_event_agent_consumes_trailing_usage_and_records_provider_ttft():
+    clock_values = iter([10.0, 10.011, 10.050, 10.060, 10.080])
+    registry = ToolRegistry(
+        [
+            ToolDefinition(
+                name="fallback.search",
+                description="Resolve with fallback.",
+                parameters={
+                    "type": "object",
+                    "properties": {"query": {"type": "string"}},
+                    "required": ["query"],
+                },
+                handler=lambda event: {"query": event.arguments["query"]},
+                argument_resolver=lambda event, context: {},
+            )
+        ]
+    )
+    event = ToolCallEvent(
+        id="event-1",
+        name="fallback.search",
+        arguments={},
+        raw_arguments="{}",
+    )
+    agent = EventAgent(
+        enabled_tools=[event.name],
+        registry=registry,
+        chat_client=TrailingUsageChatClient(),
+        monotonic_clock=lambda: next(clock_values),
+    )
+
+    batch = await agent.handle_many([event], history=[])
+
+    assert batch.results[0].status.value == "success"
+    assert agent.fallback_model_calls([event]) == [
+        {
+            "event_id": "event-1",
+            "event_name": "fallback.search",
+            "usage": TokenUsage(
+                prompt_tokens=3,
+                completion_tokens=4,
+                total_tokens=7,
+                cached_tokens=1,
+            ),
+            "ttft_ms": 11,
+            "elapsed_ms": 50,
+        }
+    ]
+
+
+@pytest.mark.asyncio
+async def test_event_agent_records_failed_fallback_resolution_attempt_usage():
+    clock_values = iter([2.0, 2.005, 2.020])
+    registry = ToolRegistry(
+        [
+            ToolDefinition(
+                name="fallback.search",
+                description="Resolve with fallback.",
+                parameters={
+                    "type": "object",
+                    "properties": {"query": {"type": "string"}},
+                    "required": ["query"],
+                },
+                handler=lambda event: {"query": event.arguments["query"]},
+                argument_resolver=lambda event, context: {},
+            )
+        ]
+    )
+    event = ToolCallEvent(
+        id="event-failed",
+        name="fallback.search",
+        arguments={},
+        raw_arguments="{}",
+    )
+    agent = EventAgent(
+        enabled_tools=[event.name],
+        registry=registry,
+        chat_client=FailedResolutionUsageChatClient(),
+        monotonic_clock=lambda: next(clock_values),
+    )
+
+    batch = await agent.handle_many([event], history=[])
+
+    assert batch.results[0].status.value == "invalid_arguments"
+    assert agent.fallback_model_calls([event])[0] == {
+        "event_id": "event-failed",
+        "event_name": "fallback.search",
+        "usage": TokenUsage(prompt_tokens=5, completion_tokens=2, total_tokens=7),
+        "ttft_ms": 5,
+        "elapsed_ms": 20,
+    }
+
+
+@pytest.mark.asyncio
+async def test_event_agent_keeps_parallel_fallback_metrics_isolated_by_event():
+    registry = ToolRegistry(
+        [
+            ToolDefinition(
+                name=name,
+                description="Resolve with fallback.",
+                parameters={
+                    "type": "object",
+                    "properties": {"query": {"type": "string"}},
+                    "required": ["query"],
+                },
+                handler=lambda event: {"query": event.arguments["query"]},
+                argument_resolver=lambda event, context: {},
+            )
+            for name in ("fallback.first", "fallback.second")
+        ]
+    )
+    events = [
+        ToolCallEvent(id=f"event-{index}", name=name, arguments={}, raw_arguments="{}")
+        for index, name in enumerate(
+            ("fallback.first", "fallback.second"),
+            start=1,
+        )
+    ]
+    agent = EventAgent(
+        enabled_tools=[event.name for event in events],
+        registry=registry,
+        chat_client=TrailingUsageChatClient(),
+        params=EventAgentParams(max_parallel_events=2),
+    )
+
+    await agent.handle_many(events, history=[])
+
+    metrics = agent.fallback_model_calls(events)
+    assert [(metric["event_id"], metric["event_name"]) for metric in metrics] == [
+        ("event-1", "fallback.first"),
+        ("event-2", "fallback.second"),
+    ]
+    assert [metric["usage"].total_tokens for metric in metrics] == [7, 7]
+
+
 @pytest.mark.asyncio
 async def test_event_agent_rejects_fallback_provider_call_for_wrong_tool_name():
     chat_client = WrongNameToolCallingChatClient({"message": "wrong"})

tests/test_event_kernel.py

@@ -868,6 +868,72 @@ async def test_kernel_normalizes_handler_exceptions():
     assert result.error == "event handler failed: boom"
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize("handler_kind", ["async", "sync", "failure"])
+async def test_kernel_measures_only_actual_handler_latency(handler_kind: str):
+    clock_values = iter([10.0, 10.025])
+
+    async def async_handler(request: EventRequest) -> dict[str, Any]:
+        return {"query": request.arguments["query"]}
+
+    def sync_handler(request: EventRequest) -> dict[str, Any]:
+        return {"query": request.arguments["query"]}
+
+    def failing_handler(request: EventRequest) -> dict[str, Any]:
+        raise RuntimeError("boom")
+
+    handler = {
+        "async": async_handler,
+        "sync": sync_handler,
+        "failure": failing_handler,
+    }[handler_kind]
+    result = await EventKernel(
+        EventRegistry([_definition(handler=handler)]),
+        monotonic_clock=lambda: next(clock_values),
+    ).execute(
+        EventRequest(
+            id="event-1",
+            name="example.lookup",
+            arguments={"query": "value"},
+            source=EventSource.PROVIDER_RESOLVED,
+        )
+    )
+
+    assert result.tool_latency_ms == 25
+    if handler_kind == "failure":
+        assert result.status is EventStatus.HANDLER_ERROR
+    else:
+        assert result.status is EventStatus.SUCCESS
+
+
+@pytest.mark.asyncio
+async def test_kernel_handler_latency_excludes_argument_fallback_time():
+    clock_values = iter([20.0, 20.012])
+
+    async def fallback(*args: Any) -> ResolvedEventArguments:
+        return ResolvedEventArguments(
+            event_name="example.lookup",
+            arguments={"query": "fallback"},
+            raw_arguments='{"query":"fallback"}',
+        )
+
+    result = await EventKernel(
+        EventRegistry([_definition()]),
+        argument_fallback=fallback,
+        monotonic_clock=lambda: next(clock_values),
+    ).execute(
+        EventRequest(
+            id="event-1",
+            name="example.lookup",
+            arguments={},
+            raw_arguments="{}",
+        )
+    )
+
+    assert result.used_fallback is True
+    assert result.tool_latency_ms == 12
+
+
 @pytest.mark.asyncio
 async def test_kernel_validates_complete_draft_2020_12_schema():
     definition = _definition(

tests/test_sqlite_store.py

@@ -14,7 +14,10 @@ def test_sqlite_store_persists_session_messages_audit_and_usage(tmp_path):
 
     session_id = store.create_session(
         title="Debug first turn",
-        config={"chat_agent": {"model": "chat-model"}},
+        config={
+            "tool_invocation_mode": "chat_agent_tools",
+            "chat_agent": {"model": "chat-model"},
+        },
     )
     store.start_turn(session_id, turn_index=1, user_message="debug this")
     store.append_message(
@@ -47,7 +50,7 @@ def test_sqlite_store_persists_session_messages_audit_and_usage(tmp_path):
         ttft_ms=120,
         elapsed_ms=450,
     )
-    store.complete_turn(session_id, turn_index=1)
+    store.complete_turn(session_id, turn_index=1, wall_time_ms=425)
 
     sessions = store.list_sessions()
     assert sessions == [
@@ -67,7 +70,26 @@ def test_sqlite_store_persists_session_messages_audit_and_usage(tmp_path):
     ]
     assert store.list_audit_logs(session_id)[0]["event"] == "chat_agent_request"
     usage = store.usage_summary(session_id)
-    assert usage["calls"][0]["total_tokens"] == 10
+    assert usage["calls"][0] | {"created_at": "ignored"} == {
+        "id": usage["calls"][0]["id"],
+        "turn_index": 1,
+        "round_index": 1,
+        "prompt_tokens": 3,
+        "completion_tokens": 7,
+        "total_tokens": 10,
+        "cached_tokens": 2,
+        "ttft_ms": 120,
+        "elapsed_ms": 450,
+        "mode": "chat_agent_tools",
+        "agent": "chat_agent",
+        "call_kind": "chat_completion",
+        "event_id": None,
+        "event_name": None,
+        "used_fallback": False,
+        "tool_latency_ms": None,
+        "metric_key": None,
+        "created_at": "ignored",
+    }
     assert usage["turns"] == [
         {
             "turn_index": 1,
@@ -76,9 +98,222 @@ def test_sqlite_store_persists_session_messages_audit_and_usage(tmp_path):
             "total_tokens": 10,
             "cached_tokens": 2,
             "elapsed_ms": 450,
+            "call_count": 1,
+            "fallback_count": 0,
+            "tool_count": 0,
+            "turn_wall_time_ms": 425,
         }
     ]
-    assert usage["session"]["total_tokens"] == 10
+    assert usage["session"] == {
+        "prompt_tokens": 3,
+        "completion_tokens": 7,
+        "total_tokens": 10,
+        "cached_tokens": 2,
+        "elapsed_ms": 450,
+        "call_count": 1,
+        "fallback_count": 0,
+        "tool_count": 0,
+        "turn_wall_time_ms": 425,
+    }
+
+
+def test_sqlite_store_persists_typed_calls_without_double_counting_metrics(tmp_path):
+    store = SQLiteSessionStore(tmp_path / "agent_lab.sqlite3")
+    session_id = store.create_session(
+        title="comparable metrics",
+        config={"tool_invocation_mode": "dual_agent"},
+    )
+    store.start_turn(session_id, turn_index=1, user_message="search")
+
+    store.append_usage(
+        session_id,
+        turn_index=1,
+        round_index=1,
+        usage=TokenUsage(prompt_tokens=2, completion_tokens=3, total_tokens=5),
+        ttft_ms=10,
+        elapsed_ms=30,
+        metadata={"metric_key": "chat:1:1"},
+    )
+    fallback_metadata = {
+        "agent": "event_agent",
+        "call_kind": "argument_fallback",
+        "event_id": "event-1",
+        "event_name": "mock_search",
+        "used_fallback": True,
+        "metric_key": "fallback:1:1:event-1",
+    }
+    for _ in range(2):
+        store.append_usage(
+            session_id,
+            turn_index=1,
+            round_index=1,
+            usage=TokenUsage(prompt_tokens=4, completion_tokens=6, total_tokens=10),
+            ttft_ms=5,
+            elapsed_ms=20,
+            metadata=fallback_metadata,
+        )
+    store.append_usage(
+        session_id,
+        turn_index=1,
+        round_index=1,
+        usage=TokenUsage(prompt_tokens=100, completion_tokens=100, total_tokens=200),
+        ttft_ms=None,
+        elapsed_ms=80,
+        metadata={
+            "agent": "event_kernel",
+            "call_kind": "tool_execution",
+            "event_id": "event-1",
+            "event_name": "mock_search",
+            "used_fallback": True,
+            "tool_latency_ms": 80,
+            "metric_key": "tool:1:1:event-1",
+        },
+    )
+    store.complete_turn(session_id, turn_index=1, wall_time_ms=95)
+
+    usage = store.usage_summary(session_id)
+
+    assert len(usage["calls"]) == 3
+    assert [call["mode"] for call in usage["calls"]] == ["dual_agent"] * 3
+    assert [call["agent"] for call in usage["calls"]] == [
+        "chat_agent",
+        "event_agent",
+        "event_kernel",
+    ]
+    assert [call["call_kind"] for call in usage["calls"]] == [
+        "chat_completion",
+        "argument_fallback",
+        "tool_execution",
+    ]
+    assert usage["calls"][1]["event_id"] == "event-1"
+    assert usage["calls"][1]["event_name"] == "mock_search"
+    assert usage["calls"][1]["used_fallback"] is True
+    assert usage["calls"][2]["tool_latency_ms"] == 80
+    assert usage["calls"][2]["total_tokens"] == 0
+    assert usage["turns"] == [
+        {
+            "turn_index": 1,
+            "prompt_tokens": 6,
+            "completion_tokens": 9,
+            "total_tokens": 15,
+            "cached_tokens": 0,
+            "elapsed_ms": 130,
+            "call_count": 3,
+            "fallback_count": 1,
+            "tool_count": 1,
+            "turn_wall_time_ms": 95,
+        }
+    ]
+    assert usage["session"] == {
+        "prompt_tokens": 6,
+        "completion_tokens": 9,
+        "total_tokens": 15,
+        "cached_tokens": 0,
+        "elapsed_ms": 130,
+        "call_count": 3,
+        "fallback_count": 1,
+        "tool_count": 1,
+        "turn_wall_time_ms": 95,
+    }
+    assert store.list_sessions()[0]["total_tokens"] == 15
+
+
+def test_sqlite_store_metric_keys_are_unique_within_each_session(tmp_path):
+    store = SQLiteSessionStore(tmp_path / "agent_lab.sqlite3")
+    for session_id in ("session-a", "session-b"):
+        store.create_session(title=session_id, config={}, session_id=session_id)
+        store.start_turn(session_id, turn_index=1, user_message="test")
+        for _ in range(2):
+            store.append_usage(
+                session_id,
+                turn_index=1,
+                round_index=1,
+                usage=TokenUsage(total_tokens=1),
+                ttft_ms=None,
+                elapsed_ms=1,
+                metadata={"metric_key": "chat:1:1"},
+            )
+
+    assert len(store.usage_summary("session-a")["calls"]) == 1
+    assert len(store.usage_summary("session-b")["calls"]) == 1
+
+
+def test_sqlite_store_migrates_legacy_usage_and_turn_metrics_additively(tmp_path):
+    database_path = tmp_path / "legacy-usage.sqlite3"
+    connection = sqlite3.connect(database_path)
+    connection.executescript(
+        """
+        CREATE TABLE sessions (
+            id TEXT PRIMARY KEY,
+            title TEXT NOT NULL,
+            created_at TEXT NOT NULL,
+            updated_at TEXT NOT NULL,
+            config_json TEXT NOT NULL
+        );
+        CREATE TABLE turns (
+            session_id TEXT NOT NULL,
+            turn_index INTEGER NOT NULL,
+            user_message TEXT NOT NULL,
+            started_at TEXT NOT NULL,
+            completed_at TEXT,
+            PRIMARY KEY (session_id, turn_index)
+        );
+        CREATE TABLE usage_stats (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            session_id TEXT NOT NULL,
+            turn_index INTEGER NOT NULL,
+            round_index INTEGER NOT NULL,
+            prompt_tokens INTEGER NOT NULL,
+            completion_tokens INTEGER NOT NULL,
+            total_tokens INTEGER NOT NULL,
+            cached_tokens INTEGER NOT NULL,
+            ttft_ms INTEGER,
+            elapsed_ms INTEGER NOT NULL,
+            created_at TEXT NOT NULL
+        );
+        INSERT INTO sessions VALUES (
+            'legacy', 'Legacy', 'now', 'now',
+            '{"tool_invocation_mode":"chat_agent_tools"}'
+        );
+        INSERT INTO turns VALUES ('legacy', 1, 'old', 'now', 'now');
+        INSERT INTO usage_stats (
+            session_id, turn_index, round_index, prompt_tokens,
+            completion_tokens, total_tokens, cached_tokens,
+            ttft_ms, elapsed_ms, created_at
+        ) VALUES ('legacy', 1, 1, 1, 2, 3, 0, 4, 5, 'now');
+        """
+    )
+    connection.commit()
+    connection.close()
+
+    store = SQLiteSessionStore(database_path)
+    usage = store.usage_summary("legacy")
+
+    assert usage["calls"][0]["agent"] == "chat_agent"
+    assert usage["calls"][0]["call_kind"] == "chat_completion"
+    assert usage["calls"][0]["mode"] == "chat_agent_tools"
+    assert usage["calls"][0]["used_fallback"] is False
+    assert usage["turns"][0]["turn_wall_time_ms"] is None
+    migrated = sqlite3.connect(database_path)
+    usage_columns = {row[1] for row in migrated.execute("PRAGMA table_info(usage_stats)")}
+    turn_columns = {row[1] for row in migrated.execute("PRAGMA table_info(turns)")}
+    indexes = {row[1] for row in migrated.execute("PRAGMA index_list(usage_stats)")}
+    metric_index_sql = migrated.execute(
+        "SELECT sql FROM sqlite_master WHERE name = 'usage_stats_metric_key_uq'"
+    ).fetchone()[0]
+    migrated.close()
+    assert {
+        "agent",
+        "call_kind",
+        "event_id",
+        "event_name",
+        "used_fallback",
+        "tool_latency_ms",
+        "metric_key",
+    } <= usage_columns
+    assert "wall_time_ms" in turn_columns
+    assert "usage_stats_metric_key_uq" in indexes
+    assert "session_id, metric_key" in metric_index_sql
 
 
 def test_sqlite_store_session_list_totals_are_not_multiplied_by_turns(tmp_path):
@@ -296,6 +531,35 @@ def test_sqlite_store_migration_is_safe_across_concurrent_initialization(
         )
         """
     )
+    connection.execute(
+        """
+        CREATE TABLE turns (
+            session_id TEXT NOT NULL,
+            turn_index INTEGER NOT NULL,
+            user_message TEXT NOT NULL,
+            started_at TEXT NOT NULL,
+            completed_at TEXT,
+            PRIMARY KEY (session_id, turn_index)
+        )
+        """
+    )
+    connection.execute(
+        """
+        CREATE TABLE usage_stats (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            session_id TEXT NOT NULL,
+            turn_index INTEGER NOT NULL,
+            round_index INTEGER NOT NULL,
+            prompt_tokens INTEGER NOT NULL,
+            completion_tokens INTEGER NOT NULL,
+            total_tokens INTEGER NOT NULL,
+            cached_tokens INTEGER NOT NULL,
+            ttft_ms INTEGER,
+            elapsed_ms INTEGER NOT NULL,
+            created_at TEXT NOT NULL
+        )
+        """
+    )
     connection.execute(
         """
         INSERT INTO messages
@@ -356,3 +620,24 @@ def test_sqlite_store_migration_is_safe_across_concurrent_initialization(
         "PRAGMA table_info(messages)"
     ).fetchall()
     assert [column[1] for column in columns].count("tool_calls_json") == 1
+    migrated = real_connect(database_path)
+    turn_columns = [row[1] for row in migrated.execute("PRAGMA table_info(turns)")]
+    usage_columns = [
+        row[1] for row in migrated.execute("PRAGMA table_info(usage_stats)")
+    ]
+    metric_indexes = [
+        row[1] for row in migrated.execute("PRAGMA index_list(usage_stats)")
+    ]
+    migrated.close()
+    assert turn_columns.count("wall_time_ms") == 1
+    for column_name in (
+        "agent",
+        "call_kind",
+        "event_id",
+        "event_name",
+        "used_fallback",
+        "tool_latency_ms",
+        "metric_key",
+    ):
+        assert usage_columns.count(column_name) == 1
+    assert metric_indexes.count("usage_stats_metric_key_uq") == 1

tests/test_websocket_api.py

@@ -350,6 +350,10 @@ def test_session_api_creates_lists_and_returns_empty_replay_data(tmp_path):
             "total_tokens": 0,
             "cached_tokens": 0,
             "elapsed_ms": 0,
+            "call_count": 0,
+            "fallback_count": 0,
+            "tool_count": 0,
+            "turn_wall_time_ms": None,
         },
     }
 
@@ -1639,6 +1643,13 @@ def test_static_app_loads_session_replay_and_usage():
     assert '`/api/sessions/${sessionId}/usage`' in js
     assert "function renderAuditReplay(" in js
     assert "function renderSessionUsage(" in js
+    assert "session.turn_wall_time_ms ?? session.elapsed_ms" in js
+    assert "turn.turn_wall_time_ms ?? turn.elapsed_ms" in js
+    assert "call.mode" in js
+    assert "call.agent" in js
+    assert "call.call_kind" in js
+    assert "call.used_fallback" in js
+    assert "call.tool_latency_ms" in js
     assert "sessionUsageSummary" in js
     assert "Finish current turn before switching sessions" in js
     assert "await loadSessionReplay(session.id, { preserveStatus: true });" in create_session