Fix session creation race condition and add capacity tests

README.md

@@ -28,8 +28,9 @@ Create a `.env` file in the project root (or export these in your shell):
 ```bash
 ANTHROPIC_API_KEY=<your-anthropic-api-key> # if using anthropic models
 HF_TOKEN=<your-hugging-face-token>
-GITHUB_TOKEN=<github-personal-access-token> 
+GITHUB_TOKEN=<github-personal-access-token>
 ```
+
 If no `HF_TOKEN` is set, the CLI will prompt you to paste one on first launch. To get a GITHUB_TOKEN follow the tutorial [here](https://docs.github.com/en/authentication/keeping-your-account-and-data-secure/managing-your-personal-access-tokens#creating-a-fine-grained-personal-access-token).
 
 ### Usage
@@ -183,6 +184,10 @@ The agent emits the following events via `event_queue`:
 - `undo_complete` - Undo operation completed
 - `shutdown` - Agent shutting down
 
+## Session Limits
+
+Session creation is concurrency-safe and enforced server-side. If the Space is at capacity, `/api/session` returns `503` instead of oversubscribing the limit. Per-user limits are enforced the same way, so bursts of simultaneous requests cannot create extra sessions past the configured cap.
+
 ## Development
 
 ### Adding Built-in Tools

agent/core/session.py

@@ -119,6 +119,11 @@ def __init__(
         # Key absent → not probed yet; fall back to the raw preference.
         self.model_effective_effort: dict[str, str | None] = {}
 
+        # Session-scoped local tool file read tracking (enforces read-before-write safety
+        # per session, preventing cross-session isolation bugs). Maps resolved file paths
+        # to True if they have been read in this session.
+        self._local_files_read: set[str] = set()
+
     async def send_event(self, event: Event) -> None:
         """Send event back to client and log to trajectory"""
         await self.event_queue.put(event)

agent/tools/local_tools.py

@@ -126,7 +126,7 @@ async def _bash_handler(args: dict[str, Any], **_kw) -> tuple[str, bool]:
         return f"bash error: {e}", False
 
 
-async def _read_handler(args: dict[str, Any], **_kw) -> tuple[str, bool]:
+async def _read_handler(args: dict[str, Any], session: Any = None, **_kw) -> tuple[str, bool]:
     file_path = args.get("path", "")
     if not file_path:
         return "No path provided.", False
@@ -140,7 +140,11 @@ async def _read_handler(args: dict[str, Any], **_kw) -> tuple[str, bool]:
     except Exception as e:
         return f"read error: {e}", False
 
-    _files_read.add(_resolve_path(file_path))
+    resolved = _resolve_path(file_path)
+    if session and hasattr(session, '_local_files_read'):
+        session._local_files_read.add(resolved)
+    else:
+        _files_read.add(resolved)
 
     lines = raw_content.splitlines()
     offset = max((args.get("offset") or 1), 1)
@@ -156,20 +160,27 @@ async def _read_handler(args: dict[str, Any], **_kw) -> tuple[str, bool]:
     return "\n".join(numbered), True
 
 
-async def _write_handler(args: dict[str, Any], **_kw) -> tuple[str, bool]:
+async def _write_handler(args: dict[str, Any], session: Any = None, **_kw) -> tuple[str, bool]:
     file_path = args.get("path", "")
     content = args.get("content", "")
     if not file_path:
         return "No path provided.", False
     p = Path(file_path)
-    if p.exists() and _resolve_path(file_path) not in _files_read:
+    resolved = _resolve_path(file_path)
+
+    # Check if file was read in this session (or globally as fallback)
+    files_read = session._local_files_read if session and hasattr(session, '_local_files_read') else _files_read
+    if p.exists() and resolved not in files_read:
         return (
             f"You must read {file_path} before overwriting it. "
             f"Use the read tool first to see current contents."
         ), False
     try:
         _atomic_write(p, content)
-        _files_read.add(_resolve_path(file_path))
+        if session and hasattr(session, '_local_files_read'):
+            session._local_files_read.add(resolved)
+        else:
+            _files_read.add(resolved)
         msg = f"Wrote {len(content)} bytes to {file_path}"
         # Syntax validation for Python files
         if p.suffix == ".py":
@@ -182,7 +193,7 @@ async def _write_handler(args: dict[str, Any], **_kw) -> tuple[str, bool]:
         return f"write error: {e}", False
 
 
-async def _edit_handler(args: dict[str, Any], **_kw) -> tuple[str, bool]:
+async def _edit_handler(args: dict[str, Any], session: Any = None, **_kw) -> tuple[str, bool]:
     from agent.tools.edit_utils import apply_edit, validate_python
 
     file_path = args.get("path", "")
@@ -199,7 +210,10 @@ async def _edit_handler(args: dict[str, Any], **_kw) -> tuple[str, bool]:
     p = Path(file_path)
     if not p.exists():
         return f"File not found: {file_path}", False
-    if _resolve_path(file_path) not in _files_read:
+
+    resolved = _resolve_path(file_path)
+    files_read = session._local_files_read if session and hasattr(session, '_local_files_read') else _files_read
+    if resolved not in files_read:
         return (
             f"You must read {file_path} before editing it. "
             f"Use the read tool first to see current contents."

backend/session_manager.py

@@ -121,6 +121,8 @@ def __init__(self, config_path: str | None = None) -> None:
         self.config = load_config(config_path or DEFAULT_CONFIG_PATH)
         self.sessions: dict[str, AgentSession] = {}
         self._lock = asyncio.Lock()
+        self._pending_session_count = 0
+        self._pending_sessions_by_user: dict[str, int] = {}
 
     def _count_user_sessions(self, user_id: str) -> int:
         """Count active sessions owned by a specific user."""
@@ -130,6 +132,28 @@ def _count_user_sessions(self, user_id: str) -> int:
             if s.user_id == user_id and s.is_active
         )
 
+    def _count_pending_user_sessions(self, user_id: str) -> int:
+        """Count session slots reserved but not yet activated for a user."""
+        return self._pending_sessions_by_user.get(user_id, 0)
+
+    def _reserve_session_slot(self, user_id: str) -> None:
+        """Reserve a session slot before the blocking constructors run."""
+        self._pending_session_count += 1
+        if user_id != "dev":
+            self._pending_sessions_by_user[user_id] = (
+                self._pending_sessions_by_user.get(user_id, 0) + 1
+            )
+
+    def _release_session_slot(self, user_id: str) -> None:
+        """Release a previously reserved session slot."""
+        self._pending_session_count = max(0, self._pending_session_count - 1)
+        if user_id != "dev":
+            pending = self._pending_sessions_by_user.get(user_id, 0) - 1
+            if pending > 0:
+                self._pending_sessions_by_user[user_id] = pending
+            else:
+                self._pending_sessions_by_user.pop(user_id, None)
+
     async def create_session(
         self,
         user_id: str = "dev",
@@ -153,23 +177,27 @@ async def create_session(
             SessionCapacityError: If the server or user has reached the
                 maximum number of concurrent sessions.
         """
+        reserved = False
+
         # ── Capacity checks ──────────────────────────────────────────
         async with self._lock:
-            active_count = self.active_session_count
+            active_count = self.active_session_count + self._pending_session_count
             if active_count >= MAX_SESSIONS:
                 raise SessionCapacityError(
                     f"Server is at capacity ({active_count}/{MAX_SESSIONS} sessions). "
                     "Please try again later.",
                     error_type="global",
                 )
             if user_id != "dev":
-                user_count = self._count_user_sessions(user_id)
+                user_count = self._count_user_sessions(user_id) + self._count_pending_user_sessions(user_id)
                 if user_count >= MAX_SESSIONS_PER_USER:
                     raise SessionCapacityError(
                         f"You have reached the maximum of {MAX_SESSIONS_PER_USER} "
                         "concurrent sessions. Please close an existing session first.",
                         error_type="per_user",
                     )
+            self._reserve_session_slot(user_id)
+            reserved = True
 
         session_id = str(uuid.uuid4())
 
@@ -198,7 +226,14 @@ def _create_session_sync():
             logger.info(f"Session initialized in {t1 - t0:.2f}s")
             return tool_router, session
 
-        tool_router, session = await asyncio.to_thread(_create_session_sync)
+        try:
+            tool_router, session = await asyncio.to_thread(_create_session_sync)
+        except Exception:
+            async with self._lock:
+                if reserved:
+                    self._release_session_slot(user_id)
+                    reserved = False
+            raise
 
         # Create wrapper
         agent_session = AgentSession(
@@ -214,11 +249,21 @@ def _create_session_sync():
             self.sessions[session_id] = agent_session
 
         # Start the agent loop task
-        task = asyncio.create_task(
-            self._run_session(session_id, submission_queue, event_queue, tool_router)
-        )
+        try:
+            task = asyncio.create_task(
+                self._run_session(session_id, submission_queue, event_queue, tool_router)
+            )
+        except Exception:
+            async with self._lock:
+                self.sessions.pop(session_id, None)
+            raise
         agent_session.task = task
 
+        async with self._lock:
+            if reserved:
+                self._release_session_slot(user_id)
+                reserved = False
+
         logger.info(f"Created session {session_id} for user {user_id}")
         return session_id
 

tests/unit/conftest.py

@@ -0,0 +1,110 @@
+"""Pytest configuration and fixtures for unit tests.
+
+This module sets up comprehensive stubs for agent dependencies to allow
+isolated unit testing without requiring the full runtime stack.
+"""
+
+import sys
+from unittest.mock import MagicMock
+from types import ModuleType
+
+
+def create_fake_package(name):
+    """Create a fake package and all its submodules."""
+    module = ModuleType(name)
+    sys.modules[name] = module
+    return module
+
+
+def _install_stubs():
+    """Install comprehensive stubs for all external dependencies.
+
+    This must run at import time because some tests import modules that pull
+    in agent dependencies at module import, before pytest hooks execute.
+    """
+    # Web frameworks
+    fastapi_module = create_fake_package('fastapi')
+    fastapi_module.FastAPI = MagicMock
+    fastapi_module.testclient = MagicMock()
+    fastapi_module.testclient.TestClient = MagicMock
+    sys.modules['fastapi'] = fastapi_module
+    sys.modules['fastapi.testclient'] = fastapi_module.testclient
+    sys.modules['starlette'] = MagicMock()
+
+    # HTTP and networking
+    sys.modules['httpx'] = MagicMock()
+    sys.modules['aiohttp'] = MagicMock()
+    sys.modules['requests'] = MagicMock()
+
+    # Data and models
+    sys.modules['pydantic'] = MagicMock()
+    sys.modules['pydantic.types'] = MagicMock()
+    sys.modules['numpy'] = MagicMock()
+    sys.modules['pandas'] = MagicMock()
+
+    # Hugging Face
+    sys.modules['huggingface_hub'] = MagicMock()
+    sys.modules['datasets'] = MagicMock()
+    sys.modules['transformers'] = MagicMock()
+
+    # Notebook support
+    sys.modules['nbformat'] = MagicMock()
+    sys.modules['nbconvert'] = MagicMock()
+    sys.modules['jupyter'] = MagicMock()
+
+    # Text processing
+    sys.modules['thefuzz'] = MagicMock()
+    sys.modules['thefuzz.fuzz'] = MagicMock()
+
+    # MCP stubs
+    mcp = create_fake_package('mcp')
+    mcp.types = MagicMock()
+    sys.modules['mcp.types'] = mcp.types
+
+    # FastMCP stubs - needs to be a real package, not just MagicMock
+    fastmcp = create_fake_package('fastmcp')
+    fastmcp.Client = MagicMock
+    fastmcp.exceptions = create_fake_package('fastmcp.exceptions')
+    fastmcp.exceptions.ToolError = Exception
+    fastmcp.types = create_fake_package('fastmcp.types')
+    fastmcp.mcp_config = MagicMock()  # Import location for MCPServerConfig loading
+    sys.modules['fastmcp.client'] = fastmcp
+    sys.modules['fastmcp.exceptions'] = fastmcp.exceptions
+    sys.modules['fastmcp.types'] = fastmcp.types
+    sys.modules['fastmcp.mcp_config'] = fastmcp.mcp_config
+
+    # LLM and AI stubs
+    sys.modules['litellm'] = MagicMock()
+    sys.modules['anthropic'] = MagicMock()
+    sys.modules['openai'] = MagicMock()
+
+    # Sandbox/container stubs
+    sys.modules['docker'] = MagicMock()
+    sys.modules['docker.client'] = MagicMock()
+
+    # Other common imports
+    sys.modules['dotenv'] = MagicMock()
+    sys.modules['yaml'] = MagicMock()
+    sys.modules['toml'] = MagicMock()
+
+
+# Install stubs IMMEDIATELY on module load (before pytest collection)
+# Keep this at module scope: importing test modules can trigger dependency
+# imports before pytest hooks run, so the stubs must already be present.
+_install_stubs()
+
+
+def pytest_configure(config):
+    """Pytest hook called before test collection starts.
+
+    We ensure stubs are already installed above, but this hook
+    can be used for any other early setup if needed.
+    """
+    pass
+
+
+
+
+
+
+