diff --git a/CHANGELOG.md b/CHANGELOG.md index 6dc644f..3c39366 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/PROTOCOL-TUI.md b/PROTOCOL-TUI.md index e6b0f05..f86bd76 100644 --- a/PROTOCOL-TUI.md +++ b/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 diff --git a/README.md b/README.md index f7c2e88..063fe9c 100644 --- a/README.md +++ b/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 diff --git a/claude_pool.py b/claude_pool.py index 616ccc2..a06a260 100644 --- a/claude_pool.py +++ b/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,6 +125,11 @@ 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 @@ -125,7 +137,9 @@ def __init__(self, process: asyncio.subprocess.Process) -> None: _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() diff --git a/pyproject.toml b/pyproject.toml index 068a717..b631cd6 100644 --- a/pyproject.toml +++ b/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" diff --git a/tests/fake_claude.py b/tests/fake_claude.py index da14f41..994bda4 100755 --- a/tests/fake_claude.py +++ b/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: diff --git a/tests/fake_claude_tui.py b/tests/fake_claude_tui.py index 3349fa8..3fd5ba1 100644 --- a/tests/fake_claude_tui.py +++ b/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?") diff --git a/tests/test_pool.py b/tests/test_pool.py index 85d34c8..9325b6f 100644 --- a/tests/test_pool.py +++ b/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)) diff --git a/tests/test_tui_worker.py b/tests/test_tui_worker.py index 7b159b4..ab99c44 100644 --- a/tests/test_tui_worker.py +++ b/tests/test_tui_worker.py @@ -15,6 +15,7 @@ from claude_pool import ( AskTimeout, Result, + WorkerAuthError, WorkerCrashError, WorkerStartError, _LIVE_PGIDS, @@ -232,10 +233,37 @@ async def scenario() -> None: def test_tui_worker_startup_auth_error_carries_login_tail() -> None: async def scenario() -> None: - with pytest.raises(WorkerStartError) as raised: + with pytest.raises(WorkerAuthError) as raised: await spawn_fake(env={"FAKE_TUI_STARTUP": "autherr"}) assert "Please run /login" in raised.value.stderr_tail + assert raised.value.marker == "invalid api key" + + run(scenario()) + + +def test_tui_worker_auth_output_fails_before_ready_timeout() -> None: + async def scenario() -> None: + started = time.monotonic() + with pytest.raises(WorkerAuthError) as raised: + await spawn_fake(env={"FAKE_TUI_STARTUP": "authstall"}, ready_timeout=10.0) + + assert raised.value.marker == "failed to authenticate" + assert "OAuth session expired" in raised.value.stdout_tail + assert time.monotonic() - started < 2.0 + await assert_no_tui_reader_threads() + + run(scenario()) + + +def test_tui_worker_rate_limit_output_does_not_trigger_auth_error() -> None: + async def scenario() -> None: + worker = await spawn_fake(env={"FAKE_TUI_STARTUP": "ratelimit"}) + try: + result_message, _rate_limit = await worker.ask("healthy", timeout=5.0) + assert result_message["result"] == "healthy" + finally: + await worker.retire() run(scenario()) diff --git a/tests/test_worker.py b/tests/test_worker.py index d121412..1f76af1 100644 --- a/tests/test_worker.py +++ b/tests/test_worker.py @@ -11,7 +11,7 @@ import pytest -from claude_pool import AskTimeout, Result, WorkerCrashError, _TailBuffer, _Worker +from claude_pool import AskTimeout, Result, WorkerAuthError, WorkerCrashError, _TailBuffer, _Worker ROOT = Path(__file__).resolve().parents[1] @@ -218,10 +218,41 @@ async def scenario() -> None: try: try: await worker.ask("hello", timeout=5.0) - except WorkerCrashError as exc: + except WorkerAuthError as exc: assert "Please run /login" in exc.stderr_tail + assert exc.marker == "invalid api key" + assert exc.stdout_tail == "" else: - raise AssertionError("expected WorkerCrashError") + raise AssertionError("expected WorkerAuthError") + finally: + await worker.kill() + + run(scenario()) + + +def test_worker_auth_stderr_interrupts_blocked_spawn_prompt() -> None: + async def scenario() -> None: + worker = await spawn_fake(env={"FAKE_CLAUDE_STARTUP": "authstall"}) + started = time.monotonic() + try: + with pytest.raises(WorkerAuthError) as raised: + await worker.ask("hello", timeout=10.0) + + assert raised.value.marker == "failed to authenticate" + assert "OAuth session expired" in raised.value.stderr_tail + assert time.monotonic() - started < 2.0 + finally: + await worker.kill() + + run(scenario()) + + +def test_worker_rate_limit_stderr_does_not_trigger_auth_error() -> None: + async def scenario() -> None: + worker = await spawn_fake(env={"FAKE_CLAUDE_STARTUP": "ratelimitstall"}) + try: + with pytest.raises(AskTimeout): + await worker.ask("hello", timeout=0.3) finally: await worker.kill()