feat: pluggable driver system with lifecycle management, middleware, and event bus

 - BaseDriver accepts DriverContext (DI facade) with backward compat for Bridge
 - DriverManager handles auto-restart with backoff, health monitoring, hot-reload
 - EventBus for lifecycle + message events (bridge.message, driver.status)
 - Middleware chains (pre-receive, pre-send) hooked into Bridge routing
 - Plugin loader: built-in, pip entry points (nextbridge.drivers), local dirs
 - CLI hook registration so drivers can add subcommands without touching main.py
 - Bridge exposes senders_snapshot(), rules_snapshot() for control-plane drivers
 - DB aggregate helpers (stats, recent_mappings, list_user_bindings, session)
 - Opt-in password-protected admin API (/_nextbridge/drivers, /admin/reload)

Author: ntkrnl64

drivers/__init__.py

@@ -1,5 +1,9 @@
+from __future__ import annotations
+
 from abc import ABC, abstractmethod
-from typing import TYPE_CHECKING, Generic, TypeVar
+from dataclasses import dataclass, field
+from enum import Enum, auto
+from typing import TYPE_CHECKING, Any, Generic, TypeVar
 
 from pydantic import BaseModel
 
@@ -8,41 +12,135 @@
 
 if TYPE_CHECKING:
     from services.bridge import Bridge
+    from services.driver_context import DriverContext
 
 T = TypeVar("T", bound=BaseModel)
 
+DRIVER_API_VERSION = 2
+
+
+# ------------------------------------------------------------------
+# Capability & metadata declarations
+# ------------------------------------------------------------------
+
+
+class DriverCapability(Enum):
+    SEND = auto()
+    RECEIVE = auto()
+    WEBHOOK = auto()
+    ATTACHMENTS = auto()
+    MENTIONS = auto()
+    REPLIES = auto()
+
+
+@dataclass
+class DriverMeta:
+    platform: str = ""
+    display_name: str = ""
+    version: str = "0.0.0"
+    api_version: int = DRIVER_API_VERSION
+    capabilities: set[DriverCapability] = field(
+        default_factory=lambda: {DriverCapability.SEND, DriverCapability.RECEIVE}
+    )
+    author: str = ""
+    url: str = ""
+
+
+class DriverHealth(Enum):
+    UNKNOWN = "unknown"
+    STARTING = "starting"
+    HEALTHY = "healthy"
+    DEGRADED = "degraded"
+    UNHEALTHY = "unhealthy"
+    STOPPED = "stopped"
+
+
+# ------------------------------------------------------------------
+# Base driver
+# ------------------------------------------------------------------
+
 
 class BaseDriver(ABC, Generic[T]):
-    """Abstract base class for all platform drivers."""
+    """Abstract base class for all platform drivers.
+
+    The third constructor parameter accepts either a legacy ``Bridge``
+    instance (API v1) or a ``DriverContext`` (API v2).  Existing drivers
+    that pass ``bridge`` continue to work unchanged.
+    """
 
-    def __init__(self, instance_id: str, config: T, bridge: "Bridge"):
+    meta: DriverMeta = DriverMeta()
+
+    def __init__(
+        self,
+        instance_id: str,
+        config: T,
+        ctx_or_bridge: DriverContext | Bridge | Any,
+    ) -> None:
         self.instance_id = instance_id
         self.config: T = config
-        self.bridge = bridge
-        self.http_server = None
 
+        from services.driver_context import DriverContext
+
+        if isinstance(ctx_or_bridge, DriverContext):
+            self._ctx = ctx_or_bridge
+            self.bridge = ctx_or_bridge.bridge
+        else:
+            self._ctx = DriverContext(bridge=ctx_or_bridge)
+            self.bridge = ctx_or_bridge
+
+        self.http_server = None
         self.logger = log.get_logger(f"[{instance_id}]", instance=True)
 
-        # Media download proxy used by downstream attachment fetching.
-        # Default follows driver-level proxy, but can be overridden by
-        # per-driver media_proxy.
         base_proxy = get_proxy(getattr(config, "proxy", UNSET))
         self._media_proxy = get_proxy(getattr(config, "media_proxy", UNSET), base_proxy)
 
+        self._health = DriverHealth.UNKNOWN
+
+    # ------------------------------------------------------------------
+    # Proxy helpers
+    # ------------------------------------------------------------------
+
     def _source_proxy_from_kwargs(self, kwargs: dict) -> str | None:
-        # Keep explicit None from kwargs (disable proxy for this send call).
         if "source_proxy" in kwargs:
             return kwargs.get("source_proxy")
         return self._media_proxy
 
     def attach_http_server(self, http_server) -> None:
         self.http_server = http_server
 
+    # ------------------------------------------------------------------
+    # Lifecycle
+    # ------------------------------------------------------------------
+
     @abstractmethod
     async def start(self):
         """Start the driver (connect, authenticate, begin listening).
         Long-running drivers should loop indefinitely here."""
 
+    async def stop(self):
+        """Graceful shutdown.  Override in drivers that hold resources."""
+
     @abstractmethod
     async def send(self, channel: dict, text: str, **kwargs) -> str | list[str] | None:
         """Send *text* to the given *channel* on this platform."""
