diff --git a/.claude/notes/agents.md b/.claude/notes/agents.md
index 9857d839..b6d1e147 100644
--- a/.claude/notes/agents.md
+++ b/.claude/notes/agents.md
@@ -855,12 +855,29 @@ shared `agents/_skills.py` resolver: the SDK wants ONE parent directory, not a l
 ran 7 steps). The adapter counts calls from the event stream and abandons the host when call N+1
 opens; `len(turnUsages)` from the `result` then replaces the estimate as `num_turns`.
 
+**A call opens on its first text, or on its `usage` frame when it streams none — never on a tool
+event.** The host streams a call's text, flushes the call's `usage` frame, then runs the call's
+tools, and it runs a batched reply's tools one at a time: `tool_call`, `tool_result`, `tool_call`,
+`tool_result`. The adapter once opened call N+1 when every open tool had returned. With that order
+the open set empties after each tool, so every tool counted as a call: a downstream suite saw a turn
+with 16 round-trips and 96 tool calls cut at `max_turns: 80`. Now a `thinking` or `message` event
+opens a call unless the open call has already streamed text, and a `usage` frame opens one only for
+a call that streamed none. A crossing call that streams text is cut at that text, before any of its
+tools. A tool-only crossing call is cut at its own frame, and its first tool can start before the kill:
+the host starts a call's tools right after the frame, and the stream gives no earlier boundary. A host
+without `usage` frames counts only the calls that stream text; such a host is below the 1.203.0 floor,
+which is also the first release that takes the `auth` init option.
+Only `_TurnState.on_text` / `on_tool_event` / `on_usage_frame` / `on_result` write the count, so the rule
+lives in one place. `on_result` compares the live count with `len(turnUsages)` and warns when they
+differ: the order is read from the 1.203.1 host and SDK source, not from a live transcript, so a host
+release that reorders the frames shows up in the logs instead of silently moving the cap.
+
 **A cut turn keeps the usage of its finished calls.** A turn cut by `max_turns` or a stop gets no
 `result`, so the adapter sums the per-round-trip `usage` frames; the `result`'s total replaces the sum
-when it arrives. Ordering (from SDK/backend source, not a live transcript): call N's `usage` frame
-always precedes the `tool_result` that opens call N+1. The frame is NOT a call boundary — it arrives
-before its call's tools run — so the cap stays on "the previous call's tools have all returned". The
-adapter warns when a cut turn had finished calls but no usage.
+when it arrives. Ordering (from the host and SDK source, not a live transcript): call N's `usage` frame
+precedes every tool event of call N. A crossing call that streams text is cut at that text, so its usage
+is dropped; a tool-only crossing call is cut at its own frame, so its usage is booked. The adapter warns
+when a cut turn had finished calls but no usage.
 
 **One `AssistantMessage` per `communicate()`.** `isStepStart` and `turnUsages` would allow a
 per-round-trip split; today `isStepStart` only merges streamed deltas. `close_window` opens the one
@@ -957,8 +974,9 @@ introduced or should silently "fix" by picking one number over the other.
 
 ### Delegate agent golden-master coverage
 
-`tests/_fixtures/golden_streams/delegate_fixtures.py` scripts the host's stdout frames for four
-scenarios: a tool call, a `max_turns` cut, an `error`-frame crash and an orphaned tool. The `event`,
+`tests/_fixtures/golden_streams/delegate_fixtures.py` scripts the host's stdout frames for six
+scenarios: a tool call, a `max_turns` cut, an `error`-frame crash, an orphaned tool, a tool-only call
+that crosses the cap, and a batched reply whose tools run one at a time. The `event`,
 `result` and `error` frames mirror the live `1.202.1` transcript. The `usage` frame and the `result`
 fields new in `1.203.0` follow the host source of UiPath/Autopilot#6656 (`src/usage.ts` and
 `handleSend`): every usage object carries all four buckets with `input_tokens` exclusive of cache,
diff --git a/.claude/notes/lint-rules.md b/.claude/notes/lint-rules.md
index b7ab4f06..79b6e1ea 100644
--- a/.claude/notes/lint-rules.md
+++ b/.claude/notes/lint-rules.md
@@ -845,7 +845,7 @@ assertion for each. A defaulted `AgentStartEvent` lands ~365 days before the clo
 first window, so the head blows any sane UPPER bound by the whole offset. A defaulted
 `AgentEndEvent` lands ~365 days BEFORE its own last message, so `decompose_turn` clamps the
 negative to `0.0` — "measured, and instant", which passes an upper bound. Only a strict
-`> 0.0` catches it. It holds on all three clocked harnesses because real work separates a
+`> 0.0` catches it. It holds on every clocked harness because real work separates a
 turn's last flush from its end event. The margin is smallest on antigravity, which holds
 its process across turns: it measures 0.007-0.035 ms there, 7-35 ticks of the 1 us
 resolution that both `datetime` and `time.monotonic()` have on Linux, macOS and Windows.
diff --git a/.claude/notes/timing.md b/.claude/notes/timing.md
index 13a9b8b9..ae210f60 100644
--- a/.claude/notes/timing.md
+++ b/.claude/notes/timing.md
@@ -16,6 +16,11 @@ removes:
   transition or an NTP step inside a turn lands directly in a generation window — an
   hour-long jump in a millisecond field. Nightly runs start at 04:18 and run for hours,
   so it is reachable rather than theoretical. A monotonic-derived stamp cannot express it.
+- Delegate stamped its one window and its tool spans with raw `datetime.now()` but measured
+  `duration_seconds` with `time.monotonic()`. On Windows the wall clock moved 0.24-0.37 ms
+  further than the monotonic one inside a sub-millisecond turn, so the tail alone came out
+  longer than the whole turn and `test_delegate_golden[a_tool_call]` failed at random. One
+  window has no inter-window drift to correct, but it still shares the turn's duration.
 
 It is an EXTRACTION, not an invention: antigravity already captured this exact pair at
 the top of `communicate` and simply did not use it for later stamps.
@@ -205,7 +210,7 @@ Magnitudes are only real where a scripted clock makes them real, so each case dr
 harness's own reducer with a hand-moved clock and feeds its output through a real
 `EventCollector`, the seam production uses to compute the head and the tail. The module
 reuses the three clock-injection idioms the per-harness suites already use rather than
-inventing a fourth: an injected `TurnClock` stand-in (pi, antigravity, claude-code), a
+inventing a fourth: an injected `TurnClock` stand-in (pi, antigravity, claude-code, delegate), a
 `datetime` subclass patched onto the module (opencode), and scripted SDK epoch-millisecond
 stamps (codex). OpenCode needs a `datetime` SUBCLASS rather than a stub because the
 reducer also calls `datetime.fromtimestamp` through the same module global to convert the
diff --git a/docs/agents/DELEGATE.md b/docs/agents/DELEGATE.md
index f26a27ac..ce403a32 100644
--- a/docs/agents/DELEGATE.md
+++ b/docs/agents/DELEGATE.md
@@ -168,8 +168,8 @@ Run-limit semantics per harness: [Run-Limit Parity](HARNESS_PARITY.md).
 ## Known Limitations
 
 1. **No multi-generation transcript splitting.** This agent builds one `AssistantMessage` per `communicate()` call, not one per backend round-trip. The host does send the signals a split needs (`isStepStart` on `message` events and per-round-trip `turnUsages` on `result`), but this agent does not use them yet.
