""" Platform Adapter Registry Allows platform adapters (built-in and plugin) to self-register so the gateway can discover and instantiate them without hardcoded if/elif chains. Built-in adapters continue to use the existing if/elif in _create_adapter() for now. Plugin adapters register here via PluginContext.register_platform() and are looked up first -- if nothing is found the gateway falls through to the legacy code path. Usage (plugin side): from gateway.platform_registry import platform_registry, PlatformEntry platform_registry.register(PlatformEntry( name="irc", label="IRC", adapter_factory=lambda cfg: IRCAdapter(cfg), check_fn=check_requirements, validate_config=lambda cfg: bool(cfg.extra.get("server")), required_env=["IRC_SERVER"], install_hint="pip install irc", )) Usage (gateway side): adapter = platform_registry.create_adapter("irc", platform_config) """ import logging from dataclasses import dataclass, field from typing import Any, Callable, Optional logger = logging.getLogger(__name__) @dataclass class PlatformEntry: """Metadata and factory for a single platform adapter.""" # Identifier used in config.yaml (e.g. "irc", "viber"). name: str # Human-readable label (e.g. "IRC", "Viber"). label: str # Factory callable: receives a PlatformConfig, returns an adapter instance. # Using a factory instead of a bare class lets plugins do custom init # (e.g. passing extra kwargs, wrapping in try/except). adapter_factory: Callable[[Any], Any] # Returns True when the platform's dependencies are available. check_fn: Callable[[], bool] # Optional: given a PlatformConfig, is it properly configured? # If None, the registry skips config validation and lets the adapter # fail at connect() time with a descriptive error. validate_config: Optional[Callable[[Any], bool]] = None # Optional: given a PlatformConfig, is the platform connected/enabled? # Used by ``GatewayConfig.get_connected_platforms()`` and setup UI status. # If None, falls back to ``validate_config`` or ``check_fn``. is_connected: Optional[Callable[[Any], bool]] = None # Env vars this platform needs (for ``hermes setup`` display). required_env: list = field(default_factory=list) # Hint shown when check_fn returns False. install_hint: str = "" # Optional setup function for interactive configuration. # Signature: () -> None (prompts user, saves env vars). # If None, falls back to _setup_standard_platform (needs token_var + vars) # or a generic "set these env vars" display. setup_fn: Optional[Callable[[], None]] = None # "builtin" or "plugin" source: str = "plugin" # Name of the plugin manifest that registered this entry (empty for # built-ins). Used by ``hermes gateway setup`` to auto-enable the # owning plugin when the user configures its platform. plugin_name: str = "" # ── Auth env var names (for _is_user_authorized integration) ── # E.g. "IRC_ALLOWED_USERS" — checked for comma-separated user IDs. allowed_users_env: str = "" # E.g. "IRC_ALLOW_ALL_USERS" — if truthy, all users authorized. allow_all_env: str = "" # ── Message limits ── # Max message length for smart-chunking. 0 = no limit. max_message_length: int = 0 # ── Privacy ── # If True, session descriptions redact PII (phone numbers, etc.) pii_safe: bool = False # ── Display ── # Emoji for CLI/gateway display (e.g. "💬") emoji: str = "🔌" # Whether this platform should appear in _UPDATE_ALLOWED_PLATFORMS # (allows /update command from this platform). allow_update_command: bool = True # ── LLM guidance ── # Platform hint injected into the system prompt (e.g. "You are on IRC. # Do not use markdown."). Empty string = no hint. platform_hint: str = "" class PlatformRegistry: """Central registry of platform adapters. Thread-safe for reads (dict lookups are atomic under GIL). Writes happen at startup during sequential discovery. """ def __init__(self) -> None: self._entries: dict[str, PlatformEntry] = {} def register(self, entry: PlatformEntry) -> None: """Register a platform adapter entry. If an entry with the same name exists, it is replaced (last writer wins -- this lets plugins override built-in adapters if desired). """ if entry.name in self._entries: prev = self._entries[entry.name] logger.info( "Platform '%s' re-registered (was %s, now %s)", entry.name, prev.source, entry.source, ) self._entries[entry.name] = entry logger.debug("Registered platform adapter: %s (%s)", entry.name, entry.source) def unregister(self, name: str) -> bool: """Remove a platform entry. Returns True if it existed.""" return self._entries.pop(name, None) is not None def get(self, name: str) -> Optional[PlatformEntry]: """Look up a platform entry by name.""" return self._entries.get(name) def all_entries(self) -> list[PlatformEntry]: """Return all registered platform entries.""" return list(self._entries.values()) def plugin_entries(self) -> list[PlatformEntry]: """Return only plugin-registered platform entries.""" return [e for e in self._entries.values() if e.source == "plugin"] def is_registered(self, name: str) -> bool: return name in self._entries def create_adapter(self, name: str, config: Any) -> Optional[Any]: """Create an adapter instance for the given platform name. Returns None if: - No entry registered for *name* - check_fn() returns False (missing deps) - validate_config() returns False (misconfigured) - The factory raises an exception """ entry = self._entries.get(name) if entry is None: return None if not entry.check_fn(): hint = f" ({entry.install_hint})" if entry.install_hint else "" logger.warning( "Platform '%s' requirements not met%s", entry.label, hint, ) return None if entry.validate_config is not None: try: if not entry.validate_config(config): logger.warning( "Platform '%s' config validation failed", entry.label, ) return None except Exception as e: logger.warning( "Platform '%s' config validation error: %s", entry.label, e, ) return None try: adapter = entry.adapter_factory(config) return adapter except Exception as e: logger.error( "Failed to create adapter for platform '%s': %s", entry.label, e, exc_info=True, ) return None # Module-level singleton platform_registry = PlatformRegistry()