Categorize all spawn failures into actionable failure types

src/bernstein/core/spawner.py

@@ -39,6 +39,7 @@
 from bernstein.core.prometheus import agent_spawn_duration, merge_duration
 from bernstein.core.router import ProviderHealthStatus, RouterError, TierAwareRouter
 from bernstein.core.sandbox import DockerSandbox, spawn_in_sandbox
+from bernstein.core.spawn_errors import RetryStrategy, classify_spawn_error
 from bernstein.core.team_state import TeamStateStore
 from bernstein.core.traces import AgentTrace, TraceStore, finalize_trace, new_trace
 from bernstein.core.worktree import WorktreeError, WorktreeManager, WorktreeSetupConfig
@@ -1459,8 +1460,26 @@ def _spawn_for_tasks_internal(self, tasks: list[Task], model_override: str | Non
                         except RouterError:
                             provider_name = None
                     except (SpawnError, Exception) as exc:
+                        categorized = classify_spawn_error(exc, provider=provider_name)
                         attempt_errors.append(f"{adapter_name}: {exc}")
 
+                        # Fail-fast for permanent and operator-fix errors — no
+                        # point trying alternate providers when the binary is
+                        # missing or credentials are invalid.
+                        if categorized.retry_strategy in (
+                            RetryStrategy.NO_RETRY,
+                            RetryStrategy.RETRY_AFTER_FIX,
+                        ):
+                            logger.warning(
+                                "Spawn failure is non-retryable (strategy=%s session=%s adapter=%s): %s",
+                                categorized.retry_strategy.value,
+                                session_id,
+                                adapter_name,
+                                exc,
+                            )
+                            self._adapter_health.record_failure(adapter_name)
+                            break
+
                         # Check for auth error (T499)
                         is_auth_error = False
                         log_path = spawn_cwd / ".sdd" / "logs" / f"{session_id}.log"
@@ -1488,10 +1507,11 @@ def _spawn_for_tasks_internal(self, tasks: list[Task], model_override: str | Non
 
                         self._adapter_health.record_failure(adapter_name)
                         logger.warning(
-                            "Agent spawn failed (session=%s provider=%s adapter=%s): %s",
+                            "Agent spawn failed (session=%s provider=%s adapter=%s strategy=%s): %s",
                             session_id,
                             provider_name,
                             adapter_name,
+                            categorized.retry_strategy.value,
                             exc,
                         )
                         if self._router is None or provider_name is None:

src/bernstein/core/worktree.py

@@ -26,6 +26,7 @@
 
 from bernstein.core.git_ops import branch_delete, worktree_add, worktree_list, worktree_remove
 from bernstein.core.platform_compat import process_alive
+from bernstein.core.worktree_isolation import validate_worktree_isolation
 
 if TYPE_CHECKING:
     import threading
@@ -298,8 +299,22 @@ def create(self, session_id: str) -> Path:
         worker_pid = os.getpid()
         write_worktree_lock(self.repo_root, session_id, pid=worker_pid)
 
+        allowed_symlinks: tuple[str, ...] = ()
         if self._setup_config is not None:
             setup_worktree_env(self.repo_root, worktree_path, self._setup_config)
+            allowed_symlinks = self._setup_config.symlink_dirs
+
+        isolation_result = validate_worktree_isolation(
+            worktree_path,
+            self.repo_root,
+            allowed_symlink_dirs=allowed_symlinks,
+        )
+        if not isolation_result.passed:
+            self.cleanup(session_id)
+            raise WorktreeError(
+                f"Worktree isolation violated for session '{session_id}': "
+                + "; ".join(isolation_result.violations)
+            )
 
         return worktree_path