fix(copilot): serialize session creation to prevent permission-denied race condition

src/conductor/providers/copilot.py

@@ -174,6 +174,7 @@ def __init__(
         self._mcp_servers = mcp_servers or {}
         self._started = False
         self._start_lock = asyncio.Lock()
+        self._session_create_lock = asyncio.Lock()
         self._idle_recovery_config = idle_recovery_config or IdleRecoveryConfig()
         self._temperature = temperature
         self._session_ids: dict[str, str] = {}
@@ -470,9 +471,15 @@ async def _execute_sdk_call(
                     )
                     session = None
 
-            # Fall back to creating a new session
+            # Fall back to creating a new session.
+            # Serialize session creation to prevent a race condition in the
+            # Copilot SDK where overlapping create_session calls (e.g. from
+            # for-each groups with max_concurrent > 1) can cause the CLI to
+            # send permission.request for a session not yet registered in
+            # _sessions, raising "Permission denied".  See #27 / #29.
             if session is None:
-                session = await self._client.create_session(session_config)
+                async with self._session_create_lock:
+                    session = await self._client.create_session(session_config)
 
             # Track session ID for checkpoint persistence
             sid = getattr(session, "session_id", None)

tests/test_providers/test_copilot.py

@@ -1,5 +1,6 @@
 """Unit tests for the CopilotProvider implementation."""
 
+import asyncio
 import contextlib
 from typing import Any
 
@@ -556,3 +557,54 @@ def test_log_parse_recovery_truncates_long_error(
             max_attempts=5,
             error=long_error,
         )
+
+
+class TestSessionCreateLock:
+    """Tests for session creation serialization (issue #29).
+
+    Verifies that concurrent create_session calls are serialized via
+    _session_create_lock to prevent the permission-denied race condition
+    described in issue #27.
+    """
+
+    def test_session_create_lock_exists(self) -> None:
+        """Test that CopilotProvider has a _session_create_lock attribute."""
+        provider = CopilotProvider(mock_handler=stub_handler)
+        assert hasattr(provider, "_session_create_lock")
+        assert isinstance(provider._session_create_lock, asyncio.Lock)
+
+    @pytest.mark.asyncio
+    async def test_session_creation_is_serialized(self) -> None:
+        """Test that concurrent create_session calls do not overlap.
+
+        Simulates concurrent SDK calls by patching _execute_sdk_call to
+        acquire the lock and verify no two create_session calls run at
+        the same time.
+        """
+        provider = CopilotProvider(mock_handler=stub_handler)
+
+        # Track whether sessions are created concurrently
+        active_creates = 0
+        max_concurrent_creates = 0
+        creation_order: list[str] = []
+
+        original_lock = provider._session_create_lock
+
+        async def mock_create_session(name: str) -> None:
+            nonlocal active_creates, max_concurrent_creates
+            async with original_lock:
+                active_creates += 1
+                max_concurrent_creates = max(max_concurrent_creates, active_creates)
+                creation_order.append(name)
+                # Simulate session creation taking some time
+                await asyncio.sleep(0.05)
+                active_creates -= 1
+
+        # Launch multiple concurrent mock session creations
+        tasks = [asyncio.create_task(mock_create_session(f"agent_{i}")) for i in range(5)]
+        await asyncio.gather(*tasks)
+
+        # With the lock, at most one create_session should run at a time
+        assert max_concurrent_creates == 1
+        # All sessions should have been created
+        assert len(creation_order) == 5