feat: configurable TUI ready-gate timeout and optional spawn concurrency cap

CHANGELOG.md

@@ -4,6 +4,15 @@ All notable changes to this project are documented in this file.
 
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 
+## [Unreleased]
+
+### Added
+
+- `ClaudePool(tui_ready_timeout=...)` and `claude-pool serve --tui-ready-timeout` to configure
+  the TUI worker readiness deadline. The default is unchanged at 30 seconds.
+- `ClaudePool(spawn_concurrency=...)` and `claude-pool serve --spawn-concurrency` to cap
+  simultaneous worker cold starts. The default is unchanged: unbounded.
+
 ## [0.2.2] - 2026-06-13
 
 ### Fixed

README.md

@@ -119,6 +119,32 @@ Choose a backend with `ClaudePool(backend=...)` or `claude-pool serve --backend`
 | `stream-json` | You want structured metadata including usage, cost, duration, and rate-limit events. | Uses Claude Code print mode (`claude -p`). This is the default. |
 | `tui` | You need to avoid print mode and drive plain `claude` in a pty. | Text-first result metadata, best-effort usage extraction, and heavier worker startup. |
 
+## Startup Tuning
+
+Two constructor arguments control worker cold starts. Both default to the
+original behavior.
+
+| Argument | Default | Meaning |
+| --- | --- | --- |
+| `tui_ready_timeout` | `30.0` | Seconds a TUI worker may take to signal readiness through its SessionStart hook before its spawn fails with `WorkerStartError`. |
+| `spawn_concurrency` | `None` | Maximum simultaneous worker cold starts across the pool. `None` leaves spawning unbounded. |
+
+On slow hosts such as single-board computers, several TUI workers cold-starting
+at the same time can push each one past the 30-second readiness deadline, so a
+process restart fails with `WorkerStartError` even though each worker would
+have started fine on its own. Raising `tui_ready_timeout` and capping
+`spawn_concurrency` keeps those restarts calm:
+
+```python
+pool = ClaudePool(backend="tui", warm=2, tui_ready_timeout=75.0, spawn_concurrency=2)
+```
+
+The daemon accepts the same knobs:
+
+```sh
+claude-pool serve --backend tui --warm 2 --tui-ready-timeout 75 --spawn-concurrency 2
+```
+
 ## How It Works
 
 ```text

claude_pool.py

@@ -39,6 +39,7 @@
 _TUI_FIRST_PASTE_SETTLE = 0.75
 _TUI_PASTE_TO_CR = 0.3
 _TUI_CR_RETRY_AFTER = 3.0
+_TUI_READY_TIMEOUT = 30.0
 _LIVE_PGIDS: set[int] = set()
 logger = logging.getLogger("claude_pool")
 
@@ -438,6 +439,8 @@ async def spawn(
         argv: Sequence[str],
         cwd: str | None = None,
         env: Mapping[str, str] | None = None,
+        *,
+        ready_timeout: float = _TUI_READY_TIMEOUT,
     ) -> _TuiWorker:
         if os.name != "posix" or pty is None:
             raise NotImplementedError("TUI workers require POSIX ptys")
@@ -512,7 +515,7 @@ async def spawn(
                 session_id=session_id,
             )
             master_fd = None
-            await worker._wait_ready()
+            await worker._wait_ready(ready_timeout)
             return worker
         except BaseException:
             if worker is not None:
@@ -669,8 +672,8 @@ def _drain_pty(self) -> None:
             with self._tail_lock:
                 self._pty_tail.append(chunk)
 
-    async def _wait_ready(self) -> None:
-        deadline = time.monotonic() + 30.0
+    async def _wait_ready(self, timeout: float) -> None:
+        deadline = time.monotonic() + timeout
         trust_answered = False
         while True:
             if self.process.returncode is not None:
@@ -993,11 +996,19 @@ def __init__(
         max_idle: float = 900.0,
         default_timeout: float = 600.0,
         backend: str = "stream-json",
+        tui_ready_timeout: float = _TUI_READY_TIMEOUT,
+        spawn_concurrency: int | None = None,
     ) -> None:
         """Create a pool configuration without starting workers.
 
         ``backend`` selects the worker transport: ``"stream-json"`` for Claude
         Code print mode, or ``"tui"`` for plain Claude Code in a pty.
+
+        ``tui_ready_timeout`` is the number of seconds a TUI worker may take to
+        signal readiness through its SessionStart hook before its spawn fails
+        with ``WorkerStartError``. ``spawn_concurrency`` caps how many worker
+        cold starts may run at the same time across the pool; ``None`` keeps
+        spawning unbounded.
         """
         if backend not in {"stream-json", "tui"}:
             raise ClaudePoolError(f"unknown backend: {backend}")
@@ -1033,6 +1044,10 @@ def __init__(
         self._max_workers = max(1, max_workers)
         self._max_idle = max_idle
         self._default_timeout = default_timeout
+        self._tui_ready_timeout = tui_ready_timeout
+        self._spawn_semaphore = (
+            asyncio.Semaphore(max(1, spawn_concurrency)) if spawn_concurrency is not None else None
+        )
         self._warm: deque[_Worker | _TuiWorker] = deque()
         self._semaphore = asyncio.Semaphore(self._max_workers)
         self._closed = False
@@ -1376,9 +1391,20 @@ async def _checkout_worker(self) -> tuple[_Worker | _TuiWorker, bool]:
         return await self._spawn_worker(), False
 
     async def _spawn_worker(self) -> _Worker | _TuiWorker:
+        if self._spawn_semaphore is not None:
+            async with self._spawn_semaphore:
+                return await self._spawn_worker_now()
+        return await self._spawn_worker_now()
+
+    async def _spawn_worker_now(self) -> _Worker | _TuiWorker:
         try:
             if self._backend == "tui":
-                return await _TuiWorker.spawn(self._tui_argv, cwd=self._cwd, env=self._env)
+                return await _TuiWorker.spawn(
+                    self._tui_argv,
+                    cwd=self._cwd,
+                    env=self._env,
+                    ready_timeout=self._tui_ready_timeout,
+                )
             return await _Worker.spawn(self._argv, cwd=self._cwd, env=self._env)
         except OSError as exc:
             raise WorkerStartError(str(exc)) from exc
@@ -1537,6 +1563,8 @@ def _pool_kwargs_from_args(args: argparse.Namespace) -> dict[str, Any]:
         "max_idle": args.max_idle,
         "default_timeout": args.default_timeout,
         "backend": args.backend,
+        "tui_ready_timeout": args.tui_ready_timeout,
+        "spawn_concurrency": args.spawn_concurrency,
     }
 
 
@@ -1886,6 +1914,8 @@ def _add_profile_arguments(parser: argparse.ArgumentParser) -> None:
     parser.add_argument("--max-workers", type=int, default=4)
     parser.add_argument("--max-idle", type=float, default=900.0)
     parser.add_argument("--default-timeout", type=float, default=600.0)
+    parser.add_argument("--tui-ready-timeout", type=float, default=_TUI_READY_TIMEOUT)
+    parser.add_argument("--spawn-concurrency", type=int)
     parser.add_argument("--socket")
 
 

tests/test_backend_parity.py

@@ -10,7 +10,14 @@
 import pytest
 
 import claude_pool
-from claude_pool import AskTimeout, ClaudePool, ClaudePoolError, PoolClosed, _LIVE_PGIDS
+from claude_pool import (
+    AskTimeout,
+    ClaudePool,
+    ClaudePoolError,
+    PoolClosed,
+    WorkerStartError,
+    _LIVE_PGIDS,
+)
 
 
 ROOT = Path(__file__).resolve().parents[1]
@@ -53,6 +60,17 @@ def pool_kwargs(config: dict[str, Any], **overrides: Any) -> dict[str, Any]:
     return kwargs
 
 
+def tui_pool_kwargs(**overrides: Any) -> dict[str, Any]:
+    return pool_kwargs(
+        {
+            "backend": "tui",
+            "claude_bin": sys.executable,
+            "extra_args": [str(TUI_FAKE)],
+        },
+        **overrides,
+    )
+
+
 async def wait_for_warm(pool: ClaudePool, count: int) -> None:
     deadline = time.monotonic() + 10.0
     while time.monotonic() < deadline:
@@ -201,6 +219,59 @@ async def scenario() -> None:
     run(scenario())
 
 
+def test_pool_tui_ready_timeout_is_configurable() -> None:
+    async def scenario() -> None:
+        pool = ClaudePool(
+            **tui_pool_kwargs(tui_ready_timeout=0.5, env={"FAKE_TUI_SLOW_START": "10"})
+        )
+        try:
+            started = time.monotonic()
+            with pytest.raises(WorkerStartError, match="did not become ready"):
+                await pool.ask("never-ready")
+
+            assert time.monotonic() - started < 5.0
+        finally:
+            await pool.aclose()
+
+    run(scenario())
+
+
+def test_pool_spawn_concurrency_serializes_cold_starts() -> None:
+    async def scenario() -> None:
+        pool = ClaudePool(**tui_pool_kwargs(spawn_concurrency=1, env={"FAKE_TUI_SLOW_START": "1"}))
+        workers: list[Any] = []
+        try:
+            started = time.monotonic()
+            workers = list(await asyncio.gather(pool._spawn_worker(), pool._spawn_worker()))
+
+            assert time.monotonic() - started >= 2.0
+        finally:
+            for worker in workers:
+                await worker.kill()
+            await pool.aclose()
+            await assert_no_live_process_groups()
+
+    run(scenario())
+
+
+def test_pool_spawn_concurrency_default_allows_overlapping_cold_starts() -> None:
+    async def scenario() -> None:
+        pool = ClaudePool(**tui_pool_kwargs(env={"FAKE_TUI_SLOW_START": "1"}))
+        workers: list[Any] = []
+        try:
+            started = time.monotonic()
+            workers = list(await asyncio.gather(pool._spawn_worker(), pool._spawn_worker()))
+
+            assert time.monotonic() - started < 2.0
+        finally:
+            for worker in workers:
+                await worker.kill()
+            await pool.aclose()
+            await assert_no_live_process_groups()
+
+    run(scenario())
+
+
 def test_unknown_backend_raises_claude_pool_error() -> None:
     with pytest.raises(ClaudePoolError, match="unknown backend: nope"):
         ClaudePool(backend="nope")

tests/test_cli.py

@@ -160,6 +160,26 @@ def test_serve_and_ask_round_trip(tmp_path: Path) -> None:
         wait_for_no_fake_processes()
 
 
+def test_serve_accepts_spawn_tuning_flags(tmp_path: Path) -> None:
+    process, socket_path = start_server(
+        tmp_path,
+        "--warm",
+        "0",
+        "--tui-ready-timeout",
+        "45",
+        "--spawn-concurrency",
+        "2",
+    )
+    try:
+        completed = run_cli(["ask", "tuned", "--socket", str(socket_path)])
+
+        assert completed.returncode == 0
+        assert completed.stdout.strip() == "tuned"
+    finally:
+        stop_server(process)
+        wait_for_no_fake_processes()
+
+
 def test_ask_response_includes_duration_and_rate_limit(tmp_path: Path) -> None:
     process, socket_path = start_server(tmp_path, "--warm", "0")
     try:

tests/test_tui_worker.py

@@ -40,11 +40,15 @@ async def spawn_fake(
     *,
     env: dict[str, str] | None = None,
     argv: list[str] | None = None,
+    ready_timeout: float | None = None,
 ) -> _TuiWorker:
     full_env = os.environ.copy()
     if env:
         full_env.update(env)
-    return await _TuiWorker.spawn(argv or fake_command(), env=full_env)
+    kwargs: dict[str, Any] = {}
+    if ready_timeout is not None:
+        kwargs["ready_timeout"] = ready_timeout
+    return await _TuiWorker.spawn(argv or fake_command(), env=full_env, **kwargs)
 
 
 async def assert_process_group_gone(pgid: int) -> None:
@@ -191,6 +195,31 @@ async def scenario() -> None:
     run(scenario())
 
 
+def test_tui_worker_ready_timeout_shorter_than_startup_fails_fast() -> None:
+    async def scenario() -> None:
+        started = time.monotonic()
+        with pytest.raises(WorkerStartError, match="did not become ready"):
+            await spawn_fake(env={"FAKE_TUI_SLOW_START": "10"}, ready_timeout=0.5)
+
+        assert time.monotonic() - started < 5.0
+        await assert_no_tui_reader_threads()
+
+    run(scenario())
+
+
+def test_tui_worker_ready_timeout_covers_slow_startup() -> None:
+    async def scenario() -> None:
+        worker = await spawn_fake(env={"FAKE_TUI_SLOW_START": "1"}, ready_timeout=15.0)
+        try:
+            result_message, _rate_limit = await worker.ask("slow-but-ready", timeout=5.0)
+
+            assert result_message["result"] == "slow-but-ready"
+        finally:
+            await worker.retire()
+
+    run(scenario())
+
+
 def test_tui_worker_startup_exit_raises_worker_start_error() -> None:
     async def scenario() -> None:
         with pytest.raises(WorkerStartError) as raised: