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

Author: declan-scale

examples/tutorials/10_async/10_temporal/020_state_machine/project/workflows/deep_research/performing_deep_research.py

@@ -13,21 +13,28 @@
 
 logger = make_logger(__name__)
 
+# These reference MCP servers still import the mcp 1.x API (``McpError``), which
+# mcp 2.0.0 renamed to ``MCPError``. uvx gives each server its own isolated env and
+# resolves ``mcp`` unpinned there, ignoring the version this project pins, so without
+# this constraint every server dies at import and the agent silently makes zero tool
+# calls. Drop the pin once the servers support mcp 2.x.
+_MCP_PIN = ["--with", "mcp<2"]
+
 MCP_SERVERS = [
     StdioServerParameters(
         command="uvx",
-        args=["mcp-server-time", "--local-timezone", "America/Los_Angeles"],
+        args=[*_MCP_PIN, "mcp-server-time", "--local-timezone", "America/Los_Angeles"],
     ),
     StdioServerParameters(
         command="uvx",
-        args=["openai-websearch-mcp"],
+        args=[*_MCP_PIN, "openai-websearch-mcp"],
         env={
             "OPENAI_API_KEY": os.environ.get("OPENAI_API_KEY", "")
         }
     ),
     StdioServerParameters(
         command="uvx",
-        args=["mcp-server-fetch"],
+        args=[*_MCP_PIN, "mcp-server-fetch"],
     ),
 ]
 

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:

src/agentex/lib/core/tracing/obs_ids.py

@@ -22,7 +22,7 @@
 from __future__ import annotations
 
 import os
-from typing import Dict, Optional, Tuple
+from typing import Dict, Tuple, Optional
 
 __all__ = ("get_obs_mode", "obs_correlation")
 
@@ -55,7 +55,7 @@ def _lgtm_ids() -> Optional[Tuple[str, str]]:
 
 def _ddtrace_ids() -> Optional[Tuple[str, str]]:
     try:
-        from ddtrace import tracer
+        from ddtrace.trace import tracer
     except ImportError:
         return None
     ctx = tracer.current_trace_context()

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
 # ---------------------------------------------------------------------------