Enable parallel subagent execution.

Author: beacoder

README.md

@@ -15,7 +15,8 @@ A Python port of the Emacs [gptel-agent-harness](https://github.com/beacoder/gpt
 - **Tools** — Agent (sub-agents), TodoWrite, Glob (git-aware), Grep
   (git grep → rg → grep), Read, Insert, Edit (incl. unified diffs), Write,
   Mkdir, Bash, Skill, Question, and PlanExit (registered while in plan
-  mode) — all OpenAI-compatible tool schemas.
+  mode) — all OpenAI-compatible tool schemas.  Multiple Agent calls in
+  one round run concurrently (up to `PARALLEL_SUBAGENT_MAX`, default 4).
 - **Default agent prompts** — the main agent and sub-agents each get a
   distinct default system prompt bundled with the package
   (`prompts/agent.txt`, `prompts/subagent.txt`), with YAML frontmatter

python_agent_harness/agent.py

@@ -202,11 +202,65 @@ def _execute_tool_call(self, call: ToolCall) -> str:
             args = {}
         return self.session.execute_tool(call.name, args, call_id=call.id)
 
+    def _deliver_tool_result(self, p: ToolCall, result: str) -> None:
+        """Append one tool result message for call P (parent thread only)."""
+        p.result = result
+        if hasattr(self.session, "take_diff"):
+            p.diff = self.session.take_diff(p.id)
+        self.messages.append(
+            Message(
+                role="tool",
+                content=result,
+                tool_call_id=p.id,
+                name=p.name,
+            )
+        )
+        if self.top_level and not self._is_cancelled():
+            # only the top-level loop mirrors its messages onto the
+            # shared session: a sub-agent runs inside the parent's
+            # tool round and must never clobber the parent's
+            # conversation history (the TUI renders from it)
+            self.session.last_messages = list(self.messages)
+
+    def _run_subagents_parallel(
+        self, calls: list[ToolCall], results: dict[str, str]
+    ) -> None:
+        """Run several Agent calls concurrently, filling RESULTS.
+
+        Each sub-agent is fully isolated (its own agent loop, message
+        list, client stream, and thread-local diff slot), so Agent calls
+        issued in the same round execute in parallel.  Delivery happens
+        later, in original tool-call order, by the parent thread.
+        """
+        from concurrent.futures import ThreadPoolExecutor, as_completed
+
+        with ThreadPoolExecutor(
+            max_workers=min(len(calls), config.PARALLEL_SUBAGENT_MAX),
+            thread_name_prefix="subagent",
+        ) as pool:
+            futures = {pool.submit(self._execute_tool_call, p): p for p in calls}
+            for fut in as_completed(futures):
+                p = futures[fut]
+                try:
+                    results[p.id] = sanitize_tool_result(fut.result())
+                except Exception as e:  # noqa: BLE001 - containment boundary
+                    results[p.id] = (
+                        f"Error: Task {p.name!r} crashed in a sub-agent "
+                        f"thread — {e}"
+                    )
+
     def _run_tool_round(self) -> None:
         """Execute all pending tool calls; deliver results as messages.
 
         The assistant message carrying the tool calls was already
         appended by the main loop; here we add the per-call results.
+
+        Agent calls (sub-agent spawns) run CONCURRENTLY in a thread
+        pool — sub-agents are isolated by design, so several Agent calls
+        in one round execute in parallel.  All other tools stay
+        sequential: they are fast and mutate shared session state
+        (undo, diff slots).  Results are delivered in the original
+        tool-call order regardless of execution order.
         """
         pending = list(self.pending)
         if not pending:
@@ -217,30 +271,27 @@ def _run_tool_round(self) -> None:
             # worker must never mirror its partial history over the next
             # run's `session.last_messages`.
             return
+        agent_calls = [p for p in pending if p.name == "Agent"]
+        results: dict[str, str] = {}
         for p in pending:
+            if p.name == "Agent":
+                continue
             if self._is_cancelled():
                 # cancelled mid-round: stop running further tools; the
                 # results already delivered stay local to this (dead) run
                 self.pending = []
                 return
