plugins/_browser/helpers/config.py

from __future__ import annotations

from pathlib import Path
from typing import TYPE_CHECKING, Any

if TYPE_CHECKING:
    from agent import Agent


PLUGIN_NAME = "_browser"
MODEL_PRESET_KEY = "model_preset"
BROWSER_MODEL_ACTIVE_KEY = "_browser_model_active"
DEFAULT_HOMEPAGE_KEY = "default_homepage"
AUTOFOCUS_ACTIVE_PAGE_KEY = "autofocus_active_page"
TAB_SCOPE_KEY = "browser_tab_scope"
MAX_OPEN_TABS_KEY = "max_open_tabs"
RUNTIME_BACKEND_KEY = "runtime_backend"
HOST_BROWSER_PRIVACY_POLICY_KEY = "host_browser_privacy_policy"
HOST_BROWSER_PROFILE_MODE_KEY = "host_browser_profile_mode"
HOST_BROWSER_SELECTION_KEY = "host_browser_selection"
PROXY_SERVER_KEY = "proxy_server"
PROXY_BYPASS_KEY = "proxy_bypass"
PROXY_USERNAME_KEY = "proxy_username"
PROXY_PASSWORD_KEY = "proxy_password"
KEYBOARD_LAYOUT_KEY = "keyboard_layout"
KEYBOARD_VARIANT_KEY = "keyboard_variant"
RUNTIME_BACKENDS = {"container", "host_required"}
BROWSER_TAB_SCOPES = {"per_context", "shared"}
HOST_BROWSER_PRIVACY_POLICIES = {"enforce_local", "warn", "allow"}
HOST_BROWSER_PROFILE_MODES = {"existing", "agent"}
DEFAULT_BROWSER_TAB_SCOPE = "per_context"
DEFAULT_MAX_OPEN_TABS = 32
MIN_MAX_OPEN_TABS = 1
HARD_MAX_OPEN_TABS = 50
DEFAULT_HOST_BROWSER_PRIVACY_POLICY = "allow"


def _normalize_extension_paths(value: Any) -> list[str]:
    if isinstance(value, str):
        candidates = value.replace("\r\n", "\n").replace("\r", "\n").split("\n")
    elif isinstance(value, (list, tuple, set)):
        candidates = list(value)
    else:
        candidates = []

    normalized_paths: list[str] = []
    seen: set[str] = set()
    for entry in candidates:
        raw_path = str(entry or "").strip()
        if not raw_path:
            continue
        normalized = str(Path(raw_path).expanduser())
        if normalized in seen:
            continue
        seen.add(normalized)
        normalized_paths.append(normalized)
    return normalized_paths


def _normalize_model_preset(value: Any) -> str:
    return str(value or "").strip()


def _normalize_host_browser_selection(value: Any) -> str:
    raw = str(value or "").strip()
    if not raw:
        return ""
    endpoint_like = "://" in raw or (
        raw.rpartition(":")[0] and raw.rpartition(":")[2].isdigit()
    )
    if endpoint_like:
        return "".join(ch for ch in raw if ch.isprintable() and not ch.isspace())[:2048]
    normalized = raw.lower().replace(" ", "_")
    return "".join(ch for ch in normalized if ch.isalnum() or ch in {"_", "-", ":", ".", "/"})[:200]


def _normalize_default_homepage(value: Any) -> str:
    homepage = str(value or "").strip()
    return homepage or "about:blank"


def _normalize_bool(value: Any, default: bool = True) -> bool:
    if value is None:
        return default
    if isinstance(value, bool):
        return value
    if isinstance(value, (int, float)):
        return bool(value)
    normalized = str(value).strip().lower()
    if normalized in {"1", "true", "yes", "on", "enabled"}:
        return True
    if normalized in {"0", "false", "no", "off", "disabled"}:
        return False
    return default


def _normalize_int(value: Any, *, default: int, minimum: int, maximum: int) -> int:
    try:
        number = int(value)
    except (TypeError, ValueError):
        number = default
    return max(minimum, min(maximum, number))


def _normalize_xkb_token(value: Any) -> str:
    return "".join(
        ch for ch in str(value or "").strip().lower() if ch.isalnum() or ch in {"_", "-"}
    )[:32]


