fix(early-stop): decide skill activation on the tool call, not its result

Armed skill-activation rows that clearly trigger a skill were not
early-stopping and burned their full turn budget. Two facts combined to
break the stop at scale:

1. The live watcher evaluated armed criteria on the tool *result*
   (`ToolEndEvent`). When a turn is cut short (e.g. a timeout), the
   `Skill` tool call is left unresolved: the deadline check discards the
   pending result and `finalize` force-closes the call as UNRESOLVED,
   which the watcher (correctly) ignores. So the decision, though fully
   determined by the call itself, was never observed and the run kept
   going.
2. The final `skill_triggered` scorer used "any engagement" while the
   live verdict used "first engagement". Because the run kept going, the
   agent could engage additional skills, and each extra engagement was
   counted as a false positive against its own criterion.

This change:

- (early_stop) Evaluates armed criteria on the tool *call*
  (`ToolStartEvent`) in addition to the result. For an observable
  criterion the verdict is fully determined by the call's inputs (which
  skill / which command), so the watcher latches the instant the call is
  dispatched. The agent polls `should_stop` immediately after dispatching
  each message, so the loop breaks before a cut-short turn can strip the
  result. The in-flight call (not yet in the collector, which reduces
  commands from `ToolEndEvent`) is appended to the partial trajectory and
  reported at `tool_call_index + 1`; the counter still advances on the
  resolved `ToolEndEvent`. UNRESOLVED ends remain ignored, so an orphaned
  tool closed post-loop can never latch a false stop.

- (skill_triggered) Scores the final check on the FIRST engaged skill,
  matching the live verdict, via a shared `_first_engaged_skill_names`
  helper. A second skill invoked alongside or after the first is no longer
  counted as a competing activation, so it is not a false positive. The
  live verdict and the authoritative score now agree by construction,
  whether or not the run stopped early. Removes the now-unused
  `_engaged_skill` helper.

- (formatting) Skips the SDK `RateLimitEvent` (an out-of-band throttling
  notice) in `format_messages` alongside `StreamEvent`, so it no longer
  logs a spurious "unhandled SDK message type" warning. Cosmetic; does not
  affect scoring.

Adds unit coverage for the tool-call decision (call fires before/without a
result, latches before a later UNRESOLVED end, index accounting), the
orchestrator cut on a resultless call, first-engagement scoring including
the parallel-call case, and the rate-limit event skip.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: anonymous

src/coder_eval/criteria/skill_triggered.py

@@ -44,16 +44,22 @@
 def _engaged_skill_names(cmd: CommandTelemetry) -> set[str]:
     """All skill names engaged by ONE command, agent-agnostically (any-skill).
 
-    Generalizes ``_engaged_skill`` (which asks about one named skill) so a
-    live verdict can detect a *competing* skill engagement. Collects:
-
-    - the Claude ``Skill`` tool call's ``skill`` parameter, namespace-stripped
-      via ``.split(":")[-1]`` (the SDK reports ``plugin:skill-name``); and
-    - every ``skills/<name>/`` or ``skills\\<name>\\`` path segment appearing
-      in any string parameter (the Codex / file-read signal — repo layout or
-      the ``.agents/skills`` sandbox symlink).
-
-    Returns the (possibly empty) set of engaged skill names for this command.
+    Detects both engagement signals so the criterion scores identically across
+    agents, and returns the (possibly empty) set of engaged skill names:
+
+    - Claude: an explicit ``Skill`` tool call carries the skill in
+      ``parameters['skill']``, optionally namespaced (e.g.
+      ``plugin:uipath-agents``); the namespace is stripped via ``.split(":")[-1]``.
+    - Codex (and any non-Claude agent): no ``Skill`` tool exists, so a skill is
+      engaged by reading its files off disk via shell. Both the repo layout
+      (``.../skills/<name>/...``) and the sandbox symlink
+      (``.agents/skills/<name>/...``) contain the substring ``skills/<name>/``,
+      matched here in any string parameter (Bash ``parameters['command']`` or a
+      file-path parameter). The trailing separator required by ``_SKILL_PATH_RE``
+      prevents prefix collisions (``uipath-agents`` vs ``uipath-agents-foo``).
+
+    Returning the full set (rather than a single-skill yes/no) lets callers detect
+    a *competing* skill engagement.
     """
     names: set[str] = set()
     if cmd.tool_name == "Skill":
