feat: --max-startup-parallel

Author: Gusarich

src/cli.py

@@ -264,6 +264,12 @@ def create_parser(cls) -> argparse.ArgumentParser:
             default=None,
             help="Max parallel instances (auto if omitted)",
         )
+        g_limits.add_argument(
+            "--max-startup-parallel",
+            type=int,
+            default=None,
+            help="Max parallel startup initializations (default: min(10,max-parallel))",
+        )
         g_limits.add_argument(
             "--timeout", type=int, default=3600, help="Timeout per instance (seconds)"
         )
@@ -844,6 +850,7 @@ def _add(field: str, reason: str, example: str = ""):
                 _add("import_conflict_policy", "must be fail|overwrite|suffix", "fail")
         except Exception:
             pass
+        # Allow 'auto' for both
         try:
             mpi = full_config.get("orchestration", {}).get(
                 "max_parallel_instances", "auto"
@@ -864,6 +871,26 @@ def _add(field: str, reason: str, example: str = ""):
                 "must be integer or 'auto'",
                 "auto",
             )
+        try:
+            mps = full_config.get("orchestration", {}).get(
+                "max_parallel_startup", "auto"
+            )
+            if isinstance(mps, str) and mps.lower() == "auto":
+                pass
+            else:
+                v2 = int(mps)
+                if v2 <= 0:
+                    _add(
+                        "orchestration.max_parallel_startup",
+                        "must be > 0 or 'auto'",
+                        "auto",
+                    )
+        except Exception:
+            _add(
+                "orchestration.max_parallel_startup",
+                "must be integer or 'auto'",
+                "auto",
+            )
         # Strategy exists
         try:
             strategy = full_config.get("strategy", args.strategy)
@@ -1598,6 +1625,10 @@ async def run_orchestrator(self, args: argparse.Namespace) -> int:
             cli_config.setdefault("orchestration", {})[
                 "max_parallel_instances"
             ] = args.max_parallel
+        if getattr(args, "max_startup_parallel", None):
+            cli_config.setdefault("orchestration", {})[
+                "max_parallel_startup"
+            ] = args.max_startup_parallel
         if hasattr(args, "timeout") and args.timeout:
             cli_config.setdefault("runner", {})["timeout"] = args.timeout
         if hasattr(args, "force_commit") and args.force_commit:
@@ -1800,6 +1831,9 @@ def _red(k, v):
 
         # Get orchestration settings from merged config
         max_parallel = full_config["orchestration"]["max_parallel_instances"]
+        max_startup_parallel = full_config["orchestration"].get(
+            "max_parallel_startup", "auto"
+        )
         state_dir = full_config.get("orchestration", {}).get(
             "state_dir"
         ) or full_config.get("state_dir", Path("./pitaya_state"))
@@ -1820,17 +1854,38 @@ def _red(k, v):
         # Respect global session volume consent by setting env for runner
         allow_global_session = bool(getattr(args, "allow_global_session_volume", False))
 
-        # Resolve max_parallel without host resource calculations
-        try:
-            if isinstance(max_parallel, str):
-                # Accept numeric strings; treat non-numeric (e.g., 'auto') as default 5
+        # Resolve parallelism
+        import os as _os
+
+        def _cpu_default() -> int:
+            try:
+                return max(2, int(_os.cpu_count() or 2))
+            except Exception:
+                return 2
+
+        # Total parallel: auto -> cpu-based
+        if isinstance(max_parallel, str):
+            if max_parallel.lower() == "auto":
+                max_parallel_val = _cpu_default()
+            else:
                 max_parallel_val = int(max_parallel)
-            elif isinstance(max_parallel, int):
-                max_parallel_val = max_parallel
+        elif isinstance(max_parallel, int):
+            max_parallel_val = max(1, max_parallel)
+        else:
+            max_parallel_val = _cpu_default()
+
+        # Startup parallel: auto -> min(10, total)
+        if isinstance(max_startup_parallel, str):
+            if max_startup_parallel.lower() == "auto":
+                max_startup_parallel_val = min(10, max_parallel_val)
             else:
-                max_parallel_val = 5
-        except Exception:
-            max_parallel_val = 5
+                max_startup_parallel_val = int(max_startup_parallel)
+        elif isinstance(max_startup_parallel, int):
+            max_startup_parallel_val = max(1, max_startup_parallel)
+        else:
+            max_startup_parallel_val = min(10, max_parallel_val)
+        # Clamp startup to not exceed total
+        max_startup_parallel_val = min(max_startup_parallel_val, max_parallel_val)
 
         # Proxy automatic egress defaults removed
 
