@@ -80,6 +80,21 @@ intentionally brief and out of scope; trimming for DISPLAY belongs in the render
   table's to state, not this file's. Full table + rationale:
   docs/agents/HARNESS_PARITY.md.
 
+## Codex compaction usage metadata
+
+Codex emits usage for the compaction model response, then emits another usage update
+after estimating the replacement context. The estimate has zero input, output, cache
+and reasoning deltas. Within a compaction lifecycle, that later bookkeeping update
+still updates usage and transcript capture but does not begin another model call.
+The first compaction usage update counts even when its reported deltas are zero;
+completion clears the lifecycle before ordinary model and tool calls resume.
+
+The source contract is visible in the pinned CLI's `openai/codex` tag `rust-v0.156.1`:
+`compact.rs` calls `update_token_usage_info` and then `recompute_token_usage`, each of
+which emits a `TokenCount` event. The app-server forwards both as
+`thread/tokenUsage/updated`. An ordinary tool call's next-call reservation remains
+unchanged. An estimate outside compaction keeps the existing accounting behavior.
+
 ## Shared turn lifecycle
 
 Every adapter drives the same skeleton, on the base class: `_begin_turn()` resets the
 
@@ -553,8 +553,11 @@ Each harness finds the call boundary in its own stream (the table above):
   `error_max_turns` stop. The CLI does not apply it on every route, so the harness
   also counts distinct main-thread `message_id`s and ends the turn itself when call
   N+1 begins.
-- **Codex** sends `thread/tokenUsage/updated` once per call, after that call's
-  tools finish. An item starts as its tool runs, so waiting for the next call's
+- **Codex** sends `thread/tokenUsage/updated` after a call's tools finish. Context
+  compaction also sends a zero-delta context-size estimate; that bookkeeping
+  update does not count as another call. The compaction model call itself counts,
+  unlike Claude Code's compaction, which emits no main-thread assistant message.
+  An item starts as its tool runs, so waiting for the next call's
   first item would let one tool of call N+1 act. A call that ran tools therefore
   opens the next call at its `tokenUsage`, since the results always go back to the
   model, and the cap fires before call N+1 can run anything.
 