def _normalize_choice(value: Any, *, allowed: set[str], default: str) -> str:
    normalized = str(value or "").strip().lower().replace("-", "_")
    if normalized in allowed:
        return normalized
    return default


def _normalize_runtime_backend(value: Any) -> str:
    normalized = str(value or "").strip().lower().replace("-", "_")
    if normalized == "host_when_available":
        return "host_required"
    return _normalize_choice(normalized, allowed=RUNTIME_BACKENDS, default="container")


def _model_config_summary(config: dict[str, Any] | None) -> str:
    if not isinstance(config, dict):
        return ""
    provider = str(config.get("provider", "") or "").strip()
    model_name = str(config.get("name", "") or "").strip()
    return " / ".join(part for part in (provider, model_name) if part)


def normalize_browser_config(settings: dict[str, Any] | None) -> dict[str, Any]:
    raw = settings if isinstance(settings, dict) else {}
    extension_paths = _normalize_extension_paths(raw.get("extension_paths", []))
    keyboard_layout = _normalize_xkb_token(raw.get(KEYBOARD_LAYOUT_KEY, ""))
    keyboard_variant = (
        _normalize_xkb_token(raw.get(KEYBOARD_VARIANT_KEY, "")) if keyboard_layout else ""
    )
    return {
        "extension_paths": extension_paths,
        DEFAULT_HOMEPAGE_KEY: _normalize_default_homepage(
            raw.get(DEFAULT_HOMEPAGE_KEY, raw.get("starting_page", "about:blank"))
        ),
        AUTOFOCUS_ACTIVE_PAGE_KEY: _normalize_bool(
            raw.get(AUTOFOCUS_ACTIVE_PAGE_KEY, True),
            default=True,
        ),
        TAB_SCOPE_KEY: _normalize_choice(
            raw.get(TAB_SCOPE_KEY, DEFAULT_BROWSER_TAB_SCOPE),
            allowed=BROWSER_TAB_SCOPES,
            default=DEFAULT_BROWSER_TAB_SCOPE,
        ),
        MAX_OPEN_TABS_KEY: _normalize_int(
            raw.get(MAX_OPEN_TABS_KEY, DEFAULT_MAX_OPEN_TABS),
            default=DEFAULT_MAX_OPEN_TABS,
            minimum=MIN_MAX_OPEN_TABS,
            maximum=HARD_MAX_OPEN_TABS,
        ),
        RUNTIME_BACKEND_KEY: _normalize_runtime_backend(
            raw.get(RUNTIME_BACKEND_KEY, "container")
        ),
        HOST_BROWSER_PRIVACY_POLICY_KEY: _normalize_choice(
            raw.get(HOST_BROWSER_PRIVACY_POLICY_KEY, DEFAULT_HOST_BROWSER_PRIVACY_POLICY),
            allowed=HOST_BROWSER_PRIVACY_POLICIES,
            default=DEFAULT_HOST_BROWSER_PRIVACY_POLICY,
        ),
        HOST_BROWSER_PROFILE_MODE_KEY: _normalize_choice(
            raw.get(HOST_BROWSER_PROFILE_MODE_KEY, "existing"),
            allowed=HOST_BROWSER_PROFILE_MODES,
            default="existing",
        ),
        HOST_BROWSER_SELECTION_KEY: _normalize_host_browser_selection(
            raw.get(HOST_BROWSER_SELECTION_KEY, raw.get("host_browser_choice", ""))
        ),
        PROXY_SERVER_KEY: str(raw.get(PROXY_SERVER_KEY, "") or "").strip()[:2048],
        PROXY_BYPASS_KEY: str(raw.get(PROXY_BYPASS_KEY, "") or "").strip()[:4096],
        PROXY_USERNAME_KEY: str(raw.get(PROXY_USERNAME_KEY, "") or "")[:1024],
        PROXY_PASSWORD_KEY: str(raw.get(PROXY_PASSWORD_KEY, "") or "")[:4096],
        KEYBOARD_LAYOUT_KEY: keyboard_layout,
        KEYBOARD_VARIANT_KEY: keyboard_variant,
        MODEL_PRESET_KEY: _normalize_model_preset(raw.get(MODEL_PRESET_KEY, "")),
    }


