feat(codex): republish todo_list revisions as they are ticked off

Codex does not reopen its plan: one `todo_list` item is opened at the
start of a turn and revised in place through `item.updated`, with
`item.completed` only arriving at the very end. The tap dropped those
updates, so a consumer saw the plan exactly twice — as first written
(nothing done) and then, at end of turn, fully finished.

Forward each revision as a `ToolResponseContent` under the item's
existing `tool_call_id`. Consumers that key responses by call id (the
GenAI Ops Hub chat does) now render the checklist filling in as the turn
runs, and the last response received is still the final state.

Scoped to `todo_list` via `_PROGRESSIVE_TOOL_ITEMS`: the other tool item
types carry no intermediate state worth acting on, and republishing
`command_execution` output on every chunk would be pure noise. Updates
for an item that was never opened are ignored — there is no request for
them to answer.

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

Author: declan-scale

src/agentex/lib/adk/_modules/_codex_sync.py

@@ -60,7 +60,13 @@
                              a synthetic ToolRequestContent Full is emitted before the response.
   mcp_tool_call           -> same as command_execution
   web_search              -> same as command_execution
-  todo_list               -> same as command_execution
+  todo_list               -> same as command_execution, plus:
+                               item.updated -> StreamTaskMessageFull(ToolResponseContent)
+                             Codex ticks one in-place todo_list item through
+                             item.updated; each revision is republished under
+                             the same tool_call_id so consumers can render the
+                             checklist filling in rather than jumping to its
+                             final state at end of turn.
   collab_tool_call        -> same as command_execution
   error (item type)       -> StreamTaskMessageFull(TextContent, error text) on completed only
 
@@ -75,9 +81,10 @@
                           without this module needing to know about spans.
   item.updated (reasoning): the intermediate cumulative text is discarded;
                             only item.completed carries the final text.
