diff --git a/README.md b/README.md index ce2c37f..42f62cd 100644 --- a/README.md +++ b/README.md @@ -85,6 +85,33 @@ curl -X POST http://localhost:8000/v1/chat/completions \ }' ``` +## Error Responses + +Errors use the OpenAI error shape: `{"error": {"message", "type", "code"}}`. + +| Status | `error.code` | Meaning | +| --- | --- | --- | +| 400 | `model_not_supported` | Claude rejected every attempted model. | +| 409 | `session_busy` | The session already has an active Claude process. | +| 429 | `usage_limit_reached` | The Claude subscription usage limit is reached. | +| 429 | `rate_limit_exceeded` | This gateway's own per-client rate limit was hit. | +| 503 | `claude_unavailable` | Claude Code failed to start for another reason. | + +`usage_limit_reached` is reported separately from `claude_unavailable` so callers can +fall back to another provider instead of retrying. When the CLI reports a reset time, +the response adds a `Retry-After` header and a `reset_at` field: + +```json +{ + "error": { + "message": "Claude subscription usage limit reached.", + "type": "rate_limit_error", + "code": "usage_limit_reached", + "reset_at": "2027-03-20T09:46:40+00:00" + } +} +``` + ## Configuration Common settings are in `claude_code_api/core/config.py`: diff --git a/claude_code_api/api/chat.py b/claude_code_api/api/chat.py index 5f41d99..01642d7 100644 --- a/claude_code_api/api/chat.py +++ b/claude_code_api/api/chat.py @@ -2,6 +2,7 @@ import hashlib import json +from datetime import datetime, timezone from typing import Any, Dict, Optional, Tuple import structlog @@ -12,7 +13,9 @@ from claude_code_api.core.claude_manager import ( ClaudeModelNotSupportedError, ClaudeSessionConflictError, + ClaudeUsageLimitError, create_project_directory, + raise_for_usage_limit, ) from claude_code_api.core.session_manager import SessionManager from claude_code_api.models.claude import get_default_model, validate_claude_model @@ -33,6 +36,8 @@ ) logger = structlog.get_logger() + +MAX_RETRY_AFTER_SECONDS = 86400 router = APIRouter() CHAT_COMPLETION_RESPONSES = { @@ -49,17 +54,25 @@ }, 400: {"model": ErrorResponse}, 422: {"model": ErrorResponse}, + 429: {"model": ErrorResponse}, 503: {"model": ErrorResponse}, 500: {"model": ErrorResponse}, } def _http_error( - status_code: int, message: str, error_type: str, code: str + status_code: int, + message: str, + error_type: str, + code: str, + extra: Optional[Dict[str, Any]] = None, + headers: Optional[Dict[str, str]] = None, ) -> HTTPException: + error: Dict[str, Any] = {"message": message, "type": error_type, "code": code} + if extra: + error.update(extra) return HTTPException( - status_code=status_code, - detail={"error": {"message": message, "type": error_type, "code": code}}, + status_code=status_code, detail={"error": error}, headers=headers ) @@ -116,6 +129,33 @@ def _extract_prompts(request: ChatCompletionRequest) -> Tuple[str, str]: return user_prompt, system_prompt +def _usage_limit_error( + error: ClaudeUsageLimitError, session_id: Optional[str] = None +) -> HTTPException: + logger.warning( + "Claude usage limit reached", + session_id=session_id, + reset_at=error.reset_at.isoformat() if error.reset_at else None, + ) + extra: Dict[str, Any] = {} + headers: Dict[str, str] = {} + if error.reset_at: + extra["reset_at"] = error.reset_at.isoformat() + retry_after = (error.reset_at - datetime.now(timezone.utc)).total_seconds() + if retry_after > 0: + headers["Retry-After"] = str( + min(int(retry_after) + 1, MAX_RETRY_AFTER_SECONDS) + ) + return _http_error( + status.HTTP_429_TOO_MANY_REQUESTS, + "Claude subscription usage limit reached.", + "rate_limit_error", + "usage_limit_reached", + extra=extra or None, + headers=headers or None, + ) + + async def _resolve_session( session_manager: SessionManager, request: ChatCompletionRequest, @@ -147,6 +187,7 @@ async def _collect_non_streaming_response( project_id: str, ) -> Dict[str, Any]: messages, parser = await _gather_claude_messages(claude_process) + raise_for_usage_limit(getattr(claude_process, "terminal_error", None)) _log_message_summary(messages) usage_summary = OpenAIConverter.calculate_usage(parser) @@ -356,6 +397,8 @@ def _register_cli_session(cli_session_id: str): "invalid_request_error", "model_not_supported", ) from e + except ClaudeUsageLimitError as e: + raise _usage_limit_error(e, session_id) from e except Exception as e: logger.error( "Failed to create Claude session", session_id=session_id, error=str(e) @@ -403,6 +446,8 @@ def _register_cli_session(cli_session_id: str): except HTTPException: # Re-raise HTTP exceptions raise + except ClaudeUsageLimitError as e: + raise _usage_limit_error(e, session_id) from e except Exception as e: logger.error( "Unexpected error in chat completion", diff --git a/claude_code_api/core/claude_manager.py b/claude_code_api/core/claude_manager.py index 66330e4..3f2e360 100644 --- a/claude_code_api/core/claude_manager.py +++ b/claude_code_api/core/claude_manager.py @@ -1,10 +1,13 @@ """Claude Code process management.""" import asyncio +import contextlib import json import os +import re import subprocess from collections import deque +from datetime import datetime, timezone from typing import Any, AsyncGenerator, Callable, Dict, List, Optional import structlog @@ -16,6 +19,8 @@ logger = structlog.get_logger() +READER_DRAIN_SECONDS = 1.0 + class ClaudeProcess: """Manages a single Claude Code process.""" @@ -39,6 +44,7 @@ def __init__( self._on_cli_session_id = on_cli_session_id self._on_end = on_end self.last_error: Optional[str] = None + self._result_error: Optional[str] = None self._stderr_tail: deque[str] = deque(maxlen=20) async def start( @@ -49,6 +55,7 @@ async def start( ) -> bool: """Start Claude Code process and wait for completion.""" self.last_error = None + self._result_error = None try: # Prepare real command - using exact format from working Claudia example cmd = [settings.claude_binary_path] @@ -153,6 +160,8 @@ async def _read_output(self): if not data: continue + self._record_output_error(data) + # Extract Claude's session ID from the first message if not claude_session_id and data.get("session_id"): claude_session_id = data["session_id"] @@ -176,6 +185,15 @@ async def _read_output(self): if self._on_end: self._on_end(self) + def _record_output_error(self, data: Dict[str, Any]) -> None: + """Capture errors Claude reports on stdout instead of stderr.""" + if data.get("type") != "result" or not data.get("is_error"): + return + + result_text = data.get("result") + if isinstance(result_text, str) and result_text.strip(): + self._result_error = result_text.strip() + async def _read_error(self): """Read stderr from process.""" try: @@ -192,6 +210,20 @@ async def _read_error(self): except Exception as e: logger.error("Error reading stderr", error=str(e)) + @property + def terminal_error(self) -> Optional[str]: + """Error Claude reported on stdout, if the run ended in one.""" + return self._result_error + + async def _drain_readers(self) -> None: + """Let buffered stdout and stderr be consumed before reporting an error.""" + for task in (self._output_task, self._error_task): + if task and not task.done(): + with contextlib.suppress(asyncio.TimeoutError, asyncio.CancelledError): + await asyncio.wait_for( + asyncio.shield(task), timeout=READER_DRAIN_SECONDS + ) + async def _verify_startup(self) -> bool: """Detect early process failures so API can return actionable errors.""" if not self.process: @@ -203,10 +235,14 @@ async def _verify_startup(self) -> bool: while loop.time() < deadline: return_code = self.process.returncode if return_code is None: + if self._result_error: + break await asyncio.sleep(0.05) continue - if return_code == 0: + await self._drain_readers() + + if return_code == 0 and not self._result_error: return True error_text = self._compose_process_error(return_code) @@ -219,11 +255,24 @@ async def _verify_startup(self) -> bool: ) return False + if self._result_error: + self.last_error = self._result_error + logger.error( + "Claude reported a terminal error", + session_id=self.session_id, + error=self._result_error, + ) + return False + return True def _compose_process_error(self, return_code: int) -> str: - if self._stderr_tail: - return f"Claude exited with code {return_code}: {' | '.join(self._stderr_tail)}" + details: List[str] = [] + if self._result_error: + details.append(self._result_error) + details.extend(self._stderr_tail) + if details: + return f"Claude exited with code {return_code}: {' | '.join(details)}" return f"Claude exited with code {return_code}" async def get_output(self) -> AsyncGenerator[Dict[str, Any], None]: @@ -315,6 +364,22 @@ class ClaudeModelNotSupportedError(ClaudeManagerError): """Raised when Claude rejects a requested model.""" +class ClaudeUsageLimitError(ClaudeManagerError): + """Raised when the Claude subscription usage limit is reached.""" + + def __init__(self, message: str, reset_at: Optional[datetime] = None): + super().__init__(message) + self.reset_at = reset_at + + +def raise_for_usage_limit(error_message: Optional[str]) -> None: + """Raise when Claude reported an exhausted subscription.""" + if error_message and _is_usage_limit_error(error_message): + raise ClaudeUsageLimitError( + error_message, reset_at=_extract_usage_limit_reset(error_message) + ) + + def _is_model_rejection_error(error_message: str) -> bool: lowered = (error_message or "").lower() patterns = ( @@ -328,6 +393,30 @@ def _is_model_rejection_error(error_message: str) -> bool: return any(pattern in lowered for pattern in patterns) +def _is_usage_limit_error(error_message: str) -> bool: + return "usage limit reached" in (error_message or "").lower() + + +def _epoch_to_datetime(value: str) -> Optional[datetime]: + try: + return datetime.fromtimestamp(int(value), tz=timezone.utc) + except (OverflowError, OSError, ValueError): + return None + + +def _extract_usage_limit_reset(error_message: str) -> Optional[datetime]: + message = error_message or "" + marker = message.lower().find("usage limit reached") + if marker >= 0: + message = message[marker:] + + for match in re.finditer(r"(?:resets?_at[=:]\s*|\|)(\d{10})(?!\d)", message): + reset = _epoch_to_datetime(match.group(1)) + if reset: + return reset + return None + + def _resolve_opus_45_fallback(model_id: Optional[str]) -> Optional[str]: if not model_id or not model_id.startswith("claude-opus-4-6-"): return None @@ -473,6 +562,7 @@ async def _start_with_fallback_models( return process last_error = process.last_error or last_error + raise_for_usage_limit(last_error) if not _is_model_rejection_error(last_error): raise ClaudeProcessStartError(last_error) diff --git a/claude_code_api/main.py b/claude_code_api/main.py index 8d5c630..45a6198 100644 --- a/claude_code_api/main.py +++ b/claude_code_api/main.py @@ -125,8 +125,14 @@ def custom_openapi(): async def http_exception_handler(request, exc): """Custom handler for HTTP exceptions to support OpenAI error format.""" if isinstance(exc.detail, dict) and "error" in exc.detail: - return JSONResponse(status_code=exc.status_code, content=exc.detail) - return JSONResponse(status_code=exc.status_code, content={"detail": exc.detail}) + return JSONResponse( + status_code=exc.status_code, content=exc.detail, headers=exc.headers + ) + return JSONResponse( + status_code=exc.status_code, + content={"detail": exc.detail}, + headers=exc.headers, + ) @app.exception_handler(RequestValidationError) diff --git a/claude_code_api/models/openai.py b/claude_code_api/models/openai.py index f538096..966d86a 100644 --- a/claude_code_api/models/openai.py +++ b/claude_code_api/models/openai.py @@ -277,6 +277,9 @@ class ErrorDetail(BaseModel): message: str = Field(..., description="Human-readable error message") type: str = Field(..., description="Error type") code: Optional[str] = Field(None, description="Error code") + reset_at: Optional[str] = Field( + None, description="ISO timestamp when the usage limit resets" + ) class ErrorResponse(BaseModel): diff --git a/tests/test_end_to_end.py b/tests/test_end_to_end.py index 4a43044..5e892d5 100644 --- a/tests/test_end_to_end.py +++ b/tests/test_end_to_end.py @@ -13,6 +13,7 @@ import os import shutil import tempfile +from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Any, Dict, List @@ -23,6 +24,7 @@ from claude_code_api.core.claude_manager import ( ClaudeModelNotSupportedError, ClaudeSessionConflictError, + ClaudeUsageLimitError, ) from claude_code_api.core.config import settings from claude_code_api.core.session_manager import SessionManager @@ -433,6 +435,114 @@ async def fake_create_session(*args, **kwargs): data = response.json() assert data["error"]["code"] == "model_not_supported" + def test_chat_completion_returns_usage_limit_reached(self, client, monkeypatch): + """Return HTTP 429 when the Claude subscription usage limit is reached.""" + claude_manager = client.app.state.claude_manager + reset_at = datetime.now(timezone.utc) + timedelta(hours=2) + + async def fake_create_session(*args, **kwargs): + raise ClaudeUsageLimitError("Usage limit reached", reset_at=reset_at) + + monkeypatch.setattr(claude_manager, "create_session", fake_create_session) + + request_data = { + "model": DEFAULT_MODEL, + "messages": [{"role": "user", "content": "Hi"}], + "stream": False, + } + + response = client.post("/v1/chat/completions", json=request_data) + assert response.status_code == 429 + data = response.json() + assert data["error"]["code"] == "usage_limit_reached" + assert data["error"]["type"] == "rate_limit_error" + assert data["error"]["reset_at"] == reset_at.isoformat() + assert 0 < int(response.headers["retry-after"]) <= 7200 + + def test_streaming_usage_limit_returns_error_before_stream( + self, client, monkeypatch + ): + """Return HTTP 429 for streaming requests instead of opening a broken stream.""" + claude_manager = client.app.state.claude_manager + + async def fake_create_session(*args, **kwargs): + raise ClaudeUsageLimitError("Usage limit reached") + + monkeypatch.setattr(claude_manager, "create_session", fake_create_session) + + request_data = { + "model": DEFAULT_MODEL, + "messages": [{"role": "user", "content": "Hi"}], + "stream": True, + } + + response = client.post("/v1/chat/completions", json=request_data) + assert response.status_code == 429 + assert response.headers["content-type"].startswith("application/json") + assert response.json()["error"]["code"] == "usage_limit_reached" + + def test_usage_limit_omits_retry_after_for_stale_reset(self, client, monkeypatch): + """Do not advertise Retry-After when the reported reset already passed.""" + claude_manager = client.app.state.claude_manager + reset_at = datetime.now(timezone.utc) - timedelta(hours=1) + + async def fake_create_session(*args, **kwargs): + raise ClaudeUsageLimitError("Usage limit reached", reset_at=reset_at) + + monkeypatch.setattr(claude_manager, "create_session", fake_create_session) + + response = client.post( + "/v1/chat/completions", + json={ + "model": DEFAULT_MODEL, + "messages": [{"role": "user", "content": "Hi"}], + "stream": False, + }, + ) + assert response.status_code == 429 + assert "retry-after" not in response.headers + + def test_usage_limit_clamps_far_future_retry_after(self, client, monkeypatch): + """Clamp Retry-After so a bad reset time cannot disable retries.""" + claude_manager = client.app.state.claude_manager + reset_at = datetime.now(timezone.utc) + timedelta(days=400) + + async def fake_create_session(*args, **kwargs): + raise ClaudeUsageLimitError("Usage limit reached", reset_at=reset_at) + + monkeypatch.setattr(claude_manager, "create_session", fake_create_session) + + response = client.post( + "/v1/chat/completions", + json={ + "model": DEFAULT_MODEL, + "messages": [{"role": "user", "content": "Hi"}], + "stream": False, + }, + ) + assert response.status_code == 429 + assert int(response.headers["retry-after"]) == 86400 + + def test_usage_limit_after_startup_maps_to_429(self, client, monkeypatch): + """A limit raised while collecting output must not become a 500.""" + from claude_code_api.api import chat as chat_api + + async def fake_collect(*args, **kwargs): + raise ClaudeUsageLimitError("Usage limit reached") + + monkeypatch.setattr(chat_api, "_collect_non_streaming_response", fake_collect) + + response = client.post( + "/v1/chat/completions", + json={ + "model": DEFAULT_MODEL, + "messages": [{"role": "user", "content": "Hi"}], + "stream": False, + }, + ) + assert response.status_code == 429 + assert response.json()["error"]["code"] == "usage_limit_reached" + class TestConversationFlow: """Test conversation flow and session management.""" diff --git a/tests/test_usage_limit.py b/tests/test_usage_limit.py new file mode 100644 index 0000000..065e5d2 --- /dev/null +++ b/tests/test_usage_limit.py @@ -0,0 +1,189 @@ +"""Tests for Claude subscription usage limit detection.""" + +import asyncio +from datetime import datetime, timezone + +import pytest + +from claude_code_api.core import claude_manager as cm + + +def test_detects_usage_limit_message(): + assert cm._is_usage_limit_error("Usage limit reached") is True + + +def test_detects_usage_limit_with_reset_suffix(): + error = "Claude AI usage limit reached|1774000000" + assert cm._is_usage_limit_error(error) is True + + +def test_ignores_unrelated_errors(): + assert cm._is_usage_limit_error("invalid model") is False + assert cm._is_usage_limit_error("Claude exited with code 1") is False + assert cm._is_usage_limit_error("") is False + + +def test_extracts_reset_time_from_resets_at(): + reset = cm._extract_usage_limit_reset("Usage limit reached resets_at=1774000000") + assert reset == datetime(2026, 3, 20, 9, 46, 40, tzinfo=timezone.utc) + + +def test_extracts_reset_time_from_pipe_suffix(): + reset = cm._extract_usage_limit_reset("Claude AI usage limit reached|1774000000") + assert reset == datetime(2026, 3, 20, 9, 46, 40, tzinfo=timezone.utc) + + +def test_missing_reset_time_returns_none(): + assert cm._extract_usage_limit_reset("Usage limit reached") is None + + +@pytest.mark.asyncio +async def test_create_session_raises_usage_limit_error(monkeypatch, tmp_path): + manager = cm.ClaudeManager() + + async def fake_start(self, prompt, model=None, system_prompt=None): + self.last_error = ( + "Claude exited with code 1: Usage limit reached resets_at=1774000000" + ) + return False + + monkeypatch.setattr(cm.ClaudeProcess, "start", fake_start) + + with pytest.raises(cm.ClaudeUsageLimitError) as exc_info: + await manager.create_session( + session_id="sess-limit", + project_path=str(tmp_path), + prompt="hello", + ) + + assert exc_info.value.reset_at == datetime( + 2026, 3, 20, 9, 46, 40, tzinfo=timezone.utc + ) + + +def test_out_of_range_reset_time_returns_none(): + error = "Usage limit reached resets_at=99999999999999999999" + assert cm._extract_usage_limit_reset(error) is None + + +def test_reset_time_ignores_unrelated_pipe_numbers(): + error = ( + "Claude exited with code 1: trace|1234567890 | " + "Claude AI usage limit reached|1774000000" + ) + assert cm._extract_usage_limit_reset(error) == datetime( + 2026, 3, 20, 9, 46, 40, tzinfo=timezone.utc + ) + + +def test_reset_time_skips_unparsable_candidate(): + error = ( + "Claude exited with code 1: warn|1700000000123 | " + "Claude AI usage limit reached|1774000000" + ) + assert cm._extract_usage_limit_reset(error) == datetime( + 2026, 3, 20, 9, 46, 40, tzinfo=timezone.utc + ) + + +@pytest.mark.asyncio +async def test_result_error_on_stdout_is_captured(tmp_path): + process = cm.ClaudeProcess(session_id="sess", project_path=str(tmp_path)) + process._record_output_error( + { + "type": "result", + "subtype": "success", + "is_error": True, + "result": "Usage limit reached", + } + ) + assert process._compose_process_error(1).endswith("Usage limit reached") + + +@pytest.mark.asyncio +async def test_successful_result_does_not_record_error(tmp_path): + process = cm.ClaudeProcess(session_id="sess", project_path=str(tmp_path)) + process._record_output_error( + {"type": "result", "subtype": "success", "is_error": False, "result": "hello"} + ) + assert process._compose_process_error(1) == "Claude exited with code 1" + + +@pytest.mark.asyncio +async def test_usage_limit_on_stdout_skips_model_fallback(monkeypatch, tmp_path): + manager = cm.ClaudeManager() + attempts = [] + + async def fake_start(self, prompt, model=None, system_prompt=None): + attempts.append(model) + self._record_output_error( + {"type": "result", "is_error": True, "result": "Usage limit reached"} + ) + self.last_error = self._compose_process_error(1) + return False + + monkeypatch.setattr(cm.ClaudeProcess, "start", fake_start) + + with pytest.raises(cm.ClaudeUsageLimitError): + await manager.create_session( + session_id="sess-fallback", + project_path=str(tmp_path), + prompt="hello", + model="claude-opus-4-6-20260205", + ) + + assert len(attempts) == 1 + + +class _ExitedProcess: + def __init__(self, returncode): + self.returncode = returncode + + +@pytest.mark.asyncio +async def test_startup_fails_when_result_reports_error_with_zero_exit(tmp_path): + process = cm.ClaudeProcess(session_id="sess", project_path=str(tmp_path)) + process.process = _ExitedProcess(0) + process._record_output_error( + {"type": "result", "is_error": True, "result": "Usage limit reached"} + ) + + assert await process._verify_startup() is False + assert "Usage limit reached" in process.last_error + + +@pytest.mark.asyncio +async def test_startup_drains_pending_output_before_composing_error(tmp_path): + process = cm.ClaudeProcess(session_id="sess", project_path=str(tmp_path)) + process.process = _ExitedProcess(1) + + async def late_reader(): + await asyncio.sleep(0.2) + process._record_output_error( + {"type": "result", "is_error": True, "result": "Usage limit reached"} + ) + + process._output_task = asyncio.create_task(late_reader()) + + assert await process._verify_startup() is False + assert "Usage limit reached" in process.last_error + + +@pytest.mark.asyncio +async def test_late_usage_limit_is_raised_not_returned_as_content(monkeypatch): + """A limit reported after startup must not surface as a successful reply.""" + from claude_code_api.api import chat as chat_api + from claude_code_api.utils.parser import ClaudeOutputParser + + class _Process: + terminal_error = "Usage limit reached resets_at=1774000000" + + async def fake_gather(_process): + return [], ClaudeOutputParser() + + monkeypatch.setattr(chat_api, "_gather_claude_messages", fake_gather) + + with pytest.raises(cm.ClaudeUsageLimitError): + await chat_api._collect_non_streaming_response( + _Process(), None, "sess", "claude-haiku-4-5-20251001", "proj" + )