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"], ), ] 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/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() 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 # ---------------------------------------------------------------------------