feat(copilot): Execute parallel tool calls concurrently (#12165)

When the LLM returns multiple tool calls in a single response (e.g.
multiple web fetches for a research task), they now execute concurrently
instead of sequentially. This can dramatically reduce latency for
multi-tool turns.

**Before:** Tool calls execute one after another — 7 web fetches × 2s
each = 14s total
**After:** All tool calls fire concurrently — 7 web fetches = ~2s total

### Changes

- **`service.py`**: New `_execute_tool_calls_parallel()` function that
spawns tool calls as concurrent `asyncio` tasks, collecting stream
events via `asyncio.Queue`
- **`service.py`**: `_yield_tool_call()` now accepts an optional
`session_lock` parameter for concurrent-safe session mutations
- **`base.py`**: Session lock exposed via `contextvars` so tools that
need it can access it without interface changes
- **`run_agent.py`**: Rate-limit counters (`successful_agent_runs`,
`successful_agent_schedules`) protected with the session lock to prevent
race conditions

### Concurrency Safety

| Shared State | Risk | Mitigation |
|---|---|---|
| `session.messages` (long-running tools only) | Race on append + upsert
| `session_lock` wraps mutations |
| `session.successful_agent_runs` counter | Bypass max-runs check |
`session_lock` wraps read-check-increment |
| Tool-internal state (DB queries, API calls) | None — stateless | No
mitigation needed |

### Testing

- Added `parallel_tool_calls_test.py` with tests for:
  - Parallel timing verification (sum vs max of delays)
  - Single tool call regression
  - Retryable error propagation
  - Shared session lock verification
  - Cancellation cleanup

Closes SECRT-2016

---------

Co-authored-by: Zamil Majdy <zamil.majdy@agpt.co>

Author: Otto-AGPT

autogpt_platform/backend/backend/copilot/parallel_tool_calls_test.py

@@ -0,0 +1,272 @@
+"""Tests for parallel tool call execution in CoPilot.
+
+These tests mock _yield_tool_call to avoid importing the full copilot stack
+which requires Prisma, DB connections, etc.
+"""
+
+import asyncio
+import time
+from typing import Any, cast
+
+import pytest
+
+
+@pytest.mark.asyncio
+async def test_parallel_tool_calls_run_concurrently():
+    """Multiple tool calls should complete in ~max(delays), not sum(delays)."""
+    # Import here to allow module-level mocking if needed
+    from backend.copilot.response_model import (
+        StreamToolInputAvailable,
+        StreamToolOutputAvailable,
+    )
+    from backend.copilot.service import _execute_tool_calls_parallel
+
+    n_tools = 3
+    delay_per_tool = 0.2
+    tool_calls = [
+        {
+            "id": f"call_{i}",
+            "type": "function",
+            "function": {"name": f"tool_{i}", "arguments": "{}"},
+        }
+        for i in range(n_tools)
+    ]
+
+    # Minimal session mock
+    class FakeSession:
+        session_id = "test"
+        user_id = "test"
+
+        def __init__(self):
+            self.messages = []
+
+    original_yield = None
+
+    async def fake_yield(tc_list, idx, sess, lock=None):
+        yield StreamToolInputAvailable(
+            toolCallId=tc_list[idx]["id"],
+            toolName=tc_list[idx]["function"]["name"],
+            input={},
+        )
+        await asyncio.sleep(delay_per_tool)
+        yield StreamToolOutputAvailable(
+            toolCallId=tc_list[idx]["id"],
+            toolName=tc_list[idx]["function"]["name"],
+            output="{}",
+        )
+
+    import backend.copilot.service as svc
+
+    original_yield = svc._yield_tool_call
+    svc._yield_tool_call = fake_yield
+    try:
+        start = time.monotonic()
+        events = []
+        async for event in _execute_tool_calls_parallel(
+            tool_calls, cast(Any, FakeSession())
+        ):
+            events.append(event)
+        elapsed = time.monotonic() - start
+    finally:
+        svc._yield_tool_call = original_yield
+
+    assert len(events) == n_tools * 2
+    # Parallel: should take ~delay, not ~n*delay
+    assert elapsed < delay_per_tool * (
+        n_tools - 0.5
+    ), f"Took {elapsed:.2f}s, expected parallel (~{delay_per_tool}s)"
+
+
+@pytest.mark.asyncio
+async def test_single_tool_call_works():
+    """Single tool call should work identically."""
+    from backend.copilot.response_model import (
+        StreamToolInputAvailable,
+        StreamToolOutputAvailable,
+    )
+    from backend.copilot.service import _execute_tool_calls_parallel
+
+    tool_calls = [
+        {
+            "id": "call_0",
+            "type": "function",
+            "function": {"name": "t", "arguments": "{}"},
+        }
+    ]
+
+    class FakeSession:
+        session_id = "test"
+        user_id = "test"
+
+        def __init__(self):
+            self.messages = []
+
+    async def fake_yield(tc_list, idx, sess, lock=None):
+        yield StreamToolInputAvailable(toolCallId="call_0", toolName="t", input={})
+        yield StreamToolOutputAvailable(toolCallId="call_0", toolName="t", output="{}")
+
+    import backend.copilot.service as svc
+
+    orig = svc._yield_tool_call
+    svc._yield_tool_call = fake_yield
+    try:
+        events = [
+            e
+            async for e in _execute_tool_calls_parallel(
+                tool_calls, cast(Any, FakeSession())
+            )
+        ]
+    finally:
+        svc._yield_tool_call = orig
+
+    assert len(events) == 2
+
+
+@pytest.mark.asyncio
+async def test_retryable_error_propagates():
+    """Retryable errors should be raised after all tools finish."""
+    from backend.copilot.response_model import StreamToolOutputAvailable
+    from backend.copilot.service import _execute_tool_calls_parallel
+
+    tool_calls = [
+        {
+            "id": f"call_{i}",
+            "type": "function",
+            "function": {"name": f"t_{i}", "arguments": "{}"},
+        }
+        for i in range(2)
+    ]
+
+    class FakeSession:
+        session_id = "test"
+        user_id = "test"
+
+        def __init__(self):
+            self.messages = []
+
+    async def fake_yield(tc_list, idx, sess, lock=None):
+        if idx == 1:
+            raise KeyError("bad")
+        from backend.copilot.response_model import StreamToolInputAvailable
+
+        yield StreamToolInputAvailable(
+            toolCallId=tc_list[idx]["id"], toolName="t_0", input={}
+        )
+        await asyncio.sleep(0.05)
+        yield StreamToolOutputAvailable(
+            toolCallId=tc_list[idx]["id"], toolName="t_0", output="{}"
+        )
+
+    import backend.copilot.service as svc
+
+    orig = svc._yield_tool_call
+    svc._yield_tool_call = fake_yield
+    try:
+        events = []
+        with pytest.raises(KeyError):
+            async for event in _execute_tool_calls_parallel(
+                tool_calls, cast(Any, FakeSession())
+            ):
+                events.append(event)
+        # First tool's events should still be yielded
+        assert any(isinstance(e, StreamToolOutputAvailable) for e in events)
+    finally:
+        svc._yield_tool_call = orig
+
+
+@pytest.mark.asyncio
+async def test_session_lock_shared():
+    """All parallel tools should receive the same lock instance."""
+    from backend.copilot.response_model import (
+        StreamToolInputAvailable,
+        StreamToolOutputAvailable,
+    )
+    from backend.copilot.service import _execute_tool_calls_parallel
+
+    tool_calls = [
+        {
+            "id": f"call_{i}",
+            "type": "function",
+            "function": {"name": f"t_{i}", "arguments": "{}"},
+        }
+        for i in range(3)
+    ]
+
+    class FakeSession:
+        session_id = "test"
+        user_id = "test"
+
+        def __init__(self):
+            self.messages = []
+
+    observed_locks = []
+
+    async def fake_yield(tc_list, idx, sess, lock=None):
+        observed_locks.append(lock)
+        yield StreamToolInputAvailable(
+            toolCallId=tc_list[idx]["id"], toolName=f"t_{idx}", input={}
+        )
+        yield StreamToolOutputAvailable(
+            toolCallId=tc_list[idx]["id"], toolName=f"t_{idx}", output="{}"
+        )
+
+    import backend.copilot.service as svc
+
+    orig = svc._yield_tool_call
+    svc._yield_tool_call = fake_yield
+    try:
+        async for _ in _execute_tool_calls_parallel(
+            tool_calls, cast(Any, FakeSession())
+        ):
+            pass
+    finally:
+        svc._yield_tool_call = orig
+
+    assert len(observed_locks) == 3
+    assert observed_locks[0] is observed_locks[1] is observed_locks[2]
+    assert isinstance(observed_locks[0], asyncio.Lock)
+
+
+@pytest.mark.asyncio
+async def test_cancellation_cleans_up():
+    """Generator close should cancel in-flight tasks."""
+    from backend.copilot.response_model import StreamToolInputAvailable
+    from backend.copilot.service import _execute_tool_calls_parallel
+
+    tool_calls = [
+        {
+            "id": f"call_{i}",
+            "type": "function",
+            "function": {"name": f"t_{i}", "arguments": "{}"},
+        }
+        for i in range(2)
+    ]
+
+    class FakeSession:
+        session_id = "test"
+        user_id = "test"
+
+        def __init__(self):
+            self.messages = []
+
+    started = asyncio.Event()
+
+    async def fake_yield(tc_list, idx, sess, lock=None):
+        yield StreamToolInputAvailable(
+            toolCallId=tc_list[idx]["id"], toolName=f"t_{idx}", input={}
+        )
+        started.set()
+        await asyncio.sleep(10)  # simulate long-running
+
+    import backend.copilot.service as svc
+
+    orig = svc._yield_tool_call
+    svc._yield_tool_call = fake_yield
+    try:
+        gen = _execute_tool_calls_parallel(tool_calls, cast(Any, FakeSession()))
+        await gen.__anext__()  # get first event
+        await started.wait()
+        await gen.aclose()  # close generator
+    finally:
+        svc._yield_tool_call = orig
+    # If we get here without hanging, cleanup worked