| 1 | from __future__ import annotations |
| 2 | |
| 3 | from pathlib import Path |
| 4 | from typing import TYPE_CHECKING, Any |
| 5 | |
| 6 | if TYPE_CHECKING: |
| 7 | from agent import Agent |
| 8 | |
| 9 | |
| 10 | PLUGIN_NAME = "_browser" |
| 11 | MODEL_PRESET_KEY = "model_preset" |
| 12 | BROWSER_MODEL_ACTIVE_KEY = "_browser_model_active" |
| 13 | DEFAULT_HOMEPAGE_KEY = "default_homepage" |
| 14 | AUTOFOCUS_ACTIVE_PAGE_KEY = "autofocus_active_page" |
| 15 | TAB_SCOPE_KEY = "browser_tab_scope" |
| 16 | MAX_OPEN_TABS_KEY = "max_open_tabs" |
| 17 | RUNTIME_BACKEND_KEY = "runtime_backend" |
| 18 | HOST_BROWSER_PRIVACY_POLICY_KEY = "host_browser_privacy_policy" |
| 19 | HOST_BROWSER_PROFILE_MODE_KEY = "host_browser_profile_mode" |
| 20 | HOST_BROWSER_SELECTION_KEY = "host_browser_selection" |
| 21 | PROXY_SERVER_KEY = "proxy_server" |
| 22 | PROXY_BYPASS_KEY = "proxy_bypass" |
| 23 | PROXY_USERNAME_KEY = "proxy_username" |
| 24 | PROXY_PASSWORD_KEY = "proxy_password" |
| 25 | KEYBOARD_LAYOUT_KEY = "keyboard_layout" |
| 26 | KEYBOARD_VARIANT_KEY = "keyboard_variant" |
| 27 | RUNTIME_BACKENDS = {"container", "host_required"} |
| 28 | BROWSER_TAB_SCOPES = {"per_context", "shared"} |
| 29 | HOST_BROWSER_PRIVACY_POLICIES = {"enforce_local", "warn", "allow"} |
| 30 | HOST_BROWSER_PROFILE_MODES = {"existing", "agent"} |
| 31 | DEFAULT_BROWSER_TAB_SCOPE = "per_context" |
| 32 | DEFAULT_MAX_OPEN_TABS = 32 |
| 33 | MIN_MAX_OPEN_TABS = 1 |
| 34 | HARD_MAX_OPEN_TABS = 50 |
| 35 | DEFAULT_HOST_BROWSER_PRIVACY_POLICY = "allow" |
| 36 | |
| 37 | |
| 38 | def _normalize_extension_paths(value: Any) -> list[str]: |
| 39 | if isinstance(value, str): |
| 40 | candidates = value.replace("\r\n", "\n").replace("\r", "\n").split("\n") |
| 41 | elif isinstance(value, (list, tuple, set)): |
| 42 | candidates = list(value) |
| 43 | else: |
| 44 | candidates = [] |
| 45 | |
| 46 | normalized_paths: list[str] = [] |
| 47 | seen: set[str] = set() |
| 48 | for entry in candidates: |
| 49 | raw_path = str(entry or "").strip() |
| 50 | if not raw_path: |
| 51 | continue |
| 52 | normalized = str(Path(raw_path).expanduser()) |
| 53 | if normalized in seen: |
| 54 | continue |
| 55 | seen.add(normalized) |
| 56 | normalized_paths.append(normalized) |
| 57 | return normalized_paths |
| 58 | |
| 59 | |
| 60 | def _normalize_model_preset(value: Any) -> str: |
| 61 | return str(value or "").strip() |
| 62 | |
| 63 | |
| 64 | def _normalize_host_browser_selection(value: Any) -> str: |
| 65 | raw = str(value or "").strip() |
| 66 | if not raw: |
| 67 | return "" |
| 68 | endpoint_like = "://" in raw or ( |
| 69 | raw.rpartition(":")[0] and raw.rpartition(":")[2].isdigit() |
| 70 | ) |
| 71 | if endpoint_like: |
| 72 | return "".join(ch for ch in raw if ch.isprintable() and not ch.isspace())[:2048] |
| 73 | normalized = raw.lower().replace(" ", "_") |
| 74 | return "".join(ch for ch in normalized if ch.isalnum() or ch in {"_", "-", ":", ".", "/"})[:200] |
| 75 | |
| 76 | |
| 77 | def _normalize_default_homepage(value: Any) -> str: |
| 78 | homepage = str(value or "").strip() |
| 79 | return homepage or "about:blank" |
| 80 | |
| 81 | |
| 82 | def _normalize_bool(value: Any, default: bool = True) -> bool: |
| 83 | if value is None: |
| 84 | return default |
| 85 | if isinstance(value, bool): |
| 86 | return value |
| 87 | if isinstance(value, (int, float)): |
| 88 | return bool(value) |
| 89 | normalized = str(value).strip().lower() |
| 90 | if normalized in {"1", "true", "yes", "on", "enabled"}: |
| 91 | return True |
| 92 | if normalized in {"0", "false", "no", "off", "disabled"}: |
| 93 | return False |
| 94 | return default |
| 95 | |
| 96 | |
| 97 | def _normalize_int(value: Any, *, default: int, minimum: int, maximum: int) -> int: |
| 98 | try: |
| 99 | number = int(value) |
| 100 | except (TypeError, ValueError): |
| 101 | number = default |
| 102 | return max(minimum, min(maximum, number)) |
| 103 | |
| 104 | |
| 105 | def _normalize_xkb_token(value: Any) -> str: |
| 106 | return "".join( |
| 107 | ch for ch in str(value or "").strip().lower() if ch.isalnum() or ch in {"_", "-"} |
| 108 | )[:32] |
| 109 | |
| 110 | |
| 111 | def _normalize_choice(value: Any, *, allowed: set[str], default: str) -> str: |
| 112 | normalized = str(value or "").strip().lower().replace("-", "_") |
| 113 | if normalized in allowed: |
| 114 | return normalized |
| 115 | return default |
| 116 | |
| 117 | |
| 118 | def _normalize_runtime_backend(value: Any) -> str: |
| 119 | normalized = str(value or "").strip().lower().replace("-", "_") |
| 120 | if normalized == "host_when_available": |
| 121 | return "host_required" |
| 122 | return _normalize_choice(normalized, allowed=RUNTIME_BACKENDS, default="container") |
| 123 | |
| 124 | |
| 125 | def _model_config_summary(config: dict[str, Any] | None) -> str: |
| 126 | if not isinstance(config, dict): |
| 127 | return "" |
| 128 | provider = str(config.get("provider", "") or "").strip() |
| 129 | model_name = str(config.get("name", "") or "").strip() |
| 130 | return " / ".join(part for part in (provider, model_name) if part) |
| 131 | |
| 132 | |
| 133 | def normalize_browser_config(settings: dict[str, Any] | None) -> dict[str, Any]: |
| 134 | raw = settings if isinstance(settings, dict) else {} |
| 135 | extension_paths = _normalize_extension_paths(raw.get("extension_paths", [])) |
| 136 | keyboard_layout = _normalize_xkb_token(raw.get(KEYBOARD_LAYOUT_KEY, "")) |
| 137 | keyboard_variant = ( |
| 138 | _normalize_xkb_token(raw.get(KEYBOARD_VARIANT_KEY, "")) if keyboard_layout else "" |
| 139 | ) |
| 140 | return { |
| 141 | "extension_paths": extension_paths, |
| 142 | DEFAULT_HOMEPAGE_KEY: _normalize_default_homepage( |
| 143 | raw.get(DEFAULT_HOMEPAGE_KEY, raw.get("starting_page", "about:blank")) |
| 144 | ), |
| 145 | AUTOFOCUS_ACTIVE_PAGE_KEY: _normalize_bool( |
| 146 | raw.get(AUTOFOCUS_ACTIVE_PAGE_KEY, True), |
| 147 | default=True, |
| 148 | ), |
| 149 | TAB_SCOPE_KEY: _normalize_choice( |
| 150 | raw.get(TAB_SCOPE_KEY, DEFAULT_BROWSER_TAB_SCOPE), |
| 151 | allowed=BROWSER_TAB_SCOPES, |
| 152 | default=DEFAULT_BROWSER_TAB_SCOPE, |
| 153 | ), |
| 154 | MAX_OPEN_TABS_KEY: _normalize_int( |
| 155 | raw.get(MAX_OPEN_TABS_KEY, DEFAULT_MAX_OPEN_TABS), |
| 156 | default=DEFAULT_MAX_OPEN_TABS, |
| 157 | minimum=MIN_MAX_OPEN_TABS, |
| 158 | maximum=HARD_MAX_OPEN_TABS, |
| 159 | ), |
| 160 | RUNTIME_BACKEND_KEY: _normalize_runtime_backend( |
| 161 | raw.get(RUNTIME_BACKEND_KEY, "container") |
| 162 | ), |
| 163 | HOST_BROWSER_PRIVACY_POLICY_KEY: _normalize_choice( |
| 164 | raw.get(HOST_BROWSER_PRIVACY_POLICY_KEY, DEFAULT_HOST_BROWSER_PRIVACY_POLICY), |
| 165 | allowed=HOST_BROWSER_PRIVACY_POLICIES, |
| 166 | default=DEFAULT_HOST_BROWSER_PRIVACY_POLICY, |
| 167 | ), |
| 168 | HOST_BROWSER_PROFILE_MODE_KEY: _normalize_choice( |
| 169 | raw.get(HOST_BROWSER_PROFILE_MODE_KEY, "existing"), |
| 170 | allowed=HOST_BROWSER_PROFILE_MODES, |
| 171 | default="existing", |
| 172 | ), |
| 173 | HOST_BROWSER_SELECTION_KEY: _normalize_host_browser_selection( |
| 174 | raw.get(HOST_BROWSER_SELECTION_KEY, raw.get("host_browser_choice", "")) |
| 175 | ), |
| 176 | PROXY_SERVER_KEY: str(raw.get(PROXY_SERVER_KEY, "") or "").strip()[:2048], |
| 177 | PROXY_BYPASS_KEY: str(raw.get(PROXY_BYPASS_KEY, "") or "").strip()[:4096], |
| 178 | PROXY_USERNAME_KEY: str(raw.get(PROXY_USERNAME_KEY, "") or "")[:1024], |
| 179 | PROXY_PASSWORD_KEY: str(raw.get(PROXY_PASSWORD_KEY, "") or "")[:4096], |
| 180 | KEYBOARD_LAYOUT_KEY: keyboard_layout, |
| 181 | KEYBOARD_VARIANT_KEY: keyboard_variant, |
| 182 | MODEL_PRESET_KEY: _normalize_model_preset(raw.get(MODEL_PRESET_KEY, "")), |
| 183 | } |
| 184 | |
| 185 | |
| 186 | def browser_runtime_config(settings: dict[str, Any] | None) -> dict[str, Any]: |
| 187 | config = normalize_browser_config(settings) |
| 188 | return { |
| 189 | "extension_paths": config["extension_paths"], |
| 190 | PROXY_SERVER_KEY: config[PROXY_SERVER_KEY], |
| 191 | PROXY_BYPASS_KEY: config[PROXY_BYPASS_KEY], |
| 192 | PROXY_USERNAME_KEY: config[PROXY_USERNAME_KEY], |
| 193 | PROXY_PASSWORD_KEY: config[PROXY_PASSWORD_KEY], |
| 194 | KEYBOARD_LAYOUT_KEY: config[KEYBOARD_LAYOUT_KEY], |
| 195 | KEYBOARD_VARIANT_KEY: config[KEYBOARD_VARIANT_KEY], |
| 196 | } |
| 197 | |
| 198 | |
| 199 | def get_browser_config(agent: "Agent | None" = None) -> dict[str, Any]: |
| 200 | from helpers import plugins |
| 201 | |
| 202 | return normalize_browser_config(plugins.get_plugin_config(PLUGIN_NAME, agent=agent) or {}) |
| 203 | |
| 204 | |
| 205 | def get_browser_model_preset_name( |
| 206 | agent: "Agent | None" = None, |
| 207 | settings: dict[str, Any] | None = None, |
| 208 | ) -> str: |
| 209 | config = ( |
| 210 | normalize_browser_config(settings) |
| 211 | if settings is not None |
| 212 | else get_browser_config(agent=agent) |
| 213 | ) |
| 214 | return str(config.get(MODEL_PRESET_KEY, "") or "").strip() |
| 215 | |
| 216 | |
| 217 | def get_browser_model_preset_options( |
| 218 | agent: "Agent | None" = None, |
| 219 | settings: dict[str, Any] | None = None, |
| 220 | ) -> list[dict[str, Any]]: |
| 221 | from plugins._model_config.helpers import model_config |
| 222 | |
| 223 | selected_name = get_browser_model_preset_name(agent=agent, settings=settings) |
| 224 | options: list[dict[str, Any]] = [] |
| 225 | found_selected = False |
| 226 | |
| 227 | for preset in model_config.get_presets(): |
| 228 | name = str(preset.get("name", "") or "").strip() |
| 229 | if not name: |
| 230 | continue |
| 231 | if name == selected_name: |
| 232 | found_selected = True |
| 233 | chat_cfg = preset.get("chat", {}) if isinstance(preset, dict) else {} |
| 234 | if not isinstance(chat_cfg, dict): |
| 235 | chat_cfg = {} |
| 236 | summary = _model_config_summary(chat_cfg) |
| 237 | options.append( |
| 238 | { |
| 239 | "name": name, |
| 240 | "label": name, |
| 241 | "missing": False, |
| 242 | "summary": summary, |
| 243 | } |
| 244 | ) |
| 245 | |
| 246 | if selected_name and not found_selected: |
| 247 | options.append( |
| 248 | { |
| 249 | "name": selected_name, |
| 250 | "label": f"{selected_name} (missing)", |
| 251 | "missing": True, |
| 252 | "summary": "", |
| 253 | } |
| 254 | ) |
| 255 | |
| 256 | return options |
| 257 | |
| 258 | |
| 259 | def get_browser_main_model_summary(agent: "Agent | None" = None) -> str: |
| 260 | from plugins._model_config.helpers import model_config |
| 261 | |
| 262 | return _model_config_summary(model_config.get_chat_model_config(agent)) |
| 263 | |
| 264 | |
| 265 | def resolve_browser_model_selection( |
| 266 | agent: "Agent | None" = None, |
| 267 | settings: dict[str, Any] | None = None, |
| 268 | ) -> dict[str, Any]: |
| 269 | from plugins._model_config.helpers import model_config |
| 270 | |
| 271 | preset_name = get_browser_model_preset_name(agent=agent, settings=settings) |
| 272 | if preset_name: |
| 273 | preset = model_config.get_preset_by_name(preset_name) |
| 274 | if isinstance(preset, dict): |
| 275 | if hasattr(model_config, "build_config_from_preset"): |
| 276 | default_preset = model_config.get_preset_by_name( |
| 277 | model_config.DEFAULT_PRESET_NAME |
| 278 | ) or {} |
| 279 | base_config = ( |
| 280 | model_config.preset_to_config(default_preset) |
| 281 | if hasattr(model_config, "preset_to_config") |
| 282 | else {} |
| 283 | ) |
| 284 | preset_config = model_config.build_config_from_preset( |
| 285 | preset, |
| 286 | base_config, |
| 287 | strip_api_key=False, |
| 288 | slots=("chat",), |
| 289 | ) |
| 290 | chat_cfg = preset_config.get("chat_model", {}) |
| 291 | else: |
| 292 | chat_cfg = preset.get("chat", {}) |
| 293 | if isinstance(chat_cfg, dict) and ( |
| 294 | str(chat_cfg.get("provider", "") or "").strip() |
| 295 | or str(chat_cfg.get("name", "") or "").strip() |
| 296 | ): |
| 297 | return { |
| 298 | "config": chat_cfg, |
| 299 | "source_kind": "preset", |
| 300 | "source_label": f"Preset '{preset_name}' via _model_config", |
| 301 | "selected_preset_name": preset_name, |
| 302 | "preset_status": "active", |
| 303 | "warning": "", |
| 304 | } |
| 305 | return { |
| 306 | "config": model_config.get_chat_model_config(agent), |
| 307 | "source_kind": "main", |
| 308 | "source_label": "Main Model via _model_config", |
| 309 | "selected_preset_name": preset_name, |
| 310 | "preset_status": "invalid", |
| 311 | "warning": ( |
| 312 | f"Configured browser preset '{preset_name}' does not define a chat model. " |
| 313 | "Falling back to the Main Model." |
| 314 | ), |
| 315 | } |
| 316 | |
| 317 | return { |
| 318 | "config": model_config.get_chat_model_config(agent), |
| 319 | "source_kind": "main", |
| 320 | "source_label": "Main Model via _model_config", |
| 321 | "selected_preset_name": preset_name, |
| 322 | "preset_status": "missing", |
| 323 | "warning": ( |
| 324 | f"Configured browser preset '{preset_name}' was not found. " |
| 325 | "Falling back to the Main Model." |
| 326 | ), |
| 327 | } |
| 328 | |
| 329 | return { |
| 330 | "config": model_config.get_chat_model_config(agent), |
| 331 | "source_kind": "main", |
| 332 | "source_label": "Main Model via _model_config", |
| 333 | "selected_preset_name": "", |
| 334 | "preset_status": "none", |
| 335 | "warning": "", |
| 336 | } |
| 337 | |
| 338 | |
| 339 | def activate_browser_model(agent: "Agent") -> dict[str, Any]: |
| 340 | selection = resolve_browser_model_selection(agent=agent) |
| 341 | agent.set_data( |
| 342 | BROWSER_MODEL_ACTIVE_KEY, |
| 343 | selection["selected_preset_name"] |
| 344 | if selection["source_kind"] == "preset" |
| 345 | else "", |
| 346 | ) |
| 347 | return selection |
| 348 | |
| 349 | |
| 350 | def clear_browser_model(agent: "Agent") -> None: |
| 351 | agent.set_data(BROWSER_MODEL_ACTIVE_KEY, "") |
| 352 | |
| 353 | |
| 354 | def browser_model_is_active(agent: "Agent") -> bool: |
| 355 | return bool(agent.get_data(BROWSER_MODEL_ACTIVE_KEY)) |
| 356 | |
| 357 | |
| 358 | def resolve_browser_model( |
| 359 | agent: "Agent", |
| 360 | settings: dict[str, Any] | None = None, |
| 361 | fallback: Any = None, |
| 362 | ): |
| 363 | selection = resolve_browser_model_selection(agent=agent, settings=settings) |
| 364 | if selection["source_kind"] == "main": |
| 365 | clear_browser_model(agent) |
| 366 | return fallback if fallback is not None else agent.get_chat_model() |
| 367 | |
| 368 | import models |
| 369 | from plugins._model_config.helpers import model_config |
| 370 | |
| 371 | model_config_object = model_config.build_model_config( |
| 372 | selection["config"], |
| 373 | models.ModelType.CHAT, |
| 374 | ) |
| 375 | return models.get_chat_model( |
| 376 | model_config_object.provider, |
| 377 | model_config_object.name, |
| 378 | model_config=model_config_object, |
| 379 | **model_config_object.build_kwargs(), |
| 380 | ) |
| 381 | |
| 382 | |
| 383 | def describe_browser_extensions(settings: dict[str, Any] | None) -> dict[str, Any]: |
| 384 | config = normalize_browser_config(settings) |
| 385 | path_details: list[dict[str, Any]] = [] |
| 386 | for extension_path in config["extension_paths"]: |
| 387 | path = Path(extension_path) |
| 388 | exists = path.exists() |
| 389 | is_dir = path.is_dir() if exists else False |
| 390 | path_details.append( |
| 391 | { |
| 392 | "path": extension_path, |
| 393 | "exists": exists, |
| 394 | "is_dir": is_dir, |
| 395 | "loadable": exists and is_dir, |
| 396 | } |
| 397 | ) |
| 398 | |
| 399 | active_paths = [item["path"] for item in path_details if item["loadable"]] |
| 400 | invalid_paths = [item["path"] for item in path_details if not item["loadable"]] |
| 401 | active = bool(active_paths) |
| 402 | |
| 403 | warnings: list[str] = [] |
| 404 | if config["extension_paths"] and not active_paths: |
| 405 | warnings.append( |
| 406 | "None of the enabled extension directories are readable unpacked folders." |
| 407 | ) |
| 408 | elif invalid_paths: |
| 409 | warnings.append( |
| 410 | "Some configured extension directories are missing or not directories, so they will be skipped." |
| 411 | ) |
| 412 | |
| 413 | return { |
| 414 | "active": active, |
| 415 | "configured_paths": config["extension_paths"], |
| 416 | "active_paths": active_paths, |
| 417 | "invalid_paths": invalid_paths, |
| 418 | "path_details": path_details, |
| 419 | "active_path_count": len(active_paths), |
| 420 | "warnings": warnings, |
| 421 | } |
| 422 | |
| 423 | |
| 424 | def build_browser_launch_config(settings: dict[str, Any] | None) -> dict[str, Any]: |
| 425 | config = normalize_browser_config(settings) |
| 426 | extensions = describe_browser_extensions(config) |
| 427 | args = ["--hide-crash-restore-bubble"] |
| 428 | channel: str | None = None |
| 429 | browser_mode = "chromium" |
| 430 | proxy = None |
| 431 | |
| 432 | if extensions["active"]: |
| 433 | joined_paths = ",".join(extensions["active_paths"]) |
| 434 | args.extend( |
| 435 | [ |
| 436 | f"--disable-extensions-except={joined_paths}", |
| 437 | f"--load-extension={joined_paths}", |
| 438 | ] |
| 439 | ) |
| 440 | |
| 441 | if config[PROXY_SERVER_KEY]: |
| 442 | proxy = {"server": config[PROXY_SERVER_KEY]} |
| 443 | for config_key, proxy_key in ( |
| 444 | (PROXY_BYPASS_KEY, "bypass"), |
| 445 | (PROXY_USERNAME_KEY, "username"), |
| 446 | (PROXY_PASSWORD_KEY, "password"), |
| 447 | ): |
| 448 | if config[config_key]: |
| 449 | proxy[proxy_key] = config[config_key] |
| 450 | |
| 451 | return { |
| 452 | "args": args, |
| 453 | "proxy": proxy, |
| 454 | "browser_mode": browser_mode, |
| 455 | "channel": channel, |
| 456 | "extensions": extensions, |
| 457 | "requires_full_browser": True, |
| 458 | } |