-2. **`max_turns` is enforced by coder_eval, not by the host.** The host's `maxSteps` option on `send` does not stop the turn (confirmed live: `maxSteps: 2` ran 7 steps and only reported `maxStepsReached: true`), so this agent does not send it. See [Run-Limit Parity](HARNESS_PARITY.md).
-   A turn cut at `max_turns`, or by a cooperative early stop, keeps the token usage and cost of every model call that finished before the cut, as on the other agents. The host sends a `usage` frame for each call before that call's tool results. The call in progress at the cut has no usage. A host that does not send `usage` frames reports usage only on its final `result` frame, which a cut turn never gets: that turn has no usage, and the agent logs a warning.
+2. **`max_turns` is enforced by coder_eval, not by the host.** The host's `maxSteps` option on `send` does not stop the turn (confirmed live: `maxSteps: 2` ran 7 steps and only reported `maxStepsReached: true`), so this agent does not send it. It counts a call on the call's first streamed text, or on its `usage` frame when the call streams none. The host runs a batched reply's tools one at a time, so tool events do not mark a call. See [Run-Limit Parity](HARNESS_PARITY.md).
+   A turn cut at `max_turns`, or by a cooperative early stop, keeps the token usage and cost of every model call that finished before the cut, as on the other agents. The host sends a `usage` frame for each call before that call's tool results. Which side of the cut the crossing call lands on depends on its shape. A call that streams text is cut at that text, before its `usage` frame and before any of its tools, so its usage is dropped. A tool-only call is cut at its own `usage` frame, so its usage is booked, and its first tool can start before the kill: the host starts a call's tools right after the frame, and the stream gives no earlier boundary. When the `result` arrives, the agent compares its own call count with the host's `turnUsages` and logs a warning if they differ, because a change in the host's frame order would move the cap. A host that does not send `usage` frames reports usage only on its final `result` frame, which a cut turn never gets: that turn has no usage, and the agent logs a warning.
 3. **Only three backend failures get a specific diagnosis.** A Cloudflare WAF block page is reported as a content-filter failure and is not retried: the same prompt or tool result is blocked again. An SSE connect timeout is reported as a connection failure and is retried. A session conflict ("A reply is already being generated") is retried in a new conversation when no turn of that conversation has finished. After a finished turn, or with a `session_id` set in the config, the retry stays in the same conversation, so the agent never continues without its earlier turns: the retry succeeds once the backend releases the conversation (it stops the reply when the host disconnects), or the task ends as an error. There is no first-response stall detection: a stalled turn ends at its turn timeout. Every other crash ends the turn as an `AgentCrashError`, which is retried unless its message matches a category that is not retried (a timeout, or an auth, billing or content-filter error).
 4. **`sdk_options` accepts only `effort`.** Reasoning effort uses the same `sdk_options.effort` key as Claude Code, so `-D agent.sdk_options.effort=high` works for both agents. Any other key, or an `effort` that is not a string, is a validation error. The host checks the tier itself: it logs a tier it does not recognize and uses the model's default. `project_id`, `session_id` and `enable_computer_use` are typed config fields.
 5. **`enable_computer_use: true` requires local permissions and is unavailable on Linux.** macOS needs Accessibility + Screen Recording grants; the SDK throws unconditionally on Linux when this is enabled.
diff --git a/docs/agents/HARNESS_PARITY.md b/docs/agents/HARNESS_PARITY.md
index b7d3790e..d63cea6a 100644
--- a/docs/agents/HARNESS_PARITY.md
+++ b/docs/agents/HARNESS_PARITY.md
@@ -12,7 +12,7 @@ This page is the contract for what each run limit means per harness, plus the sh
 
 | Limit | claude-code | codex | antigravity | opencode | pi | delegate |
 |---|---|---|---|---|---|---|
-| `run_limits.max_turns` (main-thread model API calls on every harness, see below) | native CLI cap, plus a harness backstop when call N+1 begins | a call runs from its first item to its `thread/tokenUsage/updated`, and one that ran tools opens the next call there | a call runs from its first new MODEL step to the step carrying its usage | one `step_start` step | one `turn_start` turn | a call opens when the previous call's tools have all returned |
+| `run_limits.max_turns` (main-thread model API calls on every harness, see below) | native CLI cap, plus a harness backstop when call N+1 begins | a call runs from its first item to its `thread/tokenUsage/updated`, and one that ran tools opens the next call there | a call runs from its first new MODEL step to the step carrying its usage | one `step_start` step | one `turn_start` turn | a call opens on its first streamed text, or on its `usage` frame when it streams none; a tool-only call N+1's first tool can start before the kill |
 | `run_limits.turn_timeout` | watchdog, SIGKILL on the CLI subprocess | watchdog + cooperative interrupt | watchdog, plus an earlier internal poll deadline at 80% of it (see below) | deadline enforced in-loop and on the final reap; SIGTERM→SIGKILL on the CLI's whole process group | deadline enforced in-loop and on the final reap; SIGTERM→SIGKILL on the CLI's whole process group | deadline checked both between reads and while blocked inside one (`asyncio.wait_for`); force-kills the host subprocess and drops the handle so the next turn respawns |
 | `run_limits.task_timeout` | orchestrator-level, agent-agnostic | orchestrator-level, agent-agnostic | orchestrator-level, agent-agnostic | orchestrator-level, agent-agnostic | orchestrator-level, agent-agnostic | orchestrator-level, agent-agnostic |
 | `run_limits.stop_early` | cooperative `should_stop` | cooperative `should_stop` | cooperative `should_stop` | cooperative `should_stop` (event granularity) | cooperative `should_stop` (event granularity — Pi streams incrementally) | cooperative `should_stop`, polled per forwarded SDK event; the host is abandoned (no interrupt command exists) and a fresh one spawns for the next turn |
@@ -34,8 +34,8 @@ wall clock its numbers account for.
 | `generation_completed_at` | set | `None` — see below | `None` | `None` | `None` | set |
 | `message_id` source | SDK `message_id`; `None` when the stream carries none; `subagent-<tool_use_id>` for a synthesized sub-agent terminal | synthetic `turn_id-msg-N`, shared across the sub-messages of one generation; `turn_id-subagent-N` for recovered sub-agent generations | synthetic `turn_id-msg-N`, one per generation | CLI `messageID`; `None` when absent | CLI `responseId`; `None` when absent | synthetic `<turn_id>-msg-0` (exactly one per turn) |
 | `Σ generation + ∪ tool + head + tail ≈ turn duration` | yes [^identity] | yes [^identity] | yes [^identity] | yes [^identity] | yes [^identity] | yes, trivially — one window IS the turn |
