@@ -267,10 +267,15 @@ matter how much the run actually billed. So the CLI harnesses crash rather than
   arm keys on a step the CLI reported as FINISHED, its own claim that a generation
   completed, rather than on `usage.is_empty()` alone, which would also condemn a stream
   cut before any step could finish.
-- **A terminal provider error** (Pi, OpenCode) — `pi -p` exits 0 after exhausting its
+- **A terminal provider error** (Pi, OpenCode, Codex) — `pi -p` exits 0 after exhausting its
   internal retries, so without the crash the turn books as a clean COMPLETED
   (`FinalStatus.FAILURE`, category "failed"), silently depressing the measured pass rate.
   Crashing routes it to `FinalStatus.ERROR`, which is excluded from outcomes.
+  Codex reports the same thing differently: its `turn/completed` arrives normally but
+  carries `turn.status == failed` and a `TurnError` (for example `rateLimitExceeded`
+  after its own stream retries ran out). The harness reads that status and crashes, with
+  the provider's message, so `categorize_error` routes a rate limit to
+  `AGENT_RATE_LIMIT` and its retry budget like any other agent crash.
 - **A CLI that closed its stream but would not exit** within the grace period.
 
 Every arm is gated on `stopped_early` / `max_turns_exhausted`, because an intentional cut
 
@@ -259,6 +259,23 @@ def _status_value(status: Any) -> str:
     return str(value).lower() if value is not None else ""
 
 
+def _failed_turn_reason(turn: Any) -> str | None:
+    """The crash message for a Turn codex ended as ``failed``, or None for any other status.
+
+    Codex ends a turn ``failed`` when it gives up on the provider, e.g. a
+    ``rateLimitExceeded`` it will not retry. The message carries the provider's
+    own text, so ``categorize_error`` routes it (``rate limit`` ->
+    ``AGENT_RATE_LIMIT``) like any other agent crash.
+    """
+    if turn is None or _status_value(getattr(turn, "status", None)) != "failed":
+        return None
+    error = getattr(turn, "error", None)
+    message = str(getattr(error, "message", "") or "").strip() or "no error message"
+    kind, _ = _codex_error_kind(error)
+    suffix = f" (codexErrorInfo={kind})" if kind else ""
+    return f"Codex turn failed: {message}{suffix}"
+
+
 def _record_command_text(argv: Any) -> str:
     """The command a CommandExecution record ran, without the shell wrapper Codex adds.
 
@@ -384,24 +401,29 @@ def _message_uncached_input(m: AssistantMessage) -> int:
 _ZSH_PROFILE_NAMES = frozenset({".zshenv", ".zprofile", ".zshrc"})
 
 
-def _provider_error_from(payload: Any) -> ProviderError:
-    """Map a codex ``ErrorNotification`` payload onto a ``ProviderError``.
+def _codex_error_kind(error: Any) -> tuple[str | None, int | None]:
+    """``(kind, http_status)`` of a codex ``TurnError``'s ``codexErrorInfo``.
 
     ``codexErrorInfo`` is either a bare category (enum) or a one-key object whose
     key is the category and whose value may carry ``httpStatusCode``.
     """
-    error = getattr(payload, "error", None)
-    kind: str | None = None
-    http_status: int | None = None
-    info = getattr(getattr(error, "codex_error_info", None), "root", None)
+    info: Any = getattr(getattr(error, "codex_error_info", None), "root", None)
     if isinstance(info, Enum):
-        kind = str(info.value)
-    elif hasattr(info, "model_dump"):
+        return str(info.value), None
+    if hasattr(info, "model_dump"):
         dumped = info.model_dump(by_alias=True, mode="json")
         if len(dumped) == 1:
             kind, body = next(iter(dumped.items()))
             if isinstance(body, dict) and isinstance(body.get("httpStatusCode"), int):
