From da3b5754f45ac464bf2a43eefde81db974fe300a Mon Sep 17 00:00:00 2001 From: Declan Brady Date: Mon, 27 Jul 2026 22:47:35 -0500 Subject: [PATCH 1/3] feat(codex): republish todo_list revisions as they are ticked off MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) --- src/agentex/lib/adk/_modules/_codex_sync.py | 47 ++++++++- tests/lib/adk/test_codex_sync.py | 108 ++++++++++++++++++++ 2 files changed, 151 insertions(+), 4 deletions(-) diff --git a/src/agentex/lib/adk/_modules/_codex_sync.py b/src/agentex/lib/adk/_modules/_codex_sync.py index 951622b13..b71ba9dac 100644 --- a/src/agentex/lib/adk/_modules/_codex_sync.py +++ b/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: diff --git a/tests/lib/adk/test_codex_sync.py b/tests/lib/adk/test_codex_sync.py index 644688dfb..85e67beb9 100644 --- a/tests/lib/adk/test_codex_sync.py +++ b/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 # --------------------------------------------------------------------------- From acbb9d4be0e2a24f80bf5b71e76c0ba4af143cd5 Mon Sep 17 00:00:00 2001 From: Declan Brady Date: Tue, 28 Jul 2026 09:33:19 -0500 Subject: [PATCH 2/3] fix(tracing): satisfy ruff and pyright on obs_ids imports CI lint has been red on `next` since c67e1774 added this file: - ruff I001: typing import block not sorted per the repo's isort config - pyright reportPrivateImportUsage: `tracer` is not re-exported from `ddtrace`, so it must be imported from `ddtrace.trace` The `except ImportError` guard still covers ddtrace being absent, and `ddtrace.trace.tracer` exists in the pinned ddtrace 4.10.1. Co-Authored-By: Claude Opus 5 (1M context) --- src/agentex/lib/core/tracing/obs_ids.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/agentex/lib/core/tracing/obs_ids.py b/src/agentex/lib/core/tracing/obs_ids.py index 1be803a84..99c6b2555 100644 --- a/src/agentex/lib/core/tracing/obs_ids.py +++ b/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() From 023e1719588404106c444a2dd7e2af87c7534df7 Mon Sep 17 00:00:00 2001 From: Declan Brady Date: Tue, 28 Jul 2026 10:52:19 -0500 Subject: [PATCH 3/3] fix(tutorials): pin mcp<2 for uvx-spawned MCP servers in 020_state_machine mcp 2.0.0 renamed `McpError` to `MCPError` and split types out into a separate `mcp_types` package. `mcp-server-time` and `mcp-server-fetch` still import the 1.x name, so both now die at import: ImportError: cannot import name 'McpError' from 'mcp.shared.exceptions' ... -> mcp.shared.exceptions.McpError: Connection closed uvx builds each server its own isolated environment and resolves `mcp` unpinned there, so the `mcp==1.23.0` this project installs does not apply. With every server dead the deep research agent emits "Starting deep research..." and then makes zero tool calls, so test_agent.py's `uses_tool_requests` assertion fails (and the streaming test times out waiting for the same content). This is why the job passed at 03:48 today and failed at 14:08 with no code change in between: mcp 2.0.0 landed in the gap. Constraining uvx to mcp<2 restores all three servers. Verified by completing an MCP initialize + tools/list handshake against the exact argv this file now produces: mcp-server-time tools=['get_current_time', 'convert_time'] openai-websearch-mcp tools=['openai_web_search'] mcp-server-fetch tools=['fetch'] The 30s test timeout is left alone; both tests passed in 21.1s total when the servers worked. Co-Authored-By: Claude Opus 5 (1M context) --- .../deep_research/performing_deep_research.py | 13 ++++++++++--- 1 file changed, 10 insertions(+), 3 deletions(-) diff --git a/examples/tutorials/10_async/10_temporal/020_state_machine/project/workflows/deep_research/performing_deep_research.py b/examples/tutorials/10_async/10_temporal/020_state_machine/project/workflows/deep_research/performing_deep_research.py index 954a7566b..7bcb5a6d5 100644 --- a/examples/tutorials/10_async/10_temporal/020_state_machine/project/workflows/deep_research/performing_deep_research.py +++ b/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"], ), ]