From 8d79f556c60e1b314773ac0333a693e075dc5f74 Mon Sep 17 00:00:00 2001 From: Alessandro <155005371+3clyp50@users.noreply.github.com> Date: Thu, 9 Jul 2026 17:34:48 +0200 Subject: [PATCH] Fix parallel child log typing Resolve tool-specific log objects for all parallel child jobs, removing the code_execution_tool-only special case and using native get_log_object() when available, with generic fallback. Update parallel_tools docs to reflect the shared logging contract and add coverage for wait/ fallback behavior in parallel-tool tests. --- helpers/parallel_tools.py | 93 +++++++++++++++++++++----------- helpers/parallel_tools.py.dox.md | 2 +- tests/test_parallel_tool.py | 80 ++++++++++++++++++++++++++- 3 files changed, 143 insertions(+), 32 deletions(-) diff --git a/helpers/parallel_tools.py b/helpers/parallel_tools.py index 5b62bdc5b..a65999c20 100644 --- a/helpers/parallel_tools.py +++ b/helpers/parallel_tools.py @@ -497,24 +497,7 @@ async def execute_tool_call(agent: "Agent", tool_name: str, tool_args: dict[str, if tool_name == "parallel": raise ValueError("`parallel` cannot be nested inside a parallel worker.") - tool = None - try: - import helpers.mcp_handler as mcp_helper - - tool = mcp_helper.MCPConfig.get_instance().get_tool(agent, tool_name) - except ImportError: - tool = None - except Exception as exc: - PrintStyle.warning(f"Failed to initialize MCP tool '{tool_name}' for parallel job: {exc}") - - if not tool: - tool = agent.get_tool( - name=tool_name, - method=None, - args=tool_args, - message=json.dumps({"tool_name": tool_name, "tool_args": tool_args}), - loop_data=agent.loop_data, - ) + tool = _resolve_parallel_tool(agent, tool_name, tool_args, strict=True) if not tool: raise ValueError(f"Tool '{tool_name}' not found or could not be initialized.") @@ -544,6 +527,58 @@ async def execute_tool_call(agent: "Agent", tool_name: str, tool_args: dict[str, agent.loop_data.current_tool = None +def _resolve_parallel_tool( + agent: "Agent", + tool_name: str, + tool_args: dict[str, Any], + *, + strict: bool = False, +): + message = json.dumps({"tool_name": tool_name, "tool_args": tool_args}) + + tool = None + try: + import helpers.mcp_handler as mcp_helper + + tool = mcp_helper.MCPConfig.get_instance().get_tool(agent, tool_name) + except ImportError: + tool = None + except Exception as exc: + if strict: + raise + PrintStyle.warning(f"Failed to initialize MCP tool '{tool_name}' for parallel job: {exc}") + + if not tool: + get_tool = getattr(agent, "get_tool", None) + if not callable(get_tool): + if strict: + raise ValueError(f"Tool '{tool_name}' not found or could not be initialized.") + return None + try: + tool = get_tool( + name=tool_name, + method=None, + args=tool_args, + message=message, + loop_data=getattr(agent, "loop_data", None), + ) + except Exception as exc: + if strict: + raise + PrintStyle.warning(f"Failed to initialize tool '{tool_name}' for parallel job: {exc}") + tool = None + + if not tool: + return None + + try: + tool.args = dict(tool_args) + except Exception: + pass + + return tool + + async def _cancel_job( job: ParallelJob, *, @@ -600,18 +635,16 @@ def _log_parallel_child_started(agent: "Agent", job: ParallelJob) -> None: ) return - if job.tool_name == "code_execution_tool": - runtime = job.tool_args.get("runtime", "unknown") - session = job.tool_args.get("session", None) - session_text = f"[{session}] " if session or session == 0 else "" - job.log_item = agent.context.log.log( - type="code_exe", - heading=f"icon://terminal {session_text}code_execution_tool - {runtime}", - content="", - kvps=job.tool_args, - id=job.log_id, - ) - return + tool = _resolve_parallel_tool(agent, job.tool_name, job.tool_args) + if tool is not None: + try: + job.log_item = tool.get_log_object() + if job.log_item is not None: + return + except Exception as exc: + PrintStyle.warning( + f"Failed to derive parallel child log for {job.tool_name}: {exc}" + ) heading = f"icon://construction {agent.agent_name}: Using tool '{job.tool_name}'" job.log_item = agent.context.log.log( diff --git a/helpers/parallel_tools.py.dox.md b/helpers/parallel_tools.py.dox.md index 2aedf01ab..ddaa74201 100644 --- a/helpers/parallel_tools.py.dox.md +++ b/helpers/parallel_tools.py.dox.md @@ -31,7 +31,7 @@ - Direct tool background context cleanup removes both the in-memory context and any transient chat folder left on disk. - Parent-visible child log items are created for each wrapped call so the WebUI can inspect concurrent children separately while the wrapper result remains model-history-only. - Child tool logs mirror normal tool-call visible args; job ids remain available through wrapper results and prompt extras rather than visible process-step args. -- Wrapped `code_execution_tool` child logs use the normal `code_exe` WebUI message type and terminal heading so they render like direct code execution calls. +- Wrapped tool child logs use each tool's native `get_log_object()` output when available, preserving special log rendering (for example: `code_execution_tool` uses `code_exe`, `wait` uses `progress`, MCP tools use `mcp`, and regular tools use `tool`). - Job IDs are stable handles for later await, collect, or cancel operations. - Prompt extras must stay bounded and expose only job IDs, tool names, status, and compact result/error summaries. diff --git a/tests/test_parallel_tool.py b/tests/test_parallel_tool.py index bfb65e66e..ef758d3cf 100644 --- a/tests/test_parallel_tool.py +++ b/tests/test_parallel_tool.py @@ -384,7 +384,7 @@ async def test_parallel_subordinate_jobs_are_visible_child_logs_not_scheduler_ta @pytest.mark.asyncio -async def test_parallel_direct_tool_jobs_log_normal_tool_metadata(monkeypatch) -> None: +async def test_parallel_direct_tool_jobs_fallback_to_generic_tool_log_type(monkeypatch) -> None: class FakeDeferredTask: def __init__(self, thread_name=None) -> None: self.thread_name = thread_name @@ -404,6 +404,7 @@ async def test_parallel_direct_tool_jobs_log_normal_tool_metadata(monkeypatch) - pass monkeypatch.setattr(parallel_tools, "DeferredTask", FakeDeferredTask) + monkeypatch.setattr(parallel_tools, "_resolve_parallel_tool", lambda *_args, **_kwargs: None) agent = _FakeAgent() jobs = await parallel_tools.start_parallel_jobs( @@ -445,7 +446,28 @@ async def test_parallel_code_execution_child_uses_code_exe_log_type(monkeypatch) def kill(self): pass + class FakeCodeExecutionTool: + def __init__(self, agent, args): + self.agent = agent + self.args = args + + def get_log_object(self): + runtime = self.args.get("runtime", "unknown") + session = self.args.get("session", None) + session_text = f"[{session}] " if session or session == 0 else "" + return self.agent.context.log.log( + type="code_exe", + heading=f"icon://terminal {session_text}code_execution_tool - {runtime}", + content="", + kvps=self.args, + ) + monkeypatch.setattr(parallel_tools, "DeferredTask", FakeDeferredTask) + monkeypatch.setattr( + parallel_tools, + "_resolve_parallel_tool", + lambda _agent, _tool_name, args: FakeCodeExecutionTool(_agent, args), + ) agent = _FakeAgent() jobs = await parallel_tools.start_parallel_jobs( @@ -473,6 +495,62 @@ async def test_parallel_code_execution_child_uses_code_exe_log_type(monkeypatch) } +@pytest.mark.asyncio +async def test_parallel_wait_child_uses_wait_log_type(monkeypatch) -> None: + class FakeDeferredTask: + def __init__(self, thread_name=None) -> None: + self.thread_name = thread_name + + def start_task(self, func, *args): + return self + + def is_ready(self): + return False + + def is_alive(self): + return True + + def kill(self): + pass + + class FakeWaitTool: + def __init__(self, agent, args): + self.agent = agent + self.args = args + + def get_log_object(self): + return self.agent.context.log.log( + type="progress", + heading="icon://timer Wait: Waiting...", + content="", + kvps=self.args, + ) + + monkeypatch.setattr(parallel_tools, "DeferredTask", FakeDeferredTask) + monkeypatch.setattr( + parallel_tools, + "_resolve_parallel_tool", + lambda _agent, _tool_name, args: FakeWaitTool(_agent, args), + ) + agent = _FakeAgent() + + jobs = await parallel_tools.start_parallel_jobs( + agent, # type: ignore[arg-type] + [ + parallel_tools.NormalizedToolCall( + index=0, + tool_name="wait", + tool_args={"seconds": 1}, + ) + ], + ) + + assert jobs[0].kind == "tool" + assert agent.context.log.items[0].type == "progress" + assert agent.context.log.items[0].heading == "icon://timer Wait: Waiting..." + assert agent.context.log.items[0].kvps == {"seconds": 1} + + @pytest.mark.asyncio async def test_parallel_tool_keeps_wrapper_out_of_visible_log() -> None: from tools.parallel import ParallelTool