mirror of
https://github.com/agent0ai/agent-zero.git
synced 2026-10-19 15:33:54 +00:00
5a1698ce62
Keep quick settings focused on the model preset, Host/Container choice, native page zoom, and installed Chrome extensions. Show a dimmed host configuration link only in Host mode, and match the Browser and Zoom labels to the preset label. Move extension creation, installation, and review into Browser config. Open Chromium management pages for browsing data, history, passwords, autofill, extensions, and downloads through the existing Browser surface. Honor native download folder and save-dialog preferences, and add annotation screenshots as an opt-in setting that defaults off. Update the Browser contracts and regression coverage. Validation: 201 browser/tool tests and 29 host-connector tests passed, with the host suite isolated from existing module stubs. Live checks on localhost:32083 covered settings navigation, runtime selection, zoom, annotation attachments, download preferences, matching typography, and narrow layouts.
485 lines
17 KiB
Python
485 lines
17 KiB
Python
from __future__ import annotations
|
|
|
|
import math
|
|
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"
|
|
ANNOTATION_SCREENSHOTS_KEY = "annotation_screenshots"
|
|
TAB_SCOPE_KEY = "browser_tab_scope"
|
|
MAX_OPEN_TABS_KEY = "max_open_tabs"
|
|
EVALUATE_TIMEOUT_KEY = "evaluate_timeout_seconds"
|
|
DEFAULT_EVALUATE_TIMEOUT_SECONDS = 30.0
|
|
MIN_EVALUATE_TIMEOUT_SECONDS = 0.1
|
|
MAX_EVALUATE_TIMEOUT_SECONDS = 60.0
|
|
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_evaluate_timeout(value: Any) -> float:
|
|
try:
|
|
timeout = float(value) if not isinstance(value, bool) else 0.0
|
|
except (TypeError, ValueError, OverflowError):
|
|
timeout = 0.0
|
|
if not math.isfinite(timeout) or not (
|
|
MIN_EVALUATE_TIMEOUT_SECONDS <= timeout <= MAX_EVALUATE_TIMEOUT_SECONDS
|
|
):
|
|
raise ValueError(
|
|
f"{EVALUATE_TIMEOUT_KEY} must be between "
|
|
f"{MIN_EVALUATE_TIMEOUT_SECONDS:g} and {MAX_EVALUATE_TIMEOUT_SECONDS:g} seconds"
|
|
)
|
|
return timeout
|
|
|
|
|
|
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,
|
|
),
|
|
ANNOTATION_SCREENSHOTS_KEY: _normalize_bool(
|
|
raw.get(ANNOTATION_SCREENSHOTS_KEY), default=False,
|
|
),
|
|
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,
|
|
),
|
|
EVALUATE_TIMEOUT_KEY: normalize_evaluate_timeout(
|
|
raw.get(EVALUATE_TIMEOUT_KEY, DEFAULT_EVALUATE_TIMEOUT_SECONDS)
|
|
),
|
|
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,
|
|
}
|