feat: fail fast on CLI authentication errors (v0.2.4)

CHANGELOG.md

@@ -4,6 +4,13 @@ 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/).
 
+## [0.2.4] - 2026-07-21
+
+### Added
+
+- Authentication-shaped worker startup failures now fail fast with a typed
+  `WorkerAuthError` carrying the matched marker and bounded output tails.
+
 ## [0.2.3] - 2026-07-19
 
 ### Added

PROTOCOL-TUI.md

@@ -94,8 +94,9 @@ A workspace trust dialog can appear before the prompt. The observed dialog is
 answerable by sending a single carriage return to accept the default.
 
 Authentication failures can surface as TUI startup text followed by process
-exit. The worker maps that to `WorkerStartError` with the drained pty tail so
-callers can match text such as `Invalid API key` or `/login`.
+exit. The worker maps known authentication-failure markers to `WorkerAuthError`
+with the matched marker and drained pty tail, without waiting for the full
+readiness timeout.
 
 Mid-session usage-limit or policy text is ordinary assistant output from the
 TUI. It is returned as normal `Result.text`; callers classify that text the same

README.md

@@ -145,6 +145,11 @@ The daemon accepts the same knobs:
 claude-pool serve --backend tui --warm 2 --tui-ready-timeout 75 --spawn-concurrency 2
 ```
 
+## Authentication Failures
+
+Authentication failures fail fast with `WorkerAuthError`. Headless deployments should prefer
+`CLAUDE_CODE_OAUTH_TOKEN` via the `env=` parameter.
+
 ## How It Works
 
 ```text
@@ -212,11 +217,13 @@ output. Claude Code can put error text in the same field as successful output.
 | `ClaudePoolError` | Base class for pool errors and closed sessions. |
 | `PoolClosed` | Work is requested after a pool has closed. |
 | `WorkerStartError` | A worker process cannot be started. |
+| `WorkerAuthError` | Worker startup output matches a known authentication-failure marker. |
 | `WorkerCrashError` | A worker exits before producing a result. |
 | `AskTimeout` | A prompt exceeds its timeout and the worker is killed. |
 
 `WorkerStartError` and `WorkerCrashError` expose `stderr_tail`, a bounded tail
-of worker stderr.
+of worker stderr. `WorkerAuthError` also exposes the matched `marker` and a bounded
+`stdout_tail`.
 
 ## Supported Platforms
 

claude_pool.py

@@ -40,6 +40,13 @@
 _TUI_PASTE_TO_CR = 0.3
 _TUI_CR_RETRY_AFTER = 3.0
 _TUI_READY_TIMEOUT = 30.0
+AUTH_FAILURE_MARKERS = (
+    "failed to authenticate",
+    "oauth session expired",
+    "invalid api key",
+    "please run /login",
+    "not logged in",
+)
 _LIVE_PGIDS: set[int] = set()
 logger = logging.getLogger("claude_pool")
 
@@ -118,14 +125,21 @@ def text(self) -> str:
         return self._data.decode(errors="replace")
 
 
+def _auth_failure_marker(*tails: str) -> str | None:
+    lowered = "\n".join(tails).lower()
+    return next((marker for marker in AUTH_FAILURE_MARKERS if marker.lower() in lowered), None)
+
+
 class _Worker:
     def __init__(self, process: asyncio.subprocess.Process) -> None:
         self.process = process
         self._pgid = process.pid
         _LIVE_PGIDS.add(self._pgid)
         self.spawned_at = time.monotonic()
         self.idle_since = self.spawned_at
+        self._stdout = _TailBuffer(_STDERR_LIMIT)
         self._stderr = _TailBuffer(_STDERR_LIMIT)
+        self._auth_failure = asyncio.Event()
         self._stderr_task = asyncio.create_task(self._drain_stderr())
         self._ask_lock = asyncio.Lock()
         self._killed = False
