from __future__ import annotations
import tempfile
import typing as t
from contextlib import suppress
from pathlib import Path
from session_buddy.di.container import depends
from .config import SessionPaths
from .constants import CLAUDE_DIR_KEY, COMMANDS_DIR_KEY, LOGS_DIR_KEY
_configured = False
def get_sync_typed[T](key: type[T]) -> T:
"""Type-safe wrapper for depends.get_sync.
This helper provides proper type information for the dependency injection
container, avoiding 'no-any-return' type checker errors.
Args:
key: The class type to retrieve from the DI container
Returns:
The instance of type T from the DI container
Example:
>>> from session_buddy.core.session_manager import SessionLifecycleManager
>>> manager = get_sync_typed(SessionLifecycleManager) # Properly typed
"""
result: t.Any = depends.get_sync(key)
# Type assertion: we trust the DI container to return the correct type
return t.cast(T, result) # Use PEP 695 type param, not string literal
def configure(*, force: bool = False) -> None:
"""Register default dependencies for the session-buddy MCP stack.
This function sets up the dependency injection container with type-safe
configuration and singleton instances for the session management system.
Args:
force: If True, re-registers all dependencies even if already configured.
Used primarily for testing to reset singleton state.
Example:
>>> from session_buddy.di import configure
>>> configure() # First call registers dependencies
>>> configure() # Subsequent calls are no-ops unless force=True
>>> configure(force=True) # Re-registers all dependencies
"""
global _configured
if _configured and not force:
return
# Register type-safe path configuration
paths = SessionPaths.from_home()
paths.ensure_directories()
depends.set(SessionPaths, paths)
# Register services with type-safe path access
_register_logger(paths.logs_dir, force)
_register_session_logger(paths.logs_dir, force) # Register SessionLogger
_register_permissions_manager(paths.claude_dir, force)
_register_quality_scorer(force) # Register quality scorer BEFORE lifecycle manager
_register_code_formatter(force) # Register code formatter BEFORE hooks manager
_register_workflow_metrics_engine(force) # Register workflow metrics engine
_register_lifecycle_manager(force)
_register_hooks_manager(force) # Register HooksManager for automation
_configured = True
def reset() -> None:
"""Reset dependencies to defaults."""
# Reset singleton instances that have class-level state
with suppress(ImportError, AttributeError):
from session_buddy.core.permissions import SessionPermissionsManager
SessionPermissionsManager.reset_singleton()
configure(force=True)
def _register_logger(logs_dir: Path, force: bool) -> None:
"""Register ACB logger adapter with the given logs directory.
Args:
logs_dir: Directory for session log files
force: If True, re-registers even if already registered
Note:
This function configures ACB's logger adapter but does NOT register it
in the DI container to avoid type resolution conflicts. Components
should use direct logging imports instead of DI lookup.
"""
# Skip registration entirely - we don't need to register the logger in DI
# Components should use `import logging; logger = logging.getLogger(__name__)`
# This avoids DI type confusion when crackerjack runs
def _register_session_logger(logs_dir: Path, force: bool) -> None:
"""Register SessionLogger with the given logs directory.
Args:
logs_dir: Directory for session log files
force: If True, re-registers even if already registered
"""
from session_buddy.utils.logging import SessionLogger
if not force:
with suppress(Exception):
existing = depends.get_sync(SessionLogger)
if isinstance(existing, SessionLogger):
return
# Create SessionLogger instance with fallback to temp logs if needed
try:
session_logger = SessionLogger(logs_dir)
except Exception:
tmp_logs = Path(tempfile.gettempdir()) / "session-buddy" / "logs"
tmp_logs.mkdir(parents=True, exist_ok=True)
depends.set(LOGS_DIR_KEY, tmp_logs)
session_logger = SessionLogger(tmp_logs)
depends.set(SessionLogger, session_logger)
def _register_permissions_manager(claude_dir: Path, force: bool) -> None:
"""Register SessionPermissionsManager with the given Claude directory.
Args:
claude_dir: Root Claude directory for session data
force: If True, re-registers even if already registered
Note:
Accepts Path directly instead of resolving from string keys,
following ACB's type-based dependency injection pattern.
"""
from session_buddy.core.permissions import SessionPermissionsManager
if not force:
with suppress(Exception): # Catch all DI resolution errors
existing = depends.get_sync(SessionPermissionsManager)
if isinstance(existing, SessionPermissionsManager):
return
# Create and register permissions manager instance
permissions_manager = SessionPermissionsManager(claude_dir)
depends.set(SessionPermissionsManager, permissions_manager)
def _register_quality_scorer(force: bool) -> None:
"""Register QualityScorer with the DI container.
Args:
force: If True, re-registers even if already registered
Note:
The QualityScorer interface breaks the circular dependency between
session_manager.py (core) and server.py (MCP). The MCP layer provides
the concrete implementation (MCPQualityScorer) while the core layer
depends only on the abstraction.
"""
from session_buddy.core.quality_scoring import QualityScorer
if not force:
with suppress(Exception):
existing = depends.get_sync(QualityScorer)
if existing is not None:
return
# Create and register MCP quality scorer implementation
# Import here to avoid circular dependency
try:
from session_buddy.mcp.quality_scorer import MCPQualityScorer
quality_scorer = MCPQualityScorer()
except ImportError:
# Fallback to default implementation if MCP layer not available
from session_buddy.core.quality_scoring import DefaultQualityScorer
quality_scorer = DefaultQualityScorer()
depends.set(QualityScorer, quality_scorer)
def _register_code_formatter(force: bool) -> None:
"""Register CodeFormatter with the DI container.
Args:
force: If True, re-registers even if already registered
Note:
The CodeFormatter interface breaks the circular dependency between
hooks.py (core) and server.py (MCP layer). The MCP layer provides
the concrete implementation (MCPCodeFormatter) while the core layer
depends only on the abstraction.
"""
from session_buddy.core.hooks import CodeFormatter
if not force:
with suppress(Exception):
existing = depends.get_sync(CodeFormatter)
if existing is not None:
return
# Create and register MCP code formatter implementation
# Import here to avoid circular dependency
try:
from session_buddy.mcp.code_formatter import MCPCodeFormatter
code_formatter = MCPCodeFormatter()
except ImportError:
from session_buddy.core.hooks import DefaultCodeFormatter
code_formatter = DefaultCodeFormatter()
depends.set(CodeFormatter, code_formatter)
def _register_workflow_metrics_engine(force: bool) -> None:
"""Register WorkflowMetricsEngine with the DI container.
Args:
force: If True, re-registers even if already registered
Note:
The WorkflowMetricsEngine provides comprehensive workflow analytics
including session velocity, quality trends, and performance metrics.
"""
from session_buddy.core.workflow_metrics import WorkflowMetricsEngine
if not force:
with suppress(Exception):
existing = depends.get_sync(WorkflowMetricsEngine)
if isinstance(existing, WorkflowMetricsEngine):
return
# Create and register workflow metrics engine instance
workflow_metrics_engine = WorkflowMetricsEngine()
depends.set(WorkflowMetricsEngine, workflow_metrics_engine)
def _register_lifecycle_manager(force: bool) -> None:
"""Register SessionLifecycleManager with the DI container.
Args:
force: If True, re-registers even if already registered
"""
from session_buddy.core.quality_scoring import QualityScorer
from session_buddy.core.session_manager import SessionLifecycleManager
if not force:
with suppress(Exception): # Catch all DI resolution errors
existing = depends.get_sync(SessionLifecycleManager)
if isinstance(existing, SessionLifecycleManager):
return
# Get quality scorer from DI (must be registered first)
quality_scorer = depends.get_sync(QualityScorer)
# Create and register lifecycle manager instance with injected scorer
lifecycle_manager = SessionLifecycleManager(quality_scorer=quality_scorer)
depends.set(SessionLifecycleManager, lifecycle_manager)
def _register_hooks_manager(force: bool) -> None:
"""Register HooksManager with the DI container.
Args:
force: If True, re-registers even if already registered
Note:
The HooksManager provides automation capabilities through priority-based
async hook execution with error handling and causal chain tracking.
"""
from session_buddy.core.hooks import HooksManager
if not force:
with suppress(Exception): # Catch all DI resolution errors
existing = depends.get_sync(HooksManager)
if isinstance(existing, HooksManager):
return
# Create and register hooks manager instance
from session_buddy.core.hooks import CodeFormatter
# Get code formatter from DI (must be registered first)
code_formatter = depends.get_sync(CodeFormatter)
# Create and register hooks manager instance with injected formatter
hooks_manager = HooksManager(formatter=code_formatter)
depends.set(HooksManager, hooks_manager)
__all__ = [
# Legacy string keys (deprecated - use SessionPaths instead)
"CLAUDE_DIR_KEY",
"COMMANDS_DIR_KEY",
"LOGS_DIR_KEY",
"SessionPaths",
"configure",
"depends",
"get_sync_typed",
"reset",
]