-| clock basis for recorded stamps | one `TurnClock` per turn | SDK epoch ms (`_ms_to_dt`) — the subprocess's own clock, unreachable from the host, for BOTH window bounds and tool spans | one `TurnClock` per turn | **MIXED**: window bounds on the host `datetime.now()` (`:362`, `:696`); tool spans on CLI epoch ms (`_epoch_ms_to_dt`, `:406`/`:462`) | one `TurnClock` per turn | raw `datetime.now()` (no `TurnClock` — CE064 does not require one, since there is only one window and no inter-window drift it could correct) |
-| turn bracket (`AgentStartEvent` / `AgentEndEvent`) stamp | the same `TurnClock` (**CE064**) | raw `datetime.now()` — consistent with its epoch-ms bounds | the same `TurnClock` (**CE064**) | raw `datetime.now()` — consistent with its epoch-ms tool spans | the same `TurnClock` (**CE064**) | raw `datetime.now()` — consistent with its own single-window bounds |
+| clock basis for recorded stamps | one `TurnClock` per turn | SDK epoch ms (`_ms_to_dt`) — the subprocess's own clock, unreachable from the host, for BOTH window bounds and tool spans | one `TurnClock` per turn | **MIXED**: window bounds on the host `datetime.now()` (`:362`, `:696`); tool spans on CLI epoch ms (`_epoch_ms_to_dt`, `:406`/`:462`) | one `TurnClock` per turn | one `TurnClock` per turn (its raw `datetime.now()` bounds drifted from the monotonic `duration_seconds` on Windows and put the tail past the whole turn) |
+| turn bracket (`AgentStartEvent` / `AgentEndEvent`) stamp | the same `TurnClock` (**CE064**) | raw `datetime.now()` — consistent with its epoch-ms bounds | the same `TurnClock` (**CE064**) | raw `datetime.now()` — consistent with its epoch-ms tool spans | the same `TurnClock` (**CE064**) | the same `TurnClock` (**CE064**) |
 | window built by `timing.py::close_window` | yes | yes | yes | yes | yes | yes |
 
 [^identity]: "yes" is load-bearing, and THREE sensors check it, each seeing
@@ -126,8 +126,8 @@ Two consequences worth stating, because both are behaviour changes:
   (not on `message_id`, which OpenCode and Pi can legitimately leave `None`),
   subtracts the overlap once, and re-apportions so the parts still sum.
 
-**Two clock bases remain, and the row above says which.** Antigravity, Pi and
-claude-code derive every recorded wall stamp from one `TurnClock` per turn, so
+**Two clock bases remain, and the row above says which.** Antigravity, Pi,
+claude-code and Delegate derive every recorded wall stamp from one `TurnClock` per turn, so
 a turn's bounds and the tool spans subtracted from them cannot disagree, and
 neither can be moved by a DST transition or an NTP step inside the turn.
 Antigravity needed it first: its span was monotonic while its tool intervals
@@ -555,16 +555,18 @@ Each harness finds the call boundary in its own stream (the table above):
   step, so a rise in that total closes the call, and the next call opens with a MODEL
   step at a new `step_index`.
 - **OpenCode** and **Pi** stream one `step_start` or `turn_start` per call.
-- **Delegate**'s event stream marks a round-trip only on a text reply
-  (`isStepStart`), and a tool-only reply streams only its tool call, with no text
-  before it. The host's own `maxSteps` option does not stop the turn, so the
-  adapter does not send it and enforces the cap itself. So the next call opens when every tool the
-  previous call announced has returned, since those results go back to the model,
-  and the cap fires before call N+1 can run anything. A reply that announces
-  several tools before their results counts once. If a tool never returns, the
-  next call opens at the model's next text or new tool call instead.
-  The host sends each call's `usage` frame before that call's tool results, so
-  the calls under the cap keep their tokens and cost.
+- **Delegate** streams a call's text, then writes the call's `usage` frame, then
+  runs the call's tools. It runs a batched reply's tools one at a time, so one call
+  streams `tool_call`, `tool_result`, `tool_call`, `tool_result`, and a tool event
+  cannot mark a call boundary. A call therefore opens on its first `thinking` or
+  `message` event. A tool-only reply streams no text, so it opens on its `usage`
+  frame. A call N+1 that streams text is cut at that text, before any of its tools.
+  A tool-only call N+1 is cut at its `usage` frame, and its first tool can start
+  before the kill: the host starts a call's tools right after the frame, and the
+  stream gives no earlier boundary. The host's own `maxSteps` option does not stop
+  the turn, so the adapter does not send it and enforces the cap itself. A host that
+  sends no `usage` frames counts only the calls that stream text. The calls under
+  the cap keep their tokens and cost.
 
 Every harness enforces the cap on the same loop boundary as the
 cooperative early stop, then kills or cancels the in-flight turn so the cap stops
diff --git a/src/coder_eval/agents/delegate_agent.py b/src/coder_eval/agents/delegate_agent.py
index 72d8b544..1879ae26 100644
--- a/src/coder_eval/agents/delegate_agent.py
+++ b/src/coder_eval/agents/delegate_agent.py
@@ -39,7 +39,6 @@
 import uuid
 from collections import deque
 from collections.abc import Callable
-from datetime import datetime
 from pathlib import Path
 from typing import Any, ClassVar, NoReturn
 from urllib.parse import urlparse
@@ -76,7 +75,7 @@
     TurnEndStatus,
     TurnStartEvent,
 )
-from coder_eval.timing import close_window
+from coder_eval.timing import TurnClock, close_window
 from coder_eval.utils import process_plugins
 
 from .registry import AgentRegistry
@@ -424,12 +423,14 @@ class _TurnState:
     richer transcript would need.
     """
 
-    def __init__(self, *, iteration: int, user_input: str, model: str | None) -> None:
+    def __init__(self, *, iteration: int, user_input: str, model: str | None, clock: TurnClock) -> None:
         self.iteration = iteration
         self.user_input = user_input
         self.turn_id = f"delegate-{iteration}"
-        self.started_at = time.monotonic()
-        self.started_dt = datetime.now()
+        # Every recorded stamp and duration_seconds derive from this one clock, so the
+        # time buckets tile the turn. Rationale: .claude/notes/timing.md § TurnClock
+        self.clock = clock
+        self.started_dt = clock.now()
 
         self.content_blocks: list[ContentBlock] = []
         self.text_parts: list[str] = []
@@ -437,13 +438,10 @@ def __init__(self, *, iteration: int, user_input: str, model: str | None) -> Non
         self.open_tools: dict[str, CommandTelemetry] = {}
         self.sequence = 0
         self.message_events = 0
-        # Backend round-trips begun. A tool-only reply streams no text, so each call
-        # after the first opens once the previous call's tools have all returned.
+        # Backend round-trips begun. Only the on_* methods below write these two fields.
+        # Rationale: .claude/notes/agents.md § Delegate agent
         self.api_calls = 0
-        # A result arrived while other tools were still open, so the next call is not
-        # counted yet. If the model speaks or calls a new tool first, the next call has
-        # begun; the open tools stay open, because the SDK can still deliver their results.
-        self.results_incomplete = False
+        self.call_streaming = False
 
         self.model_used: str | None = model
         # The sum of the per-call `usage` frames, until the `result` replaces it with the turn total.
@@ -457,6 +455,43 @@ def __init__(self, *, iteration: int, user_input: str, model: str | None) -> Non
     def agent_output(self) -> str:
         return self.final_response if self.final_response is not None else "".join(self.text_parts)
 
+    def on_text(self) -> None:
+        """A ``thinking`` or ``message`` event. The first one of a call opens the call."""
+        if not self.call_streaming:
+            self.api_calls += 1
+            self.call_streaming = True
+
+    def on_tool_event(self) -> None:
+        """A ``tool_call`` or ``tool_result``. It ends the call's text and never opens a call.
+
+        The host runs a batched reply's tools one at a time, so one call streams
+        ``tool_call``, ``tool_result``, ``tool_call``, ``tool_result``.
+        """
+        self.call_streaming = False
+
+    def on_usage_frame(self) -> None:
+        """A call's ``usage`` frame, which the host writes right before it starts the call's tools.
+
+        It opens the call only when the call streamed no text. A tool-only call N+1
+        is cut here, and its first tool can start before the kill, because the
+        stream gives no earlier boundary.
+        """
+        if self.call_streaming:
+            self.call_streaming = False
+        else:
+            self.api_calls += 1
+
+    def on_result(self, reported_calls: int) -> None:
+        """Replace the live count with the host's ``turnUsages`` count, and warn when they differ."""
+        if self.api_calls != reported_calls:
+            logger.warning(
+                "delegate: the event stream showed %d model call(s), but the host reported %d; "
+                + "max_turns counts from the stream, so the host's frame order has changed",
+                self.api_calls,
+                reported_calls,
+            )
+        self.api_calls = reported_calls
+
 
 @AgentRegistry.register(AgentKind.DELEGATE, DelegateAgentConfig)
 class DelegateAgent(Agent[DelegateAgentConfig]):
