diff --git a/agent/git_operations.py b/agent/git_operations.py
index a00455e..cffa18d 100644
--- a/agent/git_operations.py
+++ b/agent/git_operations.py
@@ -164,6 +164,46 @@ class GitOperations:
             logger.error(f"Error creating result branch: {e}")
             return False
 
+    async def add_task_worktree(self, path: str, task_id: str) -> Optional[str]:
+        """Create an isolated git worktree at `path` on a fresh per-task result branch.
+
+        Cut from this agent's clone (off origin/<base> or HEAD), the worktree has the full repo
+        contents plus its own index/HEAD, so one agent can run several tasks concurrently without
+        them clobbering a shared checkout. Returns the branch name, or None on failure.
+        """
+        try:
+            root = await self.repo_root() or self.workspace_dir
+            safe_task_id = task_id.replace("/", "-")[:12]
+            branch = f"agent/{self.agent_id}/{safe_task_id}-{int(time.time())}"
+
+            await self._run_git_command(["git", "fetch", "origin"], cwd=root)
+            base_ref = f"origin/{self.base_branch}"
+            if (await self._run_git_command(["git", "rev-parse", "--verify", base_ref], cwd=root)).returncode != 0:
+                base_ref = "HEAD"
+                logger.warning(f"Base branch origin/{self.base_branch} not found; worktree from HEAD")
+
+            Path(path).parent.mkdir(parents=True, exist_ok=True)
+            result = await self._run_git_command(
+                ["git", "worktree", "add", "-b", branch, path, base_ref], cwd=root,
+            )
+            if result.returncode != 0:
+                logger.error(f"git worktree add failed: {result.stderr}")
+                return None
+            logger.info(f"Created task worktree {path} on branch {branch}")
+            return branch
+        except Exception as e:
+            logger.error(f"Error creating task worktree: {e}")
+            return None
+
+    async def remove_task_worktree(self, path: str) -> None:
+        """Tear down a per-task worktree (best-effort). The branch ref is kept (already pushed)."""
+        try:
+            root = await self.repo_root() or self.workspace_dir
+            await self._run_git_command(["git", "worktree", "remove", "--force", path], cwd=root)
+            logger.info(f"Removed task worktree {path}")
+        except Exception as e:
+            logger.warning(f"Failed to remove task worktree {path}: {e}")
+
     async def commit_changes(self, message: str) -> Optional[str]:
         try:
             cwd = await self.repo_root() or self.workspace_dir
diff --git a/agent/main.py b/agent/main.py
index 9ecb370..1b20594 100644
--- a/agent/main.py
+++ b/agent/main.py
@@ -72,11 +72,10 @@ class AgentRuntimeDisconnected(RuntimeError):
 class Agent:
     """Agent that connects to orchestrator and executes tasks."""
 
-    # One task at a time per agent: tasks execute against the shared cloned repo root (see
-    # _execute_assignment, Option B 2026-06-16) and the per-agent git checkout (workspace_git,
-    # branch/commit/push) is single-repo and not concurrency-safe. Swarm parallelism comes from
-    # MULTIPLE agents, not multiple tasks per agent. Override via env only on isolated workspaces.
-    MAX_CONCURRENT_TASKS = int(os.getenv("MAX_CONCURRENT_TASKS", "1"))
+    # An agent may run several tasks at once; each executes in its OWN git worktree cut from this
+    # agent's clone (see _execute_assignment / GitOperations.add_task_worktree), so concurrent tasks
+    # never share a checkout/index. Swarm parallelism = many agents × this per-agent concurrency.
+    MAX_CONCURRENT_TASKS = int(os.getenv("MAX_CONCURRENT_TASKS", "4"))
     TASK_TIMEOUT_SECONDS = int(os.getenv("TASK_TIMEOUT_SECONDS", "60"))
     PEER_REPLY_TIMEOUT_SECONDS = int(os.getenv("PEER_REPLY_TIMEOUT_SECONDS", "20"))
     HEARTBEAT_INTERVAL_SECONDS = 15
@@ -115,14 +114,20 @@ class Agent:
         self._peer_executor: Optional[TaskExecutor] = None
 
         self.workspace_git = GitOperations(str(self.workspace_dir), self.agent_id)
+        # Serializes only the brief shared-repo git plumbing (worktree add/remove + fetch) so
+        # concurrent tasks don't race on .git locks. Task EXECUTION stays parallel.
+        self._git_admin_lock = asyncio.Lock()
 
     def available_slots(self) -> int:
         # Return how many additional tasks this agent can currently accept.
         return max(0, self.MAX_CONCURRENT_TASKS - len(self.active_tasks))
 
     def task_workspace(self, task_id: str) -> Path:
