fix: four harness bugs found by the tb2.1 fusion-vs-opus differential

eval/harbor/clawcodex_agent.py

@@ -723,6 +723,7 @@ def populate_context_post_run(self, context: AgentContext) -> None:
         """
         events = self._parse_stream_events()
         result_event = self._last_result_event(events)
+        usage_event = self._last_usage_event(events)
 
         # Surface an EARLY STOP where a human reading the trial will see it.
         # clawcodex now marks a run the agent loop cut short with a non-success
@@ -752,7 +753,9 @@ def populate_context_post_run(self, context: AgentContext) -> None:
             }
 
         try:
-            trajectory, totals = self._build_trajectory(events, result_event)
+            trajectory, totals = self._build_trajectory(
+                events, result_event, usage_event
+            )
         except Exception as exc:  # noqa: BLE001 — never fail a trial over this
             self.logger.debug(f"clawcodex trajectory build failed: {exc}")
             trajectory, totals = None, None
@@ -772,15 +775,19 @@ def populate_context_post_run(self, context: AgentContext) -> None:
             context.n_output_tokens = totals["completion"]
             if totals["cost"] is not None:
                 context.cost_usd = totals["cost"]
-        elif result_event:
-            usage = result_event.get("usage")
-            if isinstance(usage, dict):
-                input_tokens = usage.get("input_tokens") or 0
-                cache_read = usage.get("cache_read_input_tokens") or 0
-                cache_creation = usage.get("cache_creation_input_tokens") or 0
-                context.n_input_tokens = input_tokens + cache_read + cache_creation
-                context.n_cache_tokens = cache_read
-                context.n_output_tokens = usage.get("output_tokens") or 0
+        elif result_event and isinstance(result_event.get("usage"), dict):
+            prompt, cached, completion = self._usage_columns(result_event["usage"])
+            context.n_input_tokens = prompt
+            context.n_cache_tokens = cached
+            context.n_output_tokens = completion
+        elif usage_event is not None:
+            # Killed trial: no copy-back, no result event. The stream log is
+            # a live bind mount so its last cumulative usage line survived,
+            # and it is the only measurement of what this run spent.
+            prompt, cached, completion = self._usage_columns(usage_event["usage"])
+            context.n_input_tokens = prompt
+            context.n_cache_tokens = cached
+            context.n_output_tokens = completion
 
         if trajectory is None:
             return
@@ -819,17 +826,65 @@ def _last_result_event(events: list[dict[str, Any]]) -> dict[str, Any] | None:
                 return event
         return None
 
+    @staticmethod
+    def _last_usage_event(events: list[dict[str, Any]]) -> dict[str, Any] | None:
+        """The most recent incremental ``usage`` event, or None.
+
+        The metrics lane of last resort, for a trial that was KILLED. Harbor
+        cancels the exec on timeout, so the container copy-back never runs
+        (no session data) and clawcodex never reaches its terminal ``result``
+        event (no result usage) — the two sources below. Every killed trial
+        therefore reported no tokens and no cost at all: 21 of 46 in one
+        terminal-bench job (2026-08-02), and precisely the longest, most
+        expensive ones, because those are what time out. Job totals are summed
+        from per-trial values, so the headline cost was a floor biased low by
+        exactly the trials that cost the most.
+
+        The stream log survives a kill (it is a live bind mount of the host
+        trial dir), so the last cumulative usage line in it is a real
+        measurement of everything the run had spent up to the kill.
+        """
+        for event in reversed(events):
+            if event.get("type") == "usage" and isinstance(event.get("usage"), dict):
+                return event
+        return None
+
+    @staticmethod
+    def _usage_columns(usage: dict[str, Any]) -> tuple[int, int, int]:
+        """``(prompt, cached, completion)`` from a stream-json usage dict.
+
+        ``input_tokens`` is only the NON-cached part of the prompt, so a total
+        built from input+output alone silently omits every cached token — the
+        bug #786 fixed. One helper so the result-event and usage-event lanes
+        cannot drift apart on that arithmetic again.
+        """
+        cache_read = usage.get("cache_read_input_tokens") or 0
+        cache_creation = usage.get("cache_creation_input_tokens") or 0
+        prompt = (usage.get("input_tokens") or 0) + cache_read + cache_creation
+        return prompt, cache_read, (usage.get("output_tokens") or 0)
+
     def _final_metrics(
         self,
         result_event: dict[str, Any] | None,
         total_steps: int,
         totals: dict[str, Any] | None,
+        usage_event: dict[str, Any] | None = None,
     ) -> FinalMetrics | None:
-        """``totals`` = authoritative session billing totals (preferred).
-        Falls back to the stream-json usage when no session cost block was
-        synced; that fallback now carries cumulative cache counters, so it is
-        no longer short by the cached portion — though it still omits
-        subagent and compaction tokens, which only the cost block sees."""
+        """Three lanes, in descending order of authority:
+
+        1. ``totals`` — the session's BILLING totals. Includes subagent and
+           compaction tokens, which nothing else sees.
+        2. the terminal ``result`` event's usage — main loop only, but
+           carries cumulative cache counters so it is not short by the
+           cached portion.
+        3. the last incremental ``usage`` event — the only one of the three
+           that survives a KILLED run, where the copy-back never ran and no
+           result event was ever emitted. See ``_last_usage_event``.
+
+        Cost is unavailable in lane 3 (the incremental event carries tokens
+        only); harbor prices those itself, and a token count is what the
+        killed trials were missing entirely.
+        """
         usage = (result_event or {}).get("usage")
         if not isinstance(usage, dict):
             usage = {}
@@ -839,17 +894,17 @@ def _final_metrics(
             cached = totals["cached"]
             cost = totals["cost"]
         elif result_event:
-            input_tokens = usage.get("input_tokens") or 0
-            cache_read = usage.get("cache_read_input_tokens") or 0
-            cache_creation = usage.get("cache_creation_input_tokens") or 0
-            prompt = input_tokens + cache_read + cache_creation
-            completion = usage.get("output_tokens") or 0
-            cached = cache_read
+            prompt, cached, completion = self._usage_columns(usage)
             cost = result_event.get("total_cost_usd")
+        elif usage_event:
+            prompt, cached, completion = self._usage_columns(usage_event["usage"])
+            cost = None
         else:
             return None
         extra: dict[str, Any] = {}
         num_turns = (result_event or {}).get("num_turns")
+        if not isinstance(num_turns, int) and usage_event:
+            num_turns = usage_event.get("num_turns")
         if isinstance(num_turns, int):
             extra["num_turns"] = num_turns
         duration_ms = (result_event or {}).get("duration_ms")
@@ -992,6 +1047,7 @@ def _build_trajectory(
         self,
         events: list[dict[str, Any]],
         result_event: dict[str, Any] | None,
+        usage_event: dict[str, Any] | None = None,
     ) -> tuple[Trajectory | None, dict[str, Any] | None]:
         """Returns ``(trajectory, billing_totals)`` — the totals are surfaced
         so the caller can also set the leaderboard token/cost columns."""
@@ -1023,7 +1079,9 @@ def _build_trajectory(
             agent=agent,
             steps=steps,
             notes=notes,
-            final_metrics=self._final_metrics(result_event, len(steps), totals),
+            final_metrics=self._final_metrics(
+                result_event, len(steps), totals, usage_event
+            ),
         )
         return trajectory, totals
 

src/cli_core/__init__.py

@@ -18,6 +18,7 @@
     StreamJsonWriter,
     SystemEvent,
     ToolResultEvent,
+    UsageEvent,
     ToolUseEvent,
     UserInputMessage,
 )
@@ -34,6 +35,7 @@
     "StreamJsonWriter",
     "SystemEvent",
     "ToolResultEvent",
+    "UsageEvent",
     "ToolUseEvent",
     "UserInputMessage",
 ]

src/cli_core/structured_io.py

@@ -82,6 +82,33 @@ class ToolResultEvent(HeadlessEvent):
     is_error: bool = False
 
 
+@dataclass
+class UsageEvent(HeadlessEvent):
+    """Running token totals, emitted as the run proceeds.
+
+    ``ResultEvent`` carries the authoritative usage, but only at the very end.
+    A run that is killed — an eval harness hitting its per-task ceiling, a
+    SIGKILL, a lost connection — never emits one, so everything it spent
+    became unmeasurable. Measured on terminal-bench 2.1 (2026-08-02): 21 of
+    46 trials in one job reported no tokens and no cost at all, and they were
+    exactly the killed ones, i.e. the longest and most expensive. Job totals
+    are summed from per-trial values, so the headline cost was a floor biased
+    low by precisely the trials that cost the most.
+
+    Emitted per assistant message (one model round trip), which is the
+    granularity that survives a kill mid-turn. Cumulative, not per-message, so
+    a consumer only needs the LAST one it saw and can ignore the rest.
+
+    Additive: a new event ``type``, so existing consumers that switch on the
+    types they know are unaffected. It does not replace or change
+    ``ResultEvent.usage``.
+    """
+
+    type: str = "usage"
+    usage: dict[str, Any] = field(default_factory=dict)
+    num_turns: int = 0
+
+
 @dataclass
 class ResultEvent(HeadlessEvent):
     type: str = "result"

src/entrypoints/headless.py

@@ -44,6 +44,7 @@
     SystemEvent,
     ToolResultEvent,
     ToolUseEvent,
+    UsageEvent,
     UserInputMessage,
     cli_error,
     ndjson_safe_dumps,
@@ -617,6 +618,13 @@ def _with_goal_continuations(
     aggregate_tool_events: list[dict] = []
     num_turns_total = 0
     usage_total: dict[str, int] = {}
+    # Running totals for the incremental UsageEvent lane. Deliberately a
+    # SEPARATE accumulator from ``usage_total`` above: that one folds
+    # ``result.usage`` once per completed turn, this one folds each assistant
+    # message's usage as it lands. Feeding both from either source would
+    # double-count, and this lane exists precisely for runs that never reach
+    # the per-turn accounting.
+    live_usage: dict[str, int] = {}
     exit_code = 0
     # Terminal reason of the LAST agent-loop turn, when it stopped the run
     # early rather than the model finishing (``tool_failure_loop``,
@@ -710,6 +718,34 @@ def _persist(msg: Any) -> None:
                                     msg.content,
                                     usage=getattr(msg, "usage", None),
                                 )
+                                # Publish the running total as we go. The
+                                # ResultEvent below is authoritative but only
+                                # exists if the run REACHES the end; a killed
+                                # run (eval ceiling, SIGKILL, dropped
+                                # connection) emitted nothing, so everything
+                                # it spent was unmeasurable. Per assistant
+                                # message is the granularity that survives a
+                                # kill mid-turn — a turn that never completes
+                                # never reaches the per-turn accounting below.
+                                #
+                                # Tracked SEPARATELY from ``usage_total``,
+                                # which folds ``result.usage`` once per turn:
+                                # accumulating both from one source would
+                                # double-count. This lane feeds the stream
+                                # only; the ResultEvent is untouched.
+                                _msg_usage = getattr(msg, "usage", None)
+                                if (
+                                    writer is not None
+                                    and msg.role == "assistant"
+                                    and _msg_usage
+                                ):
+                                    _accumulate_usage(live_usage, _msg_usage)
+                                    writer.write(
+                                        UsageEvent(
+                                            usage=dict(live_usage),
+                                            num_turns=num_turns_total,
+                                        )
+                                    )
                                 # Claude Code's stream-json exposes signed
                                 # thinking blocks as assistant content events.
                                 # Emit only the private blocks here; the visible

src/providers/fusion_provider.py

@@ -124,6 +124,15 @@ def expired(self) -> bool:
         return time.monotonic() >= self.expires_at
 
 
+class _EmptyVisionResponse(RuntimeError):
+    """The vision model answered 200 with no usable text.
+
+    Distinguished from every other vision failure because it is the one that
+    is worth a retry: the provider is reachable and fast, it just produced
+    nothing this turn. See :meth:`FusionProvider._describe`.
+    """
+
+
 class _Budget:
     """Per-request limits on the rewrite: call count, wall clock, and abort.
 
