refactor(webui): split settings backend by domain

nanobot/webui/settings_capabilities.py

@@ -0,0 +1,804 @@
+"""Capability settings domain logic for Web, media, network, and API features."""
+
+from __future__ import annotations
+
+import asyncio
+import os
+import re
+from dataclasses import dataclass
+from typing import TYPE_CHECKING, Any, Awaitable, Callable, TypedDict
+
+from nanobot.agent.tools.web import SEARCH_PROVIDER_OPTIONS
+from nanobot.api.runtime import ApiRuntime, ApiStartOptions
+from nanobot.audio.transcription import resolve_transcription_config
+from nanobot.audio.transcription_registry import (
+    resolve_transcription_provider,
+    transcription_provider_names,
+)
+from nanobot.config.schema import Config
+from nanobot.optional_features import (
+    OptionalFeatureError,
+    extra_installed,
+    optional_dependency_groups,
+)
+from nanobot.providers.image_generation import (
+    get_image_gen_provider,
+    image_gen_provider_names,
+)
+from nanobot.providers.registry import find_by_name
+from nanobot.security.network import is_loopback_host
+from nanobot.webui.settings_contracts import (
+    QueryParams,
+    SettingsRequest,
+    SettingsRouteResult,
+    WebUISettingsError,
+    parse_bool,
+    query_first,
+    query_first_alias,
+)
+from nanobot.webui.settings_models import (
+    OAuthStatusReader,
+    mask_secret_hint,
+    provider_configured_for_settings,
+)
+from nanobot.webui.workspaces import (
+    read_webui_default_access_mode,
+)
+
+if TYPE_CHECKING:
+    from nanobot.webui.settings_services import WebUISettingsServices
+
+SettingsOperation = Callable[..., dict[str, Any]]
+
+
+@dataclass(frozen=True)
+class CapabilitySettingsOperations:
+    update_web_search: SettingsOperation
+    update_api: SettingsOperation
+    update_image: SettingsOperation
+    update_transcription: SettingsOperation
+    update_network: SettingsOperation
+    nanobot_features_action: SettingsOperation
+    api_runtime: Callable[[], ApiRuntime]
+    reload_image: Callable[[], Awaitable[dict[str, Any]]]
+
+
+class CapabilitySettingsPayload(TypedDict):
+    web_search: dict[str, Any]
+    web: dict[str, Any]
+    api: dict[str, Any]
+    observability: dict[str, Any]
+    image_generation: dict[str, Any]
+    transcription: dict[str, Any]
+
+
+_WEB_SEARCH_PROVIDER_OPTIONS = SEARCH_PROVIDER_OPTIONS
+_WEB_SEARCH_PROVIDER_BY_NAME = {
+    provider["name"]: provider for provider in _WEB_SEARCH_PROVIDER_OPTIONS
+}
+_IMAGE_GENERATION_ASPECT_RATIOS = {
+    "1:1",
+    "3:4",
+    "9:16",
+    "4:3",
+    "16:9",
+    "3:2",
+    "2:3",
+    "21:9",
+}
+
+
+def _image_generation_provider_rows(
+    config: Config,
+    *,
+    oauth_status: OAuthStatusReader,
+) -> list[dict[str, Any]]:
+    rows: list[dict[str, Any]] = []
+    for name in image_gen_provider_names():
+        image_provider = get_image_gen_provider(name)
+        spec = find_by_name(name)
+        provider_config = getattr(config.providers, name, None)
+        configured = (
+            provider_configured_for_settings(spec, provider_config, oauth_status)
+            if spec is not None and provider_config is not None
+            else bool(getattr(provider_config, "api_key", None))
+        )
+        rows.append(
+            {
+                "name": name,
+                "label": spec.label if spec is not None else name,
+                "configured": configured,
+                "auth_type": "oauth" if spec is not None and spec.is_oauth else "api_key",
+                "api_key_hint": mask_secret_hint(getattr(provider_config, "api_key", None)),
+                "api_base": getattr(provider_config, "api_base", None),
+                "default_api_base": (
+                    spec.default_api_base if spec and spec.default_api_base else None
+                ),
+                "models": list(image_provider.model_options) if image_provider else [],
+                "default_model": (
+                    image_provider.model_options[0]
+                    if image_provider and image_provider.model_options
+                    else None
+                ),
+            }
+        )
+    return rows
+
+
+def _transcription_provider_rows(config: Config) -> list[dict[str, Any]]:
+    rows: list[dict[str, Any]] = []
+    for name in transcription_provider_names():
+        spec = find_by_name(name)
+        provider_config = getattr(config.providers, name, None)
+        rows.append(
+            {
+                "name": name,
+                "label": spec.label if spec is not None else name,
+                "configured": bool(getattr(provider_config, "api_key", None)),
+                "api_key_hint": mask_secret_hint(getattr(provider_config, "api_key", None)),
+                "api_base": getattr(provider_config, "api_base", None),
+                "default_api_base": (
+                    spec.default_api_base if spec and spec.default_api_base else None
+                ),
+            }
+        )
+    return rows
+
+
+def capability_settings_payload(
+    config: Config,
+    *,
+    oauth_status: OAuthStatusReader,
+) -> CapabilitySettingsPayload:
+    search_config = config.tools.web.search
+    image_config = config.tools.image_generation
+    transcription = resolve_transcription_config(config)
+    search_provider = (
+        search_config.provider
+        if search_config.provider in _WEB_SEARCH_PROVIDER_BY_NAME
+        else "duckduckgo"
+    )
+    image_providers = _image_generation_provider_rows(config, oauth_status=oauth_status)
+    selected_image_provider = next(
+        (
+            provider
+            for provider in image_providers
+            if provider["name"] == image_config.provider
+        ),
+        None,
+    )
+    return {
+        "web_search": {
+            "provider": search_provider,
+            "api_key_hint": mask_secret_hint(search_config.api_key),
+            "base_url": search_config.base_url or None,
+            "max_results": search_config.max_results,
+            "timeout": search_config.timeout,
+            "providers": list(_WEB_SEARCH_PROVIDER_OPTIONS),
+        },
+        "web": {
+            "enable": config.tools.web.enable,
+            "proxy": config.tools.web.proxy,
+            "user_agent": config.tools.web.user_agent,
+            "search": {
+                "max_results": search_config.max_results,
+                "timeout": search_config.timeout,
+            },
+            "fetch": {
+                "use_jina_reader": config.tools.web.fetch.use_jina_reader,
+            },
+        },
+        "api": {
+            "host": config.api.host,
+            "port": config.api.port,
+            "timeout": config.api.timeout,
+            "api_key_hint": mask_secret_hint(config.api.api_key),
+        },
+        "observability": {
+            "provider": "langfuse",
+            "configured": bool(
+                os.environ.get("LANGFUSE_SECRET_KEY")
+                and os.environ.get("LANGFUSE_PUBLIC_KEY")
+            ),
+            "base_url": os.environ.get("LANGFUSE_BASE_URL")
+            or "https://cloud.langfuse.com",
+        },
+        "image_generation": {
+            "enabled": image_config.enabled,
+            "provider": image_config.provider,
+            "provider_configured": bool(
+                selected_image_provider and selected_image_provider["configured"]
+            ),
+            "model": image_config.model,
+            "default_aspect_ratio": image_config.default_aspect_ratio,
+            "default_image_size": image_config.default_image_size,
+            "max_images_per_turn": image_config.max_images_per_turn,
+            "save_dir": image_config.save_dir,
+            "providers": image_providers,
+        },
+        "transcription": {
+            "enabled": transcription.enabled,
+            "provider": transcription.provider,
+            "provider_configured": transcription.configured,
+            "model": transcription.model,
+            "language": transcription.language,
+            "max_duration_sec": transcription.max_duration_sec,
+            "max_upload_mb": transcription.max_upload_mb,
+            "providers": _transcription_provider_rows(config),
+        },
+    }
+
+
+def update_network_safety_settings(
+    config: Config,
+    query: QueryParams,
+) -> tuple[bool, str | None]:
+    raw_allow = (
+        query_first_alias(
+            query,
+            "webui_allow_local_service_access",
+            "webuiAllowLocalServiceAccess",
+        )
+        or query_first_alias(
+            query,
+            "allow_local_preview_access",
+            "allowLocalPreviewAccess",
+        )
+    )
+    raw_default_access_mode = query_first_alias(
+        query,
+        "webui_default_access_mode",
+        "webuiDefaultAccessMode",
+    )
+    if raw_allow is None and raw_default_access_mode is None:
+        raise WebUISettingsError(
+            "webui_allow_local_service_access or webui_default_access_mode is required"
+        )
+
+    changed = False
+    if raw_allow is not None:
+        allow_local = parse_bool(raw_allow, "webui_allow_local_service_access")
+        if config.tools.webui_allow_local_service_access != allow_local:
+            config.tools.webui_allow_local_service_access = allow_local
+            changed = True
+
+    default_access_mode: str | None = None
+    if raw_default_access_mode is not None:
+        default_access_mode = raw_default_access_mode.strip().lower()
+        if default_access_mode == "restricted":
+            default_access_mode = "default"
+        if default_access_mode not in {"default", "full"}:
+            raise WebUISettingsError(
+                "webui_default_access_mode must be default or full"
+            )
+    return changed, default_access_mode
+
+
+def update_web_search_settings(config: Config, query: QueryParams) -> tuple[bool, bool]:
+    provider_name = (query_first(query, "provider") or "").strip().lower()
+    provider_option = _WEB_SEARCH_PROVIDER_BY_NAME.get(provider_name)
+    if provider_option is None:
+        raise WebUISettingsError("unknown web search provider")
+
+    search_config = config.tools.web.search
+    web_config = config.tools.web
+    previous_provider = search_config.provider
+    changed = False
+    restart_required = False
+
+    def set_search_value(attr: str, value: object) -> None:
+        nonlocal changed
+        if getattr(search_config, attr) != value:
+            setattr(search_config, attr, value)
+            changed = True
+
+    def set_fetch_value(attr: str, value: object) -> None:
+        nonlocal changed
+        if getattr(web_config.fetch, attr) != value:
+            setattr(web_config.fetch, attr, value)
+            changed = True
+
+    if search_config.provider != provider_name:
+        search_config.provider = provider_name
+        changed = True
+
+    credential = provider_option["credential"]
+    if credential == "none":
+        set_search_value("api_key", "")
+        set_search_value("base_url", "")
+    elif credential == "base_url":
+        base_url = query_first_alias(query, "base_url", "baseUrl")
+        base_url = base_url.strip() if base_url is not None else None
+        if not base_url and previous_provider == provider_name and search_config.base_url:
+            base_url = search_config.base_url
+        if not base_url:
+            raise WebUISettingsError("base_url is required")
+        set_search_value("base_url", base_url)
+        set_search_value("api_key", "")
+    elif credential in {"api_key", "optional_api_key"}:
+        raw_api_key = query_first_alias(query, "api_key", "apiKey")
+        api_key = raw_api_key.strip() if raw_api_key is not None else None
+        if api_key is None and previous_provider == provider_name and search_config.api_key:
+            api_key = search_config.api_key
+        if credential == "api_key" and not api_key:
+            raise WebUISettingsError("api_key is required")
+        set_search_value("api_key", api_key or "")
+        set_search_value("base_url", "")
+    else:
+        raise WebUISettingsError("unknown web search credential type")
+
+    max_results = query_first_alias(query, "max_results", "maxResults")
+    if max_results is not None:
+        try:
+            parsed = int(max_results)
+        except ValueError:
+            raise WebUISettingsError("max_results must be an integer") from None
+        if parsed < 1 or parsed > 10:
+            raise WebUISettingsError("max_results must be between 1 and 10")
+        set_search_value("max_results", parsed)
+
+    timeout = query_first(query, "timeout")
+    if timeout is not None:
+        try:
+            parsed_timeout = int(timeout)
+        except ValueError:
+            raise WebUISettingsError("timeout must be an integer") from None
+        if parsed_timeout < 1 or parsed_timeout > 120:
+            raise WebUISettingsError("timeout must be between 1 and 120")
+        set_search_value("timeout", parsed_timeout)
+
+    use_jina_reader = query_first_alias(query, "use_jina_reader", "useJinaReader")
+    if use_jina_reader is not None:
+        previous_jina_reader = web_config.fetch.use_jina_reader
+        set_fetch_value("use_jina_reader", parse_bool(use_jina_reader, "use_jina_reader"))
+        if web_config.fetch.use_jina_reader != previous_jina_reader:
+            restart_required = True
+    return changed, restart_required
+
+
+def update_api_settings(config: Config, query: QueryParams) -> None:
+    """Update the managed OpenAI-compatible API configuration."""
+    api = config.api
+    host = query_first(query, "host")
+    if host is not None:
+        host = host.strip()
+        if not host:
+            raise WebUISettingsError("host is required")
+        api.host = host
+
+    port = query_first(query, "port")
+    if port is not None:
+        try:
+            parsed_port = int(port)
+        except ValueError:
+            raise WebUISettingsError("port must be an integer") from None
+        if parsed_port < 1 or parsed_port > 65535:
+            raise WebUISettingsError("port must be between 1 and 65535")
+        api.port = parsed_port
+
+    timeout = query_first(query, "timeout")
+    if timeout is not None:
+        try:
+            parsed_timeout = float(timeout)
+        except ValueError:
+            raise WebUISettingsError("timeout must be a number") from None
+        if parsed_timeout < 1 or parsed_timeout > 3600:
+            raise WebUISettingsError("timeout must be between 1 and 3600")
+        api.timeout = parsed_timeout
+
+    api_key = query_first_alias(query, "api_key", "apiKey")
+    if api_key is not None:
+        api.api_key = api_key.strip()
+    if not is_loopback_host(api.host) and not api.api_key.strip():
+        raise WebUISettingsError(
+            "an API key is required when the API is available on the network"
+        )
+
+
+def update_image_generation_settings(
+    config: Config,
+    query: QueryParams,
+    *,
+    oauth_status: OAuthStatusReader,
+) -> bool:
+    image_config = config.tools.image_generation
+    changed = False
+
+    provider_name = query_first(query, "provider")
+    if provider_name is not None:
+        provider_name = provider_name.strip().lower()
+        if not provider_name:
+            raise WebUISettingsError("image generation provider is required")
+        if get_image_gen_provider(provider_name) is None:
+            raise WebUISettingsError("unknown image generation provider")
+        if image_config.provider != provider_name:
+            image_config.provider = provider_name
+            changed = True
+
+    enabled = query_first(query, "enabled")
+    if enabled is not None:
+        parsed_enabled = parse_bool(enabled, "enabled")
+        if image_config.enabled != parsed_enabled:
+            image_config.enabled = parsed_enabled
+            changed = True
+
+    model = query_first(query, "model")
+    if model is not None:
+        model = model.strip()
+        if not model:
+            raise WebUISettingsError("image generation model is required")
+        if len(model) > 200:
+            raise WebUISettingsError("image generation model is too long")
+        if image_config.model != model:
+            image_config.model = model
+            changed = True
+
+    default_aspect_ratio = query_first_alias(
+        query,
+        "default_aspect_ratio",
+        "defaultAspectRatio",
+    )
+    if default_aspect_ratio is not None:
+        default_aspect_ratio = default_aspect_ratio.strip()
+        if default_aspect_ratio not in _IMAGE_GENERATION_ASPECT_RATIOS:
+            raise WebUISettingsError("unsupported image generation aspect ratio")
+        if image_config.default_aspect_ratio != default_aspect_ratio:
+            image_config.default_aspect_ratio = default_aspect_ratio
+            changed = True
+
+    default_image_size = query_first_alias(
+        query,
+        "default_image_size",
+        "defaultImageSize",
+    )
+    if default_image_size is not None:
+        default_image_size = default_image_size.strip()
+        if not default_image_size:
+            raise WebUISettingsError("default image size is required")
+        if len(default_image_size) > 32 or not all(
+            char.isascii() and (char.isalnum() or char in {"x", "X", ":", "-", "_"})
+            for char in default_image_size
+        ):
+            raise WebUISettingsError("unsupported image generation size")
+        if image_config.default_image_size != default_image_size:
+            image_config.default_image_size = default_image_size
+            changed = True
+
+    max_images_per_turn = query_first_alias(
+        query,
+        "max_images_per_turn",
+        "maxImagesPerTurn",
+    )
+    if max_images_per_turn is not None:
+        try:
+            parsed_max = int(max_images_per_turn)
+        except ValueError:
+            raise WebUISettingsError("max_images_per_turn must be an integer") from None
+        if parsed_max < 1 or parsed_max > 8:
+            raise WebUISettingsError("max_images_per_turn must be between 1 and 8")
+        if image_config.max_images_per_turn != parsed_max:
+            image_config.max_images_per_turn = parsed_max
+            changed = True
+
+    if image_config.enabled:
+        selected_provider = next(
+            (
+                provider
+                for provider in _image_generation_provider_rows(
+                    config,
+                    oauth_status=oauth_status,
+                )
+                if provider["name"] == image_config.provider
+            ),
+            None,
+        )
+        if not selected_provider or not selected_provider["configured"]:
+            raise WebUISettingsError("image generation provider is not configured")
+    return changed
+
+
+def update_transcription_settings(config: Config, query: QueryParams) -> bool:
+    transcription = config.transcription
+    changed = False
+
+    enabled = query_first(query, "enabled")
+    if enabled is not None:
+        parsed_enabled = parse_bool(enabled, "enabled")
+        if transcription.enabled != parsed_enabled:
+            transcription.enabled = parsed_enabled
+            changed = True
+
+    provider = query_first(query, "provider")
+    if provider is not None:
+        provider = provider.strip().lower()
+        provider_spec = resolve_transcription_provider(provider)
+        if provider_spec is None:
+            raise WebUISettingsError("unknown transcription provider")
+        provider = provider_spec.name
+        if transcription.provider != provider:
+            transcription.provider = provider
+            changed = True
+
+    model = query_first(query, "model")
+    if model is not None:
+        model = model.strip() or None
+        if model is not None and len(model) > 200:
+            raise WebUISettingsError("transcription model is too long")
+        if transcription.model != model:
+            transcription.model = model
+            changed = True
+
+    language = query_first(query, "language")
+    if language is not None:
+        language = language.strip().lower() or None
+        if language is not None and not re.fullmatch(r"[a-z]{2,3}", language):
+            raise WebUISettingsError(
+                "transcription language must be 2-3 lowercase letters"
+            )
+        if transcription.language != language:
+            transcription.language = language
+            changed = True
+
+    max_duration_sec = query_first_alias(query, "max_duration_sec", "maxDurationSec")
+    if max_duration_sec is not None:
+        try:
+            parsed_duration = int(max_duration_sec)
+        except ValueError:
+            raise WebUISettingsError("max_duration_sec must be an integer") from None
+        if parsed_duration < 1 or parsed_duration > 600:
+            raise WebUISettingsError("max_duration_sec must be between 1 and 600")
+        if transcription.max_duration_sec != parsed_duration:
+            transcription.max_duration_sec = parsed_duration
+            changed = True
+
+    max_upload_mb = query_first_alias(query, "max_upload_mb", "maxUploadMb")
+    if max_upload_mb is not None:
+        try:
+            parsed_upload = int(max_upload_mb)
+        except ValueError:
+            raise WebUISettingsError("max_upload_mb must be an integer") from None
+        if parsed_upload < 1 or parsed_upload > 100:
+            raise WebUISettingsError("max_upload_mb must be between 1 and 100")
+        if transcription.max_upload_mb != parsed_upload:
+            transcription.max_upload_mb = parsed_upload
+            changed = True
+    return changed
+
+
+def network_safety_payload(config: Config) -> dict[str, Any]:
+    """Return the network-related fields embedded in the advanced DTO."""
+    return {
+        "webui_allow_local_service_access": config.tools.webui_allow_local_service_access,
+        "allow_local_preview_access": config.tools.webui_allow_local_service_access,
+        "webui_default_access_mode": read_webui_default_access_mode(),
+        "private_service_protection_enabled": True,
+        "ssrf_whitelist_count": len(config.tools.ssrf_whitelist),
+    }
+
+
+def masked_api_secret(value: str) -> str | None:
+    value = value.strip()
+    if not value:
+        return None
+    return f"{value[:3]}...{value[-4:]}" if len(value) > 8 else "configured"
+
+
+def api_runtime_message(message: str) -> str:
+    known = {
+        "api_exited_during_startup": "API server exited during startup. Check its log for details.",
+        "api_stop_timeout": "API server did not stop in time.",
+        "api_state_stale": "API server state was stale; try starting it again.",
+    }
+    if message in known:
+        return known[message]
+    if message.startswith("api_"):
+        return f"API server {message.removeprefix('api_').replace('_', ' ')}"
+    return message.replace("_", " ")
+
+
+def api_service_payload(
+    settings: WebUISettingsServices,
+    runtime: ApiRuntime,
+    *,
+    last_action: str | None = None,
+) -> dict[str, Any]:
+    config = settings.config.load()
+    status = runtime.status()
+    extras = optional_dependency_groups()
+    connect_host = (
+        "127.0.0.1" if config.api.host in {"0.0.0.0", "::"} else config.api.host
+    )
+    payload = {
+        "installed": extra_installed("api", extras.get("api")),
+        "running": status.running,
+        "managed": status.running,
+        "host": config.api.host,
+        "port": config.api.port,
+        "timeout": config.api.timeout,
+        "api_key_hint": masked_api_secret(config.api.api_key),
+        "endpoint": f"http://{connect_host}:{config.api.port}/v1",
+        "command": "nanobot serve",
+        "log_path": str(status.log_path),
+    }
+    if last_action:
+        payload["last_action"] = last_action
+    return payload
+
+
+class CapabilitySettingsHandler:
+    """Handle capability commands after transport authentication and decoding."""
+
+    def __init__(self, settings: WebUISettingsServices, logger: Any) -> None:
+        self.settings = settings
+        self.logger = logger
+
+    async def handle(
+        self,
+        action: str,
+        request: SettingsRequest,
+        operations: CapabilitySettingsOperations,
+    ) -> SettingsRouteResult:
+        if action == "api-status":
+            return SettingsRouteResult.success(
+                api_service_payload(self.settings, operations.api_runtime())
+            )
+        if action == "api-start":
+            return await self._start_api(request, operations)
+        if action == "api-stop":
+            return await self._stop_api(operations)
+
+        mutation = {
+            "web-search-update": (
+                operations.update_web_search,
+                "browser",
+                False,
+            ),
+            "transcription-update": (
+                operations.update_transcription,
+                None,
+                False,
+            ),
+            "network-update": (
+                operations.update_network,
+                "runtime",
+                False,
+            ),
+            "image-update": (
+                operations.update_image,
+                "image",
+                True,
+            ),
+        }.get(action)
+        if mutation is None:
+            return SettingsRouteResult.failure(404, "unknown settings action")
+
+        operation, section, apply_image_reload = mutation
+        try:
+            payload = self.settings.mutate(operation, request.query)
+        except WebUISettingsError as exc:
+            return SettingsRouteResult.failure(exc.status, exc.message)
+        if apply_image_reload:
+            payload, image_restart_cleared = await self.apply_image_runtime_change(
+                payload,
+                operations.reload_image,
+            )
+        else:
+            image_restart_cleared = False
+        return SettingsRouteResult.success(
+            payload,
+            decorate_restart=True,
+            restart_section=section,
+            clear_restart_section=("image" if image_restart_cleared else None),
+        )
+
+    async def apply_image_runtime_change(
+        self,
+        payload: dict[str, Any],
+        reload_image: Callable[[], Awaitable[dict[str, Any]]],
+    ) -> tuple[dict[str, Any], bool]:
+        """Hot-apply image settings, preserving restart fallback on failure."""
+        if not payload.get("requires_restart"):
+            return payload, False
+        try:
+            result = await reload_image()
+        except Exception:
+            self.logger.exception("failed to hot-reload image generation settings")
+            return payload, False
+
+        applied = bool(result.get("ok")) and not result.get("requires_restart")
+        updated = dict(payload)
+        updated["requires_restart"] = not applied
+        if not applied:
+            self.logger.warning(
+                "image generation settings were saved but require restart: {}",
+                result.get("message") or "hot reload failed",
+            )
+        return updated, applied
+
+    async def _start_api(
+        self,
+        request: SettingsRequest,
+        operations: CapabilitySettingsOperations,
+    ) -> SettingsRouteResult:
+        api_key = (request.payload or {}).get("api_key")
+        if api_key is not None and not isinstance(api_key, str):
+            return SettingsRouteResult.failure(
+                400,
+                "API service API key must be a string",
+            )
+        try:
+            await asyncio.to_thread(
+                self.settings.mutate,
+                operations.nanobot_features_action,
+                "enable",
+                {"name": ["api"]},
+                allow_install=self._allow_feature_package_install(request),
+            )
+            self.settings.mutate(operations.update_api, request.query)
+            config = self.settings.config.load()
+            runtime = operations.api_runtime()
+            options = ApiStartOptions(
+                host=config.api.host,
+                port=config.api.port,
+                workspace=str(config.workspace_path),
+                config_path=str(self.settings.config.path),
+            )
+            current = runtime.status()
+            result = await asyncio.to_thread(
+                runtime.restart if current.running else runtime.start_background,
+                options,
+            )
+            if not result.ok:
+                return SettingsRouteResult.failure(
+                    500,
+                    api_runtime_message(result.message),
+                )
+        except (WebUISettingsError, OptionalFeatureError) as exc:
+            return SettingsRouteResult.failure(
+                getattr(exc, "status", 400),
+                getattr(exc, "message", str(exc)),
+            )
+        except Exception as exc:
+            self.logger.exception("failed to start managed API service")
+            return SettingsRouteResult.failure(500, str(exc))
+        return SettingsRouteResult.success(
+            api_service_payload(
+                self.settings,
+                operations.api_runtime(),
+                last_action="started",
+            )
+        )
+
+    async def _stop_api(
+        self,
+        operations: CapabilitySettingsOperations,
+    ) -> SettingsRouteResult:
+        runtime = operations.api_runtime()
+        try:
+            result = await asyncio.to_thread(runtime.stop)
+        except Exception as exc:
+            self.logger.exception("failed to stop managed API service")
+            return SettingsRouteResult.failure(500, str(exc))
+        if not result.ok and result.message != "api_not_running":
+            return SettingsRouteResult.failure(
+                500,
+                api_runtime_message(result.message),
+            )
+        return SettingsRouteResult.success(
+            api_service_payload(
+                self.settings,
+                operations.api_runtime(),
+                last_action="stopped",
+            )
+        )
+
+    def _allow_feature_package_install(self, request: SettingsRequest) -> bool:
+        if request.local_browser:
+            return True
+        try:
+            return bool(
+                self.settings.config.load().tools.webui_allow_remote_package_install
+            )
+        except Exception:
+            self.logger.exception("failed to load remote package install policy")
+            return False