-        # Compute the dedicated per-task working directory inside the agent workspace.
-        return self.workspace_dir / ".agent_tasks" / task_id
+        # Per-task git worktree path. Kept OUTSIDE the repo root (default /tmp/agent-worktrees) so
+        # the main checkout never sees it as untracked, and namespaced by agent_id to avoid
+        # collisions. `git worktree add` requires the leaf to not pre-exist, so we don't mkdir it.
+        base = Path(os.getenv("AGENT_WORKTREE_BASE", "/tmp/agent-worktrees"))
+        return base / self.agent_id / task_id.replace("/", "-")
 
     async def safe_send(self, payload: dict):
         # Serialize and send a websocket message while holding a lock to prevent concurrent writes.
@@ -386,22 +391,28 @@ class Agent:
             await self.send_status_update("busy", task_id, "Starting task execution")
 
             start_time = time.time()
+            task_workspace = self.task_workspace(task_id)
 
             git_enabled = False
+            task_git = None
             try:
-                # Execute against the cloned repository root (Option B, 2026-06-16). The previous
-                # per-task subdir (/workspace/.agent_tasks/<task_id>) was created empty and never
-                # seeded with the repo, so the executor saw an empty workspace and every SWE task
-                # failed with "Empty workspace: no files detected". Using the repo root means the
-                # model reads the real source and git (workspace_git, rooted at the same dir)
-                # commits the actual edits. One-task-per-agent (MAX_CONCURRENT_TASKS) keeps this safe.
-                executor = TaskExecutor(agent_id=self.agent_id, workspace_dir=str(self.workspace_dir))
-                git_enabled = await self.workspace_git.is_git_workspace()
-                if git_enabled:
-                    branch_created = await self.workspace_git.create_result_branch(task_id)
-                    if not branch_created:
-                        logger.warning("Failed to create task result branch; continuing without git push")
-                        git_enabled = False
+                # Each task runs in its OWN git worktree cut from this agent's clone, so an agent can
+                # run several tasks concurrently without sharing a checkout/index. The worktree holds
+                # the full repo contents on a fresh result branch; the executor reads/writes the real
+                # source there and the per-task GitOperations (task_git) commits/pushes that branch.
+                # This fixes the prior empty per-task subdir → "Empty workspace: no files detected".
+                # Falls back to the repo root (no isolated branch) only if worktree creation fails.
+                if await self.workspace_git.is_git_workspace():
+                    async with self._git_admin_lock:  # serialize shared-repo git plumbing only
+                        result_branch = await self.workspace_git.add_task_worktree(str(task_workspace), task_id)
+                    if result_branch:
+                        git_enabled = True
+                        task_git = GitOperations(str(task_workspace), self.agent_id)
+                        task_git.result_branch = result_branch
+                    else:
+                        logger.warning("Failed to create task worktree; executing on repo root without git push")
+                exec_dir = str(task_workspace) if git_enabled else str(self.workspace_dir)
+                executor = TaskExecutor(agent_id=self.agent_id, workspace_dir=exec_dir)
 
                 result = await asyncio.wait_for(
                     executor.execute_task(
@@ -409,7 +420,7 @@ class Agent:
                         description=description,
                         context={
                             **context,
-                            "workspace_dir": str(self.workspace_dir),
+                            "workspace_dir": exec_dir,
                             "repo_workspace_dir": str(self.workspace_dir),
                             "git_repo_url": self.git_repo_url,
                             "agent_id": self.agent_id,
@@ -425,12 +436,12 @@ class Agent:
                 if result.get("success"):
                     self.last_summary = self._summarize_result(result) or self.last_summary
                 if result.get("success") and not awaiting_handoff:
-                    if git_enabled:
-                        commit_sha = await self.workspace_git.commit_changes(
+                    if git_enabled and task_git:
+                        commit_sha = await task_git.commit_changes(
                             message=f"Task {task_id}: {description[:50]}"
                         )
                         if commit_sha:
-                            branch_name = await self.workspace_git.push_results()
+                            branch_name = await task_git.push_results()
                             result["git_branch"] = branch_name
                             result["commit_sha"] = commit_sha
                         else:
@@ -473,6 +484,9 @@ class Agent:
                 await self.send_task_result(task_id, False, {"error": str(e), "success": False})
                 await self.send_status_update("idle", None, f"Task failed: {e}")
             finally:
+                if git_enabled:
+                    async with self._git_admin_lock:
+                        await self.workspace_git.remove_task_worktree(str(task_workspace))
                 self.active_tasks.pop(task_id, None)
                 ACTIVE_TASKS.set(len(self.active_tasks))
                 self.current_task_id = None