""" Command Bus - dispatches commands to their handlers. Commands represent intent to change the system state. Each command has exactly one handler. Supports a middleware pipeline that intercepts command dispatch. Middleware executes in registration order before the handler. Note: Command classes are defined in application.commands to maintain proper layer separation (infrastructure should not define business commands). """ from abc import ABC, abstractmethod from collections.abc import Callable from typing import Any, Protocol, TYPE_CHECKING from mcp_hangar.domain.contracts.command import CommandHandler # noqa: F401 -- re-exported for backward compat from mcp_hangar.application.ports.bus import ICommandBus from mcp_hangar.logging_config import get_logger from mcp_hangar.observability.tracing import get_tracer from mcp_hangar.application.ports.bus import HandlerNotRegisteredError from .caller_rate_limit import charge if TYPE_CHECKING: from ..application.commands import Command from ..domain.security.rate_limiter import RateLimiter logger = get_logger(__name__) class CommandBusMiddleware(ABC): """Middleware that intercepts command dispatch. Middleware can inspect, validate, or reject commands before they reach handlers. """ @abstractmethod def __call__(self, command: "Command", next_handler: Callable[["Command"], Any]) -> Any: """Process the command, optionally calling next_handler to continue. Args: command: The command being dispatched. next_handler: Call this to continue the middleware chain. Returns: The result from the handler (or from rejection). Raises: Any exception to reject the command. """ class CommandBus(ICommandBus): """ Dispatches commands to their registered handlers. Each command type can have exactly one handler. The bus is responsible for routing commands to the appropriate handler. Supports a middleware pipeline executed before handler dispatch. """ def __init__(self): self._handlers: dict[type, CommandHandler] = {} self._middleware: list[CommandBusMiddleware] = [] def register(self, command_type: type, handler: CommandHandler) -> None: """ Register a handler for a command type. Args: command_type: The type of command to handle handler: The handler instance Raises: ValueError: If a handler is already registered for this command type """ if command_type in self._handlers: raise ValueError(f"Handler already registered for {command_type.__name__}") self._handlers[command_type] = handler logger.debug("command_handler_registered", command_type=command_type.__name__) def unregister(self, command_type: type) -> bool: """ Unregister a handler for a command type. Returns: True if handler was removed, False if not found """ if command_type in self._handlers: del self._handlers[command_type] return True return False def add_middleware(self, middleware: CommandBusMiddleware) -> None: """Add middleware to the command bus pipeline. Middleware is executed in registration order before the handler. Args: middleware: The middleware to add. """ self._middleware.append(middleware) logger.debug("command_bus_middleware_added", middleware=type(middleware).__name__) def send(self, command: "Command") -> Any: """ Send a command through middleware pipeline to its handler. Args: command: The command to execute Returns: The result from the handler Raises: ValueError: If no handler is registered for this command type """ command_type = type(command) handler = self._handlers.get(command_type) if handler is None: raise HandlerNotRegisteredError(f"No handler registered for {command_type.__name__}") logger.debug("command_dispatching", command_type=command_type.__name__) tracer = get_tracer(__name__) # Build middleware chain (innermost = handler.handle) def final_handler(cmd: "Command") -> Any: with tracer.start_as_current_span(f"handler.{command_type.__name__}") as h_span: h_span.set_attribute("command.type", command_type.__name__) return handler.handle(cmd) # Wrap in middleware (reverse order so first-registered runs first) chain = final_handler for mw in reversed(self._middleware): def make_step(middleware: CommandBusMiddleware, next_step: Callable) -> Callable: def step(cmd: "Command") -> Any: return middleware(cmd, next_step) return step chain = make_step(mw, chain) return chain(command) def has_handler(self, command_type: type) -> bool: """Check if a handler is registered for the command type.""" return command_type in self._handlers class RefusalLog(Protocol): """What this middleware needs of the security handler: a record per refusal. Declared here rather than imported so the infrastructure layer keeps no dependency on the layers above it; `bootstrap.runtime` passes the real handler in. """ def log_rate_limit_exceeded( self, mcp_server_id: str | None = None, limit: int = 0, window_seconds: int = 0, source_ip: str | None = None, *, scope: str = "", key_kind: str = "", key: str = "", ) -> None: ... class RateLimitMiddleware(CommandBusMiddleware): """Middleware that enforces rate limiting on all commands. Charges each command, before dispatch, to its caller's budget when `rate_limit.per_caller` sets one, and to the budget all callers share. Raises RateLimitExceeded when either is used up (see `caller_rate_limit`). A refusal is recorded by the security handler, once, the way a tool-level one is (#1495). """ def __init__(self, rate_limiter: "RateLimiter", *, security_handler: RefusalLog | None = None): """Initialize with rate limiter. Args: rate_limiter: Rate limiter instance to check against. security_handler: Where a refusal is recorded. None records none, which is what a bus built without one (a test, a bus wired by hand) has always done. """ self._rate_limiter = rate_limiter self._security_handler = security_handler def __call__(self, command: "Command", next_handler: Callable[["Command"], Any]) -> Any: """Charge the command to its caller's budget and the shared one, then dispatch it.""" tracer = get_tracer(__name__) with tracer.start_as_current_span("rate_limit.check") as rl_span: # One budget per command type, shared by every caller, and with # `rate_limit.per_caller` one per caller under it (#1471). key = type(command).__name__ rl_span.set_attribute("rate_limit.key", key) refusal = charge(self._rate_limiter, key) rl_span.set_attribute("rate_limit.allowed", refusal is None) if refusal is not None: rl_span.set_attribute("rate_limit.scope", refusal.scope) if refusal is not None: # Update Prometheus metrics try: from mcp_hangar import metrics as prometheus_metrics prometheus_metrics.RATE_LIMIT_HITS_TOTAL.inc(result="rejected") if hasattr(self._rate_limiter, "get_stats"): stats = self._rate_limiter.get_stats() prometheus_metrics.RATE_LIMIT_ACTIVE_BUCKETS.set(stats.get("active_buckets", 0)) except Exception: # noqa: BLE001 -- fault-barrier: metrics failure must not block rate limit enforcement pass # One record per refusal, with bounded fields only: whose budget was # used up, and the command type it is named after (#1495). No # argument value and no caller's own text. if self._security_handler is not None: try: self._security_handler.log_rate_limit_exceeded( limit=refusal.limit, window_seconds=refusal.window_seconds, scope=refusal.scope, key_kind="command", key=key, ) except Exception: # noqa: BLE001 -- fault-barrier: a logging failure must not block rate limit enforcement logger.debug("rate_limit_refusal_not_recorded", command_type=key) raise refusal # Update allowed metric try: from mcp_hangar import metrics as prometheus_metrics prometheus_metrics.RATE_LIMIT_HITS_TOTAL.inc(result="allowed") if hasattr(self._rate_limiter, "get_stats"): stats = self._rate_limiter.get_stats() prometheus_metrics.RATE_LIMIT_ACTIVE_BUCKETS.set(stats.get("active_buckets", 0)) except Exception: # noqa: BLE001 -- fault-barrier: metrics failure must not block command dispatch pass return next_handler(command) # Global command bus instance _command_bus: CommandBus | None = None def get_command_bus() -> CommandBus: """Get the global command bus instance.""" global _command_bus if _command_bus is None: _command_bus = CommandBus() return _command_bus def reset_command_bus() -> None: """Reset the global command bus (for testing).""" global _command_bus _command_bus = None