nanobot/webui/settings_contracts.py

@@ -0,0 +1,82 @@
+"""Stable request and error contracts shared by WebUI settings domains."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any
+
+QueryParams = dict[str, list[str]]
+
+
+@dataclass(frozen=True)
+class SettingsRequest:
+    """Transport-neutral input decoded by the settings route facade."""
+
+    query: QueryParams
+    payload: dict[str, Any] | None = None
+    local_browser: bool = False
+
+
+@dataclass(frozen=True)
+class SettingsRouteResult:
+    """Transport-neutral result returned by a settings domain handler."""
+
+    payload: dict[str, Any] | None = None
+    status: int = 200
+    error: str | None = None
+    decorate_restart: bool = False
+    restart_section: str | None = None
+    clear_restart_section: str | None = None
+    restart_payload_key: str | None = None
+
+    @classmethod
+    def success(
+        cls,
+        payload: dict[str, Any],
+        *,
+        decorate_restart: bool = False,
+        restart_section: str | None = None,
+        clear_restart_section: str | None = None,
+        restart_payload_key: str | None = None,
+    ) -> SettingsRouteResult:
+        return cls(
+            payload=payload,
+            decorate_restart=decorate_restart,
+            restart_section=restart_section,
+            clear_restart_section=clear_restart_section,
+            restart_payload_key=restart_payload_key,
+        )
+
+    @classmethod
+    def failure(cls, status: int, error: str) -> SettingsRouteResult:
+        return cls(status=status, error=error)
+
+
+class WebUISettingsError(ValueError):
+    """User-facing settings validation failure."""
+
+    def __init__(self, message: str, *, status: int = 400) -> None:
+        super().__init__(message)
+        self.message = message
+        self.status = status
+
+
+def query_first(query: QueryParams, key: str) -> str | None:
+    values = query.get(key)
+    return values[0] if values else None
+
+
+def query_first_alias(query: QueryParams, snake: str, camel: str) -> str | None:
+    value = query_first(query, snake)
+    return query_first(query, camel) if value is None else value
+
+
+def query_has_alias(query: QueryParams, snake: str, camel: str) -> bool:
+    return snake in query or camel in query
+
+
+def parse_bool(value: str, field: str) -> bool:
+    normalized = value.strip().lower()
+    if normalized not in {"1", "0", "true", "false", "yes", "no"}:
+        raise WebUISettingsError(f"{field} must be boolean")
+    return normalized in {"1", "true", "yes"}

nanobot/webui/settings_system.py

@@ -0,0 +1,957 @@
+"""System and channel settings domain logic."""
+
+from __future__ import annotations
+
+import asyncio
+import inspect
+import re
+import time
+from collections.abc import Callable, Iterable
+from dataclasses import dataclass
+from pathlib import Path
+from typing import TYPE_CHECKING, Any, TypedDict, cast
+from zoneinfo import ZoneInfo
+
+from nanobot.channels._setup import channel_setup_spec
+from nanobot.channels.connect import ChannelConnectError
+from nanobot.channels.contracts import (
+    RouteFieldType,
+    channel_instance_config,
+    channel_update_instance_config,
+)
+from nanobot.config.schema import Config
+from nanobot.optional_features import OptionalFeatureError, with_channel_runtime_status
+from nanobot.security.workspace_access import workspace_sandbox_status
+from nanobot.webui.settings_capabilities import network_safety_payload
+from nanobot.webui.settings_contracts import (
+    QueryParams,
+    SettingsRequest,
+    SettingsRouteResult,
+    WebUISettingsError,
+    query_first,
+    query_first_alias,
+)
+from nanobot.webui.token_usage import token_usage_payload
+
+if TYPE_CHECKING:
+    from nanobot.webui.settings_services import WebUISettingsServices
+
+LoadChannelPlugin = Callable[[str], Any]
+ListPendingPairings = Callable[[], Iterable[dict[str, Any]]]
+SettingsOperation = Callable[..., Any]
+
+
+@dataclass(frozen=True)
+class SystemSettingsOperations:
+    cli_apps_payload: SettingsOperation
+    cli_apps_action: SettingsOperation
+    nanobot_features_payload: SettingsOperation
+    nanobot_features_action: SettingsOperation
+    nanobot_feature_instance_target: SettingsOperation
+    validate_channel_config: SettingsOperation
+    load_channel_plugin: LoadChannelPlugin
+    list_pending: ListPendingPairings
+    approve_code: SettingsOperation
+    deny_code: SettingsOperation
+    mcp_presets_action: SettingsOperation
+    reload_mcp: SettingsOperation
+    check_for_update: SettingsOperation
+    channel_feature_action: SettingsOperation | None = None
+    channel_runtime_status: Callable[[], dict[str, Any]] | None = None
+
+
+class SystemSettingsPayload(TypedDict):
+    runtime: dict[str, Any]
+    usage: dict[str, Any]
+    advanced: dict[str, Any]
+    version: dict[str, Any]
+    docs: dict[str, Any]
+
+
+_DOCS_STABLE_VERSION_RE = re.compile(r"^\d+\.\d+\.\d+(?:\.post\d+)?$")
+_DOCS_LATEST_URL = "https://nanobot.wiki/docs/latest"
+_SKIP_FIELD = object()
+
+
+def docs_version(version: str) -> str:
+    """Map package versions to the matching public docs path."""
+    normalized = version.strip()
+    if _DOCS_STABLE_VERSION_RE.fullmatch(normalized):
+        return normalized
+    return "latest"
+
+
+def docs_payload(version: str) -> dict[str, Any]:
+    selected_version = docs_version(version)
+    base_url = f"https://nanobot.wiki/docs/{selected_version}"
+    return {
+        "version": selected_version,
+        "base_url": base_url,
+        "chat_apps_url": f"{base_url}/getting-started/chat-apps",
+        "latest_url": _DOCS_LATEST_URL,
+    }
+
+
+def system_settings_payload(
+    config: Config,
+    *,
+    config_path: Path,
+    version: str,
+) -> SystemSettingsPayload:
+    defaults = config.agents.defaults
+    exec_config = config.tools.exec
+    sandbox_status = workspace_sandbox_status(
+        restrict_to_workspace=config.tools.restrict_to_workspace,
+        workspace=config.workspace_path,
+    )
+    return {
+        "runtime": {
+            "config_path": str(config_path.expanduser()),
+            "workspace_path": str(config.workspace_path),
+            "gateway_host": config.gateway.host,
+            "gateway_port": config.gateway.port,
+            "heartbeat": {
+                "enabled": config.gateway.heartbeat.enabled,
+                "interval_s": config.gateway.heartbeat.interval_s,
+                "keep_recent_messages": config.gateway.heartbeat.keep_recent_messages,
+            },
+            "dream": {
+                "schedule": defaults.dream.describe_schedule(),
+            },
+            "unified_session": defaults.unified_session,
+        },
+        "usage": token_usage_payload(timezone_name=defaults.timezone),
+        "advanced": {
+            "restrict_to_workspace": config.tools.restrict_to_workspace,
+            "workspace_sandbox": sandbox_status.as_dict(),
+            **network_safety_payload(config),
+            "mcp_server_count": len(config.tools.mcp_servers),
+            "exec_enabled": exec_config.enable,
+            "exec_sandbox": exec_config.sandbox or None,
+            "exec_path_prepend_set": bool(exec_config.path_prepend),
+            "exec_path_append_set": bool(exec_config.path_append),
+        },
+        "version": {"current": version},
+        "docs": docs_payload(version),
+    }
+
+
+def settings_usage_payload(config: Config) -> dict[str, Any]:
+    """Return the lightweight token usage slice for Overview refreshes."""
+    return token_usage_payload(timezone_name=config.agents.defaults.timezone)
+
+
+def update_agent_system_settings(config: Config, query: QueryParams) -> tuple[bool, bool]:
+    defaults = config.agents.defaults
+    changed = False
+    restart_required = False
+
+    timezone = query_first(query, "timezone")
+    if timezone is not None:
+        timezone = timezone.strip()
+        if not timezone:
+            raise WebUISettingsError("timezone is required")
+        try:
+            ZoneInfo(timezone)
+        except Exception:
+            raise WebUISettingsError("invalid timezone") from None
+        timezone_changed = defaults.timezone != timezone
+        if timezone_changed or defaults.timezone_mode != "manual":
+            defaults.timezone = timezone
+            defaults.timezone_mode = "manual"
+            changed = True
+            restart_required = timezone_changed
+
+    tool_hint_max_length = query_first_alias(
+        query,
+        "tool_hint_max_length",
+        "toolHintMaxLength",
+    )
+    if tool_hint_max_length is not None:
+        try:
+            parsed = int(tool_hint_max_length)
+        except ValueError:
+            raise WebUISettingsError(
+                "tool_hint_max_length must be an integer"
+            ) from None
+        if parsed < 20 or parsed > 500:
+            raise WebUISettingsError(
+                "tool_hint_max_length must be between 20 and 500"
+            )
+        if defaults.tool_hint_max_length != parsed:
+            defaults.tool_hint_max_length = parsed
+            changed = True
+            restart_required = True
+    return changed, restart_required
+
+
+def save_channel_config_values(
+    config: Config,
+    name: str,
+    raw_values: dict[str, Any],
+    instance_id: str = "default",
+    *,
+    load_channel_plugin: LoadChannelPlugin,
+) -> list[str]:
+    if not name:
+        raise WebUISettingsError("missing channel name")
+    try:
+        plugin = load_channel_plugin(name)
+    except ImportError:
+        raise WebUISettingsError(f"unknown channel '{name}'", status=404) from None
+    setup_spec = channel_setup_spec(name, plugin=plugin)
+    if setup_spec is None:
+        raise WebUISettingsError(
+            f"channel '{name}' cannot be configured from WebUI",
+            status=404,
+        )
+    field_types = setup_spec.route_field_types
+    if not raw_values:
+        return []
+
+    section = getattr(config.channels, name, None)
+    channel_config = channel_instance_config(
+        plugin,
+        section,
+        instance_id=instance_id,
+    )
+    saved: list[str] = []
+    prefix = f"channels.{name}."
+    for raw_key, raw_value in raw_values.items():
+        if not raw_key:
+            raise WebUISettingsError(
+                "channel settings payload contains an invalid key"
+            )
+        field = raw_key[len(prefix) :] if raw_key.startswith(prefix) else raw_key
+        value_type = field_types.get(field)
+        if value_type is None:
+            raise WebUISettingsError(f"'{raw_key}' cannot be configured from WebUI")
+        value = coerce_channel_value(raw_key, raw_value, value_type)
+        if value is _SKIP_FIELD:
+            continue
+        assign_channel_config_value(channel_config, field, value)
+        saved.append(raw_key)
+
+    try:
+        updated_section = channel_update_instance_config(
+            plugin,
+            section,
+            channel_config,
+            instance_id=instance_id,
+        )
+    except ValueError as exc:
+        raise WebUISettingsError(
+            f"Invalid {name} configuration: {exc}",
+            status=400,
+        ) from exc
+    setattr(config.channels, name, updated_section)
+    return saved
+
+
+def coerce_channel_value(
+    raw_key: str,
+    raw_value: Any,
+    value_type: RouteFieldType,
+) -> Any:
+    if isinstance(value_type, tuple):
+        kind = value_type[0]
+        allowed = value_type[1]
+    else:
+        kind = value_type
+        allowed = None
+
+    if kind in {"string", "secret"}:
+        value = raw_value.strip() if isinstance(raw_value, str) else str(raw_value)
+        if kind == "secret" and not value:
+            return _SKIP_FIELD
+        return value
+
+    if kind == "list":
+        if raw_value is None:
+            return []
+        if isinstance(raw_value, str):
+            return [item.strip() for item in raw_value.split(",") if item.strip()]
+        if isinstance(raw_value, list):
+            return [
+                str(item).strip()
+                for item in cast(list[Any], raw_value)
+                if str(item).strip()
+            ]
+        raise WebUISettingsError(f"'{raw_key}' must be a comma-separated list")
+
+    if kind == "int":
+        if raw_value in (None, ""):
+            return _SKIP_FIELD
+        try:
+            return int(raw_value)
+        except (TypeError, ValueError) as exc:
+            raise WebUISettingsError(f"'{raw_key}' must be a number") from exc
+
+    if kind == "bool":
+        if isinstance(raw_value, bool):
+            return raw_value
+        value = str(raw_value).strip().lower()
+        if value in {"true", "1", "yes", "on"}:
+            return True
+        if value in {"false", "0", "no", "off"}:
+            return False
+        raise WebUISettingsError(f"'{raw_key}' must be true or false")
+
+    if kind == "enum":
+        value = raw_value.strip() if isinstance(raw_value, str) else str(raw_value)
+        if not value:
+            return _SKIP_FIELD
+        if allowed is None or value not in allowed:
+            options = ", ".join(sorted(allowed or ()))
+            raise WebUISettingsError(f"'{raw_key}' must be one of: {options}")
+        return value
+
+    raise WebUISettingsError(f"'{raw_key}' has an unsupported field type")
+
+
+def assign_channel_config_value(
+    channel_config: dict[str, Any],
+    field: str,
+    value: Any,
+) -> None:
+    target = channel_config
+    parts = field.split(".")
+    for part in parts[:-1]:
+        current: object = target.get(part)
+        if not isinstance(current, dict):
+            current = {}
+            target[part] = current
+        target = cast(dict[str, Any], current)
+    target[parts[-1]] = value
+
+
+def pairing_payload(
+    list_pending: ListPendingPairings,
+    last_action: dict[str, Any] | None = None,
+    *,
+    now: float | None = None,
+) -> dict[str, Any]:
+    current_time = time.time() if now is None else now
+    requests: list[dict[str, Any]] = []
+    for item in list_pending():
+        expires_at = float(item.get("expires_at", 0) or 0)
+        created_at = float(item.get("created_at", 0) or 0)
+        requests.append(
+            {
+                "code": str(item.get("code", "")),
+                "channel": str(item.get("channel", "")),
+                "sender_id": str(item.get("sender_id", "")),
+                "created_at_ms": int(created_at * 1000) if created_at else None,
+                "expires_at_ms": int(expires_at * 1000) if expires_at else None,
+                "expires_in_seconds": (
+                    max(0, int(expires_at - current_time)) if expires_at else None
+                ),
+            }
+        )
+    payload: dict[str, Any] = {"requests": requests}
+    if last_action is not None:
+        payload["last_action"] = last_action
+    return payload
+
+
+class SystemSettingsHandler:
+    """Handle channel and system commands behind a transport-neutral request DTO."""
+
+    def __init__(self, settings: WebUISettingsServices, logger: Any) -> None:
+        self.settings = settings
+        self.logger = logger
+        self._channel_connectors: dict[str, Any] = {}
+
+    async def handle(
+        self,
+        action: str,
+        request: SettingsRequest,
+        operations: SystemSettingsOperations,
+        *,
+        channel_name: str | None = None,
+        connect_action: str | None = None,
+    ) -> SettingsRouteResult:
+        if action == "cli-list":
+            return await self._cli_apps(request, operations)
+        if action.startswith("cli-"):
+            return await self._cli_apps_action(
+                request,
+                action.removeprefix("cli-"),
+                operations,
+            )
+        if action == "features-list":
+            return await self._features(operations)
+        if action in {"features-enable", "features-disable"}:
+            return await self._features_action(
+                request,
+                action.removeprefix("features-"),
+                operations,
+            )
+        if action == "channel-validate":
+            return await self._channel_validate(request, operations)
+        if action == "channel-configure":
+            return await self._channel_configure(request, operations)
+        if action == "channel-connect" and channel_name and connect_action:
+            return await self._channel_connect(
+                request,
+                channel_name,
+                connect_action,
+                operations,
+            )
+        if action == "pairing-list":
+            return SettingsRouteResult.success(pairing_payload(operations.list_pending))
+        if action in {"pairing-approve", "pairing-deny"}:
+            return self._pairing_action(
+                request,
+                action.removeprefix("pairing-"),
+                operations,
+            )
+        if action == "mcp-list":
+            return await self._mcp_presets(request, None, operations)
+        if action.startswith("mcp-"):
+            return await self._mcp_presets(
+                request,
+                action.removeprefix("mcp-"),
+                operations,
+            )
+        if action == "version-check":
+            return await self._version_check(operations)
+        return SettingsRouteResult.failure(404, "unknown settings action")
+
+    async def _cli_apps(
+        self,
+        request: SettingsRequest,
+        operations: SystemSettingsOperations,
+    ) -> SettingsRouteResult:
+        installed_only = (query_first(request.query, "installed_only") or "").lower() in {
+            "1",
+            "true",
+            "yes",
+        }
+        try:
+            payload = await operations.cli_apps_payload(
+                installed_only=installed_only,
+                config_path=self.settings.config.path,
+            )
+        except Exception:
+            self.logger.exception("failed to load CLI Apps payload")
+            return SettingsRouteResult.failure(500, "failed to load CLI Apps")
+        return SettingsRouteResult.success(payload)
+
+    async def _cli_apps_action(
+        self,
+        request: SettingsRequest,
+        action: str,
+        operations: SystemSettingsOperations,
+    ) -> SettingsRouteResult:
+        try:
+            payload = await asyncio.to_thread(
+                operations.cli_apps_action,
+                action,
+                request.query,
+                config_path=self.settings.config.path,
+            )
+        except WebUISettingsError as exc:
+            return SettingsRouteResult.failure(exc.status, exc.message)
+        except Exception as exc:
+            status = getattr(exc, "status", 500)
+            message = getattr(exc, "message", str(exc))
+            if status >= 500:
+                self.logger.exception("CLI Apps action '{}' failed", action)
+            return SettingsRouteResult.failure(status, message)
+        return SettingsRouteResult.success(payload)
+
+    async def _features(
+        self,
+        operations: SystemSettingsOperations,
+    ) -> SettingsRouteResult:
+        try:
+            payload = await asyncio.to_thread(
+                operations.nanobot_features_payload,
+                config_path=self.settings.config.path,
+            )
+        except Exception:
+            self.logger.exception("failed to load nanobot features")
+            return SettingsRouteResult.failure(500, "failed to load nanobot features")
+        return SettingsRouteResult.success(
+            self._with_channel_runtime_status(payload, operations)
+        )
+
+    def _nanobot_features_payload(
+        self,
+        operations: SystemSettingsOperations,
+    ) -> dict[str, Any]:
+        return operations.nanobot_features_payload(config_path=self.settings.config.path)
+
+    def _nanobot_features_action(
+        self,
+        action: str,
+        query: QueryParams,
+        operations: SystemSettingsOperations,
+        *,
+        allow_install: bool = True,
+    ) -> dict[str, Any]:
+        return self.settings.mutate(
+            operations.nanobot_features_action,
+            action,
+            query,
+            allow_install=allow_install,
+        )
+
+    async def _features_action(
+        self,
+        request: SettingsRequest,
+        action: str,
+        operations: SystemSettingsOperations,
+    ) -> SettingsRouteResult:
+        try:
+            payload = await asyncio.to_thread(
+                self._nanobot_features_action,
+                action,
+                request.query,
+                operations,
+                allow_install=(
+                    action != "enable"
+                    or self.allow_feature_package_install(request)
+                ),
+            )
+        except OptionalFeatureError as exc:
+            return SettingsRouteResult.failure(exc.status, exc.message)
+        except Exception as exc:
+            status = getattr(exc, "status", 500)
+            message = getattr(exc, "message", str(exc))
+            if status >= 500:
+                self.logger.exception(
+                    "nanobot feature action '{}' failed",
+                    action,
+                )
+            return SettingsRouteResult.failure(status, message)
+        payload = await self._apply_feature_runtime_change(
+            action,
+            request.query,
+            payload,
+            operations,
+        )
+        payload = self._with_channel_runtime_status(payload, operations)
+        return SettingsRouteResult.success(
+            payload,
+            decorate_restart=True,
+            restart_section="runtime",
+        )
+
+    def _with_channel_runtime_status(
+        self,
+        payload: dict[str, Any],
+        operations: SystemSettingsOperations,
+    ) -> dict[str, Any]:
+        if operations.channel_runtime_status is None:
+            return payload
+        try:
+            return with_channel_runtime_status(
+                payload,
+                operations.channel_runtime_status(),
+            )
+        except Exception:
+            self.logger.exception("failed to load channel runtime status")
+            return payload
+
+    async def _apply_feature_runtime_change(
+        self,
+        action: str,
+        query: QueryParams,
+        payload: dict[str, Any],
+        operations: SystemSettingsOperations,
+    ) -> dict[str, Any]:
+        if operations.channel_feature_action is None:
+            return payload
+        name = (query_first(query, "name") or "").strip()
+        if not name:
+            return payload
+        try:
+            instance_id = operations.nanobot_feature_instance_target(query)
+            result = operations.channel_feature_action(action, name, instance_id)
+            if inspect.isawaitable(result):
+                result = await result
+        except Exception as exc:
+            self.logger.exception("failed to apply channel '{}' without restart", name)
+            return self.feature_runtime_fallback(
+                payload,
+                message=(
+                    f"{name} channel config was saved, but hot reload failed: {exc}"
+                ),
+            )
+
+        if not isinstance(result, dict):
+            return payload
+        result = cast(dict[str, Any], result)
+        if not result.get("handled"):
+            return payload
+
+        updated = dict(payload)
+        updated["requires_restart"] = bool(result.get("requires_restart"))
+        message = result.get("message")
+        if isinstance(message, str) and message:
+            last_action = dict(updated.get("last_action") or {})
+            previous = last_action.get("message")
+            last_action["message"] = (
+                f"{previous}. {message}"
+                if isinstance(previous, str) and previous
+                else message
+            )
+            last_action["hot_reload"] = not updated["requires_restart"]
+            if "ok" in result:
+                last_action["ok"] = bool(result["ok"])
+            updated["last_action"] = last_action
+        return updated
+
+    @staticmethod
+    def feature_runtime_fallback(
+        payload: dict[str, Any],
+        *,
+        message: str,
+    ) -> dict[str, Any]:
+        updated = dict(payload)
+        updated["requires_restart"] = True
+        last_action = dict(updated.get("last_action") or {})
+        previous = last_action.get("message")
+        last_action["message"] = (
+            f"{previous}. {message}"
+            if isinstance(previous, str) and previous
+            else message
+        )
+        last_action["hot_reload"] = False
+        updated["last_action"] = last_action
+        return updated
+
+    async def _channel_configure(
+        self,
+        request: SettingsRequest,
+        operations: SystemSettingsOperations,
+    ) -> SettingsRouteResult:
+        name = (query_first(request.query, "name") or "").strip()
+        instance_id = (
+            query_first(request.query, "instance_id") or "default"
+        ).strip()
+        enable = (query_first(request.query, "enable") or "").strip().lower() in {
+            "1",
+            "true",
+            "yes",
+        }
+        try:
+            saved = await asyncio.to_thread(
+                self._save_channel_config_values,
+                name,
+                self.parse_channel_values(request),
+                instance_id,
+                operations,
+            )
+        except WebUISettingsError as exc:
+            return SettingsRouteResult.failure(exc.status, exc.message)
+        except Exception:
+            self.logger.exception("failed to save channel '{}' settings", name)
+            return SettingsRouteResult.failure(500, "failed to save channel settings")
+
+        payload: dict[str, Any] = {
+            "name": name,
+            "saved": True,
+            "saved_keys": saved,
+        }
+        if not enable:
+            features = await asyncio.to_thread(
+                self._nanobot_features_payload,
+                operations,
+            )
+            payload["nanobot_features"] = self._with_channel_runtime_status(
+                features,
+                operations,
+            )
+            return SettingsRouteResult.success(
+                payload,
+                decorate_restart=True,
+                restart_section="runtime",
+                restart_payload_key="nanobot_features",
+            )
+
+        feature_query = {"name": [name]}
+        if instance_id:
+            feature_query["instance_id"] = [instance_id]
+        try:
+            features = await asyncio.to_thread(
+                self._nanobot_features_action,
+                "enable",
+                feature_query,
+                operations,
+                allow_install=self.allow_feature_package_install(request),
+            )
+        except OptionalFeatureError as exc:
+            return SettingsRouteResult.failure(
+                exc.status,
+                f"Settings saved, but {exc.message}",
+            )
+        except Exception as exc:
+            self.logger.exception(
+                "failed to enable channel '{}' after settings save",
+                name,
+            )
+            return SettingsRouteResult.failure(
+                500,
+                f"Settings saved, but enabling {name} failed: {exc}",
+            )
+
+        features = await self._apply_feature_runtime_change(
+            "enable",
+            feature_query,
+            features,
+            operations,
+        )
+        payload["nanobot_features"] = self._with_channel_runtime_status(
+            features,
+            operations,
+        )
+        return SettingsRouteResult.success(
+            payload,
+            decorate_restart=True,
+            restart_section="runtime",
+            restart_payload_key="nanobot_features",
+        )
+
+    async def _channel_validate(
+        self,
+        request: SettingsRequest,
+        operations: SystemSettingsOperations,
+    ) -> SettingsRouteResult:
+        name = (query_first(request.query, "name") or "").strip()
+        instance_id = (
+            query_first(request.query, "instance_id") or "default"
+        ).strip()
+        try:
+            payload = await asyncio.to_thread(
+                operations.validate_channel_config,
+                name,
+                self.parse_channel_values(request),
+                instance_id=instance_id,
+            )
+        except WebUISettingsError as exc:
+            return SettingsRouteResult.failure(exc.status, exc.message)
+        except Exception:
+            self.logger.exception("failed to validate channel '{}' settings", name)
+            return SettingsRouteResult.failure(
+                500,
+                "failed to validate channel settings",
+            )
+        return SettingsRouteResult.success(payload)
+
+    @staticmethod
+    def parse_channel_values(request: SettingsRequest) -> dict[str, Any]:
+        if request.payload is None or "values" not in request.payload:
+            return {}
+        values = request.payload.get("values")
+        if not isinstance(values, dict):
+            raise WebUISettingsError(
+                "channel settings payload must be a JSON object"
+            )
+        return cast(dict[str, Any], values)
+
+    def _save_channel_config_values(
+        self,
+        name: str,
+        raw_values: dict[str, Any],
+        instance_id: str,
+        operations: SystemSettingsOperations,
+    ) -> list[str]:
+        return self.settings.config.update(
+            lambda config: save_channel_config_values(
+                config,
+                name,
+                raw_values,
+                instance_id,
+                load_channel_plugin=operations.load_channel_plugin,
+            )
+        )
+
+    async def _channel_connect(
+        self,
+        request: SettingsRequest,
+        channel_name: str,
+        action: str,
+        operations: SystemSettingsOperations,
+    ) -> SettingsRouteResult:
+        try:
+            connector = self._channel_connectors.get(channel_name)
+            if connector is None:
+                plugin = operations.load_channel_plugin(channel_name)
+                connector = plugin.load_connector()
+                self._channel_connectors[channel_name] = connector
+        except ImportError:
+            return SettingsRouteResult.failure(
+                404,
+                f"channel '{channel_name}' does not support connect",
+            )
+
+        try:
+            payload = await connector.handle(action, request.query)
+        except ChannelConnectError as exc:
+            return SettingsRouteResult.failure(exc.status, exc.message)
+        except Exception:
+            self.logger.exception(
+                "failed to run {} WebUI connect action for {}",
+                action,
+                channel_name,
+            )
+            return SettingsRouteResult.failure(
+                500,
+                f"failed to {action} {channel_name} connection",
+            )
+
+        if payload.get("status") != "succeeded":
+            return SettingsRouteResult.success(payload)
+        payload = await self._with_channel_connect_success(
+            request,
+            channel_name,
+            payload,
+            operations,
+        )
+        return SettingsRouteResult.success(
+            payload,
+            decorate_restart=True,
+            restart_section="runtime",
+            restart_payload_key="nanobot_features",
+        )
+
+    async def _with_channel_connect_success(
+        self,
+        request: SettingsRequest,
+        channel_name: str,
+        payload: dict[str, Any],
+        operations: SystemSettingsOperations,
+    ) -> dict[str, Any]:
+        target = {"name": [channel_name]}
+        if payload.get("instance_id"):
+            target["instance_id"] = [str(payload["instance_id"])]
+        try:
+            features = await asyncio.to_thread(
+                self._nanobot_features_action,
+                "enable",
+                target,
+                operations,
+                allow_install=self.allow_feature_package_install(request),
+            )
+        except OptionalFeatureError as exc:
+            features = self.feature_runtime_fallback(
+                self._nanobot_features_payload(operations),
+                message=(
+                    f"{channel_name} connected, but enabling channel support failed: "
+                    f"{exc.message}"
+                ),
+            )
+        else:
+            features = await self._apply_feature_runtime_change(
+                "enable",
+                target,
+                features,
+                operations,
+            )
+        updated = dict(payload)
+        updated["nanobot_features"] = self._with_channel_runtime_status(
+            features,
+            operations,
+        )
+        return updated
+
+    def allow_feature_package_install(self, request: SettingsRequest) -> bool:
+        if request.local_browser:
+            return True
+        try:
+            return bool(
+                self.settings.config.load().tools.webui_allow_remote_package_install
+            )
+        except Exception:
+            self.logger.exception("failed to load remote package install policy")
+            return False
+
+    def _pairing_action(
+        self,
+        request: SettingsRequest,
+        action: str,
+        operations: SystemSettingsOperations,
+    ) -> SettingsRouteResult:
+        code = (query_first(request.query, "code") or "").strip()
+        if not code:
+            return SettingsRouteResult.failure(400, "Missing pairing code")
+        if action == "approve":
+            result = operations.approve_code(code)
+            if result is None:
+                return SettingsRouteResult.failure(
+                    404,
+                    "Pairing code not found or expired",
+                )
+            channel, sender_id = result
+            return SettingsRouteResult.success(
+                pairing_payload(
+                    operations.list_pending,
+                    {
+                        "ok": True,
+                        "action": "approve",
+                        "message": f"Approved {sender_id} for {channel}",
+                        "channel": channel,
+                        "sender_id": sender_id,
+                        "code": code,
+                    },
+                )
+            )
+
+        if not operations.deny_code(code):
+            return SettingsRouteResult.failure(
+                404,
+                "Pairing code not found or expired",
+            )
+        return SettingsRouteResult.success(
+            pairing_payload(
+                operations.list_pending,
+                {
+                    "ok": True,
+                    "action": "deny",
+                    "message": f"Denied pairing code {code}",
+                    "code": code,
+                },
+            )
+        )
+
+    async def _mcp_presets(
+        self,
+        request: SettingsRequest,
+        action: str | None,
+        operations: SystemSettingsOperations,
+    ) -> SettingsRouteResult:
+        try:
+            payload = await operations.mcp_presets_action(
+                action,
+                request.query,
+                reload_mcp=operations.reload_mcp,
+                config=self.settings.config,
+            )
+        except Exception as exc:
+            status = getattr(exc, "status", 500)
+            message = getattr(exc, "message", str(exc))
+            if status >= 500:
+                self.logger.exception(
+                    "MCP preset action '{}' failed",
+                    action or "list",
+                )
+            return SettingsRouteResult.failure(status, message)
+        return SettingsRouteResult.success(
+            payload,
+            decorate_restart=action is not None,
+            restart_section="runtime" if action is not None else None,
+        )
+
+    async def _version_check(
+        self,
+        operations: SystemSettingsOperations,
+    ) -> SettingsRouteResult:
+        try:
+            update_info = await asyncio.to_thread(operations.check_for_update)
+        except Exception:
+            self.logger.exception("version check failed")
+            return SettingsRouteResult.failure(500, "version check failed")
+        return SettingsRouteResult.success({"updateAvailable": update_info})

tests/webui/test_settings_capabilities.py

@@ -0,0 +1,62 @@
+from __future__ import annotations
+
+from typing import Any
+
+from nanobot.config.schema import Config
+from nanobot.webui.settings_capabilities import (
+    capability_settings_payload,
+    update_api_settings,
+    update_image_generation_settings,
+    update_network_safety_settings,
+    update_transcription_settings,
+    update_web_search_settings,
+)
+
+
+def _oauth_status(_spec: Any) -> dict[str, Any]:
+    return {"configured": False}
+
+
+def test_capability_domain_updates_representative_settings() -> None:
+    config = Config()
+    config.providers.openrouter.api_key = "sk-test"
+
+    web_changed, web_restart = update_web_search_settings(
+        config,
+        {
+            "provider": ["duckduckgo"],
+            "max_results": ["7"],
+            "use_jina_reader": ["false"],
+        },
+    )
+    update_api_settings(
+        config,
+        {"host": ["127.0.0.2"], "port": ["8900"], "timeout": ["90"]},
+    )
+    image_changed = update_image_generation_settings(
+        config,
+        {"enabled": ["true"], "provider": ["openrouter"]},
+        oauth_status=_oauth_status,
+    )
+    transcription_changed = update_transcription_settings(
+        config,
+        {"provider": ["openrouter"], "model": ["openai/whisper-large-v3"]},
+    )
+    network_changed, access_mode = update_network_safety_settings(
+        config,
+        {
+            "webui_allow_local_service_access": ["false"],
+            "webui_default_access_mode": ["restricted"],
+        },
+    )
+    payload = capability_settings_payload(config, oauth_status=_oauth_status)
+
+    assert (web_changed, web_restart) == (True, True)
+    assert image_changed is True
+    assert transcription_changed is True
+    assert (network_changed, access_mode) == (True, "default")
+    assert payload["web_search"]["max_results"] == 7
+    assert payload["api"]["host"] == "127.0.0.2"
+    assert payload["api"]["port"] == 8900
+    assert payload["image_generation"]["enabled"] is True
+    assert payload["transcription"]["provider"] == "openrouter"

tests/webui/test_settings_models.py

@@ -0,0 +1,58 @@
+from __future__ import annotations
+
+from typing import Any
+
+from nanobot.config.schema import Config
+from nanobot.webui.settings_models import (
+    model_settings_payload,
+    update_agent_model_settings,
+    update_provider_settings,
+)
+
+
+def _oauth_status(_spec: Any) -> dict[str, Any]:
+    return {
+        "configured": False,
+        "account": None,
+        "expires_at": None,
+        "login_supported": True,
+    }
+
+
+def test_model_domain_owns_dto_and_config_updates() -> None:
+    config = Config()
+    config.providers.openrouter.api_key = "sk-before"
+
+    agent_changed = update_agent_model_settings(
+        config,
+        {
+            "model": ["openai/gpt-5.4"],
+            "provider": ["openrouter"],
+            "context_window_tokens": ["200000"],
+        },
+        oauth_status=_oauth_status,
+    )
+    provider_changed, restart_required = update_provider_settings(
+        config,
+        {
+            "provider": ["openrouter"],
+            "api_key": ["sk-after"],
+        },
+    )
+    payload = model_settings_payload(config, oauth_status=_oauth_status)
+
+    assert agent_changed is True
+    assert provider_changed is True
+    assert restart_required is False
+    assert config.agents.defaults.model == "openai/gpt-5.4"
+    assert config.agents.defaults.provider == "openrouter"
+    assert config.agents.defaults.context_window_tokens == 200_000
+    assert config.providers.openrouter.api_key == "sk-after"
+    assert set(payload) == {
+        "agent",
+        "model_presets",
+        "model_call_order",
+        "model_call_order_editable",
+        "providers",
+    }
+    assert payload["agent"]["model"] == "openai/gpt-5.4"

tests/webui/test_settings_system.py

@@ -0,0 +1,44 @@
+from __future__ import annotations
+
+from nanobot.config.schema import Config
+from nanobot.webui.settings_system import (
+    coerce_channel_value,
+    system_settings_payload,
+    update_agent_system_settings,
+)
+
+
+def test_system_domain_owns_runtime_dto_and_agent_updates(tmp_path) -> None:
+    config = Config()
+
+    changed, restart_required = update_agent_system_settings(
+        config,
+        {
+            "timezone": ["Asia/Shanghai"],
+            "tool_hint_max_length": ["120"],
+        },
+    )
+    payload = system_settings_payload(
+        config,
+        config_path=tmp_path / "config.json",
+        version="0.3.0",
+    )
+
+    assert changed is True
+    assert restart_required is True
+    assert config.agents.defaults.timezone == "Asia/Shanghai"
+    assert config.agents.defaults.timezone_mode == "manual"
+    assert config.agents.defaults.tool_hint_max_length == 120
+    assert payload["runtime"]["config_path"] == str(tmp_path / "config.json")
+    assert payload["version"] == {"current": "0.3.0"}
+    assert payload["docs"]["version"] == "0.3.0"
+    assert set(payload) == {"runtime", "usage", "advanced", "version", "docs"}
+
+
+def test_system_domain_validates_channel_field_values() -> None:
+    assert coerce_channel_value("allow_from", "alice, bob", "list") == [
+        "alice",
+        "bob",
+    ]
+    assert coerce_channel_value("enabled", "yes", "bool") is True
+    assert coerce_channel_value("port", "8765", "int") == 8765