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

@@ -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(

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.
 

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