From b873b7e4d2173096004f8f295be0f71653067acf Mon Sep 17 00:00:00 2001 From: Kamran Mammadov Date: Tue, 11 Aug 2026 18:27:47 +0700 Subject: [PATCH] fix: close storage when application creation fails A ConfigurationError raised inside create_application() left the aiosqlite connection pool running. Its connections are non-daemon threads, so the subsequent sys.exit(1) blocked in wait_for_thread_shutdown() and the process hung at 0% CPU instead of exiting. Supervisors read that as a healthy service: launchd reports "running" with no exit code, so KeepAlive never fires, and systemd Restart= behaves the same way. Guard component creation so storage is closed before the exception propagates. This covers every failure between pool startup and run_application(), which owns the only other close() call. Add regression tests asserting storage is closed and that no non-daemon threads survive a failed startup. Fixes #211 Co-Authored-By: Claude Opus 5 --- src/main.py | 196 +++++++++++++++++++++------------------- tests/unit/test_main.py | 67 ++++++++++++++ 2 files changed, 171 insertions(+), 92 deletions(-) create mode 100644 tests/unit/test_main.py diff --git a/src/main.py b/src/main.py index 02660733..5600da33 100644 --- a/src/main.py +++ b/src/main.py @@ -103,105 +103,117 @@ async def create_application(config: Settings) -> Dict[str, Any]: storage = Storage(config.database_url) await storage.initialize() - # Create security components - providers = [] - - # Add whitelist provider if users are configured - if config.allowed_users: - providers.append(WhitelistAuthProvider(config.allowed_users)) - - # Add token provider if enabled - if config.enable_token_auth: - token_storage = InMemoryTokenStorage() # TODO: Use database storage - providers.append(TokenAuthProvider(config.auth_token_secret, token_storage)) - - # Fall back to allowing all users in development mode - if not providers and config.development_mode: - logger.warning( - "No auth providers configured" - " - creating development-only allow-all provider" + try: + # Create security components + providers = [] + + # Add whitelist provider if users are configured + if config.allowed_users: + providers.append(WhitelistAuthProvider(config.allowed_users)) + + # Add token provider if enabled + if config.enable_token_auth: + token_storage = InMemoryTokenStorage() # TODO: Use database storage + providers.append(TokenAuthProvider(config.auth_token_secret, token_storage)) + + # Fall back to allowing all users in development mode + if not providers and config.development_mode: + logger.warning( + "No auth providers configured" + " - creating development-only allow-all provider" + ) + providers.append(WhitelistAuthProvider([], allow_all_dev=True)) + elif not providers: + raise ConfigurationError("No authentication providers configured") + + auth_manager = AuthenticationManager(providers) + security_validator = SecurityValidator( + config.approved_directory, + disable_security_patterns=config.disable_security_patterns, ) - providers.append(WhitelistAuthProvider([], allow_all_dev=True)) - elif not providers: - raise ConfigurationError("No authentication providers configured") - - auth_manager = AuthenticationManager(providers) - security_validator = SecurityValidator( - config.approved_directory, - disable_security_patterns=config.disable_security_patterns, - ) - rate_limiter = RateLimiter(config) + rate_limiter = RateLimiter(config) - # Create audit storage and logger - audit_storage = InMemoryAuditStorage() # TODO: Use database storage in production - audit_logger = AuditLogger(audit_storage) + # Create audit storage and logger + # TODO: Use database storage in production + audit_storage = InMemoryAuditStorage() + audit_logger = AuditLogger(audit_storage) - # Create Claude integration components with persistent storage - session_storage = SQLiteSessionStorage(storage.db_manager) - session_manager = SessionManager(config, session_storage) + # Create Claude integration components with persistent storage + session_storage = SQLiteSessionStorage(storage.db_manager) + session_manager = SessionManager(config, session_storage) - # Create Claude SDK manager and integration facade - logger.info("Using Claude Python SDK integration") - sdk_manager = ClaudeSDKManager(config, security_validator=security_validator) + # Create Claude SDK manager and integration facade + logger.info("Using Claude Python SDK integration") + sdk_manager = ClaudeSDKManager(config, security_validator=security_validator) - claude_integration = ClaudeIntegration( - config=config, - sdk_manager=sdk_manager, - session_manager=session_manager, - ) + claude_integration = ClaudeIntegration( + config=config, + sdk_manager=sdk_manager, + session_manager=session_manager, + ) - # --- Event bus and agentic platform components --- - event_bus = EventBus() + # --- Event bus and agentic platform components --- + event_bus = EventBus() - # Event security middleware - event_security = EventSecurityMiddleware( - event_bus=event_bus, - security_validator=security_validator, - auth_manager=auth_manager, - ) - event_security.register() - - # Agent handler — translates events into Claude executions - agent_handler = AgentHandler( - event_bus=event_bus, - claude_integration=claude_integration, - default_working_directory=config.approved_directory, - default_user_id=config.allowed_users[0] if config.allowed_users else 0, - ) - agent_handler.register() - - # Create bot with all dependencies - dependencies = { - "auth_manager": auth_manager, - "security_validator": security_validator, - "rate_limiter": rate_limiter, - "audit_logger": audit_logger, - "claude_integration": claude_integration, - "storage": storage, - "event_bus": event_bus, - "project_registry": None, - "project_threads_manager": None, - } - - bot = ClaudeCodeBot(config, dependencies) - - # Notification service and scheduler need the bot's Telegram Bot instance, - # which is only available after bot.initialize(). We store placeholders - # and wire them up in run_application() after initialization. - - logger.info("Application components created successfully") - - return { - "bot": bot, - "claude_integration": claude_integration, - "storage": storage, - "config": config, - "features": features, - "event_bus": event_bus, - "agent_handler": agent_handler, - "auth_manager": auth_manager, - "security_validator": security_validator, - } + # Event security middleware + event_security = EventSecurityMiddleware( + event_bus=event_bus, + security_validator=security_validator, + auth_manager=auth_manager, + ) + event_security.register() + + # Agent handler — translates events into Claude executions + agent_handler = AgentHandler( + event_bus=event_bus, + claude_integration=claude_integration, + default_working_directory=config.approved_directory, + default_user_id=config.allowed_users[0] if config.allowed_users else 0, + ) + agent_handler.register() + + # Create bot with all dependencies + dependencies = { + "auth_manager": auth_manager, + "security_validator": security_validator, + "rate_limiter": rate_limiter, + "audit_logger": audit_logger, + "claude_integration": claude_integration, + "storage": storage, + "event_bus": event_bus, + "project_registry": None, + "project_threads_manager": None, + } + + bot = ClaudeCodeBot(config, dependencies) + + # Notification service and scheduler need the bot's Telegram Bot instance, + # which is only available after bot.initialize(). We store placeholders + # and wire them up in run_application() after initialization. + + logger.info("Application components created successfully") + + return { + "bot": bot, + "claude_integration": claude_integration, + "storage": storage, + "config": config, + "features": features, + "event_bus": event_bus, + "agent_handler": agent_handler, + "auth_manager": auth_manager, + "security_validator": security_validator, + } + except BaseException: + # Storage has already started aiosqlite's connection pool, whose + # threads are non-daemon. run_application() owns the only other + # close() call, and it never runs if we fail here, so those threads + # would keep the interpreter alive: sys.exit() then blocks forever + # in wait_for_thread_shutdown() and the process hangs instead of + # exiting. Supervisors read that as a healthy service. + logger.debug("Closing storage after failed application creation") + await storage.close() + raise async def run_application(app: Dict[str, Any]) -> None: diff --git a/tests/unit/test_main.py b/tests/unit/test_main.py new file mode 100644 index 00000000..7d2aca93 --- /dev/null +++ b/tests/unit/test_main.py @@ -0,0 +1,67 @@ +"""Tests for application startup in src.main.""" + +import asyncio +import threading +import time +from pathlib import Path +from unittest.mock import AsyncMock, patch + +import pytest + +from src.config import create_test_config +from src.exceptions import ConfigurationError +from src.main import create_application + + +def _config_without_auth_providers(tmp_path: Path): + """Config that reaches storage init and then fails auth validation.""" + return create_test_config( + database_url=f"sqlite:///{tmp_path / 'test.db'}", + approved_directory=str(tmp_path), + allowed_users=[], + enable_token_auth=False, + development_mode=False, + ) + + +async def test_create_application_closes_storage_on_configuration_error(tmp_path): + """Storage must be closed when component creation fails.""" + config = _config_without_auth_providers(tmp_path) + storage = AsyncMock() + + with patch("src.main.Storage", return_value=storage): + with pytest.raises(ConfigurationError): + await create_application(config) + + storage.initialize.assert_awaited_once() + storage.close.assert_awaited_once() + + +async def test_create_application_leaves_no_live_database_threads(tmp_path): + """Regression: a startup failure must not leave the pool's threads running. + + aiosqlite connections are non-daemon threads, so any left alive block + interpreter shutdown. sys.exit() then hangs in wait_for_thread_shutdown() + and the process never exits, which reads as "healthy" to a supervisor. + """ + config = _config_without_auth_providers(tmp_path) + before = {thread.ident for thread in threading.enumerate()} + + with pytest.raises(ConfigurationError): + await create_application(config) + + # close() signals the worker threads rather than joining them, so give + # them a moment to actually finish. + leaked = [] + deadline = time.monotonic() + 5.0 + while time.monotonic() < deadline: + leaked = [ + thread + for thread in threading.enumerate() + if thread.ident not in before and thread.is_alive() and not thread.daemon + ] + if not leaked: + break + await asyncio.sleep(0.05) + + assert not leaked, f"non-daemon threads still alive after failure: {leaked}"