-  item.updated (tool):    tool item types other than agent_message do not
-                          emit updates; item.started opens the request and
-                          item.completed closes it.
+  item.updated (tool):    only todo_list is republished (see above). For the
+                          other tool item types item.started opens the request
+                          and item.completed closes it; any updates in between
+                          carry no state a consumer could act on.
 """
 
 from __future__ import annotations
@@ -112,6 +119,11 @@ def _truncate(text: str, max_len: int = _MAX_RESULT_LENGTH) -> str:
     return str(text)[:max_len]
 
 
+# Tool items codex revises in place rather than reopening. Their item.updated
+# events carry real intermediate state, so they are forwarded as responses.
+_PROGRESSIVE_TOOL_ITEMS = frozenset({"todo_list"})
+
+
 def _tool_name_for(item_type: str, payload: dict[str, Any]) -> str:
     """Derive a canonical tool name from a codex item type."""
     if item_type == "command_execution":
@@ -467,6 +479,33 @@ def _handle_item(self, evt_type: str, item: dict[str, Any]) -> list[StreamTaskMe
                 )
                 out.append(StreamTaskMessageDone(type="done", index=req_idx))
 
+            elif evt_type == "item.updated" and item_type in _PROGRESSIVE_TOOL_ITEMS:
+                # Codex revises its plan in place: one todo_list item is opened
+                # at the start of the turn and ticked off through item.updated,
+                # with item.completed only arriving at the very end. Forwarding
+                # each revision as a response for the SAME tool_call_id lets a
+                # consumer show the checklist filling in as the turn runs; the
+                # last response received is the current state.
+                if item_id in self._tool_open:
+                    actual_type = self._tool_item_types.get(item_id, item_type)
+                    result_text, is_error = _tool_output_for(actual_type, item)
+                    resp_content: dict[str, Any] = {"result": result_text}
+                    if is_error:
+                        resp_content["is_error"] = True
+                    out.append(
+                        StreamTaskMessageFull(
+                            type="full",
+                            index=self._alloc(),
+                            content=ToolResponseContent(
+                                type="tool_response",
+                                author="agent",
+                                tool_call_id=tool_call_id,
+                                name=_tool_name_for(actual_type, item),
+                                content=resp_content,
+                            ),
+                        )
+                    )
+
             elif evt_type == "item.completed":
                 # file_change items may only emit item.completed (no started).
                 if item_id not in self._tool_open:

tests/lib/adk/test_codex_sync.py

@@ -47,6 +47,13 @@ async def _collect(stream: AsyncIterator[Any]) -> list[Any]:
     return [e async for e in stream]
 
 
+def _result_text(event: StreamTaskMessageFull) -> str:
+    content = event.content
+    assert isinstance(content, ToolResponseContent)
+    assert isinstance(content.content, dict)
+    return str(content.content["result"])
+
+
 # ---------------------------------------------------------------------------
 # Helpers
 # ---------------------------------------------------------------------------
@@ -360,6 +367,107 @@ async def test_tool_indices_request_before_response(self) -> None:
         assert req.index < resp.index
 
 
+# ---------------------------------------------------------------------------
+# Progressive tool items (todo_list)
+# ---------------------------------------------------------------------------
+
+
+def _todo_item(item_id: str, *completed: bool) -> dict[str, Any]:
+    return {
+        "id": item_id,
+        "type": "todo_list",
+        "items": [{"text": f"step {i + 1}", "completed": done} for i, done in enumerate(completed)],
+    }
+
+
+class TestTodoListUpdates:
+    async def test_each_update_republishes_the_checklist(self) -> None:
+        """Codex ticks ONE in-place todo_list item, so every item.updated is
+        forwarded as a response under the same tool_call_id. Without this a
+        consumer only sees the plan as first written and then, at end of turn,
+        as finished."""
+        events = [
+            {"type": "item.started", "item": _todo_item("item_1", False, False)},
+            {"type": "item.updated", "item": _todo_item("item_1", True, False)},
+            {"type": "item.updated", "item": _todo_item("item_1", True, True)},
+            {"type": "item.completed", "item": _todo_item("item_1", True, True)},
+        ]
+        out = await _collect(convert_codex_to_agentex_events(_aiter(events)))
+
+        requests = [
+            e for e in out if isinstance(e, StreamTaskMessageStart) and isinstance(e.content, ToolRequestContent)
+        ]
+        responses = [
+            e for e in out if isinstance(e, StreamTaskMessageFull) and isinstance(e.content, ToolResponseContent)
+        ]
+
+        assert len(requests) == 1
+        assert len(responses) == 3
+
+        request_content = requests[0].content
+        assert isinstance(request_content, ToolRequestContent)
+        for response in responses:
+            content = response.content
+            assert isinstance(content, ToolResponseContent)
+            assert content.name == "todo_list"
+            assert content.tool_call_id == request_content.tool_call_id
+
+        first, second, final = (json.loads(_result_text(r)) for r in responses)
+        assert [i["completed"] for i in first["items"]] == [True, False]
+        assert [i["completed"] for i in second["items"]] == [True, True]
+        assert [i["completed"] for i in final["items"]] == [True, True]
+
+    async def test_counts_the_call_once_across_its_updates(self) -> None:
+        events = [
+            {"type": "item.started", "item": _todo_item("item_1", False)},
+            {"type": "item.updated", "item": _todo_item("item_1", True)},
+            {"type": "item.completed", "item": _todo_item("item_1", True)},
+        ]
+        counters: dict[str, Any] = {}
+        await _collect(convert_codex_to_agentex_events(_aiter(events), on_result=counters.update))
+        assert counters.get("tool_call_count") == 1
+
+    async def test_ignores_an_update_for_an_unopened_item(self) -> None:
+        """An update with no preceding start has no request to answer; a
+        response would dangle with a tool_call_id nothing points at."""
+        events = [{"type": "item.updated", "item": _todo_item("orphan", True)}]
+        out = await _collect(convert_codex_to_agentex_events(_aiter(events)))
+        assert [e for e in out if isinstance(e, StreamTaskMessageFull)] == []
+
+    async def test_does_not_republish_other_tool_items(self) -> None:
+        events = [
+            {
+                "type": "item.started",
+                "item": {"id": "cmd1", "type": "command_execution", "command": "sleep 1"},
+            },
+            {
+                "type": "item.updated",
+                "item": {
+                    "id": "cmd1",
+                    "type": "command_execution",
+                    "command": "sleep 1",
+                    "aggregated_output": "partial",
+                },
+            },
+            {
+                "type": "item.completed",
+                "item": {
+                    "id": "cmd1",
+                    "type": "command_execution",
+                    "command": "sleep 1",
+                    "aggregated_output": "done",
+                    "exit_code": 0,
+                },
+            },
+        ]
+        out = await _collect(convert_codex_to_agentex_events(_aiter(events)))
+        responses = [
+            e for e in out if isinstance(e, StreamTaskMessageFull) and isinstance(e.content, ToolResponseContent)
+        ]
+        assert len(responses) == 1
+        assert _result_text(responses[0]) == "done"
+
+
 # ---------------------------------------------------------------------------
 # Reasoning
 # ---------------------------------------------------------------------------