@@ -768,12 +803,23 @@ def emit(event: StreamEvent) -> None:
             iteration=self._iteration,
             user_input=user_input,
             model=self.config.model,
+            clock=TurnClock(),
         )
 
         emit(
-            AgentStartEvent(task_id=self.task_id, prompt=user_input, iteration=self._iteration, model=self.config.model)
+            AgentStartEvent(
+                task_id=self.task_id,
+                prompt=user_input,
+                iteration=self._iteration,
+                model=self.config.model,
+                timestamp=state.started_dt,
+            )
+        )
+        emit(
+            TurnStartEvent(
+                task_id=self.task_id, turn_id=state.turn_id, model=self.config.model, timestamp=state.started_dt
+            )
         )
-        emit(TurnStartEvent(task_id=self.task_id, turn_id=state.turn_id, model=self.config.model))
 
         deadline = None if timeout is None else time.monotonic() + timeout
         stopped_early = False
@@ -813,14 +859,14 @@ def emit(event: StreamEvent) -> None:
                 if mtype == "error":
                     await self._crash_on_host_error(state, collector, emit, str(msg.get("message", "unknown error")))
                 if mtype == "usage":
+                    state.on_usage_frame()
                     self._add_call_usage(msg, state)
-                    continue
-                event = msg.get("event")
-                if mtype != "event" or not isinstance(event, dict):
-                    logger.debug("delegate: ignoring host message %r", mtype)
-                    continue
-
-                self._handle_event(event, state, emit)
+                else:
+                    event = msg.get("event")
+                    if mtype != "event" or not isinstance(event, dict):
+                        logger.debug("delegate: ignoring host message %r", mtype)
+                        continue
+                    self._handle_event(event, state, emit)
 
                 if max_turns is not None and state.api_calls > max_turns:
                     state.max_turns_exhausted = True
@@ -871,13 +917,10 @@ def _handle_event(self, event: dict[str, Any], state: _TurnState, emit: Callable
         if isinstance(session_id, str) and session_id:
             self._session_id = session_id
 
-        if state.api_calls == 0 and (event_type in _TEXT_EVENT_TYPES or event_type == "tool_call"):
-            state.api_calls = 1
-        elif state.results_incomplete and (
-            event_type in _TEXT_EVENT_TYPES or (event_type == "tool_call" and _tool_id(event) not in state.open_tools)
-        ):
-            state.api_calls += 1
-            state.results_incomplete = False
+        if event_type in _TEXT_EVENT_TYPES:
+            state.on_text()
+        elif event_type in ("tool_call", "tool_result"):
+            state.on_tool_event()
 
         if event_type == "message":
             text = event.get("content")
@@ -894,9 +937,6 @@ def _handle_event(self, event: dict[str, Any], state: _TurnState, emit: Callable
             self._handle_tool_call(event, state, emit)
         elif event_type == "tool_result":
             self._handle_tool_result(event, state, emit)
-            state.results_incomplete = bool(state.open_tools)
-            if not state.open_tools:
-                state.api_calls += 1
         elif event_type == "error":
             state.error_message = str(event.get("error") or "unknown error")
             logger.warning("delegate: SDK reported an error event: %s", state.error_message)
@@ -923,12 +963,13 @@ def _handle_tool_call(self, event: dict[str, Any], state: _TurnState, emit: Call
             str(event.get("toolName") or "unknown"), raw_args if isinstance(raw_args, dict) else {}
         )
         state.sequence += 1
+        started = state.clock.now()
         telemetry = CommandTelemetry(
             tool_name=tool_name,
             tool_id=tool_id,
             assistant_turn_index=state.message_events,
-            timestamp=datetime.now(),
-            execution_started_at=datetime.now(),
+            timestamp=started,
+            execution_started_at=started,
             parameters=parameters,
             sequence_number=state.sequence,
         )
@@ -953,13 +994,13 @@ def _handle_tool_result(
                 tool_name=_orphan_tool_name(event, tool_id),
                 tool_id=tool_id or str(uuid.uuid4()),
                 assistant_turn_index=state.message_events,
-                timestamp=datetime.now(),
+                timestamp=state.clock.now(),
                 parameters=_orphan_tool_args(output),
                 sequence_number=state.sequence,
             )
         if isinstance(output, dict) and isinstance(output.get("content"), str):
             output = output["content"]
-        completed = datetime.now()
+        completed = state.clock.now()
         telemetry.execution_completed_at = completed
         if telemetry.execution_started_at is not None:
             telemetry.duration_ms = (completed - telemetry.execution_started_at).total_seconds() * 1000
@@ -984,7 +1025,7 @@ def _handle_tool_result(
     def _add_call_usage(msg: dict[str, Any], state: _TurnState) -> None:
         """Add one model call's ``usage`` frame to the turn's running total.
 
-        The host writes a call's frame before any tool result that call caused,
+        The host writes a call's frame before any tool event that call caused,
         so a turn cut at ``max_turns`` or by an early stop keeps the usage of
         every call that finished.
         """
@@ -996,7 +1037,9 @@ def _add_call_usage(msg: dict[str, Any], state: _TurnState) -> None:
     def _warn_if_usage_missing(state: _TurnState, status: AgentEndStatus) -> None:
         """Warn when model calls finished but no usage arrived, so the turn books no tokens or cost.
 
-        A cut turn's in-flight call is not finished, so it is not counted.
+        A cut turn's in-flight call is not finished, so it is not counted. A host
+        with no ``usage`` frames shows only the calls that streamed text, so the
+        cut-turn figure is a floor.
         """
         if state.usage is not None:
             return
@@ -1009,8 +1052,8 @@ def _warn_if_usage_missing(state: _TurnState, status: AgentEndStatus) -> None:
                 )
         elif state.api_calls > 1:
             logger.warning(
-                "delegate: turn ended (%s) with no usage frame from the host; tokens and cost for its %d "
-                + "finished model call(s) are unknown",
+                "delegate: turn ended (%s) with no usage frame from the host; tokens and cost for at least "
+                + "%d finished model call(s) are unknown",
                 status.value,
                 state.api_calls - 1,
             )
@@ -1021,7 +1064,7 @@ def _handle_result(self, msg: dict[str, Any], state: _TurnState) -> None:
         Its ``usage`` is the turn total, so it replaces the running sum of the
         ``usage`` frames. ``turnUsages`` has one entry per backend round-trip, so
         its length is the authoritative call count and replaces the running
-        estimate.
+        estimate, with a warning when the two differ.
         """
         response = msg.get("response")
         if isinstance(response, str):
@@ -1037,7 +1080,7 @@ def _handle_result(self, msg: dict[str, Any], state: _TurnState) -> None:
             state.usage = usage
         turn_usages = msg.get("turnUsages")
         if isinstance(turn_usages, list) and turn_usages:
-            state.api_calls = len(turn_usages)
+            state.on_result(len(turn_usages))
 
     def _close_open_tools(self, state: _TurnState, emit: Callable[[StreamEvent], None]) -> None:
         for tool_id, telemetry in list(state.open_tools.items()):
@@ -1077,7 +1120,7 @@ def _finalize_turn(
 
         messages = []
         if state.content_blocks:
-            completed = datetime.now()
+            completed = state.clock.now()
             # Single generation window for the whole turn (see .claude/notes/agents.md
             # § Delegate agent): tiles from the turn's own start, since there is no
             # prior emission to tile from.
@@ -1098,16 +1141,19 @@ def _finalize_turn(
                 )
             )
 
+        ended = state.clock.now()
         emit(
             TurnEndEvent(
                 task_id=self.task_id,
                 turn_id=state.turn_id,
                 status=TurnEndStatus(status.value),
                 tokens=usage if not usage.is_empty() else None,
+                timestamp=ended,
             )
         )
         emit(
             AgentEndEvent(
+                timestamp=ended,
                 task_id=self.task_id,
                 status=status,
                 usage=usage,
@@ -1127,7 +1173,7 @@ def _finalize_turn(
                 ),
                 crashed=crashed,
                 crash_reason=crash_reason,
-                duration_seconds=time.monotonic() - state.started_at,
+                duration_seconds=(ended - state.started_dt).total_seconds(),
             )
         )
 
diff --git a/tests/_bracket_clock.py b/tests/_bracket_clock.py
index e8ba8057..7b2efd39 100644
--- a/tests/_bracket_clock.py
+++ b/tests/_bracket_clock.py
@@ -1,8 +1,8 @@
 """A ``TurnClock`` stand-in anchored far from real time, for the CE064 tests.
 
 CE064 checks only that ``timestamp=`` is PRESENT on an ``AgentStartEvent`` /
-``AgentEndEvent``; this guards the SOURCE of that stamp on the three harnesses
-that own a clock. The stand-in sits a year from ``datetime.now()``, so a
+``AgentEndEvent``; this guards the SOURCE of that stamp on every harness
+that owns a clock. The stand-in sits a year from ``datetime.now()``, so a
 bracket that fell back to ``StreamEvent``'s default fails by a year, not by a
 microsecond. It advances on the real monotonic clock, so the head and tail
 come out as small positive measurements.
diff --git a/tests/_fixtures/golden_streams/delegate_fixtures.py b/tests/_fixtures/golden_streams/delegate_fixtures.py
index 6c18ed68..68ed9075 100644
--- a/tests/_fixtures/golden_streams/delegate_fixtures.py
+++ b/tests/_fixtures/golden_streams/delegate_fixtures.py
@@ -223,8 +223,9 @@ def _build_catalogue() -> list[DelegateScenario]:
         )
     )
 