@@ -153,6 +167,10 @@ async def spawn(
     def stderr_tail(self) -> str:
         return self._stderr.text()
 
+    @property
+    def stdout_tail(self) -> str:
+        return self._stdout.text()
+
     @property
     def alive(self) -> bool:
         return self.process.returncode is None
@@ -212,7 +230,9 @@ def _signal_process_group(self) -> None:
 
     async def _ask(self, prompt: str) -> tuple[dict[str, Any], dict[str, Any] | None]:
         if self.process.stdin is None or self.process.stdout is None:
-            raise WorkerCrashError("worker pipes are unavailable", stderr_tail=self.stderr_tail)
+            raise self._crash_error("worker pipes are unavailable")
+
+        await self._raise_auth_failure_if_present()
 
         payload = {
             "type": "user",
@@ -228,14 +248,12 @@ async def _ask(self, prompt: str) -> tuple[dict[str, Any], dict[str, Any] | None
         except (BrokenPipeError, ConnectionResetError) as exc:
             await self._wait_after_crash()
             await self._finish_stderr_task(cancel=False)
-            raise WorkerCrashError(
-                "worker exited before accepting input", self.stderr_tail
-            ) from exc
+            raise self._crash_error("worker exited before accepting input") from exc
 
         rate_limit: dict[str, Any] | None = None
         while True:
             try:
-                raw = await self.process.stdout.readline()
+                raw = await self._read_stdout_line()
             except (ValueError, asyncio.LimitOverrunError) as exc:
                 await self.kill()
                 raise WorkerCrashError(
@@ -244,11 +262,14 @@ async def _ask(self, prompt: str) -> tuple[dict[str, Any], dict[str, Any] | None
             if raw == b"":
                 await self._wait_after_crash()
                 await self._finish_stderr_task(cancel=False)
-                raise WorkerCrashError("worker exited before result", stderr_tail=self.stderr_tail)
+                raise self._crash_error("worker exited before result")
+
+            self._stdout.append(raw)
 
             try:
                 message = json.loads(raw.decode(errors="replace"))
             except json.JSONDecodeError:
+                await self._raise_auth_failure_if_present()
                 continue
             if not isinstance(message, dict):
                 continue
@@ -268,6 +289,54 @@ async def _drain_stderr(self) -> None:
             if not chunk:
                 return
             self._stderr.append(chunk)
+            if _auth_failure_marker(self.stderr_tail) is not None:
+                self._auth_failure.set()
+
+    async def _read_stdout_line(self) -> bytes:
+        if self.process.stdout is None:
+            return b""
+        reader = asyncio.create_task(self.process.stdout.readline())
+        auth_waiter = asyncio.create_task(self._auth_failure.wait())
+        try:
+            done, _pending = await asyncio.wait(
+                {reader, auth_waiter}, return_when=asyncio.FIRST_COMPLETED
+            )
+            if auth_waiter in done and self._auth_failure.is_set():
+                await self._raise_auth_failure_if_present()
+            return await reader
+        finally:
+            for task in (reader, auth_waiter):
+                if not task.done():
+                    task.cancel()
+                with suppress(asyncio.CancelledError):
+                    await task
+
+    async def _raise_auth_failure_if_present(self) -> None:
+        stderr_tail = self.stderr_tail
+        stdout_tail = self.stdout_tail
+        marker = _auth_failure_marker(stderr_tail, stdout_tail)
+        if marker is None:
+            return
+        await self.kill()
+        raise WorkerAuthError(
+            "worker authentication failed",
+            marker=marker,
+            stderr_tail=stderr_tail,
+            stdout_tail=stdout_tail,
+        )
+
+    def _crash_error(self, message: str) -> WorkerCrashError | WorkerAuthError:
+        stderr_tail = self.stderr_tail
+        stdout_tail = self.stdout_tail
+        marker = _auth_failure_marker(stderr_tail, stdout_tail)
+        if marker is not None:
+            return WorkerAuthError(
+                message,
+                marker=marker,
+                stderr_tail=stderr_tail,
+                stdout_tail=stdout_tail,
+            )
+        return WorkerCrashError(message, stderr_tail=stderr_tail)
 
     async def _close_stdin(self) -> None:
         if self.process.stdin is None or self.process.stdin.is_closing():
@@ -325,6 +394,26 @@ def __init__(self, message: str, stderr_tail: str = "") -> None:
         self.stderr_tail = stderr_tail
 
 
+class WorkerAuthError(WorkerStartError):
+    """Raised when worker startup output indicates an authentication failure.
+
+    ``marker`` is the entry from ``AUTH_FAILURE_MARKERS`` that matched.
+    ``stderr_tail`` and ``stdout_tail`` contain bounded trailing process output.
+    """
+
+    def __init__(
+        self,
+        message: str,
+        *,
+        marker: str,
+        stderr_tail: str = "",
+        stdout_tail: str = "",
+    ) -> None:
+        super().__init__(message, stderr_tail=stderr_tail)
+        self.marker = marker
+        self.stdout_tail = stdout_tail
+
+
 class WorkerCrashError(ClaudePoolError):
     """Raised when a worker exits before producing a result for a turn.
 
@@ -676,10 +765,30 @@ async def _wait_ready(self, timeout: float) -> None:
         deadline = time.monotonic() + timeout
         trust_answered = False
         while True:
+            tail = self.stderr_tail
+            marker = _auth_failure_marker(tail)
+            if marker is not None:
+                await self.kill()
+                raise WorkerAuthError(
+                    "TUI worker authentication failed",
+                    marker=marker,
+                    stderr_tail=tail,
+                    stdout_tail=tail,
+                )
+
             if self.process.returncode is not None:
-                raise WorkerStartError("TUI worker exited during startup", self.stderr_tail)
+                await self.kill()
+                tail = self.stderr_tail
+                marker = _auth_failure_marker(tail)
+                if marker is not None:
+                    raise WorkerAuthError(
+                        "TUI worker authentication failed",
+                        marker=marker,
+                        stderr_tail=tail,
+                        stdout_tail=tail,
+                    )
+                raise WorkerStartError("TUI worker exited during startup", tail)
 
-            tail = self.stderr_tail
             lowered = tail.lower()
             if not trust_answered and "trust" in lowered:
                 with suppress(OSError, AskTimeout):
@@ -874,7 +983,7 @@ async def send(self, prompt: str, timeout: float | None = None) -> Result:
             ask_timeout = self._pool._default_timeout if timeout is None else timeout
             try:
                 result_message, rate_limit = await self._worker.ask(prompt, ask_timeout)
-            except (WorkerCrashError, AskTimeout) as exc:
+            except (WorkerAuthError, WorkerCrashError, AskTimeout) as exc:
                 self._usable = False
                 if self._exited:
                     raise ClaudePoolError("session closed") from exc
@@ -1434,10 +1543,24 @@ async def _replenish_once(self) -> None:
                     worker = await self._spawn_worker()
                     await asyncio.sleep(0.5)
                     if not worker.alive:
-                        error = WorkerStartError(
-                            "warm worker exited during startup",
-                            stderr_tail=worker.stderr_tail,
+                        stderr_tail = worker.stderr_tail
+                        stdout_tail = (
+                            worker.stdout_tail if isinstance(worker, _Worker) else stderr_tail
                         )
+                        marker = _auth_failure_marker(stderr_tail, stdout_tail)
+                        error: WorkerStartError
+                        if marker is not None:
+                            error = WorkerAuthError(
+                                "warm worker authentication failed",
+                                marker=marker,
+                                stderr_tail=stderr_tail,
+                                stdout_tail=stdout_tail,
+                            )
+                        else:
+                            error = WorkerStartError(
+                                "warm worker exited during startup",
+                                stderr_tail=stderr_tail,
+                            )
                         logger.warning("%s", error)
                         self._spawn_cooldown_until = time.monotonic() + 30.0
                         await worker.kill()

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "claude-pool"
-version = "0.2.3"
+version = "0.2.4"
 description = "Warm pooled workers for the Claude Code CLI."
 readme = "README.md"
 license = "MIT"

tests/fake_claude.py

@@ -134,6 +134,14 @@ def _handle_startup() -> None:
         sys.stderr.write("Invalid API key · Please run /login\n")
         sys.stderr.flush()
         raise SystemExit(1)
+    if startup == "authstall":
+        sys.stderr.write("Failed to authenticate: OAuth session expired\n")
+        sys.stderr.flush()
+        time.sleep(30)
+    if startup == "ratelimitstall":
+        sys.stderr.write("Rate limit reached; service overloaded\n")
+        sys.stderr.flush()
+        time.sleep(30)
 
 
 def _handle_prompt(prompt: str, session_id: str, num_turns: int) -> None:

tests/fake_claude_tui.py

@@ -205,6 +205,12 @@ def main() -> int:
     if startup == "autherr":
         write_screen("Invalid API key . Please run /login")
         return 1
+    if startup == "authstall":
+        write_screen("Failed to authenticate: OAuth session expired")
+        while True:
+            time.sleep(60.0)
+    if startup == "ratelimit":
+        write_screen("Rate limit reached; service overloaded")
 
     if os.environ.get("FAKE_TUI_TRUST") == "1":
         write_screen("Do you trust the files in this folder?")

tests/test_pool.py

@@ -10,7 +10,14 @@
 
 import pytest
 
-from claude_pool import ClaudePool, ClaudePoolError, PoolClosed, WorkerCrashError, WorkerStartError
+from claude_pool import (
+    ClaudePool,
+    ClaudePoolError,
+    PoolClosed,
+    WorkerAuthError,
+    WorkerCrashError,
+    WorkerStartError,
+)
 from claude_pool import _LIVE_PGIDS
 
 
@@ -248,6 +255,25 @@ async def scenario() -> None:
     run(scenario())
 
 
+def test_cold_worker_auth_failure_propagates_promptly() -> None:
+    async def scenario() -> None:
+        pool = ClaudePool(
+            **pool_kwargs(env={"FAKE_CLAUDE_STARTUP": "authstall"}, default_timeout=10.0)
+        )
+        started = time.monotonic()
+        try:
+            with pytest.raises(WorkerAuthError) as raised:
+                await pool.ask("auth-failure")
+
+            assert raised.value.marker == "failed to authenticate"
+            assert "OAuth session expired" in raised.value.stderr_tail
+            assert time.monotonic() - started < 2.0
+        finally:
+            await pool.aclose()
+
+    run(scenario())
+
+
 def test_max_workers_serializes_concurrent_asks() -> None:
     async def scenario() -> None:
         pool = ClaudePool(**pool_kwargs(max_workers=1))