@@ -557,6 +557,8 @@ def __init__(
         self.api_calls = 0
         self.in_api_call = False
         self.call_ran_tools = False
+        self.in_context_compaction = False
+        self.compaction_usage_seen = False
 
         # Finalize inputs, COMMITTED by communicate after a clean pump return.
         # Defaults are the crash values (no terminal usage; format from messages).
@@ -755,6 +757,9 @@ def on_item_started(self, notification: Any) -> None:
         if item_id is not None and started_at_ms is not None:
             self.start_ms_by_id[item_id] = started_at_ms
         root_type = getattr(root, "type", None)
+        if root_type == "contextCompaction":
+            self.in_context_compaction = True
+            self.compaction_usage_seen = False
         if not self.in_api_call and root_type not in _NON_MODEL_ITEM_TYPES:
             self._open_api_call()
         # Any item that isn't transcript content is a tool call (generic capture).
@@ -790,6 +795,8 @@ def on_item_completed(self, notification: Any) -> None:
             return
         completed_ms = getattr(notification.payload, "completed_at_ms", None)
         root_type = getattr(root, "type", None)
+        if root_type == "contextCompaction":
+            self.in_context_compaction = False
         if root_type is not None and root_type not in _CONTENT_ITEM_TYPES:
             tool_id = getattr(root, "id", None) or f"{root_type}_{self.next_sequence}"
             seq = self.seq_by_id.get(tool_id, self.next_sequence)
@@ -875,18 +882,39 @@ def on_agent_message_delta(self, notification: Any) -> None:
                 self.emit.on_event(TextChunkEvent(task_id=self.task_id, turn_id=self.turn_id, text=delta))
 
     def on_token_usage_updated(self, notification: Any) -> None:
-        """One per generation → cut a message. Carries `total` (cumulative over the
-        whole THREAD, i.e. every turn so far) and `last` (this generation's delta)."""
-        if not self.in_api_call:
-            self.api_calls += 1
-        self.in_api_call = False
-        # Tool results always go back to the model, so the next call begins here,
-        # before Codex can run the next call's tools (an item starts as its tool runs).
-        if self.call_ran_tools:
-            self._open_api_call()
+        """Capture generation usage and compaction's context-size estimates.
+
+        Rationale: .claude/notes/agents.md § Codex compaction usage metadata
+        """
+        usage = getattr(notification.payload, "token_usage", None) if notification.payload else None
+        last = getattr(usage, "last", None)
+        estimate = (
+            self.in_context_compaction
+            and self.compaction_usage_seen
+            and last is not None
+            and all(
+                (getattr(last, key, 0) or 0) == 0
+                for key in (
+                    "input_tokens",
+                    "output_tokens",
+                    "cached_input_tokens",
+                    "cache_write_input_tokens",
+                    "reasoning_output_tokens",
+                )
+            )
+        )
+        if self.in_context_compaction:
+            self.compaction_usage_seen = True
+        if not estimate:
+            if not self.in_api_call:
+                self.api_calls += 1
+            self.in_api_call = False
+            # Tool results reserve the next model call before its tools can run.
+            if self.call_ran_tools:
+                self._open_api_call()
         if notification.payload:
-            self.latest_token_usage = getattr(notification.payload, "token_usage", None)
-            self._flush_message(getattr(self.latest_token_usage, "last", None))
+            self.latest_token_usage = usage
+            self._flush_message(last)
 
     def on_turn_completed(self, notification: Any) -> bool:
         """Capture the final Turn. Returns True (break the pump) iff the payload is
 
@@ -19,7 +19,7 @@
     _CODEX_APPROVAL_MODE,
     CodexAgent,
 )
-from coder_eval.models import AgentConfig, AgentKind, parse_agent_config
+from coder_eval.models import AgentConfig, AgentKind, AssistantMessage, ReconciliationMessage, parse_agent_config
 
 
 class TestCodexAgentInitialization:
@@ -504,6 +504,7 @@ def test_get_state_returns_current_state():
 
 from openai_codex.generated.v2_all import (  # noqa: E402
     SubAgentActivityThreadItem,
+    TokenUsageBreakdown,
     Turn,
     TurnCompletedNotification,
 )
@@ -2229,6 +2230,130 @@ def _cmd_notifications(count: int) -> list:
         notifications.append(_turn_completed())
         return notifications
 
+    @staticmethod
+    def _compaction_notifications(
+        *,
+        estimate: bool = True,
+        empty_response: bool = False,
+        item_id: str = "compact",
+        cache_write: int | None = 0,
+    ) -> list:
+        item = SimpleNamespace(type="contextCompaction", id=item_id)
+        response = _token_usage(inp=10, out=5, cached=0)
+        response.payload.token_usage.last = TokenUsageBreakdown(
+            input_tokens=0 if empty_response else 10,
+            output_tokens=0 if empty_response else 5,
+            cached_input_tokens=0,
+            cache_write_input_tokens=0,
+            reasoning_output_tokens=0,
+            total_tokens=0 if empty_response else 15,
+        )
+        notifications = [_item_notification("item/started", item), response]
+        if estimate:
+            recomputed = _token_usage(inp=20, out=5, cached=0)
+            recomputed.payload.token_usage.last = TokenUsageBreakdown(
+                input_tokens=0,
+                output_tokens=0,
+                cached_input_tokens=0,
+                cache_write_input_tokens=cache_write,
+                reasoning_output_tokens=0,
+                total_tokens=20,
+            )
+            notifications.append(recomputed)
+        notifications.append(_item_notification("item/completed", item))
+        return notifications
+
+    @pytest.mark.parametrize(
+        "estimate,empty_response,cache_write",
+        [(True, False, 0), (False, False, 0), (True, True, 0), (True, False, None)],
+    )
+    async def test_compaction_estimate_does_not_spend_model_call_budget(self, estimate, empty_response, cache_write):
+        notifications = self._cmd_notifications(1)
+        notifications[5:5] = self._compaction_notifications(
+            estimate=estimate, empty_response=empty_response, cache_write=cache_write
+        )
+        agent = _started_agent(parse_agent_config(type=AgentKind.CODEX), notifications)
+
+        record = await agent.communicate("go", max_turns=3)
+
+        assert record.num_turns == 3
+        assert record.max_turns_exhausted is False
+        assert len(record.commands) == 1
+
+    async def test_repeated_compaction_resets_before_normal_tool_calls(self):
+        notifications = self._cmd_notifications(2)
+        notifications[10:10] = self._compaction_notifications(item_id="compact-2", empty_response=True)
+        notifications[5:5] = self._compaction_notifications(item_id="compact-1")
+        agent = _started_agent(parse_agent_config(type=AgentKind.CODEX), notifications)
+
+        record = await agent.communicate("go", max_turns=None)
+
+        assert record.num_turns == 5
+        assert record.max_turns_exhausted is False
+        assert len(record.commands) == 2
+
+    async def test_zero_usage_outside_compaction_still_counts(self):
+        update = _token_usage(inp=10, out=5, cached=0)
+        update.payload.token_usage.last = TokenUsageBreakdown(
+            input_tokens=0,
+            output_tokens=0,
+            cached_input_tokens=0,
+            cache_write_input_tokens=0,
+            reasoning_output_tokens=0,
+            total_tokens=20,
+        )
+        notifications = [_token_usage(inp=10, out=5, cached=0), update, _turn_completed()]
+        agent = _started_agent(parse_agent_config(type=AgentKind.CODEX), notifications)
+
+        record = await agent.communicate("go", max_turns=None)
+
+        assert record.num_turns == 2
+        assert record.max_turns_exhausted is False
+
+    async def test_final_compaction_estimate_updates_usage_and_message_buckets(self):
+        reply = SimpleNamespace(type="agentMessage", id="preface", text="Preparing a shorter context.")
+        notifications = [
+            _item_notification("item/started", reply),
+            _item_notification("item/completed", reply),
+            *self._compaction_notifications(),
+            _turn_completed(),
+        ]
+        agent = _started_agent(parse_agent_config(type=AgentKind.CODEX), notifications)
+
+        record = await agent.communicate("go", max_turns=None)
+
+        assert record.token_usage is not None
+        assert record.token_usage.uncached_input_tokens == 20
+        assert record.token_usage.output_tokens == 5
+        buckets = [
+            message for message in record.messages if isinstance(message, (AssistantMessage, ReconciliationMessage))
+        ]
+        assert sum(message.input_tokens for message in buckets) == record.token_usage.uncached_input_tokens
+        assert sum(message.output_tokens for message in buckets) == record.token_usage.output_tokens
+        assert sum(message.cache_read_tokens for message in buckets) == record.token_usage.cache_read_input_tokens
+        assert (
+            sum(message.cache_creation_tokens for message in buckets) == record.token_usage.cache_creation_input_tokens
+        )
+
+    async def test_zero_usage_after_compaction_completion_counts_next_model_call(self):
+        update = _token_usage(inp=0, out=0, cached=0)
+        update.payload.token_usage.last = SimpleNamespace(
+            input_tokens=0,
+            output_tokens=0,
+            cached_input_tokens=0,
+            cache_write_input_tokens=0,
+            reasoning_output_tokens=0,
+            total_tokens=0,
+        )
+        notifications = self._compaction_notifications()
+        notifications.extend([update, _turn_completed()])
+        agent = _started_agent(parse_agent_config(type=AgentKind.CODEX), notifications)
+
+        record = await agent.communicate("go", max_turns=2)
+
+        assert record.num_turns == 2
+        assert record.max_turns_exhausted is False
+
     async def test_cap_stops_before_the_next_call_runs_a_tool(self):
         agent = _started_agent(parse_agent_config(type=AgentKind.CODEX), self._cmd_notifications(5))