-    # (b) a max_turns cut. The host gets no chance to send a result, so the record
-    # keeps the usage frames of the calls under the cap and drops the call that crossed it.
+    # (b) a max_turns cut. The host gets no chance to send a result. The crossing call
+    # streams text first, so it is cut at that text, before its usage frame: the record
+    # keeps the usage frames of the calls under the cap and drops the crossing call's.
     scenarios.append(
         DelegateScenario(
             name="b_max_turns_cut",
@@ -268,6 +269,52 @@ def _build_catalogue() -> list[DelegateScenario]:
         )
     )
 
+    # (e) a max_turns cut where the crossing call is tool-only. It streams no text, so
+    # it is cut at its own usage frame, before the stream shows its tool: the record books its usage.
+    scenarios.append(
+        DelegateScenario(
+            name="e_tool_only_call_crosses_cap",
+            frames=[
+                _ev(type="message", content="", isStepStart=True),
+                *[
+                    frame
+                    for n in range(3)
+                    for frame in (
+                        _usage(10 * (n + 1), n + 1),
+                        _tool_call(f"t{n}", command=f"step {n}"),
+                        _tool_result(f"t{n}", f"out {n}"),
+                    )
+                ],
+                _host_result("done", [_totals(10, 1), _totals(20, 2), _totals(30, 3)]),
+            ],
+            max_turns=2,
+        )
+    )
+
+    # (f) one batched reply whose three tools the host runs one at a time. It is one
+    # call, so the turn completes under max_turns=2.
+    scenarios.append(
+        DelegateScenario(
+            name="f_batched_tools_in_sequence",
+            frames=[
+                _ev(type="message", content="", isStepStart=True),
+                _usage(100, 20),
+                *[
+                    frame
+                    for name in ("a", "b", "c")
+                    for frame in (
+                        _tool_call(f"tool-{name}", "Bash", command=f"echo {name}"),
+                        _tool_result(f"tool-{name}", name),
+                    )
+                ],
+                _ev(type="message", content="Done.", isStepStart=True),
+                _usage(50, 30),
+                _host_result("Done.", [_totals(100, 20), _totals(50, 30)]),
+            ],
+            max_turns=2,
+        )
+    )
+
     return scenarios
 
 