@@ -452,12 +461,48 @@ def _vision_timeout(self, provider: Any) -> Any:
         except Exception:  # noqa: BLE001 — the float still satisfies the SDK
             return seconds
 
-    def _describe(self, source: dict[str, Any]) -> str:
+    def _describe(self, source: dict[str, Any], budget: "_Budget | None" = None) -> str:
         """Ask the vision model about one image. Returns its description.
 
         Raises on failure; :meth:`_substitute` converts that into a text note
         so invariant 1 (no image block survives) always holds.
+
+        An EMPTY 200 gets one retry; nothing else does. The distinction is
+        what keeps the negative cache honest. A transport failure means the
+        vision provider is unreachable, and ``_substitute`` caches that
+        permanently on purpose — the docstring's arithmetic (8 images x 60 s
+        x 2 attempts on every turn, forever, uninterruptible) is why. An empty
+        completion is the opposite situation: the provider is up, answered
+        fast, and just produced nothing that turn. Caching THAT permanently
+        loses the image for the rest of the session over a transient quirk —
+        observed on terminal-bench gcode-to-text (2026-08-02), where
+        ``openai:gpt-5.6-luna`` returned no text for image 2 of 5 and the task
+        scored 0 against a baseline that solved it.
+
+        Bounded at one extra round trip per distinct image, and only while the
+        request's own call/time budget still allows it, so the outage
+        arithmetic above is unchanged.
         """