@@ -66,21 +72,22 @@ def _engaged_skill_names(cmd: CommandTelemetry) -> set[str]:
     return names
 
 
-def _engaged_skill(cmd: CommandTelemetry, skill_name: str) -> bool:
-    """True when one command engaged ``skill_name`` — agent-agnostically.
+def _first_engaged_skill_names(turn_records: list[TurnRecord]) -> set[str]:
+    """Skills engaged by the FIRST command that engages any skill (else empty).
 
-    Claude: an explicit ``Skill`` tool call carries the skill in
-    ``parameters['skill']`` (optionally namespaced, e.g. ``plugin:uipath-agents``).
-
-    Codex (and any non-Claude agent): no ``Skill`` tool exists, so the skill is
-    engaged by reading its files off disk via shell. Both the repo layout
-    (``.../skills/<skill_name>/...``) and the sandbox symlink
-    (``.agents/skills/<skill_name>/...``) contain the substring
-    ``skills/<skill_name>/``, which appears in the recorded command string
-    (Bash ``parameters['command']``) or a file-path parameter. The trailing
-    slash prevents prefix collisions (``uipath-agents`` vs ``uipath-agents-foo``).
+    Activation measures which skill the agent selects *first*, so scoring keys off
+    the first engaging command rather than "any command anywhere in the run". This
+    keeps the final check and the live verdict consistent and prevents a later,
+    incidental engagement (or a second skill invoked alongside the first) from
+    being counted as a competing activation and mis-scored as a false positive.
+    Commands are scanned in ``sequence_number`` order (turn records preserve it).
     """
-    return skill_name in _engaged_skill_names(cmd)
+    for turn in turn_records:
+        for cmd in turn.commands:
+            engaged = _engaged_skill_names(cmd)
+            if engaged:
+                return engaged
+    return set()
 
 
 @register_criterion
@@ -116,9 +123,10 @@ def _check_impl(
                 error="turn_records not provided to checker",
             )
 
-        triggered: bool = any(
-            _engaged_skill(cmd, criterion.skill_name) for turn in turn_records for cmd in turn.commands
-        )
+        # First-engagement policy (mirrors ``live_verdict``): the run is scored on
+        # the FIRST skill the agent engages, so a second skill invoked alongside or
+        # after it is not counted as a competing activation.
+        triggered: bool = criterion.skill_name in _first_engaged_skill_names(turn_records)
         expected_yes: bool = criterion.expected_skill == criterion.skill_name
         score = 1.0 if triggered == expected_yes else 0.0
         observed = _YES if triggered else _NO
@@ -152,18 +160,16 @@ def live_verdict(
 
         This covers the "wrong skill loads" case (a positive wrong signal) and
         negative rows (``expected_skill == ""`` -> any engagement of the target
-        fails that criterion). It is a *policy* made consistent by stopping: the
-        trajectory is frozen at the deciding event, so the standard checker on the
-        truncated trajectory agrees with this verdict by construction.
+        fails that criterion). ``_check_impl`` applies the SAME first-engagement
+        policy on the full trajectory, so the live verdict and the authoritative
+        score agree by construction — whether or not the run stopped early.
         """
+        engaged = _first_engaged_skill_names(turn_records)
+        if not engaged:
+            return "undecided"
         expected_yes = criterion.expected_skill == criterion.skill_name