def browser_runtime_config(settings: dict[str, Any] | None) -> dict[str, Any]:
    config = normalize_browser_config(settings)
    return {
        "extension_paths": config["extension_paths"],
        PROXY_SERVER_KEY: config[PROXY_SERVER_KEY],
        PROXY_BYPASS_KEY: config[PROXY_BYPASS_KEY],
        PROXY_USERNAME_KEY: config[PROXY_USERNAME_KEY],
        PROXY_PASSWORD_KEY: config[PROXY_PASSWORD_KEY],
        KEYBOARD_LAYOUT_KEY: config[KEYBOARD_LAYOUT_KEY],
        KEYBOARD_VARIANT_KEY: config[KEYBOARD_VARIANT_KEY],
    }


def get_browser_config(agent: "Agent | None" = None) -> dict[str, Any]:
    from helpers import plugins

    return normalize_browser_config(plugins.get_plugin_config(PLUGIN_NAME, agent=agent) or {})


def get_browser_model_preset_name(
    agent: "Agent | None" = None,
    settings: dict[str, Any] | None = None,
) -> str:
    config = (
        normalize_browser_config(settings)
        if settings is not None
        else get_browser_config(agent=agent)
    )
    return str(config.get(MODEL_PRESET_KEY, "") or "").strip()


def get_browser_model_preset_options(
    agent: "Agent | None" = None,
    settings: dict[str, Any] | None = None,
) -> list[dict[str, Any]]:
    from plugins._model_config.helpers import model_config

    selected_name = get_browser_model_preset_name(agent=agent, settings=settings)
    options: list[dict[str, Any]] = []
    found_selected = False

    for preset in model_config.get_presets():
        name = str(preset.get("name", "") or "").strip()
        if not name:
            continue
        if name == selected_name:
            found_selected = True
        chat_cfg = preset.get("chat", {}) if isinstance(preset, dict) else {}
        if not isinstance(chat_cfg, dict):
            chat_cfg = {}
        summary = _model_config_summary(chat_cfg)
        options.append(
            {
                "name": name,
                "label": name,
                "missing": False,
                "summary": summary,
            }
        )

    if selected_name and not found_selected:
        options.append(
            {
                "name": selected_name,
                "label": f"{selected_name} (missing)",
                "missing": True,
                "summary": "",
            }
        )

    return options


def get_browser_main_model_summary(agent: "Agent | None" = None) -> str:
    from plugins._model_config.helpers import model_config

    return _model_config_summary(model_config.get_chat_model_config(agent))


def resolve_browser_model_selection(
    agent: "Agent | None" = None,
    settings: dict[str, Any] | None = None,
) -> dict[str, Any]:
    from plugins._model_config.helpers import model_config

    preset_name = get_browser_model_preset_name(agent=agent, settings=settings)
    if preset_name:
        preset = model_config.get_preset_by_name(preset_name)
        if isinstance(preset, dict):
            if hasattr(model_config, "build_config_from_preset"):
                default_preset = model_config.get_preset_by_name(
                    model_config.DEFAULT_PRESET_NAME
                ) or {}
                base_config = (
                    model_config.preset_to_config(default_preset)
                    if hasattr(model_config, "preset_to_config")
                    else {}
                )
                preset_config = model_config.build_config_from_preset(
                    preset,
                    base_config,
                    strip_api_key=False,
                    slots=("chat",),
                )
                chat_cfg = preset_config.get("chat_model", {})
            else:
                chat_cfg = preset.get("chat", {})
            if isinstance(chat_cfg, dict) and (
                str(chat_cfg.get("provider", "") or "").strip()
                or str(chat_cfg.get("name", "") or "").strip()
            ):
                return {
                    "config": chat_cfg,
                    "source_kind": "preset",
                    "source_label": f"Preset '{preset_name}' via _model_config",
                    "selected_preset_name": preset_name,
                    "preset_status": "active",
                    "warning": "",
                }
            return {
                "config": model_config.get_chat_model_config(agent),
                "source_kind": "main",
                "source_label": "Main Model via _model_config",
                "selected_preset_name": preset_name,
                "preset_status": "invalid",
                "warning": (
                    f"Configured browser preset '{preset_name}' does not define a chat model. "
                    "Falling back to the Main Model."
                ),
            }

        return {
            "config": model_config.get_chat_model_config(agent),
            "source_kind": "main",
            "source_label": "Main Model via _model_config",
            "selected_preset_name": preset_name,
            "preset_status": "missing",
            "warning": (
                f"Configured browser preset '{preset_name}' was not found. "
                "Falling back to the Main Model."
            ),
        }

    return {
        "config": model_config.get_chat_model_config(agent),
        "source_kind": "main",
        "source_label": "Main Model via _model_config",
        "selected_preset_name": "",
        "preset_status": "none",
        "warning": "",
    }