diff --git a/tests/_fixtures/golden_streams/expected/delegate_e_tool_only_call_crosses_cap.json b/tests/_fixtures/golden_streams/expected/delegate_e_tool_only_call_crosses_cap.json
new file mode 100644
index 00000000..03bb2060
--- /dev/null
+++ b/tests/_fixtures/golden_streams/expected/delegate_e_tool_only_call_crosses_cap.json
@@ -0,0 +1,112 @@
+{
+  "agent_output": "",
+  "assistant_turn_count": 1,
+  "commands": [
+    {
+      "assistant_turn_index": 0,
+      "duration_ms": "<scrubbed>",
+      "error_message": null,
+      "execution_completed_at": "<scrubbed>",
+      "execution_started_at": "<scrubbed>",
+      "generation_completed_at": null,
+      "parameters": {
+        "command": "step 0"
+      },
+      "result_data": null,
+      "result_status": "success",
+      "result_summary": "out 0",
+      "result_tokens": 2,
+      "sequence_number": 1,
+      "timestamp": "<scrubbed>",
+      "tool_id": "t0",
+      "tool_name": "shell"
+    },
+    {
+      "assistant_turn_index": 0,
+      "duration_ms": "<scrubbed>",
+      "error_message": null,
+      "execution_completed_at": "<scrubbed>",
+      "execution_started_at": "<scrubbed>",
+      "generation_completed_at": null,
+      "parameters": {
+        "command": "step 1"
+      },
+      "result_data": null,
+      "result_status": "success",
+      "result_summary": "out 1",
+      "result_tokens": 2,
+      "sequence_number": 2,
+      "timestamp": "<scrubbed>",
+      "tool_id": "t1",
+      "tool_name": "shell"
+    }
+  ],
+  "crash_reason": null,
+  "crashed": false,
+  "duration_seconds": "<scrubbed>",
+  "harness_startup_ms": "<scrubbed>",
+  "harness_teardown_ms": "<scrubbed>",
+  "iteration": 1,
+  "max_turns_exhausted": true,
+  "messages": [
+    {
+      "cache_creation_tokens": 0,
+      "cache_read_tokens": 0,
+      "completed_at": "<scrubbed>",
+      "content_blocks": [
+        {
+          "block_type": "tool_use",
+          "is_error": false,
+          "sequence": 0,
+          "signature": null,
+          "text": null,
+          "thinking": null,
+          "tool_use_id": "t0"
+        },
+        {
+          "block_type": "tool_use",
+          "is_error": false,
+          "sequence": 1,
+          "signature": null,
+          "text": null,
+          "thinking": null,
+          "tool_use_id": "t1"
+        }
+      ],
+      "generation_duration_ms": "<scrubbed>",
+      "input_tokens": 60,
+      "message_id": "delegate-1-msg-0",
+      "model": "virtuoso-1-5",
+      "output_tokens": 6,
+      "parent_tool_use_id": null,
+      "reasoning_tokens": 0,
+      "role": "assistant",
+      "started_at": "<scrubbed>",
+      "stop_reason": null,
+      "tool_use_ids": [
+        "t0",
+        "t1"
+      ]
+    }
+  ],
+  "model_used": "virtuoso-1-5",
+  "num_turns": 3,
+  "provider_errors": [],
+  "result_summary": {
+    "is_error": false,
+    "result": null,
+    "stop_reason": null,
+    "subtype": "max_turns_exhausted"
+  },
+  "timestamp": "<scrubbed>",
+  "token_usage": {
+    "cache_creation_input_tokens": 0,
+    "cache_read_input_tokens": 0,
+    "input_tokens": 60,
+    "output_tokens": 6,
+    "total_cost_usd": "<scrubbed>",
+    "uncached_input_tokens": 60
+  },
+  "tool_union_ms": "<scrubbed>",
+  "user_input": "do it"
+}
diff --git a/tests/_fixtures/golden_streams/expected/delegate_f_batched_tools_in_sequence.json b/tests/_fixtures/golden_streams/expected/delegate_f_batched_tools_in_sequence.json
new file mode 100644
index 00000000..d89c7f23
--- /dev/null
+++ b/tests/_fixtures/golden_streams/expected/delegate_f_batched_tools_in_sequence.json
@@ -0,0 +1,150 @@
+{
+  "agent_output": "Done.",
+  "assistant_turn_count": 1,
+  "commands": [
+    {
+      "assistant_turn_index": 0,
+      "duration_ms": "<scrubbed>",
+      "error_message": null,
+      "execution_completed_at": "<scrubbed>",
+      "execution_started_at": "<scrubbed>",
+      "generation_completed_at": null,
+      "parameters": {
+        "command": "echo a"
+      },
+      "result_data": null,
+      "result_status": "success",
+      "result_summary": "a",
+      "result_tokens": 1,
+      "sequence_number": 1,
+      "timestamp": "<scrubbed>",
+      "tool_id": "tool-a",
+      "tool_name": "Bash"
+    },
+    {
+      "assistant_turn_index": 0,
+      "duration_ms": "<scrubbed>",
+      "error_message": null,
+      "execution_completed_at": "<scrubbed>",
+      "execution_started_at": "<scrubbed>",
+      "generation_completed_at": null,
+      "parameters": {
+        "command": "echo b"
+      },
+      "result_data": null,
+      "result_status": "success",
+      "result_summary": "b",
+      "result_tokens": 1,
+      "sequence_number": 2,
+      "timestamp": "<scrubbed>",
+      "tool_id": "tool-b",
+      "tool_name": "Bash"
+    },
+    {
+      "assistant_turn_index": 0,
+      "duration_ms": "<scrubbed>",
+      "error_message": null,
+      "execution_completed_at": "<scrubbed>",
+      "execution_started_at": "<scrubbed>",
+      "generation_completed_at": null,
+      "parameters": {
+        "command": "echo c"
+      },
+      "result_data": null,
+      "result_status": "success",
+      "result_summary": "c",
+      "result_tokens": 1,
+      "sequence_number": 3,
+      "timestamp": "<scrubbed>",
+      "tool_id": "tool-c",
+      "tool_name": "Bash"
+    }
+  ],
+  "crash_reason": null,
+  "crashed": false,
+  "duration_seconds": "<scrubbed>",
+  "harness_startup_ms": "<scrubbed>",
+  "harness_teardown_ms": "<scrubbed>",
+  "iteration": 1,
+  "max_turns_exhausted": false,
+  "messages": [
+    {
+      "cache_creation_tokens": 0,
+      "cache_read_tokens": 0,
+      "completed_at": "<scrubbed>",
+      "content_blocks": [
+        {
+          "block_type": "tool_use",
+          "is_error": false,
+          "sequence": 0,
+          "signature": null,
+          "text": null,
+          "thinking": null,
+          "tool_use_id": "tool-a"
+        },
+        {
+          "block_type": "tool_use",
+          "is_error": false,
+          "sequence": 1,
+          "signature": null,
+          "text": null,
+          "thinking": null,
+          "tool_use_id": "tool-b"
+        },
+        {
+          "block_type": "tool_use",
+          "is_error": false,
+          "sequence": 2,
+          "signature": null,
+          "text": null,
+          "thinking": null,
+          "tool_use_id": "tool-c"
+        },
+        {
+          "block_type": "text",
+          "is_error": false,
+          "sequence": 3,
+          "signature": null,
+          "text": "Done.",
+          "thinking": null,
+          "tool_use_id": null
+        }
+      ],
+      "generation_duration_ms": "<scrubbed>",
+      "input_tokens": 150,
+      "message_id": "delegate-1-msg-0",
+      "model": "virtuoso-1-5",
+      "output_tokens": 50,
+      "parent_tool_use_id": null,
+      "reasoning_tokens": 0,
+      "role": "assistant",
+      "started_at": "<scrubbed>",
+      "stop_reason": null,
+      "tool_use_ids": [
+        "tool-a",
+        "tool-b",
+        "tool-c"
+      ]
+    }
+  ],
+  "model_used": "virtuoso-1-5",
+  "num_turns": 2,
+  "provider_errors": [],
+  "result_summary": {
+    "is_error": false,
+    "result": null,
+    "stop_reason": null,
+    "subtype": "completed"
+  },
+  "timestamp": "<scrubbed>",
+  "token_usage": {
+    "cache_creation_input_tokens": 0,
+    "cache_read_input_tokens": 0,
+    "input_tokens": 150,
+    "output_tokens": 50,
+    "total_cost_usd": "<scrubbed>",
+    "uncached_input_tokens": 150
+  },
+  "tool_union_ms": "<scrubbed>",
+  "user_input": "do it"
+}
diff --git a/tests/test_delegate_agent.py b/tests/test_delegate_agent.py
index f4059161..91011ff3 100644
--- a/tests/test_delegate_agent.py
+++ b/tests/test_delegate_agent.py
@@ -30,6 +30,7 @@
 from coder_eval.models import AgentKind, DelegateAgentConfig
 from coder_eval.reports.markdown import collect_agent_settings_rows
 from coder_eval.streaming.events import AgentEndEvent, AgentEndStatus, AgentStartEvent