+        try:
+            return self._describe_once(source)
+        except _EmptyVisionResponse:
+            blocked = budget.exhausted() if budget is not None else None
+            if blocked is not None:
+                logger.warning(
+                    "[fusion] vision returned no text and %s; not retrying", blocked
+                )
+                raise
+            if budget is not None:
+                # A retry is a real network call, so charge it. Otherwise a
+                # provider stuck returning empty 200s would double the
+                # fan-out this cap exists to bound.
+                budget.calls -= 1
+            logger.info("[fusion] vision returned no text; retrying once")
+            return self._describe_once(source)
+
+    def _describe_once(self, source: dict[str, Any]) -> str:
+        """One vision call. Raises :class:`_EmptyVisionResponse` for an empty
+        200 and lets every other failure propagate as-is."""
         prompt = self._prompt()
         provider = self._vision()
         # The image is handed over in ANTHROPIC block shape and the target
@@ -502,7 +547,10 @@ def _describe(self, source: dict[str, Any]) -> str:
             # response shape drops). Treat it as a failure so the caller
             # emits an explicit note rather than an empty text block that
             # reads to the base model as "the image was blank".
-            raise RuntimeError(
+            #
+            # Typed, not a bare RuntimeError: ``_describe`` retries THIS and
+            # only this, and a string match would be a fragile way to say so.
+            raise _EmptyVisionResponse(
                 f"vision model {self._fusion.vision.selector} returned no text"
             )
         if len(text) > _MAX_DESCRIPTION_CHARS:
@@ -551,7 +599,7 @@ def _substitute(self, block: Any, budget: "_Budget") -> Any:
 
         budget.calls -= 1
         try:
-            description = self._describe(source)
+            description = self._describe(source, budget)
         except Exception as exc:  # noqa: BLE001 — invariant 1: never re-raise
             # Degrading to a note keeps the turn alive. Re-raising, or
             # leaving the image in place, reproduces the exact 400 this