@@ -1851,6 +1906,7 @@ def _red(k, v):
         # Create orchestrator
         self.orchestrator = Orchestrator(
             max_parallel_instances=max_parallel_val,
+            max_parallel_startup=max_startup_parallel_val,
             state_dir=Path(state_dir),
             logs_dir=Path(logs_dir),
             container_limits=container_limits,

src/config.py

@@ -262,8 +262,10 @@ def get_default_config() -> Dict[str, Any]:
             "tmpfs_size_mb": 512,
         },
         "orchestration": {
-            # Default concurrency: 5; use --max-parallel to override
-            "max_parallel_instances": 5,
+            # Default concurrency: auto-calculated from CPU count
+            "max_parallel_instances": "auto",
+            # Startup cap: auto -> min(10, max_parallel_instances)
+            "max_parallel_startup": "auto",
             # Branch namespace is hierarchical by default
             # Format: orc/<strategy>/<run_id>/k<short8>
             "branch_namespace": "hierarchical",

src/instance_runner/api.py

@@ -7,6 +7,7 @@
 """
 
 from pathlib import Path
+import asyncio
 from typing import Any, Callable, Dict, Optional
 
 from ..shared import (
@@ -34,6 +35,7 @@ async def run_instance(
     session_id: Optional[str] = None,
     operator_resume: bool = False,
     event_callback: Optional[Callable[[Dict[str, Any]], None]] = None,
+    startup_semaphore: Optional[asyncio.Semaphore] = None,
     timeout_seconds: int = 3600,
     container_limits: Optional[ContainerLimits] = None,
     auth_config: Optional[AuthConfig] = None,
@@ -114,6 +116,7 @@ async def run_instance(
         operator_resume=operator_resume,
         session_group_key=session_group_key,
         event_callback=event_callback,
+        startup_semaphore=startup_semaphore,
         timeout_seconds=timeout_seconds,
         container_limits=container_limits,
         auth_config=auth_config,

src/instance_runner/runner.py

@@ -84,6 +84,7 @@ async def run_instance(
     operator_resume: bool = False,
     session_group_key: Optional[str] = None,
     event_callback: Optional[Callable[[Dict[str, Any]], None]] = None,
+    startup_semaphore: Optional[asyncio.Semaphore] = None,
     timeout_seconds: int = 3600,
     container_limits: Optional[ContainerLimits] = None,
     auth_config: Optional[AuthConfig] = None,
@@ -210,6 +211,7 @@ async def run_instance(
             operator_resume=operator_resume,
             session_group_key=session_group_key,
             event_callback=event_callback,
+            startup_semaphore=startup_semaphore,
             timeout_seconds=timeout_seconds,
             container_limits=container_limits,
             auth_config=auth_config,
@@ -317,6 +319,7 @@ async def _run_instance_attempt(
     operator_resume: bool,
     session_group_key: Optional[str],
     event_callback: Optional[Callable[[Dict[str, Any]], None]],
+    startup_semaphore: Optional[asyncio.Semaphore],
     timeout_seconds: int,
     container_limits: ContainerLimits,
     auth_config: Optional[AuthConfig],
@@ -419,20 +422,43 @@ def emit_event(event_type: str, data: Dict[str, Any]) -> None:
 
             emit_event("instance.phase_completed", {"phase": "validation"})
 
-            # Phase 2: Workspace Preparation (NEW - happens BEFORE container)
+            # Phase 2: Workspace Preparation (happens BEFORE container)
             logger.info(f"Preparing isolated workspace for branch {base_branch}")
-            emit_event("instance.workspace_preparing", {"base_branch": base_branch})
-
-            logger.info("Calling git_ops.prepare_workspace...")
-            workspace_dir = await git_ops.prepare_workspace(
-                repo_path=repo_path,
-                base_branch=base_branch,
-                instance_id=instance_id,
-                run_id=run_id,
-                strategy_execution_id=strategy_execution_id,
-                container_name=container_name,
-                reuse_if_exists=operator_resume,
-            )
+            if startup_semaphore is not None:
+                # Signal that we are waiting on startup slot
+                try:
+                    emit_event(
+                        "instance.startup_waiting",
+                        {"base_branch": base_branch},
+                    )
+                except Exception:
+                    pass
+                async with startup_semaphore:
+                    emit_event(
+                        "instance.workspace_preparing", {"base_branch": base_branch}
+                    )
+                    logger.info("Calling git_ops.prepare_workspace (startup slot acquired)...")
+                    workspace_dir = await git_ops.prepare_workspace(
+                        repo_path=repo_path,
+                        base_branch=base_branch,
+                        instance_id=instance_id,
+                        run_id=run_id,
+                        strategy_execution_id=strategy_execution_id,
+                        container_name=container_name,
+                        reuse_if_exists=operator_resume,
+                    )
+            else:
+                emit_event("instance.workspace_preparing", {"base_branch": base_branch})
+                logger.info("Calling git_ops.prepare_workspace...")
+                workspace_dir = await git_ops.prepare_workspace(
+                    repo_path=repo_path,
+                    base_branch=base_branch,
+                    instance_id=instance_id,
+                    run_id=run_id,
+                    strategy_execution_id=strategy_execution_id,
+                    container_name=container_name,
+                    reuse_if_exists=operator_resume,
+                )
             logger.info(f"Workspace prepared at: {workspace_dir}")
 
             emit_event(

src/orchestration/orchestrator.py

@@ -54,6 +54,7 @@ class Orchestrator:
     def __init__(
         self,
         max_parallel_instances: Optional[int] = None,
+        max_parallel_startup: Optional[int] = None,
         state_dir: Path = Path("./pitaya_state"),
         logs_dir: Path = Path("./logs"),
         container_limits: Optional[ContainerLimits] = None,
@@ -87,6 +88,7 @@ def __init__(
             auth_config: Authentication configuration for AI tools
         """
         self.max_parallel_instances: Optional[int] = max_parallel_instances
+        self.max_parallel_startup: Optional[int] = max_parallel_startup
         self.state_dir = state_dir
         self.logs_dir = logs_dir
         self.container_limits = container_limits or ContainerLimits()
@@ -131,6 +133,8 @@ def __init__(
 
         # Resource pool semaphore (initialized after adaptive calc in initialize())
         self._resource_pool: asyncio.Semaphore = asyncio.Semaphore(1)
+        # Separate startup semaphore to throttle expensive workspace prep
+        self._startup_pool: asyncio.Semaphore = asyncio.Semaphore(1)
         # Branch namespace strategy: 'flat' (default) or 'hierarchical'
         self.branch_namespace = str(branch_namespace or "flat").lower()
         self.allow_overwrite_protected_refs = bool(allow_overwrite_protected_refs)
@@ -206,14 +210,30 @@ async def initialize(self) -> None:
         # Clean up orphaned containers from previous runs
         await self.cleanup_orphaned_containers()
 
-        # Resolve max_parallel_instances without any host CPU/memory calculations.
-        # Default to 5 if not provided by the caller/CLI.
-        if self.max_parallel_instances is None:
-            self.max_parallel_instances = 5
-            logger.info("Parallelism(default): max_parallel_instances=5")
+        # Resolve parallelism defaults: CPU-based for total; startup=min(10, total)
+        if self.max_parallel_instances is None or int(self.max_parallel_instances) <= 0:
+            try:
+                import os as _os
+
+                cpu = int(_os.cpu_count() or 2)
+            except Exception:
+                cpu = 2
+            self.max_parallel_instances = max(1, cpu)
+            logger.info(
+                f"Parallelism(default): max_parallel_instances={self.max_parallel_instances} (cpu-based)"
+            )
+        if self.max_parallel_startup is None or int(self.max_parallel_startup) <= 0:
+            self.max_parallel_startup = max(
+                1, min(10, int(self.max_parallel_instances or 1))
+            )
+            logger.info(
+                f"Parallelism(default): max_parallel_startup={self.max_parallel_startup} (min(10,total))"
+            )
 
         # Initialize resource pool semaphore now that parallelism is resolved
         self._resource_pool = asyncio.Semaphore(int(self.max_parallel_instances or 1))
+        # Initialize startup pool (caps concurrent workspace preparations)
+        self._startup_pool = asyncio.Semaphore(int(self.max_parallel_startup or self.max_parallel_instances or 1))
 
         # Start multiple background executors equal to max_parallel_instances
         num_executors = int(self.max_parallel_instances or 1)
@@ -222,7 +242,8 @@ async def initialize(self) -> None:
             self._executor_tasks.append(task)
 
         logger.info(
-            f"Pitaya initialized with {self.max_parallel_instances} max parallel instances and {num_executors} executor tasks"
+            f"Pitaya initialized with max_parallel_instances={self.max_parallel_instances} "
+            f"max_parallel_startup={self.max_parallel_startup} and {num_executors} executor tasks"
         )
         self._initialized = True
 
@@ -1499,6 +1520,11 @@ def _write_last_active():
                         "container_env_preparing",
                         "Preparing container env...",
                     )
+                elif et == "instance.startup_waiting":
+                    phase, activity = (
+                        "startup_waiting",
+                        "Waiting for startup slot...",
+                    )
                 elif et == "instance.container_env_prepared":
                     phase, activity = "container_env_prepared", "Container env ready"
                 elif et == "instance.container_created":
@@ -1608,6 +1634,7 @@ def _write_last_active():
                 ),
                 operator_resume=op_resume,
                 event_callback=event_callback,
+                startup_semaphore=self._startup_pool,
                 timeout_seconds=self.runner_timeout_seconds,
                 container_limits=ContainerLimits(
                     cpu_count=eff_cpu, memory_gb=eff_mem, memory_swap_gb=eff_mem
@@ -2555,6 +2582,7 @@ async def _run_instance_from_saved_state(self, instance_info) -> InstanceResult:
                     instance_id=instance_info.instance_id,
                 ),
                 container_name=instance_info.container_name,
+                startup_semaphore=self._startup_pool,
                 container_limits=ContainerLimits(
                     cpu_count=eff_cpu, memory_gb=eff_mem, memory_swap_gb=eff_mem
                 ),