+from tests._bracket_clock import ANCHOR, AnchoredClock, assert_bracket_on_the_clock, assert_overhead_is_measured
 from tests._fixtures.golden_streams.delegate_fixtures import (
     _ev,
     _FakeProcess,
@@ -732,7 +733,7 @@ async def test_max_turns_exhausted(self, patch_exec, tmp_path):
         agent, _proc = await _started_agent(patch_exec, events, tmp_path)
         record = await agent.communicate("hi", max_turns=1)
         assert record.max_turns_exhausted is True
-        assert [c.tool_id for c in record.commands if c.result_status == "success"] == ["t0"]
+        assert [c.tool_id for c in record.commands] == ["t0"]
         assert record.num_turns == 2
 
     @staticmethod
@@ -826,19 +827,23 @@ async def test_max_turns_counts_round_trips_not_text_chunks(self, patch_exec, tm
         assert record.max_turns_exhausted is False
         assert record.num_turns == 2
 
-    async def test_max_turns_stops_a_tool_only_reply_before_its_tool_runs(self, patch_exec, tmp_path):
-        """A tool-only reply streams no text event, only its tool call."""
+    async def test_max_turns_cuts_a_tool_only_reply_at_its_usage_frame(self, patch_exec, tmp_path):
+        """A tool-only reply streams no text event; its usage frame still precedes its tool call."""
         events = [
             _ev(type="message", content="on it", isStepStart=True),
-            *[frame for n in range(3) for frame in (_tool_call(f"t{n}"), _tool_result(f"t{n}"))],
+            *[frame for n in range(3) for frame in (_usage(10, 1), _tool_call(f"t{n}"), _tool_result(f"t{n}"))],
         ]
         agent, _proc = await _started_agent(patch_exec, events, tmp_path)
         record = await agent.communicate("hi", max_turns=2)
         assert record.max_turns_exhausted is True
-        assert [c.tool_id for c in record.commands if c.result_status == "success"] == ["t0", "t1"]
+        assert [c.tool_id for c in record.commands] == ["t0", "t1"]
         assert record.num_turns == 3
+        # The crossing call is cut at its own usage frame, so the cut turn books its usage too.
+        assert record.token_usage is not None
+        assert (record.token_usage.uncached_input_tokens, record.token_usage.output_tokens) == (30, 3)
 
-    async def test_max_turns_counts_a_batched_reply_once(self, patch_exec, tmp_path):
+    async def test_max_turns_counts_text_calls_on_a_host_without_usage_frames(self, patch_exec, tmp_path):
+        """A host below the 1.203.0 floor sends no usage frames; each call that streams text still counts once."""
         events = [
             _ev(type="message", content="", isStepStart=True),
             _tool_call("a"),
@@ -853,18 +858,99 @@ async def test_max_turns_counts_a_batched_reply_once(self, patch_exec, tmp_path)
         assert record.max_turns_exhausted is False
         assert record.num_turns == 2
 
+    async def test_max_turns_counts_a_batched_reply_once_when_the_host_runs_its_tools_in_sequence(
+        self, patch_exec, tmp_path, caplog
+    ):
+        """The host runs a batched reply's tools one at a time; tool events never open a call, so it counts once."""
+        events = [
+            _ev(type="message", content="", isStepStart=True),
+            _usage(10, 1),
+            *[frame for name in "abc" for frame in (_tool_call(name), _tool_result(name))],
+            _ev(type="message", content="done", isStepStart=True),
+            _usage(20, 2),
+            _host_result("done", [_totals(10, 1), _totals(20, 2)]),
+        ]
+        agent, _proc = await _started_agent(patch_exec, events, tmp_path)
+        with caplog.at_level(logging.WARNING):
+            record = await agent.communicate("hi", max_turns=2)
+        assert record.max_turns_exhausted is False
+        assert [c.tool_id for c in record.commands] == ["a", "b", "c"]
+        assert record.num_turns == 2
+        assert not any("event stream showed" in r.message for r in caplog.records)
+
+    async def test_stream_count_that_differs_from_turn_usages_warns(self, patch_exec, tmp_path, caplog):
+        """A usage frame after the call's first tool event counts a call twice; the result's turnUsages shows it."""
+        events = [
+            *[
+                frame
+                for n in range(2)
+                for frame in (
+                    _ev(type="message", content=f"step {n}", isStepStart=True),
+                    _tool_call(f"t{n}"),
+                    _usage(10, 1),
+                    _tool_result(f"t{n}"),
+                )
+            ],
+            _host_result("done", [_totals(10, 1), _totals(10, 1)]),
+        ]
+        agent, _proc = await _started_agent(patch_exec, events, tmp_path)
+        with caplog.at_level(logging.WARNING):
+            record = await agent.communicate("hi")
+        assert record.num_turns == 2
+        assert any("event stream showed 4 model call(s), but the host reported 2" in r.message for r in caplog.records)
+
+    async def test_tool_events_a_sub_agent_streams_do_not_open_calls(self, patch_exec, tmp_path):
+        """A sub-agent's calls write no usage frame to the main session; only its tool events reach the stream."""
+        events = [
+            _ev(type="message", content="delegating", isStepStart=True),
+            _usage(10, 1),
+            _tool_call("agent"),
+            *[frame for n in range(4) for frame in (_tool_call(f"sub{n}"), _tool_result(f"sub{n}"))],
+            _tool_result("agent"),
+            _ev(type="message", content="done", isStepStart=True),
+            _usage(10, 1),
+            _host_result("done", [_totals(10, 1), _totals(10, 1)]),
+        ]
+        agent, _proc = await _started_agent(patch_exec, events, tmp_path)
+        record = await agent.communicate("hi", max_turns=2)
+        assert record.max_turns_exhausted is False
+        assert record.num_turns == 2
+
+    async def test_early_stop_can_fire_on_a_usage_frame_before_its_tool_events(self, patch_exec, tmp_path):
+        events = [
+            _ev(type="message", content="", isStepStart=True),
+            _usage(10, 1),
+            _tool_call("t0"),
+            _tool_result("t0"),
+            _usage(10, 1),
+            _tool_call("t1"),
+            _tool_result("t1"),
+        ]
+        agent, _proc = await _started_agent(patch_exec, events, tmp_path)
+        polls = {"n": 0}
+
+        def should_stop() -> bool:
+            polls["n"] += 1
+            return polls["n"] == 5
+
+        record = await agent.communicate("hi", should_stop=should_stop)
+        assert [c.tool_id for c in record.commands] == ["t0"]
+        assert record.token_usage is not None
+        assert (record.token_usage.uncached_input_tokens, record.token_usage.output_tokens) == (20, 2)
+
     async def test_max_turns_still_counts_after_a_tool_never_returns(self, patch_exec, tmp_path):
-        """Tool b never returns, so the next call opens on its first new tool call."""
+        """Tool b never returns, so only the usage frames show the next calls."""
         events = [
+            _usage(10, 1),
             _tool_call("a"),
             _tool_call("b"),
             _tool_result("a"),
-            *[frame for n in range(3) for frame in (_tool_call(f"t{n}"), _tool_result(f"t{n}"))],
+            *[frame for n in range(3) for frame in (_usage(10, 1), _tool_call(f"t{n}"), _tool_result(f"t{n}"))],
         ]
         agent, _proc = await _started_agent(patch_exec, events, tmp_path)
         record = await agent.communicate("hi", max_turns=2)
         assert record.max_turns_exhausted is True
-        assert [c.tool_id for c in record.commands if c.result_status == "success"] == ["a", "t0"]
+        assert [c.tool_id for c in record.commands] == ["a", "b", "t0"]
         assert record.num_turns == 3
 
     async def test_communicate_before_start_raises(self):
@@ -1038,6 +1124,50 @@ def on_event(self, event: Any) -> None:
         assert sum(isinstance(e, AgentEndEvent) for e in seen) == 1
 
 
+class TestTurnClock:
+    """Every recorded stamp comes from the turn's ``TurnClock``, the basis of ``duration_seconds``.
+
+    A raw ``datetime.now()`` drifts from ``time.monotonic()`` on Windows, which put
+    the teardown bucket past the whole turn in ``test_delegate_golden[a_tool_call]``.
+    """
+
+    @staticmethod
+    def _events() -> list[bytes]:
+        return [
+            _ev(type="message", content="", isStepStart=True),
+            _usage(10, 1),
+            _tool_call("t0"),
+            _tool_result("t0"),
+            _ev(type="message", content="done", isStepStart=True),
+            _usage(10, 1),
+            _host_result("done", [_totals(10, 1), _totals(10, 1)]),
+        ]
+
+    async def test_the_turn_bracket_is_on_the_clock(self, patch_exec, tmp_path, monkeypatch):
+        monkeypatch.setattr(agent_module, "TurnClock", AnchoredClock)
+        agent, _proc = await _started_agent(patch_exec, self._events(), tmp_path)
+        seen: list[Any] = []
+
+        class _Recorder:
+            def on_event(self, event: Any) -> None:
+                seen.append(event)
+
+        await agent.communicate("hi", stream_callback=_Recorder())
+        assert_bracket_on_the_clock(seen)
+
+    async def test_windows_and_tool_spans_are_on_the_clock(self, patch_exec, tmp_path, monkeypatch):
+        monkeypatch.setattr(agent_module, "TurnClock", AnchoredClock)
+        agent, _proc = await _started_agent(patch_exec, self._events(), tmp_path)
+        record = await agent.communicate("hi")
+        assert_overhead_is_measured(record)
+        assert record.commands
+        for command in record.commands:
+            assert command.execution_started_at is not None and command.execution_started_at >= ANCHOR
+            assert command.execution_completed_at is not None and command.execution_completed_at >= ANCHOR
+        assert record.messages
+        assert all(m.started_at is not None and m.started_at >= ANCHOR for m in record.messages)
+
+
 class TestStop:
     async def test_stop_sends_destroy_and_marks_finished(self, patch_exec, tmp_path):
         agent, proc = await _started_agent(patch_exec, [], tmp_path)
diff --git a/tests/test_delegate_agent_live.py b/tests/test_delegate_agent_live.py
index c79987fb..89d6d5a5 100644
--- a/tests/test_delegate_agent_live.py
+++ b/tests/test_delegate_agent_live.py
@@ -12,6 +12,7 @@
 ``_ev`` / ``_result`` / ``_tool_call`` / ``_tool_result`` builders.
 """
 
+import logging
 import os
 import shutil
 from pathlib import Path
@@ -141,3 +142,22 @@ async def test_delegate_live_token_usage_populated(tmp_path):
     assert record.token_usage is not None
     assert record.token_usage.output_tokens > 0
     assert record.num_turns is not None and record.num_turns >= 1
+
+
+@_live
+async def test_delegate_live_call_count_matches_the_host(tmp_path, caplog):
+    """A turn that runs tools: the count max_turns uses agrees with the host's turnUsages."""
+    agent = _make_agent()
+    await agent.start(str(tmp_path))
+    try:
+        with caplog.at_level(logging.WARNING):
+            record = await agent.communicate(
+                "Run the shell commands `echo one`, `echo two` and `echo three`, then reply DONE.",
+                timeout=180,
+            )
+    finally:
+        await agent.stop()
+
+    assert record.crashed is False
+    assert record.commands
+    assert not any("event stream showed" in r.message for r in caplog.records)
diff --git a/tests/test_timing_identity_contract.py b/tests/test_timing_identity_contract.py
index 1ffbda20..d41b8f5f 100644
--- a/tests/test_timing_identity_contract.py
+++ b/tests/test_timing_identity_contract.py
@@ -7,7 +7,7 @@
 a real ``EventCollector``, so every asserted number is computed by the harness.
 
 HAZARD: a case must script every clock its reducer reads — an injected
-``TurnClock`` (pi, antigravity, claude-code), a ``datetime`` subclass on the
+``TurnClock`` (pi, antigravity, claude-code, delegate), a ``datetime`` subclass on the
 module (opencode), ``time.monotonic`` on top of the injected clock
 (claude-code), or SDK epoch-ms stamps (codex). A real clock left in makes the
 case pass by accident.
@@ -239,16 +239,15 @@ def now(tz: Any = None) -> datetime:  # type: ignore[override]
         return at(_SteppedDatetime.at_ms)
 
 
-def _delegate_turn(monkeypatch: pytest.MonkeyPatch) -> Turn:
+def _delegate_turn() -> Turn:
     """One tiled window covering the whole turn (the delegate agent builds no
     per-round-trip segments — see ``.claude/notes/agents.md`` § Delegate agent),
     with a real head (dispatch before ``_TurnState`` is constructed) and tail
-    (published after the last flush)."""
-    from coder_eval.agents import delegate_agent as delegate_module
+    (published after the last flush). The clock is injected, as for pi and
+    antigravity: the agent stamps everything off its ``TurnClock``."""
     from coder_eval.agents.delegate_agent import DelegateAgent, _TurnState
     from coder_eval.models import DelegateAgentConfig
 
-    monkeypatch.setattr(delegate_module, "datetime", _SteppedDatetime)
     agent = DelegateAgent(DelegateAgentConfig(type=AgentKind.DELEGATE), task_id="t")
     commands: list[CommandTelemetry] = []
     messages: list[TranscriptMessage] = []
@@ -259,19 +258,19 @@ def emit(e: Any) -> None:
         elif isinstance(e, AgentEndEvent):
             messages.extend(e.messages)
 
-    _SteppedDatetime.at_ms = 500  # dispatch before the turn state exists: head
-    state = _TurnState(iteration=1, user_input="go", model="m")
+    clock = _InjectedClock(at_ms=500)  # dispatch before the turn state exists: head
+    state = _TurnState(iteration=1, user_input="go", model="m", clock=clock)
 
-    _SteppedDatetime.at_ms = 700
+    clock.at_ms = 700
     agent._handle_tool_call({"type": "tool_call", "toolId": "c1", "toolName": "bash", "toolArgs": {}}, state, emit)
-    _SteppedDatetime.at_ms = 1200
+    clock.at_ms = 1200
     agent._handle_tool_result(
         {"type": "tool_result", "toolId": "c1", "toolResult": {"content": "ok"}, "toolStatus": "completed"}, state, emit
     )
-    _SteppedDatetime.at_ms = 1800
+    clock.at_ms = 1800
     agent._handle_event({"type": "message", "content": "done", "isStepStart": True}, state, emit)
 
-    _SteppedDatetime.at_ms = 2000  # last flush: the single window closes here
+    clock.at_ms = 2000  # last flush: the single window closes here
     agent._finalize_turn(state, AgentEndStatus.COMPLETED, emit)
 
     return Turn(started_ms=0.0, ended_ms=2200.0, messages=messages, commands=commands)
@@ -614,8 +613,8 @@ def test_opencode_buckets_tile_the_turn(monkeypatch: pytest.MonkeyPatch):
     assert_identity_closes(_opencode_turn(monkeypatch))
 
 
-def test_delegate_buckets_tile_the_turn(monkeypatch: pytest.MonkeyPatch):
-    assert_identity_closes(_delegate_turn(monkeypatch))
+def test_delegate_buckets_tile_the_turn():
+    assert_identity_closes(_delegate_turn())
 
 
 def test_antigravity_buckets_tile_the_turn():