-                http_status = body["httpStatusCode"]
+                return kind, body["httpStatusCode"]
+            return kind, None
+    return None, None
+
+
+def _provider_error_from(payload: Any) -> ProviderError:
+    """Map a codex ``ErrorNotification`` payload onto a ``ProviderError``."""
+    error = getattr(payload, "error", None)
+    kind, http_status = _codex_error_kind(error)
     return ProviderError(
         at=datetime.now(),
         message=str(getattr(error, "message", "") or ""),
@@ -1199,6 +1221,12 @@ def _on_turn_timeout() -> None:
                 # like every other timeout path.
                 assert timeout is not None
                 self._finalize_and_raise_timeout(state.finalize, timeout)
+            failed_reason = None if state.ended_cleanly else _failed_turn_reason(state.result_turn)
+            if failed_reason is not None:
+                # Codex gave up on the turn (a provider error it would not retry),
+                # so the agent never finished: crash, never score it.
+                # Rationale: .claude/notes/agents.md § Why a clean exit can still be a crash
+                self._finalize_and_raise_crash(state.finalize, truncate_crash_message(failed_reason))
         except (AgentCrashError, TurnTimeoutError):
             # Already funneled through finalize by the inner handlers.
             raise
 
@@ -2985,3 +2985,125 @@ def test_clean_turn_has_no_provider_errors(self):
         st, collector = self._state()
         st.finalize(AgentEndStatus.COMPLETED)
         assert collector.build_turn_record().provider_errors == []
+
+
+# The provider message codex reported in adhoc-2026-10-21_16-16-27
+# (v2/skill-case-diagnose-formal-arg-slot-ids): five retried stream disconnects,
+# then a rate limit it would not retry, then a `failed` turn/completed.
+_RATE_LIMIT_MESSAGE = (
+    "rate limit exceeded: Your requests to gpt-5.6-luna for gpt-5.6-luna in eastus2 have exceeded token rate limit."
+)
+
+
+def _error_notification(error: dict, *, will_retry: bool) -> SimpleNamespace:
+    from openai_codex.generated.v2_all import ErrorNotification
+
+    payload = ErrorNotification.model_validate(
+        {"error": error, "threadId": "t1", "turnId": "u1", "willRetry": will_retry}
+    )
+    return SimpleNamespace(method="error", payload=payload)
+
+
+def _turn_failed(error: dict | None) -> SimpleNamespace:
+    turn = Turn.model_validate(
+        {
+            "id": "turn_1",
+            "status": "failed",
+            "durationMs": 21847,
+            "error": error,
+            "items": [],
+            "itemsView": "notLoaded",
+        }
+    )
+    return SimpleNamespace(method="turn/completed", payload=TurnCompletedNotification(thread_id="th_1", turn=turn))
+
+
+def _rate_limited_turn(*, commands: int) -> list[SimpleNamespace]:
+    notes: list[SimpleNamespace] = []
+    for i in range(commands):
+        root = SimpleNamespace(
+            type="commandExecution",
+            id=f"cmd_{i}",
+            command=f"uip maestro case validate {i}",
+            exit_code=0,
+            aggregated_output="ok\n",
+            duration_ms=5,
+        )
+        notes += [_item_notification("item/started", root), _item_notification("item/completed", root)]
+    notes.append(_delta("I'm using the Maestro Case skill to repair the plan."))
+    for attempt in range(1, 6):
+        notes.append(
+            _error_notification(
+                {
+                    "message": f"Reconnecting... {attempt}/5",
+                    "codexErrorInfo": {"responseStreamDisconnected": {"httpStatusCode": None}},
+                    "additionalDetails": _RATE_LIMIT_MESSAGE,
+                },
+                will_retry=True,
+            )
+        )
+    notes.append(
+        _error_notification({"message": _RATE_LIMIT_MESSAGE, "codexErrorInfo": "rateLimitExceeded"}, will_retry=False)
+    )
+    notes.append(_turn_failed({"message": _RATE_LIMIT_MESSAGE, "codexErrorInfo": "rateLimitExceeded"}))
+    return notes
+
+
+class TestFailedTurnIsACrash:
+    """A `turn/completed` whose turn is `failed` is a crash, never a COMPLETED turn.
+
+    Before this, codex's terminal rate limit booked as COMPLETED and the row was
+    graded FAILURE: 27 of 114 rows in adhoc-2026-10-21_16-16-27 (codex,
+    gpt-5.6-luna) were scored on a turn the provider cut off.
+    """
+
+    @pytest.mark.parametrize("commands", [0, 4])
+    async def test_rate_limited_turn_raises_crash_and_keeps_the_partial(self, commands):
+        agent = _started_agent(parse_agent_config(type=AgentKind.CODEX), _rate_limited_turn(commands=commands))
+
+        with pytest.raises(AgentCrashError) as excinfo:
+            await agent.communicate("repair the case")
+
+        assert _RATE_LIMIT_MESSAGE in str(excinfo.value)
+        assert "codexErrorInfo=rateLimitExceeded" in str(excinfo.value)
+        assert agent.get_state() == AgentState.ERROR
+
+        partial = agent.pending_turn
+        assert partial is not None
+        assert partial.crashed is True
+        assert partial.crash_reason is not None and "rate limit exceeded" in partial.crash_reason
+        # The work done before the cut and the provider evidence both survive.
+        assert len(partial.commands) == commands
+        assert [(e.kind, e.will_retry) for e in partial.provider_errors] == [
+            *[("responseStreamDisconnected", True)] * 5,
+            ("rateLimitExceeded", False),
+        ]
+
+    async def test_rate_limit_crash_is_categorized_as_a_rate_limit(self):
+        from coder_eval.errors.categories import ErrorCategory
+        from coder_eval.errors.categorization import categorize_error
+
+        agent = _started_agent(parse_agent_config(type=AgentKind.CODEX), _rate_limited_turn(commands=0))
+        with pytest.raises(AgentCrashError) as excinfo:
+            await agent.communicate("repair the case")
+
+        assert categorize_error(excinfo.value, {"component": "agent"}) is ErrorCategory.AGENT_RATE_LIMIT
+
+    async def test_failed_turn_without_error_details_still_crashes(self):
+        agent = _started_agent(parse_agent_config(type=AgentKind.CODEX), [_delta("hi"), _turn_failed(None)])
+
+        with pytest.raises(AgentCrashError, match="no error message"):
+            await agent.communicate("go")
+
+    async def test_completed_turn_with_a_retried_provider_error_is_not_a_crash(self):
+        notes = [
+            _error_notification({"message": "Reconnecting... 1/5"}, will_retry=True),
+            _delta("done"),
+            _turn_completed(),
+        ]
+        agent = _started_agent(parse_agent_config(type=AgentKind.CODEX), notes)
+
+        record = await agent.communicate("go")
+
+        assert record.crashed is False
+        assert agent.pending_turn is None