-        for turn in turn_records:
-            for cmd in turn.commands:
-                engaged = _engaged_skill_names(cmd)
-                if engaged:
-                    observed_yes = criterion.skill_name in engaged
-                    return "pass" if observed_yes == expected_yes else "fail"
-        return "undecided"
+        observed_yes = criterion.skill_name in engaged
+        return "pass" if observed_yes == expected_yes else "fail"
 
     def aggregate(
         self,

src/coder_eval/formatting.py

@@ -80,9 +80,12 @@ def format_messages(
             continue
 
         type_name = type(msg).__name__
-        # StreamEvent is a known SDK type used for token-delta capture
-        # elsewhere; don't surface it as an "unhandled" warning here.
-        if type_name == "StreamEvent":
+        # Known, non-transcript SDK types: StreamEvent carries token deltas
+        # (captured elsewhere) and RateLimitEvent is an out-of-band throttling
+        # notice the SDK interleaves into the stream. Neither is transcript
+        # content, so skip both rather than surfacing an "unhandled" warning.
+        # Matched by name (not import) to stay robust across SDK versions.
+        if type_name in ("StreamEvent", "RateLimitEvent"):
             continue
         if type_name not in warned_unknown_types:
             warned_unknown_types.add(type_name)

src/coder_eval/orchestration/early_stop.py

@@ -9,13 +9,21 @@
   never a silent no-op.
 * ``EarlyStopWatcher`` — the runtime observer. A ``StreamCallback`` composed
   into the agent's event stream that maintains its own ``EventCollector``,
-  evaluates every armed criterion's ``live_verdict`` on each tool completion,
-  applies the stop rule, and exposes ``should_stop()`` (the cooperative
-  interrupt the agent polls) plus ``info`` (the ``EarlyStopInfo`` the
+  evaluates every armed criterion's ``live_verdict`` on each tool *call* (and on
+  its result), applies the stop rule, and exposes ``should_stop()`` (the
+  cooperative interrupt the agent polls) plus ``info`` (the ``EarlyStopInfo`` the
   orchestrator records). Fail-open: a raising ``live_verdict`` disarms the
   watcher and degrades to a full run — a verdict bug can never cause a *false*
   early stop.
 
+Deciding on the tool *call* (``ToolStartEvent``), not the result, is what makes
+the stop robust: for an observable criterion the verdict is fully determined by
+the call's inputs (which skill / which command), so the watcher can latch the
+instant the call is dispatched — before a cut-short turn (e.g. a timeout) can
+strip the result and leave the call unresolved. The agent polls ``should_stop``
+immediately after dispatching each message, so a latch on the call breaks the
+loop before the result message is ever pulled.
+
 Live verdicts only *trigger* the stop; the authoritative scores always come
 from the standard ``check_all`` on the frozen trajectory after the cut.
 """
@@ -28,12 +36,19 @@
 
 from coder_eval.models import EarlyStopInfo, EarlyStopReason
 from coder_eval.streaming.collector import EventCollector
-from coder_eval.streaming.events import AgentStartEvent, StreamEvent, ToolEndEvent, ToolEndStatus, TurnStartEvent
+from coder_eval.streaming.events import (
+    AgentStartEvent,
+    StreamEvent,
+    ToolEndEvent,
+    ToolEndStatus,
+    ToolStartEvent,
+    TurnStartEvent,
+)
 
 
 if TYPE_CHECKING:
     from coder_eval.criteria.base import BaseCriterion, LiveVerdict
-    from coder_eval.models import BaseSuccessCriterion, TaskDefinition
+    from coder_eval.models import BaseSuccessCriterion, CommandTelemetry, TaskDefinition
 
     # Armed pair the watcher holds: (criterion model, its checker). Lives in the
     # TYPE_CHECKING block (only annotations reference it, and those are lazy under
@@ -210,25 +225,35 @@ def for_task(cls, task: TaskDefinition) -> EarlyStopWatcher:
     # --- StreamCallback -------------------------------------------------- #
 
     def on_event(self, event: StreamEvent) -> None:
-        """Forward the event to the internal collector; evaluate on tool completions.
+        """Forward the event to the internal collector; evaluate on each tool call.
 
         Short-circuits once the decision is latched (fired or disarmed). Counts
-        ``TurnStartEvent`` for ``sdk_turn_index`` and ``ToolEndEvent`` for the
-        1-based ``tool_call_index``, and stamps the wall-clock origin at the
-        FIRST ``AgentStartEvent`` only (a retry's second AgentStart does not
+        ``TurnStartEvent`` for ``sdk_turn_index`` and each dispatched tool call
+        for the 1-based ``tool_call_index``, and stamps the wall-clock origin at
+        the FIRST ``AgentStartEvent`` only (a retry's second AgentStart does not
         reset it).
 
-        UNRESOLVED tool ends are ignored entirely (not collected, counted, or
-        evaluated). ``_ClaudeTurnState.finalize`` force-closes orphaned tools as
-        UNRESOLVED *after* the message loop has ended and the terminal status is
-        already chosen (COMPLETED / TIMEOUT / crash) — those are not live tool
-        activity and must not trip the stop rule, or a run that ran to completion
-        (or timed out / crashed) would latch a false early stop (e.g. an
-        unresolved ``Skill`` call counts as engagement because ``skill_triggered``
-        ignores ``result_status``). A legitimate stop only ever fires on an
-        OBSERVED tool result during the loop, so dropping unresolved ends can
-        never suppress a real stop; it also keeps a crashed attempt's orphan tools
-        out of the retry-persistent partial trajectory.
+        The decision is evaluated on the tool *call* (``ToolStartEvent``): for an
+        observable criterion the verdict is fully determined by the call's inputs,
+        so latching here lets the agent's post-dispatch ``should_stop`` poll break
+        the loop before a cut-short turn can strip the result. The call is not in
+        the collector yet (it reduces commands from ``ToolEndEvent``), so it is
+        passed to ``_evaluate`` as the in-flight command, reported at
+        ``tool_call_index + 1`` (it has no ``ToolEndEvent`` to count yet). The
+        matching ``ToolEndEvent`` still evaluates, which covers a verdict that only
+        becomes decidable once the result is known and is a no-op once a call has
+        already latched the stop. ``tool_call_index`` is incremented on the
+        resolved ``ToolEndEvent`` so it stays a count of completed tool calls.
+
+        UNRESOLVED tool ends are ignored entirely (not counted or evaluated).
+        ``_ClaudeTurnState.finalize`` force-closes orphaned tools as UNRESOLVED
+        *after* the message loop has ended and the terminal status is already
+        chosen (COMPLETED / TIMEOUT / crash) — those are not live tool activity
+        and must not trip the stop rule, or a run that ran to completion (or timed
+        out / crashed) without a real, in-loop decision would latch a false early
+        stop. A legitimate stop always fires on the in-loop call, so dropping
+        unresolved ends can never suppress a real stop; it also keeps a crashed
+        attempt's orphan tools out of the retry-persistent partial trajectory.
         """
         if self._info is not None or self._disarmed:
             return
@@ -237,13 +262,19 @@ def on_event(self, event: StreamEvent) -> None:
                 self._started_monotonic = time.monotonic()
         elif isinstance(event, TurnStartEvent):
             self._sdk_turn_index += 1
+        elif isinstance(event, ToolStartEvent):
+            # Decide on the call, evaluating with it appended as the in-flight
+            # command (it has no ToolEnd to count yet, so report it as +1).
+            self._evaluate(in_flight=event.tool)
+            return
         elif isinstance(event, ToolEndEvent):
             if event.status == ToolEndStatus.UNRESOLVED:
                 return
             self._tool_call_index += 1
-        self._collector.on_event(event)
-        if isinstance(event, ToolEndEvent):
+            self._collector.on_event(event)
             self._evaluate()
+            return
+        self._collector.on_event(event)
 
     def should_stop(self) -> bool:
         """The cooperative interrupt the agent polls after each dispatched message."""
@@ -261,8 +292,17 @@ def disarmed(self) -> bool:
 
     # --- Stop rule -------------------------------------------------- #
 
-    def _evaluate(self) -> None:
-        records = [self._collector.build_turn_record()]
+    def _evaluate(self, in_flight: CommandTelemetry | None = None) -> None:
+        record = self._collector.build_turn_record()
+        if in_flight is not None:
+            # The in-flight call has no ToolEnd yet, so the collector (which
+            # reduces commands from ToolEnd) has not captured it. Append it and
+            # re-sort by sequence so the verdict sees it in first-engagement order.
+            record.commands = sorted([*record.commands, in_flight], key=lambda c: c.sequence_number)
+        records = [record]
+        # An in-flight call has not been counted by a ToolEnd yet, so report it as
+        # the next (1-based) tool call.
+        tool_call_index = self._tool_call_index + (1 if in_flight is not None else 0)
         verdicts: list[LiveVerdict] = []
         for criterion, checker in self._armed:
             try:
@@ -283,7 +323,7 @@ def _evaluate(self) -> None:
         # whose stop_when permits fail decides the run.
         for (criterion, _checker), verdict in zip(self._armed, verdicts, strict=True):
             if verdict == "fail" and criterion.stop_when in ("fail", "decided"):
-                self._fire(EarlyStopReason.CRITERION_FAILED, criterion)
+                self._fire(EarlyStopReason.CRITERION_FAILED, criterion, tool_call_index=tool_call_index)
                 return
 
         # Pass-stop: EVERY armed criterion live-passes AND each permits pass.
@@ -296,13 +336,13 @@ def _evaluate(self) -> None:
             for (criterion, _checker), verdict, prev in zip(self._armed, verdicts, self._prev_verdicts, strict=True):
                 if verdict != prev:
                     deciding = criterion
-            self._fire(EarlyStopReason.CRITERION_PASSED, deciding)
+            self._fire(EarlyStopReason.CRITERION_PASSED, deciding, tool_call_index=tool_call_index)
             return
 
         # No stop this round — record the verdicts so the next round can detect flips.
         self._prev_verdicts = verdicts
 
-    def _fire(self, reason: EarlyStopReason, criterion: BaseSuccessCriterion) -> None:
+    def _fire(self, reason: EarlyStopReason, criterion: BaseSuccessCriterion, *, tool_call_index: int) -> None:
         elapsed = 0.0
         if self._started_monotonic is not None:
             elapsed = max(time.monotonic() - self._started_monotonic, 0.0)
@@ -313,7 +353,7 @@ def _fire(self, reason: EarlyStopReason, criterion: BaseSuccessCriterion) -> Non
             deciding_criterion_description=criterion.description,
             armed_criteria=[f"{c.type}: {c.description}" for c, _ in self._armed],
             sdk_turn_index=self._sdk_turn_index,
-            tool_call_index=self._tool_call_index,
+            tool_call_index=tool_call_index,
             elapsed_seconds=elapsed,
             turns_remaining_at_stop=turns_remaining,
         )
@@ -323,6 +363,6 @@ def _fire(self, reason: EarlyStopReason, criterion: BaseSuccessCriterion) -> Non
             reason.value,
             criterion.type,
             self._sdk_turn_index,
-            self._tool_call_index,
+            tool_call_index,
             elapsed,
         )

tests/test_agent.py

@@ -671,6 +671,18 @@ class BareMessage:
     assert "[ASSISTANT] Second response" in formatted
     assert formatted.count("[ASSISTANT]") == 2
 
+    # Test 12: RateLimitEvent — an out-of-band throttling notice the SDK
+    # interleaves into the stream. It is not transcript content, so it is skipped
+    # (matched by class name) and does NOT surface as an "unhandled" tag/warning.
+    class RateLimitEvent:
+        pass
+
+    formatted = agent._format_messages([RateLimitEvent()])
+    assert formatted == "[No output]"
+    # It also does not crowd out real content when interleaved.
+    formatted = agent._format_messages([RateLimitEvent(), _make_assistant("still here")])
+    assert formatted == "[ASSISTANT] still here"
+
 
 def test_format_messages_system_message_subclasses_are_filtered():
     """Regression: SystemMessage SUBCLASSES (TaskStartedMessage, etc.) must