+
+    # ------------------------------------------------------------------
+    # Health
+    # ------------------------------------------------------------------
+
+    async def health_check(self) -> DriverHealth:
+        return self._health
+
+    @property
+    def health(self) -> DriverHealth:
+        return self._health
+
+    @health.setter
+    def health(self, value: DriverHealth) -> None:
+        old = self._health
+        self._health = value
+        if old != value:
+            try:
+                self._ctx.emit("health_changed", driver=self, old=old, new=value)
+            except Exception:
+                pass

drivers/discord.py

@@ -596,28 +596,26 @@ async def _send_bot(
                 pass
 
         try:
-            # Build kwargs to only include non-None/non-empty values
-            send_kwargs = {}
-            if text:
-                send_kwargs["content"] = text
-            if discord_files:
-                send_kwargs["files"] = discord_files
-            if reference is not None:
-                send_kwargs["reference"] = reference
-
             replied_user = True
             source_mentioned_self = kwargs.get("source_mentioned_self")
             if source_mentioned_self is not None:
                 replied_user = bool(source_mentioned_self)
 
-            send_kwargs["allowed_mentions"] = discord.AllowedMentions(
+            allowed = discord.AllowedMentions(
                 everyone=self.config.allow_mentions_everyone,
                 users=self.config.allow_mentions_users,
                 roles=self.config.allow_mentions_roles,
                 replied_user=replied_user,
             )
 
-            sent = await ch.send(**send_kwargs)
+            send_kwargs: dict = {"allowed_mentions": allowed}
+            if text:
+                send_kwargs["content"] = text
+            if discord_files:
+                send_kwargs["files"] = discord_files
+            if reference is not None:
+                send_kwargs["reference"] = reference
+            sent = await ch.send(**send_kwargs)  # type: ignore[no-matching-overload]
             return str(sent.id)
         except Exception:
             self.logger.exception("send error")

drivers/qq.py

@@ -2334,8 +2334,8 @@ async def _do_upload(d=data_bytes, fn=fname, gid=group_id):
                             }
                         )
 
-        main_segments = []
-        standalone_segments = []
+        main_segments: list[dict] = []
+        standalone_segments: list[dict | list[dict]] = []
         for seg in segments:
             if seg["type"] in ("video", "record"):
                 standalone_segments.append(seg)
@@ -2349,7 +2349,7 @@ async def _do_upload(d=data_bytes, fn=fname, gid=group_id):
                 and standalone_segments
             ):
                 # If only reply segment remains, attach it to the first standalone segment
-                standalone_segments[0] = [main_segments[0], standalone_segments[0]]
+                standalone_segments[0] = [main_segments[0], standalone_segments[0]]  # ty: ignore[invalid-assignment]
                 main_segments = []
             else:
                 msg_id = await self._api_send_group_msg(group_id, main_segments)

drivers/registry.py

@@ -1,17 +1,20 @@
-"""
-Driver registry.
+"""Driver registry.
 
-Each driver module calls ``register()`` at import time.  ``main.py`` then
-auto-discovers all driver modules via ``pkgutil.iter_modules`` so no central
-list needs to be maintained — drop a file into ``drivers/`` and it's live.
+Each driver module calls ``register()`` at import time.  The plugin
+loader auto-discovers driver modules so no central list needs to be
+maintained — drop a file into ``drivers/`` and it's live.
 """
 
 from __future__ import annotations
 
+from typing import Any, Callable
+
 from pydantic import BaseModel
 
 _REGISTRY: dict[str, tuple[type[BaseModel], type]] = {}
 
+_CLI_HOOKS: list[Callable[..., Any]] = []
+
 
 def register(name: str, config_cls: type[BaseModel], driver_cls: type) -> None:
     """Register a driver under *name*.
@@ -24,7 +27,41 @@ def register(name: str, config_cls: type[BaseModel], driver_cls: type) -> None:
     _REGISTRY[name] = (config_cls, driver_cls)
 
 
+def unregister(name: str) -> bool:
+    """Remove a driver from the registry.  Returns ``True`` if it existed."""
+    return _REGISTRY.pop(name, None) is not None
+
+
 def all_drivers() -> dict[str, tuple[type[BaseModel], type]]:
     """Return a snapshot of ``{name: (config_cls, driver_cls)}`` for every
     registered driver."""
     return dict(_REGISTRY)
+
+
+def get_driver(name: str) -> tuple[type[BaseModel], type] | None:
+    """Look up a single driver by platform name."""
+    return _REGISTRY.get(name)
+
+
+def register_cli(hook: Callable[..., Any]) -> None:
+    """Register a CLI subcommand hook.
+
+    The hook is called with an ``argparse._SubParsersAction`` and should
+    add whatever subcommands the driver needs.  Called at import time,
+    same pattern as ``register()``.
+
+    Example::
+
+        def _setup_cli(subparsers):
+            p = subparsers.add_parser("mydriver", help="...")
+            sub = p.add_subparsers(dest="mydriver_command")
+            sub.add_parser("pair", help="Pair with remote")
+
+        register_cli(_setup_cli)
+    """
+    _CLI_HOOKS.append(hook)
+
+
+def all_cli_hooks() -> list[Callable[..., Any]]:
+    """Return all registered CLI hooks."""
+    return list(_CLI_HOOKS)