def activate_browser_model(agent: "Agent") -> dict[str, Any]:
    selection = resolve_browser_model_selection(agent=agent)
    agent.set_data(
        BROWSER_MODEL_ACTIVE_KEY,
        selection["selected_preset_name"]
        if selection["source_kind"] == "preset"
        else "",
    )
    return selection


def clear_browser_model(agent: "Agent") -> None:
    agent.set_data(BROWSER_MODEL_ACTIVE_KEY, "")


def browser_model_is_active(agent: "Agent") -> bool:
    return bool(agent.get_data(BROWSER_MODEL_ACTIVE_KEY))


def resolve_browser_model(
    agent: "Agent",
    settings: dict[str, Any] | None = None,
    fallback: Any = None,
):
    selection = resolve_browser_model_selection(agent=agent, settings=settings)
    if selection["source_kind"] == "main":
        clear_browser_model(agent)
        return fallback if fallback is not None else agent.get_chat_model()

    import models
    from plugins._model_config.helpers import model_config

    model_config_object = model_config.build_model_config(
        selection["config"],
        models.ModelType.CHAT,
    )
    return models.get_chat_model(
        model_config_object.provider,
        model_config_object.name,
        model_config=model_config_object,
        **model_config_object.build_kwargs(),
    )


def describe_browser_extensions(settings: dict[str, Any] | None) -> dict[str, Any]:
    config = normalize_browser_config(settings)
    path_details: list[dict[str, Any]] = []
    for extension_path in config["extension_paths"]:
        path = Path(extension_path)
        exists = path.exists()
        is_dir = path.is_dir() if exists else False
        path_details.append(
            {
                "path": extension_path,
                "exists": exists,
                "is_dir": is_dir,
                "loadable": exists and is_dir,
            }
        )

    active_paths = [item["path"] for item in path_details if item["loadable"]]
    invalid_paths = [item["path"] for item in path_details if not item["loadable"]]
    active = bool(active_paths)

    warnings: list[str] = []
    if config["extension_paths"] and not active_paths:
        warnings.append(
            "None of the enabled extension directories are readable unpacked folders."
        )
    elif invalid_paths:
        warnings.append(
            "Some configured extension directories are missing or not directories, so they will be skipped."
        )

    return {
        "active": active,
        "configured_paths": config["extension_paths"],
        "active_paths": active_paths,
        "invalid_paths": invalid_paths,
        "path_details": path_details,
        "active_path_count": len(active_paths),
        "warnings": warnings,
    }


def build_browser_launch_config(settings: dict[str, Any] | None) -> dict[str, Any]:
    config = normalize_browser_config(settings)
    extensions = describe_browser_extensions(config)
    args = ["--hide-crash-restore-bubble"]
    channel: str | None = None
    browser_mode = "chromium"
    proxy = None

    if extensions["active"]:
        joined_paths = ",".join(extensions["active_paths"])
        args.extend(
            [
                f"--disable-extensions-except={joined_paths}",
                f"--load-extension={joined_paths}",
            ]
        )

    if config[PROXY_SERVER_KEY]:
        proxy = {"server": config[PROXY_SERVER_KEY]}
        for config_key, proxy_key in (
            (PROXY_BYPASS_KEY, "bypass"),
            (PROXY_USERNAME_KEY, "username"),
            (PROXY_PASSWORD_KEY, "password"),
        ):
            if config[config_key]:
                proxy[proxy_key] = config[config_key]

    return {
        "args": args,
        "proxy": proxy,
        "browser_mode": browser_mode,
        "channel": channel,
        "extensions": extensions,
        "requires_full_browser": True,
    }