-            result = sanitize_tool_result(self._execute_tool_call(p))
-            p.result = result
-            if hasattr(self.session, "take_diff"):
-                p.diff = self.session.take_diff(p.id)
-            self.messages.append(
-                Message(
-                    role="tool",
-                    content=result,
-                    tool_call_id=p.id,
-                    name=p.name,
-                )
-            )
-            if self.top_level and not self._is_cancelled():
-                # only the top-level loop mirrors its messages onto the
-                # shared session: a sub-agent runs inside the parent's
-                # tool round and must never clobber the parent's
-                # conversation history (the TUI renders from it)
-                self.session.last_messages = list(self.messages)
+            results[p.id] = sanitize_tool_result(self._execute_tool_call(p))
+        if agent_calls:
+            self._run_subagents_parallel(agent_calls, results)
+            if self._is_cancelled():
+                self.pending = []
+                return
+        for p in pending:
+            if self._is_cancelled():
+                self.pending = []
+                return
+            self._deliver_tool_result(p, results[p.id])
         self.pending = []
         self.session.notify("tools")
 

python_agent_harness/agent_session.py

@@ -101,7 +101,15 @@ def __init__(
         self.bash_policy = BashPolicy()
         self.tool_ctx = ToolContext(self)
         self._tool_diffs: dict[str, str] = {}
-        self._active_call_id: str | None = None
+        # thread-local: parallel sub-agents each execute tools in their
+        # own pool thread; the "currently executing call" that Edit/Write
+        # attach their diff to must be per-thread, or concurrent
+        # sub-agents would clobber each other's diff slot
+        self._active_call = threading.local()
+        # serializes the interactive Bash approval prompt: parallel
+        # sub-agents may hit CONFIRM simultaneously, but the TUI can only
+        # ask one question at a time
+        self._bash_lock = threading.Lock()
         self.store = SessionStore(
             project_dir=project_dir,
             model=model,
@@ -195,19 +203,20 @@ def execute_tool(
                 except SafetyViolation as e:
                     return str(e)
 
-        self._active_call_id = call_id
+        self._active_call.call_id = call_id
         try:
             result = self.registry.execute(name, args, self.tool_ctx)
         finally:
-            self._active_call_id = None
+            self._active_call.call_id = None
 
         self.notify("tool")
         return result
 
     def record_diff(self, diff_text: str) -> None:
         """Attach a unified diff to the tool call currently executing."""
-        if self._active_call_id and diff_text:
-            self._tool_diffs[self._active_call_id] = diff_text
+        call_id = getattr(self._active_call, "call_id", None)
+        if call_id and diff_text:
+            self._tool_diffs[call_id] = diff_text
 
     def take_diff(self, call_id: str) -> str | None:
         """Pop and return the diff recorded for CALL_ID, if any."""
@@ -242,24 +251,31 @@ def guard_path(self, path: str, tool_name: str) -> None:
         check_path(path, tool_name)
 
     def verify_bash(self, command: str) -> str | None:
-        """Return an error string to deliver, or None to run."""
-        self.bash_policy.plan_mode = self.plan_mode.is_plan
-        verdict = self.bash_policy.verdict(command)
-        if verdict != "CONFIRM":
-            return verdict
-        if self.bash_approval_fn:
-            run, answer = self.bash_approval_fn(command)
-        else:
-            run, answer = self._ask_via_tui(command)
-        if answer == "allow":
-            self.bash_policy.record(command, "allow")
-            return None
-        if answer == "deny":
-            self.bash_policy.record(command, "deny")
-            return "Error: Bash command rejected by user approval (denied for this session)."
-        if run:
-            return None
-        return "Error: Bash command rejected by user approval."
+        """Return an error string to deliver, or None to run.
+
+        The interactive approval prompt is serialized: parallel
+        sub-agents may reach CONFIRM simultaneously, but the user can
+        only answer one question at a time.  Command *execution* stays
+        parallel — the lock is released before the process starts.
+        """
+        with self._bash_lock:
+            self.bash_policy.plan_mode = self.plan_mode.is_plan
+            verdict = self.bash_policy.verdict(command)
+            if verdict != "CONFIRM":
+                return verdict
+            if self.bash_approval_fn:
+                run, answer = self.bash_approval_fn(command)
+            else:
+                run, answer = self._ask_via_tui(command)
+            if answer == "allow":
+                self.bash_policy.record(command, "allow")
+                return None
+            if answer == "deny":
+                self.bash_policy.record(command, "deny")
+                return "Error: Bash command rejected by user approval (denied for this session)."
+            if run:
+                return None
+            return "Error: Bash command rejected by user approval."
 
     def _ask_via_tui(self, command: str) -> tuple[bool, str]:
         prompt = (

python_agent_harness/config.py

@@ -151,6 +151,9 @@
 
 # ---- sub-agents ---------------------------------------------------------------
 SUBAGENT_MAX_ROUNDS = 60
+# Max sub-agents that may run CONCURRENTLY in one tool round (Agent calls
+# issued together are executed in parallel; excess calls queue).
+PARALLEL_SUBAGENT_MAX = 4
 # Tools a sub-agent must NOT see or call: it runs autonomously as a
 # one-shot task inside the parent's tool round, so it cannot spawn
 # further sub-agents (Agent), ask the user questions (Question), nor

python_agent_harness/tools/agent_tool.py

@@ -5,6 +5,10 @@
 parent as a single tool result string.  Errors are contained: an
 unexpected sub-agent response becomes an error string fed to the parent,
 never a crash.
+
+Multiple Agent calls issued in the same round run CONCURRENTLY — each
+sub-agent is fully isolated (own loop, own history, own stream), so
+independent tasks can be delegated in parallel.
 """
 
 from __future__ import annotations
@@ -15,7 +19,9 @@
     "Launch a specialized sub-agent to handle complex, multi-step tasks "
     "autonomously. Sub-agents run independently and return results in one "
     "message. Use for open-ended searches, complex research, or when "
-    "uncertain about finding results in the first few tries."
+    "uncertain about finding results in the first few tries.\n\n"
+    "Multiple Agent calls issued in the same round run concurrently, so "
+    "delegate independent tasks in parallel for efficiency."
 )
 
 PARAMETERS = {

python_agent_harness/undo.py

@@ -12,6 +12,7 @@
 import random
 import shutil
 import string
+import threading
 import time
 from dataclasses import dataclass
 
@@ -34,9 +35,16 @@ def __init__(self, backup_dir: str | None = None, depth: int = config.UNDO_DEPTH
             temp_dir(), "python-agent-harness-undo"
         )
         self.entries: list[UndoEntry] = []
+        # parallel sub-agents (and the top-level round) snapshot files
+        # concurrently: the entries list must be serialized
+        self._lock = threading.Lock()
 
     def snapshot(self, path: str, tool: str) -> None:
         """Snapshot PATH before a write; records absent files separately."""
+        with self._lock:
+            self._snapshot_locked(path, tool)
+
+    def _snapshot_locked(self, path: str, tool: str) -> None:
         path = os.path.abspath(path)
         if os.path.isfile(path):
             os.makedirs(self.backup_dir, exist_ok=True)
@@ -62,13 +70,18 @@ def snapshot(self, path: str, tool: str) -> None:
 
     def record_absent(self, path: str, tool: str) -> None:
         """Record a file that did not exist before a Write."""
-        path = os.path.abspath(path)
-        if os.path.exists(path) or any(e.path == path for e in self.entries):
-            return
-        self.entries.append(UndoEntry(path, None, False, tool, time.time()))
+        with self._lock:
+            path = os.path.abspath(path)
+            if os.path.exists(path) or any(e.path == path for e in self.entries):
+                return
+            self.entries.append(UndoEntry(path, None, False, tool, time.time()))
 
     def undo_last(self) -> tuple[bool, str]:
         """Restore the newest entry. Returns (ok, message)."""
+        with self._lock:
+            return self._undo_last_locked()
+
+    def _undo_last_locked(self) -> tuple[bool, str]:
         if not self.entries:
             return False, "Nothing to undo."
         entry = self.entries[-1]
@@ -95,11 +108,12 @@ def undo_last(self) -> tuple[bool, str]:
             return False, f"Error: remove failed — {e}"
 
     def history(self) -> list[str]:
-        out = []
-        for e in reversed(self.entries):
-            stamp = time.strftime("%H:%M:%S", time.localtime(e.time))
-            out.append(f"{stamp} {e.tool} {e.path}")
-        return out
+        with self._lock:
+            out = []
+            for e in reversed(self.entries):
+                stamp = time.strftime("%H:%M:%S", time.localtime(e.time))
+                out.append(f"{stamp} {e.tool} {e.path}")
+            return out
 
 
 def temp_dir() -> str: