Add dedicated client for each subagent.

Author: beacoder

python_agent_harness/agent.py

@@ -66,10 +66,15 @@ def __init__(
         top_level: bool = True,
         system: str | None = None,
         max_rounds: int = 60,
+        client: Any | None = None,
     ) -> None:
         self.session = session
         self.messages: list[Message] = messages if messages is not None else []
         self.top_level = top_level
+        # an explicit per-run client (a dedicated clone for this
+        # sub-agent invocation, see AgentSession.run_subagent) wins
+        # over the session's shared sub-agent client
+        self._client = client
         # fall back to the session's prompt so a run never loses it;
         # sub-agent loops use the session's SUB-AGENT prompt (their own),
         # never the parent's system prompt (which carries the parent's
@@ -619,18 +624,19 @@ def safe_delta(text: str) -> None:
                 exclude=config.SUBAGENT_EXCLUDED_TOOLS if not self.top_level else ()
             )
             # sub-agent runs use their own LLM when one is configured
-            # (session.subagent_client, mirroring gptel-agent-harness-
-            # subagent-model/-backend); everything unset inherits the
-            # main agent's settings, so the sub-agent path is identical
-            # when no separate LLM is configured
+            # (a per-invocation clone of session.subagent_client,
+            # mirroring gptel-agent-harness-subagent-model/-backend);
+            # everything unset inherits the main agent's settings, so
+            # the sub-agent path is identical when no separate LLM is
+            # configured
             if self.top_level:
                 client = session.client
                 temperature = session.temperature
                 max_tokens = session.max_tokens
                 reasoning_effort = session.reasoning_effort
                 stream = session.stream
             else:
-                client = session.subagent_client
+                client = self._client or session.subagent_client
                 temperature = session.subagent_temperature
                 max_tokens = session.subagent_max_tokens
                 reasoning_effort = session.subagent_reasoning_effort
@@ -795,14 +801,19 @@ def run_agent_loop(
     top_level: bool = True,
     system: str | None = None,
     max_rounds: int = 60,
+    client: Any | None = None,
 ) -> str | None:
-    """Convenience wrapper running a full agent run (FSM)."""
+    """Convenience wrapper running a full agent run (FSM).
+
+    ``client`` (when given) overrides the session's client for this
+    run — the per-invocation sub-agent clone."""
     return AgentLoop(
         session,
         messages=messages,
         top_level=top_level,
         system=system,
         max_rounds=max_rounds,
+        client=client,
     ).run()
 
 

python_agent_harness/agent_session.py

@@ -104,8 +104,10 @@ def __init__(
         # timeout) and per-request options when a different LLM is
         # configured for sub-agents (mirrors gptel-agent-harness-
         # subagent-model/-backend); every unset option inherits the
-        # main agent's value.  The Agent tool's sub-agent loop uses
-        # these instead of the main client.
+        # main agent's value.  The sub-agent loop never uses this
+        # client directly — each Agent tool invocation clones it
+        # (see run_subagent) so concurrent sub-agents never share a
+        # Client's pool/abort state.
         self.subagent_client = subagent_client or client
         self.subagent_temperature = (
             temperature if subagent_temperature is None else subagent_temperature
@@ -132,6 +134,13 @@ def __init__(
         # serializes interactive prompts (Question tool, PlanExit
         # confirmation): the TUI can only ask one question at a time
         self._interactive_lock = threading.Lock()
+        # dedicated per-invocation sub-agent clients (see run_subagent):
+        # concurrent sub-agents each run on their own Client clone, so
+        # one sub-agent's connection failure / abort can never tear
+        # down a sibling's in-flight request on a shared client.  The
+        # active clones are tracked so cancel()/close() can reach them.
+        self._subagent_clients_lock = threading.Lock()
+        self._active_subagent_clients: list[Client] = []
         self.store = SessionStore(
             project_dir=project_dir,
             model=model,
@@ -310,8 +319,42 @@ def run_subagent(self, subagent_type: str, description: str, prompt: str) -> str
 
         The sub-agent has no TodoWrite (parent-only), so it can never
         touch the parent's todo list.
+
+        Each invocation runs on a DEDICATED client, cloned from the
+        configured sub-agent client: concurrent Agent tool calls share
+        this session, and a shared Client would race — ``_reset_http``
+        / ``abort`` swap and close the underlying httpx pool and
+        ``_aborted`` is per-request state, so one sub-agent's
+        connection failure (or a Ctrl-C) would tear down a sibling's
+        in-flight request.  The clone is tracked for cancel/close and
+        released when the sub-agent finishes.
         """
-        return run_subagent(self, description, prompt)
+        client, owned = self._new_subagent_client()
+        if owned:
+            with self._subagent_clients_lock:
+                self._active_subagent_clients.append(client)
+        try:
+            return run_subagent(self, description, prompt, client=client)
+        finally:
+            if owned:
+                with self._subagent_clients_lock:
+                    if client in self._active_subagent_clients:
+                        self._active_subagent_clients.remove(client)
+                client.close()
+
+    def _new_subagent_client(self) -> tuple[Any, bool]:
+        """A dedicated Client for one sub-agent invocation.
+
+        Real Clients are cloned (fresh httpx pool, own ``_aborted``
+        flag, same endpoint/credentials/log).  A non-Client
+        ``subagent_client`` (a test double) is passed through
+        untouched — the isolation concern does not apply to it, and
+        custom clients keep working as-is.
+        """
+        base = self.subagent_client
+        if isinstance(base, Client):
+            return base.clone(), True
+        return base, False
 
     def plan_exit(self) -> str:
         """PlanExit tool implementation.
@@ -460,6 +503,16 @@ def close(self) -> None:
             self.client.close()
         if self.subagent_client is not self.client and hasattr(self.subagent_client, "close"):
             self.subagent_client.close()
+        # defensive: sub-agent workers close their own clones in
+        # run_subagent's finally; close any stragglers (e.g. a worker
+        # still winding down after cancel) so no pool leaks
+        with self._subagent_clients_lock:
+            strays = list(self._active_subagent_clients)
+            self._active_subagent_clients.clear()
+        for c in strays:
+            if hasattr(c, "close"):
+                with contextlib.suppress(Exception):  # best effort
+                    c.close()
 
     def cancel(self) -> None:
         """Cancel the in-flight agent run (Ctrl-C).
@@ -478,10 +531,14 @@ def cancel(self) -> None:
         # A sub-agent streams on its own client when a separate LLM is
         # configured — abort BOTH pools so a blocked sub-agent read is
         # interrupted too (see Client.abort for why close() alone is
-        # not enough).  A shared client is aborted once.
+        # not enough).  A shared client is aborted once; dedicated
+        # per-invocation sub-agent clones (see run_subagent) are each
+        # aborted so every in-flight sub-agent request is interrupted.
         clients = [self.client]
         if self.subagent_client is not self.client:
             clients.append(self.subagent_client)
+        with self._subagent_clients_lock:
+            clients.extend(self._active_subagent_clients)
         for c in clients:
             if hasattr(c, "abort"):
                 with contextlib.suppress(Exception):  # best effort

python_agent_harness/client.py

@@ -11,6 +11,7 @@
 import json
 import os
 import random
+import threading
 import time
 from collections.abc import Callable, Iterator
 from pathlib import Path
@@ -21,6 +22,12 @@
 from . import config
 from .models import Message, ToolCall, ToolSpec, Usage
 
+# serializes appends to the shared LLM log file: concurrent sub-agents
+# (each with its own client but ONE shared log_path, see Client.clone)
+# finish their interactions in parallel, and interleaved write() calls
+# would corrupt the JSON stream
+_log_lock = threading.Lock()
+
 
 class ApiError(Exception):
     """Raised when the API call itself fails (network/HTTP)."""
@@ -134,7 +141,7 @@ def _log_llm_interaction(
             "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
         }
 
-        with open(log_file, "a", encoding="utf-8") as f:
+        with _log_lock, open(log_file, "a", encoding="utf-8") as f:
             f.write(json.dumps(marker, indent=2, ensure_ascii=False) + "\n")
             f.write(json.dumps(body, indent=2, ensure_ascii=False) + "\n")
     except Exception:  # noqa: BLE001 - logging must never break the agent
@@ -174,6 +181,7 @@ def __init__(
         retry_base_delay: float | None = None,
         retry_max_delay: float | None = None,
         config_path: str | None = None,
+        log_path: Path | None = None,
     ) -> None:
         self.base_url = (base_url or config.DEFAULT_BASE_URL).rstrip("/")
         self.api_key = api_key or _default_api_key()
@@ -194,11 +202,43 @@ def __init__(
         # the user asked to stop.  Cleared at the start of each chat()
         # so a fresh turn may retry normally.
         self._aborted = False
-        self.log_path: Path | None = _llm_log_path() if config.LLM_LOG_ENABLED else None
+        # an explicit log file is inherited by clones so every request
+        # of one session (main + all sub-agents) lands in a single log
+        self.log_path = (
+            log_path
+            if log_path is not None
+            else (_llm_log_path() if config.LLM_LOG_ENABLED else None)
+        )
 
     def close(self) -> None:
         self._http.close()
 
+    def clone(self) -> Client:
+        """A fresh Client with identical settings (no shared state).
+
+        Concurrent requests must never share one Client: ``_reset_http``
+        and ``abort`` swap and close the underlying httpx pool, and
+        ``_aborted`` is per-request flag state — so one request's
+        connection failure (or Ctrl-C abort) would tear down a
+        sibling's in-flight request on the same client.  Each
+        concurrent sub-agent clones its own client (see
+        ``AgentSession.run_subagent``), keeping pools and the abort
+        flag strictly per-request.  The log file is shared so one
+        session's LLM interactions stay in one log.
+        """
+        return Client(
+            base_url=self.base_url,
+            api_key=self.api_key,
+            model=self.model,
+            timeout=self.timeout,
+            verify=self.verify,
+            retry_max=self.retry_max,
+            retry_base_delay=self.retry_base_delay,
+            retry_max_delay=self.retry_max_delay,
+            config_path=self._config_path,
+            log_path=self.log_path,
+        )
+
     def abort(self) -> None:
         """Abort the in-flight request (called on cancel).
 

python_agent_harness/subagent.py

@@ -31,8 +31,14 @@ def run_subagent(
     parent_session: object,
     description: str,
     prompt: str,
+    client: object | None = None,
 ) -> str:
-    """Run a sub-agent task; return a result string (never raises)."""
+    """Run a sub-agent task; return a result string (never raises).
+
+    ``client`` (when given) is the per-invocation dedicated client
+    (see ``AgentSession.run_subagent``); the loop falls back to the
+    session's shared sub-agent client otherwise.
+    """
     session = parent_session
     try:
         messages = [Message(role="user", content=prompt)]
@@ -46,6 +52,7 @@ def run_subagent(
             top_level=False,
             system=_subagent_system_prompt(session),
             max_rounds=config.SUBAGENT_MAX_ROUNDS,
+            client=client,
         )
         if isinstance(result, str):
             return result