from dataclasses import dataclass from typing import Any from models.bot import BotInstance DEFAULT_NODE_ID = "local" DEFAULT_TRANSPORT_KIND = "direct" DEFAULT_RUNTIME_KIND = "docker" DEFAULT_CORE_ADAPTER = "nanobot" TARGET_CONFIG_KEY = "dashboardRuntime" SUPPORTED_TRANSPORT_KINDS = {"direct", "edge"} SUPPORTED_RUNTIME_KINDS = {"docker", "native"} def _normalize_target_part(value: Any, fallback: str) -> str: text = str(value or "").strip().lower() return text or fallback @dataclass(frozen=True) class ProviderTarget: node_id: str = DEFAULT_NODE_ID transport_kind: str = DEFAULT_TRANSPORT_KIND runtime_kind: str = DEFAULT_RUNTIME_KIND core_adapter: str = DEFAULT_CORE_ADAPTER @property def key(self) -> str: return ":".join([self.node_id, self.transport_kind, self.runtime_kind, self.core_adapter]) def normalize_provider_target(value: Any, fallback: ProviderTarget | None = None) -> ProviderTarget: base = fallback or ProviderTarget() if isinstance(value, ProviderTarget): raw_node_id = value.node_id raw_transport_kind = value.transport_kind raw_runtime_kind = value.runtime_kind raw_core_adapter = value.core_adapter elif isinstance(value, dict): raw_node_id = value.get("node_id", value.get("nodeId")) raw_transport_kind = value.get("transport_kind", value.get("transportKind")) raw_runtime_kind = value.get("runtime_kind", value.get("runtimeKind")) raw_core_adapter = value.get("core_adapter", value.get("coreAdapter")) else: raw_node_id = None raw_transport_kind = None raw_runtime_kind = None raw_core_adapter = None transport_kind = _normalize_target_part(raw_transport_kind, base.transport_kind) if transport_kind not in SUPPORTED_TRANSPORT_KINDS: transport_kind = base.transport_kind runtime_kind = _normalize_target_part(raw_runtime_kind, base.runtime_kind) if runtime_kind not in SUPPORTED_RUNTIME_KINDS: runtime_kind = base.runtime_kind return ProviderTarget( node_id=_normalize_target_part(raw_node_id, base.node_id), transport_kind=transport_kind, runtime_kind=runtime_kind, core_adapter=_normalize_target_part(raw_core_adapter, base.core_adapter), ) def provider_target_to_dict(target: ProviderTarget) -> dict[str, str]: return { "node_id": target.node_id, "transport_kind": target.transport_kind, "runtime_kind": target.runtime_kind, "core_adapter": target.core_adapter, } def provider_target_from_config(config_data: Any, fallback: ProviderTarget | None = None) -> ProviderTarget: if not isinstance(config_data, dict): return normalize_provider_target(None, fallback=fallback) return normalize_provider_target(config_data.get(TARGET_CONFIG_KEY), fallback=fallback) def write_provider_target_config(config_data: dict[str, Any], target: ProviderTarget) -> dict[str, Any]: config_data[TARGET_CONFIG_KEY] = { "nodeId": target.node_id, "transportKind": target.transport_kind, "runtimeKind": target.runtime_kind, "coreAdapter": target.core_adapter, } return config_data def resolve_provider_target(app_state: Any, bot: BotInstance) -> ProviderTarget: fallback = ProviderTarget( node_id=_normalize_target_part(getattr(app_state, "provider_default_node_id", None), DEFAULT_NODE_ID), transport_kind=_normalize_target_part( getattr(app_state, "provider_default_transport_kind", None), DEFAULT_TRANSPORT_KIND, ), runtime_kind=_normalize_target_part( getattr(app_state, "provider_default_runtime_kind", None), DEFAULT_RUNTIME_KIND, ), core_adapter=_normalize_target_part( getattr(app_state, "provider_default_core_adapter", None), DEFAULT_CORE_ADAPTER, ), ) resolver = getattr(app_state, "resolve_bot_provider_target", None) if callable(resolver): return normalize_provider_target(resolver(bot), fallback=fallback) return normalize_provider_target( { "node_id": getattr(bot, "node_id", None), "transport_kind": getattr(bot, "transport_kind", None), "runtime_kind": getattr(bot, "runtime_kind", None), "core_adapter": getattr(bot, "core_adapter", None), }, fallback=fallback, )