mirror of
https://github.com/Alishahryar1/free-claude-code.git
synced 2026-07-10 00:14:16 +00:00
336 lines
13 KiB
Python
336 lines
13 KiB
Python
"""Application runtime composition and lifecycle ownership."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
import os
|
|
from dataclasses import dataclass, field
|
|
from typing import TYPE_CHECKING, Any
|
|
|
|
from fastapi import FastAPI
|
|
from loguru import logger
|
|
|
|
from api.admin_urls import local_admin_url
|
|
from config.env_files import ANTHROPIC_AUTH_TOKEN_ENV, process_env_key_is_effective
|
|
from config.paths import default_claude_workspace_path
|
|
from config.settings import Settings, get_settings
|
|
from providers.exceptions import ServiceUnavailableError
|
|
from providers.runtime import ProviderRuntime
|
|
|
|
if TYPE_CHECKING:
|
|
from cli.managed import ManagedClaudeSessionManager
|
|
from messaging.platforms.ports import MessagingPlatformComponents, MessagingRuntime
|
|
from messaging.session import SessionStore
|
|
from messaging.workflow import MessagingWorkflow
|
|
|
|
_SHUTDOWN_TIMEOUT_S = 5.0
|
|
|
|
|
|
async def best_effort(
|
|
name: str,
|
|
awaitable: Any,
|
|
timeout_s: float = _SHUTDOWN_TIMEOUT_S,
|
|
*,
|
|
log_verbose_errors: bool = False,
|
|
) -> None:
|
|
"""Run a shutdown step with timeout; never raise to callers."""
|
|
try:
|
|
await asyncio.wait_for(awaitable, timeout=timeout_s)
|
|
except TimeoutError:
|
|
logger.warning("Shutdown step timed out: {} ({}s)", name, timeout_s)
|
|
except Exception as e:
|
|
if log_verbose_errors:
|
|
logger.warning(
|
|
"Shutdown step failed: {}: {}: {}",
|
|
name,
|
|
type(e).__name__,
|
|
e,
|
|
)
|
|
else:
|
|
logger.warning(
|
|
"Shutdown step failed: {}: exc_type={}",
|
|
name,
|
|
type(e).__name__,
|
|
)
|
|
|
|
|
|
def warn_if_process_auth_token(settings: Settings) -> None:
|
|
"""Warn when server auth was implicitly inherited from the shell."""
|
|
model_config = getattr(settings, "model_config", Settings.model_config)
|
|
if process_env_key_is_effective(model_config, ANTHROPIC_AUTH_TOKEN_ENV):
|
|
logger.warning(
|
|
"ANTHROPIC_AUTH_TOKEN is set in the process environment but not in "
|
|
"a configured .env file. The proxy will require that token. Add "
|
|
"ANTHROPIC_AUTH_TOKEN= to .env to disable proxy auth, or set the "
|
|
"same token in .env to make server auth explicit."
|
|
)
|
|
|
|
|
|
def log_startup_failure(settings: Settings, exc: Exception) -> None:
|
|
"""Log startup failures without traceback noise unless verbose diagnostics are enabled."""
|
|
message = startup_failure_message(settings, exc)
|
|
logger.error("Startup failed:\n{}", message)
|
|
|
|
|
|
def startup_failure_message(settings: Settings, exc: Exception) -> str:
|
|
"""Return a concise startup failure message for logs and ASGI lifespan failure."""
|
|
if isinstance(exc, ServiceUnavailableError):
|
|
return exc.message.strip() or "Server startup failed."
|
|
|
|
if settings.log_api_error_tracebacks:
|
|
return f"{type(exc).__name__}: {exc}"
|
|
|
|
return f"Server startup failed: exc_type={type(exc).__name__}"
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class AppRuntime:
|
|
"""Own optional messaging, CLI, session, and provider runtime resources."""
|
|
|
|
app: FastAPI
|
|
settings: Settings
|
|
_provider_runtime: ProviderRuntime | None = field(default=None, init=False)
|
|
messaging_runtime: MessagingRuntime | None = None
|
|
messaging_workflow: MessagingWorkflow | None = None
|
|
cli_manager: ManagedClaudeSessionManager | None = None
|
|
|
|
@classmethod
|
|
def for_app(
|
|
cls,
|
|
app: FastAPI,
|
|
settings: Settings | None = None,
|
|
) -> AppRuntime:
|
|
return cls(app=app, settings=settings or get_settings())
|
|
|
|
async def startup(self) -> None:
|
|
logger.info("Starting Claude Code Proxy...")
|
|
admin_url = local_admin_url(self.settings)
|
|
self._provider_runtime = ProviderRuntime(self.settings)
|
|
self.app.state.provider_runtime = self._provider_runtime
|
|
try:
|
|
warn_if_process_auth_token(self.settings)
|
|
await self._validate_configured_models_best_effort()
|
|
self._provider_runtime.start_model_list_refresh()
|
|
await self._start_messaging_if_configured()
|
|
self._publish_state()
|
|
logging.getLogger("uvicorn.error").info(
|
|
"Admin UI: %s (local-only)", admin_url
|
|
)
|
|
except Exception as exc:
|
|
log_startup_failure(self.settings, exc)
|
|
await best_effort(
|
|
"provider_runtime.cleanup",
|
|
self._provider_runtime.cleanup(),
|
|
log_verbose_errors=self.settings.log_api_error_tracebacks,
|
|
)
|
|
raise
|
|
|
|
async def _validate_configured_models_best_effort(self) -> None:
|
|
"""Warm validation status without blocking first-run/admin access."""
|
|
if self._provider_runtime is None:
|
|
return
|
|
try:
|
|
await self._provider_runtime.validate_configured_models()
|
|
except ServiceUnavailableError as exc:
|
|
self.app.state.startup_validation_error = exc.message
|
|
logger.warning(
|
|
"Configured provider model validation failed during startup; "
|
|
"server will continue and requests will fail at provider resolution "
|
|
"when config is incomplete. {}",
|
|
exc.message,
|
|
)
|
|
|
|
async def shutdown(self) -> None:
|
|
verbose = self.settings.log_api_error_tracebacks
|
|
if self.messaging_workflow is not None:
|
|
try:
|
|
self.messaging_workflow.session_store.flush_pending_save()
|
|
except Exception as e:
|
|
if verbose:
|
|
logger.warning("Session store flush on shutdown: {}", e)
|
|
else:
|
|
logger.warning(
|
|
"Session store flush on shutdown: exc_type={}",
|
|
type(e).__name__,
|
|
)
|
|
|
|
logger.info("Shutdown requested, cleaning up...")
|
|
if self.messaging_runtime:
|
|
await best_effort(
|
|
"messaging_runtime.stop",
|
|
self.messaging_runtime.stop(),
|
|
log_verbose_errors=verbose,
|
|
)
|
|
if self.cli_manager:
|
|
await best_effort(
|
|
"cli_manager.stop_all",
|
|
self.cli_manager.stop_all(),
|
|
log_verbose_errors=verbose,
|
|
)
|
|
if self._provider_runtime is not None:
|
|
await best_effort(
|
|
"provider_runtime.cleanup",
|
|
self._provider_runtime.cleanup(),
|
|
log_verbose_errors=verbose,
|
|
)
|
|
await self._shutdown_limiter()
|
|
logger.info("Server shut down cleanly")
|
|
|
|
async def _start_messaging_if_configured(self) -> None:
|
|
try:
|
|
from messaging.platforms.factory import (
|
|
MessagingPlatformOptions,
|
|
create_messaging_components,
|
|
)
|
|
|
|
components = create_messaging_components(
|
|
self.settings.messaging_platform,
|
|
MessagingPlatformOptions(
|
|
telegram_bot_token=self.settings.telegram_bot_token,
|
|
allowed_telegram_user_id=self.settings.allowed_telegram_user_id,
|
|
discord_bot_token=self.settings.discord_bot_token,
|
|
allowed_discord_channels=self.settings.allowed_discord_channels,
|
|
voice_note_enabled=self.settings.voice_note_enabled,
|
|
whisper_model=self.settings.whisper_model,
|
|
whisper_device=self.settings.whisper_device,
|
|
hf_token=self.settings.hf_token,
|
|
nvidia_nim_api_key=self.settings.nvidia_nim_api_key,
|
|
messaging_rate_limit=self.settings.messaging_rate_limit,
|
|
messaging_rate_window=self.settings.messaging_rate_window,
|
|
log_raw_messaging_content=self.settings.log_raw_messaging_content,
|
|
log_api_error_tracebacks=self.settings.log_api_error_tracebacks,
|
|
),
|
|
)
|
|
|
|
if components:
|
|
await self._start_messaging_workflow(components)
|
|
|
|
except ImportError as e:
|
|
if self.settings.log_api_error_tracebacks:
|
|
logger.warning("Messaging module import error: {}", e)
|
|
else:
|
|
logger.warning(
|
|
"Messaging module import error: exc_type={}",
|
|
type(e).__name__,
|
|
)
|
|
except Exception as e:
|
|
if self.settings.log_api_error_tracebacks:
|
|
logger.error("Failed to start messaging platform: {}", e)
|
|
import traceback
|
|
|
|
logger.error(traceback.format_exc())
|
|
else:
|
|
logger.error(
|
|
"Failed to start messaging platform: exc_type={}",
|
|
type(e).__name__,
|
|
)
|
|
|
|
async def _start_messaging_workflow(
|
|
self, components: MessagingPlatformComponents
|
|
) -> None:
|
|
from cli.managed import ManagedClaudeSessionManager
|
|
from messaging.session import SessionStore
|
|
from messaging.workflow import MessagingWorkflow
|
|
|
|
workspace = (
|
|
os.path.abspath(self.settings.allowed_dir)
|
|
if self.settings.allowed_dir
|
|
else os.getcwd()
|
|
)
|
|
os.makedirs(workspace, exist_ok=True)
|
|
|
|
data_path = os.path.abspath(default_claude_workspace_path())
|
|
os.makedirs(data_path, exist_ok=True)
|
|
|
|
api_url = f"http://{self.settings.host}:{self.settings.port}/v1"
|
|
allowed_dirs = [workspace] if self.settings.allowed_dir else []
|
|
plans_dir_abs = os.path.abspath(os.path.join(data_path, "plans"))
|
|
plans_directory = os.path.relpath(plans_dir_abs, workspace)
|
|
self.cli_manager = ManagedClaudeSessionManager(
|
|
workspace_path=workspace,
|
|
api_url=api_url,
|
|
allowed_dirs=allowed_dirs,
|
|
plans_directory=plans_directory,
|
|
auth_token=getattr(self.settings, "anthropic_auth_token", ""),
|
|
log_raw_cli_diagnostics=self.settings.log_raw_cli_diagnostics,
|
|
log_messaging_error_details=self.settings.log_messaging_error_details,
|
|
)
|
|
|
|
session_store = SessionStore(
|
|
storage_path=os.path.join(data_path, "sessions.json"),
|
|
message_log_cap=self.settings.max_message_log_entries_per_chat,
|
|
)
|
|
self.messaging_runtime = components.runtime
|
|
self.messaging_workflow = MessagingWorkflow(
|
|
platform_name=components.name,
|
|
outbound=components.outbound,
|
|
voice_cancellation=components.voice_cancellation,
|
|
cli_manager=self.cli_manager,
|
|
session_store=session_store,
|
|
debug_platform_edits=self.settings.debug_platform_edits,
|
|
debug_subagent_stack=self.settings.debug_subagent_stack,
|
|
log_raw_messaging_content=self.settings.log_raw_messaging_content,
|
|
log_raw_cli_diagnostics=self.settings.log_raw_cli_diagnostics,
|
|
log_messaging_error_details=self.settings.log_messaging_error_details,
|
|
)
|
|
self._restore_tree_state(session_store)
|
|
|
|
components.runtime.on_message(self.messaging_workflow.handle_message)
|
|
await components.runtime.start()
|
|
logger.info("{} platform started with messaging workflow", components.name)
|
|
|
|
def _restore_tree_state(self, session_store: SessionStore) -> None:
|
|
conversation_snapshot = session_store.load_conversation_snapshot()
|
|
if conversation_snapshot.is_empty:
|
|
return
|
|
if self.messaging_workflow is None:
|
|
return
|
|
|
|
logger.info(
|
|
"Restoring {} conversation trees...",
|
|
len(conversation_snapshot.trees),
|
|
)
|
|
from messaging.trees import TreeQueueManager
|
|
|
|
self.messaging_workflow.replace_tree_queue(
|
|
TreeQueueManager.from_snapshot(
|
|
conversation_snapshot,
|
|
queue_update_callback=self.messaging_workflow.update_queue_positions,
|
|
node_started_callback=self.messaging_workflow.mark_node_processing,
|
|
)
|
|
)
|
|
if self.messaging_workflow.tree_queue.cleanup_stale_nodes() > 0:
|
|
session_store.save_conversation_snapshot(
|
|
self.messaging_workflow.tree_queue.snapshot()
|
|
)
|
|
|
|
def _publish_state(self) -> None:
|
|
self.app.state.messaging_runtime = self.messaging_runtime
|
|
self.app.state.messaging_workflow = self.messaging_workflow
|
|
self.app.state.cli_manager = self.cli_manager
|
|
|
|
async def _shutdown_limiter(self) -> None:
|
|
verbose = self.settings.log_api_error_tracebacks
|
|
try:
|
|
from messaging.limiter import MessagingRateLimiter
|
|
except Exception as e:
|
|
if verbose:
|
|
logger.debug(
|
|
"Rate limiter shutdown skipped (import failed): {}: {}",
|
|
type(e).__name__,
|
|
e,
|
|
)
|
|
else:
|
|
logger.debug(
|
|
"Rate limiter shutdown skipped (import failed): exc_type={}",
|
|
type(e).__name__,
|
|
)
|
|
return
|
|
|
|
await best_effort(
|
|
"MessagingRateLimiter.shutdown_instance",
|
|
MessagingRateLimiter.shutdown_instance(),
|
|
timeout_s=2.0,
|
|
log_verbose_errors=verbose,
|
|
)
|