"""In-process compatibility runtime for generated reviewed adapters.""" from __future__ import annotations import importlib from collections.abc import Callable, Mapping from typing import Any from jwebot_sdk import Event, ExtensionResult, PlanKind from .descriptors import LOCAL_EXTENSIONS, extension_for_kind from .local_commands import ( LOCAL_COMMAND_ALIASES, LocalCommandEngine, local_command_parts, utc_now, pack_reply_parts, unpack_reply_parts, ) class LocalExtensionError(ValueError): """Controlled local adapter or broker contract failure.""" class _Registry: def __init__(self) -> None: self.handlers: dict[str, Callable[[Event, str], ExtensionResult]] = {} def register( self, command: str, handler: Callable[[Event, str], ExtensionResult], ) -> None: key = command.casefold() if key in self.handlers: raise LocalExtensionError(f"duplicate local command: {command}") self.handlers[key] = handler class LocalServiceBroker: """Kernel-side implementation of reviewed local service profiles.""" def __init__(self, engine: LocalCommandEngine) -> None: self.engine = engine def call( self, profile: str, *, operation: str, payload: Mapping[str, object], ) -> Mapping[str, object]: if operation != "execute": raise LocalExtensionError("unsupported local service operation") command = payload.get("command") arguments = payload.get("arguments", "") event_value = payload.get("event") if ( not isinstance(command, str) or not isinstance(arguments, str) or not isinstance(event_value, Mapping) ): raise LocalExtensionError("invalid local service payload") kind, _ = local_command_parts( f"{command} {arguments}".rstrip() ) owner = extension_for_kind(kind) if owner is None or owner.extension_id != profile: raise LocalExtensionError("local service profile mismatch") event = dict(event_value) row = { "event_id": str(event.get("event_id", "")), "group_fingerprint": str(event.get("group_fingerprint", "")), "sender": str(event.get("sender", "")), "question": f"{command} {arguments}".rstrip(), "created_at": str(event.get("observed_at") or utc_now()), } text = self.engine.handle(row, event) parts = unpack_reply_parts(text) if parts is not None: return {"text": parts[0], "texts": parts} return {"text": text} class _Context: def __init__(self, engine: LocalCommandEngine) -> None: self.commands = _Registry() self.services = LocalServiceBroker(engine) def local_extension_command_token(question: str) -> str: normalized = question.lstrip() if not normalized.startswith("/"): raise LocalExtensionError("local command must start with slash") raw = normalized[1:] folded = raw.casefold() for alias in sorted(LOCAL_COMMAND_ALIASES, key=len, reverse=True): if ( extension_for_kind(LOCAL_COMMAND_ALIASES[alias]) is not None and folded.startswith(alias.casefold()) ): return "/" + raw[: len(alias)] raise LocalExtensionError("extension command alias is unknown") class LocalExtensionRuntime: """Load generated adapters once, then dispatch without mutable call state.""" def __init__(self, engine: LocalCommandEngine) -> None: context = _Context(engine) for descriptor in LOCAL_EXTENSIONS: module_name = descriptor.extension_id.replace("-", "_") module = importlib.import_module( f".adapters.{module_name}", package=__package__, ) entrypoint = getattr(module, "plugin", None) if not callable(entrypoint): raise LocalExtensionError( f"local adapter has no entrypoint: {descriptor.extension_id}" ) entrypoint(context) expected = { command.casefold() for descriptor in LOCAL_EXTENSIONS for command in descriptor.commands } if set(context.commands.handlers) != expected: raise LocalExtensionError( "local adapter registration does not match ownership inventory" ) self._handlers = dict(context.commands.handlers) def handle( self, row: Mapping[str, Any], event: Mapping[str, Any], ) -> str: question = str(row.get("question", "")) _, arguments = local_command_parts(question) command = local_extension_command_token(question) try: handler = self._handlers[command.casefold()] except KeyError as error: raise LocalExtensionError("local command is not registered") from error observed_at = str( event.get("observed_at") or row.get("created_at") or utc_now() ) sdk_event = Event( event_id=str(row.get("event_id") or event.get("event_id") or ""), observed_at=observed_at, sender=str(row.get("sender") or event.get("sender") or ""), text=str(event.get("text") or question), group_fingerprint=str( row.get("group_fingerprint") or event.get("group_fingerprint") or "" ), mention_bot=bool(event.get("mention_bot", True)), confidence=float(event.get("confidence", 0.0)), sender_confidence=float(event.get("sender_confidence", 0.0)), body_confidence=float(event.get("body_confidence", 0.0)), source=str(event.get("source", "jwebot-edge")), ) result = handler(sdk_event, arguments) if not isinstance(result, ExtensionResult): raise LocalExtensionError("local adapter returned invalid result") if result.reply.kind is PlanKind.MULTI_TEXT: return pack_reply_parts(list(result.reply.messages)) if result.reply.kind is not PlanKind.TEXT: raise LocalExtensionError("local command returned no visible reply") return result.reply.message