Merge branch 'bs5-c' into hermes/hermes-b802e898
# Conflicts: # tests/tools/test_bot_desktop_install.py
This commit is contained in:
@@ -78,6 +78,9 @@ def _screen_install(args) -> int:
|
||||
except install.InstallBusy as exc:
|
||||
print(f"Bot Desktop: {exc}")
|
||||
return 1
|
||||
if rc == install.NO_SUDO:
|
||||
print("Bot Desktop: this host has no sudo. Run as root on the host:\n " + cmd.removeprefix("sudo "))
|
||||
return 1
|
||||
if rc != 0:
|
||||
print(f"Bot Desktop: installer exited {rc}")
|
||||
return rc or 1
|
||||
|
||||
@@ -374,6 +374,16 @@ def _rewrite_browser_navigate(td: Dict[str, Any], available: set) -> Optional[Di
|
||||
return _fn_def({**td["function"], "description": desc})
|
||||
|
||||
|
||||
def _rewrite_computer_use(td: Dict[str, Any], available: set) -> Optional[Dict[str, Any]]:
|
||||
"""Strip the Bot Screen handoff (`request_handoff` / `wait_for_human`) where no Bot Desktop can exist
|
||||
(macOS, Windows): the model would otherwise learn actions that cannot succeed on this host."""
|
||||
from tools.bot_desktop.runtime import is_supported_host
|
||||
from tools.computer_use.schema import schema_for_host
|
||||
if is_supported_host():
|
||||
return td
|
||||
return _fn_def({**td["function"], **schema_for_host(supported=False)})
|
||||
|
||||
|
||||
def _rewrite_browser_exec(td: Dict[str, Any], available: set) -> Optional[Dict[str, Any]]:
|
||||
"""browser_exec runs arbitrary host Python: a session without the terminal surface
|
||||
must not regain host execution via the browser toolset. Session-level gate rather
|
||||
@@ -461,6 +471,7 @@ _DYNAMIC_SCHEMA_REWRITERS = {
|
||||
"browser_vault_list": _rewrite_browser_vault,
|
||||
"browser_vault_fill": _rewrite_browser_vault,
|
||||
"delegate_task": _rewrite_delegate_task,
|
||||
"computer_use": _rewrite_computer_use,
|
||||
}
|
||||
|
||||
|
||||
|
||||
@@ -33,7 +33,7 @@ def test_user_pinned_profile_wins(tmp_path, monkeypatch):
|
||||
|
||||
def test_dock_browser_advertises_a_devtools_port():
|
||||
"""A human-started instance must be attachable, or the agent can never drive it afterwards."""
|
||||
assert "--remote-debugging-port=" in browser.dock_command("/opt/chrome", "/p/dir").split()[2]
|
||||
assert "--remote-debugging-port=" in browser.dock_argv("/opt/chrome", "/p/dir")[2]
|
||||
|
||||
|
||||
def _fake_running_instance(user_data_dir, pid: int, port: int) -> None:
|
||||
@@ -89,3 +89,84 @@ def test_agent_attaches_to_human_started_browser(monkeypatch):
|
||||
monkeypatch.setattr(browser, "running_instance_cdp_port", lambda d, **kw: None)
|
||||
session._run_browser_command_unfenced("t", "open", ["https://x"], 10, None, "agent-browser", info)
|
||||
assert "--cdp" not in argvs[-1] and argvs[-1][:3] == ["agent-browser", "--session", "h_abc"]
|
||||
|
||||
|
||||
def _install_browsers(tmp_path, monkeypatch, *, playwright: bool, system: bool):
|
||||
"""A Playwright build under a private PLAYWRIGHT_BROWSERS_PATH and/or a system chromium on PATH."""
|
||||
monkeypatch.delenv("AGENT_BROWSER_EXECUTABLE_PATH", raising=False)
|
||||
roots = tmp_path / "pw"
|
||||
roots.mkdir(parents=True)
|
||||
monkeypatch.setenv("PLAYWRIGHT_BROWSERS_PATH", str(roots))
|
||||
monkeypatch.setattr("tools.browser_tool_install._chromium_search_roots", lambda: [str(roots)])
|
||||
pw_exe = roots / "chromium-1200" / "chrome-linux" / "chrome"
|
||||
if playwright:
|
||||
pw_exe.parent.mkdir(parents=True)
|
||||
pw_exe.write_text("#!/bin/sh\n", encoding="utf-8")
|
||||
pw_exe.chmod(0o755)
|
||||
sys_exe = tmp_path / "bin" / "chromium"
|
||||
if system:
|
||||
sys_exe.parent.mkdir(parents=True)
|
||||
sys_exe.write_text("#!/bin/sh\n", encoding="utf-8")
|
||||
sys_exe.chmod(0o755)
|
||||
monkeypatch.setattr("shutil.which", lambda name, *a, **k: str(sys_exe) if system and name == "chromium" else None)
|
||||
return str(pw_exe), str(sys_exe)
|
||||
|
||||
|
||||
def test_unprivileged_user_under_apparmor_userns_restriction_gets_the_system_browser(tmp_path, monkeypatch):
|
||||
"""Playwright's bundled Chromium has no setuid chrome_sandbox; with
|
||||
kernel.apparmor_restrict_unprivileged_userns=1 it dies 'FATAL: No usable sandbox!' for a non-root user,
|
||||
so the dock icon is dead. A distro chromium (which ships the sandbox helper) must win there — and
|
||||
the Playwright build stays the answer when it is the only one (no --no-sandbox for non-root)."""
|
||||
pw_exe, sys_exe = _install_browsers(tmp_path, monkeypatch, playwright=True, system=True)
|
||||
monkeypatch.setattr(browser.os, "geteuid", lambda: 1000)
|
||||
monkeypatch.setattr(browser, "_userns_restricted", lambda: True)
|
||||
assert browser.executable() == sys_exe
|
||||
|
||||
monkeypatch.setattr(browser, "_userns_restricted", lambda: False)
|
||||
assert browser.executable() == pw_exe # unrestricted host: Playwright's build as before
|
||||
|
||||
_install_browsers(tmp_path / "only-pw", monkeypatch, playwright=True, system=False)
|
||||
monkeypatch.setattr(browser, "_userns_restricted", lambda: True)
|
||||
exe = browser.executable()
|
||||
assert exe and exe.endswith("chrome-linux/chrome")
|
||||
assert "--no-sandbox" not in browser.dock_argv(exe, "/p/dir")
|
||||
|
||||
|
||||
def test_root_dock_browser_starts_with_the_same_sandbox_args_as_the_agents_browser(monkeypatch):
|
||||
"""Chromium refuses to start as root without --no-sandbox; agent-browser gets that flag from one
|
||||
policy, and the dock icon (same binary, same profile) must get the very same flags or the human's
|
||||
click dies while the agent's launch works."""
|
||||
from tools import browser_tool_session as session
|
||||
|
||||
monkeypatch.setattr(session.os, "geteuid", lambda: 0)
|
||||
monkeypatch.setattr(browser.os, "geteuid", lambda: 0)
|
||||
agent_env: dict = {}
|
||||
session._apply_chromium_sandbox_args(agent_env)
|
||||
agent_flags = set(agent_env["AGENT_BROWSER_ARGS"].split(","))
|
||||
assert agent_flags, "root must inject sandbox flags for agent-browser"
|
||||
assert agent_flags <= set(browser.dock_argv("/opt/chrome", "/p/dir"))
|
||||
|
||||
|
||||
def test_status_reports_the_headed_browser_or_its_absence(monkeypatch):
|
||||
"""The official image ships only chromium_headless_shell: executable() is None and the dock silently
|
||||
has no Browser icon. Status must say so instead of leaving the pane to guess."""
|
||||
monkeypatch.setattr(runtime, "_launcher_pid", lambda: None)
|
||||
monkeypatch.setattr(runtime, "published_env", lambda: {})
|
||||
monkeypatch.setattr(runtime, "geometry", lambda: "1440x900")
|
||||
monkeypatch.setattr(browser, "executable", lambda: None)
|
||||
assert runtime.status().as_dict()["browser"] is None
|
||||
monkeypatch.setattr(browser, "executable", lambda: "/usr/bin/chromium")
|
||||
assert runtime.status().browser == "/usr/bin/chromium"
|
||||
|
||||
|
||||
def test_dock_exec_line_survives_spaces_in_the_executable_and_profile_paths():
|
||||
"""The launcher used to split the shell line on the first space to find the executable, so a
|
||||
Chromium under '/opt/Google Chrome/' or a profile under a spaced HERMES_HOME broke the dock icon.
|
||||
Exec= follows the Desktop Entry spec: each argument double-quoted, with the reserved characters
|
||||
backslash-escaped inside the quotes."""
|
||||
exe = "/opt/Google Chrome/chrome"
|
||||
profile = '/home/a b/.hermes/browser "x"/profile'
|
||||
line = browser.dock_exec_line(exe, profile)
|
||||
assert line.startswith('Exec="/opt/Google Chrome/chrome" ')
|
||||
assert r'"--user-data-dir=/home/a b/.hermes/browser \\"x\\"/profile"' in line # spec: \" quoted, then \ string-escaped
|
||||
assert "--remote-debugging-port=0" in line
|
||||
|
||||
@@ -76,3 +76,36 @@ def test_real_profile_local_browser_is_fenced_by_provenance_even_without_a_live_
|
||||
result = json.loads(browser.browser_click("e1", task_id="review"))
|
||||
assert commands == [], f"human holds the lease, yet a real-profile browser command was dispatched: {commands}"
|
||||
assert result.get("code") == "human_has_control"
|
||||
|
||||
|
||||
def test_browser_console_supervisor_fast_path_is_fenced_while_human_controls_shared_browser(monkeypatch):
|
||||
"""`browser_console(expression=...)` answers over the CDP supervisor's WebSocket without ever reaching
|
||||
`_run_browser_command`, so the fence must sit in front of that fast path too — otherwise the one
|
||||
command that reads arbitrary page state is the one command the human's takeover does not stop."""
|
||||
import tools.browser_supervisor as supervisor_mod
|
||||
|
||||
commands: list = []
|
||||
browser, _ = _wire(monkeypatch, commands)
|
||||
browser._active_sessions["review"] = {"session_name": "review", "cdp_url": "ws://127.0.0.1:9222/devtools/browser/x",
|
||||
"features": {"local": True}}
|
||||
evaluated: list = []
|
||||
|
||||
class FakeSupervisor:
|
||||
def evaluate_runtime(self, expression, **_kw):
|
||||
evaluated.append(expression)
|
||||
return {"ok": True, "result": "WHAT-THE-HUMAN-TYPED", "result_type": "string"}
|
||||
|
||||
class FakeRegistry:
|
||||
def get(self, task_id):
|
||||
return FakeSupervisor()
|
||||
|
||||
monkeypatch.setattr(supervisor_mod, "SUPERVISOR_REGISTRY", FakeRegistry())
|
||||
try:
|
||||
lease.acquire("human-viewer")
|
||||
raw = browser.browser_console(expression="document.title", task_id="review")
|
||||
finally:
|
||||
browser._active_sessions.pop("review", None)
|
||||
result = json.loads(raw)
|
||||
assert evaluated == [] and commands == [], "human holds the lease, yet the page was evaluated"
|
||||
assert "WHAT-THE-HUMAN-TYPED" not in raw
|
||||
assert result.get("code") == "human_has_control"
|
||||
|
||||
@@ -8,6 +8,8 @@ import pytest
|
||||
|
||||
from tools.bot_desktop import install, runtime
|
||||
|
||||
_REAL_INSTALL_COMMAND = runtime.install_command # captured before the fixture pins a sudo line
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _isolated_host(tmp_path, monkeypatch):
|
||||
@@ -125,3 +127,73 @@ def test_passwordless_sudo_runs_the_install_without_asking_for_a_password(monkey
|
||||
assert code == 0
|
||||
assert spawned and spawned[0][:1] == ["sudo"] and "-S" not in spawned[0]
|
||||
assert "Done" in lines
|
||||
|
||||
|
||||
@pytest.mark.linux_only
|
||||
def test_timeout_returns_and_frees_the_slot_even_when_a_descendant_survives(monkeypatch):
|
||||
"""From an unprivileged Hermes, killpg reaches the sudo leader but not a root-owned apt child; that
|
||||
child keeps the pipe's write end open, so draining stdout never sees EOF and the profile slot stays
|
||||
taken forever. The timeout must end the drain and release the slot regardless of what survived.
|
||||
Stand-in for the unkillable root child: a grandchild in its own session holding our stdout."""
|
||||
import subprocess
|
||||
import time
|
||||
|
||||
monkeypatch.setattr(install, "_sudo_nopasswd", lambda: True)
|
||||
monkeypatch.setattr(install, "_TERM_GRACE_SECONDS", 0.2, raising=False)
|
||||
real_popen = subprocess.Popen
|
||||
|
||||
def popen(argv, **kw):
|
||||
if argv[:1] != ["sudo"]:
|
||||
return real_popen(argv, **kw)
|
||||
return real_popen(["sh", "-c", "setsid sleep 30 & echo child $!; wait"], **kw)
|
||||
|
||||
monkeypatch.setattr(install.subprocess, "Popen", popen)
|
||||
lines: list[str] = []
|
||||
started = time.monotonic()
|
||||
worker = threading.Thread(target=lambda: lines.append(
|
||||
f"code {install.install_packages(ask_password=lambda: '', on_line=lines.append, timeout_seconds=0.5)}"))
|
||||
worker.start()
|
||||
worker.join(3.0)
|
||||
survivor = next((int(line.split()[1]) for line in lines if line.startswith("child ")), None)
|
||||
if survivor:
|
||||
subprocess.run(["kill", "-9", str(survivor)], check=False)
|
||||
assert not worker.is_alive(), f"install_packages hung {time.monotonic() - started:.1f}s on a surviving descendant"
|
||||
assert any(line.startswith("code ") and line != "code 0" for line in lines), lines
|
||||
assert any("timed out" in line for line in lines), lines
|
||||
install.release(install.claim()) # slot is free again
|
||||
|
||||
|
||||
def test_root_installs_without_sudo_and_without_asking(monkeypatch):
|
||||
"""The official Docker image runs Hermes as uid 0 with no sudo binary: the package manager must be
|
||||
run directly, and the password card must never be raised for a user who already is root."""
|
||||
import subprocess
|
||||
|
||||
monkeypatch.setattr(install.os, "geteuid", lambda: 0)
|
||||
monkeypatch.setattr("shutil.which", lambda name, *a, **k: None)
|
||||
monkeypatch.setattr(runtime, "package_manager", lambda: "apt")
|
||||
monkeypatch.setattr(runtime, "install_command", _REAL_INSTALL_COMMAND)
|
||||
spawned: list[list[str]] = []
|
||||
real_popen = subprocess.Popen
|
||||
|
||||
def popen(argv, **kw):
|
||||
spawned.append(list(argv))
|
||||
return real_popen(["true"], **kw)
|
||||
|
||||
monkeypatch.setattr(install.subprocess, "Popen", popen)
|
||||
code = install.install_packages(ask_password=lambda: pytest.fail("root was asked for a sudo password"),
|
||||
on_line=lambda _l: None)
|
||||
assert code == 0
|
||||
assert spawned and spawned[0][0] == "apt-get", spawned
|
||||
|
||||
|
||||
def test_no_sudo_binary_returns_the_host_command_instead_of_a_password_card(monkeypatch):
|
||||
"""Unprivileged with no sudo on the host (minimal containers): a password card would be a dead end,
|
||||
the user needs the exact command to run on the host instead."""
|
||||
monkeypatch.setattr(install.os, "geteuid", lambda: 1000)
|
||||
monkeypatch.setattr("shutil.which", lambda name, *a, **k: None)
|
||||
monkeypatch.setattr(install.subprocess, "Popen", lambda *a, **k: pytest.fail("package manager spawned"))
|
||||
lines: list[str] = []
|
||||
code = install.install_packages(ask_password=lambda: pytest.fail("password card raised without sudo"),
|
||||
on_line=lines.append)
|
||||
assert code == install.NO_SUDO
|
||||
assert any("apt-get install" in line for line in lines), lines
|
||||
|
||||
@@ -35,7 +35,7 @@ def _seed(tmp_path: Path, fake_bins: list[str], browser_exec: str = "") -> Path:
|
||||
"HERMES_BD_SOCKET": str(tmp_path / "rfb.sock"), "HERMES_BD_XAUTH": str(tmp_path / "Xauthority"),
|
||||
"HERMES_BD_ENV_FILE": str(tmp_path / "env"), "HERMES_BD_CONFIG_HOME": str(cfg),
|
||||
"HERMES_BD_SEED_ONLY": "1",
|
||||
**({"HERMES_BD_BROWSER_EXEC": browser_exec} if browser_exec else {}),
|
||||
**({"HERMES_BD_BROWSER_EXEC": browser_exec, "HERMES_BD_BROWSER_EXEC_LINE": f"Exec={browser_exec} --user-data-dir={tmp_path}/bp"} if browser_exec else {}),
|
||||
}
|
||||
subprocess.run(["bash", str(LAUNCHER)], env=env, check=True, stdin=subprocess.DEVNULL, capture_output=True, timeout=30)
|
||||
return cfg
|
||||
@@ -43,7 +43,7 @@ def _seed(tmp_path: Path, fake_bins: list[str], browser_exec: str = "") -> Path:
|
||||
|
||||
def test_dock_lists_only_programs_present_on_path(tmp_path):
|
||||
chrome = tmp_path / "bin" / "chrome" # the browser is the one runtime.py resolved, never a PATH scan
|
||||
cfg = _seed(tmp_path, ["xfce4-terminal", "chrome", "firefox"], browser_exec=f"{chrome} --user-data-dir={tmp_path}/bp")
|
||||
cfg = _seed(tmp_path, ["xfce4-terminal", "chrome", "firefox"], browser_exec=str(chrome))
|
||||
panel = ET.parse(cfg / "xfce4/xfconf/xfce-perchannel-xml/xfce4-panel.xml") # well-formed or this raises
|
||||
launcher_ids = [str(p.get("name")) for p in panel.iter("property") if p.get("value") == "launcher"]
|
||||
execs = sorted(
|
||||
|
||||
33
tests/tools/test_computer_use_schema_host.py
Normal file
33
tests/tools/test_computer_use_schema_host.py
Normal file
@@ -0,0 +1,33 @@
|
||||
"""The computer_use schema only teaches the Bot Screen handoff (`request_handoff` / `wait_for_human`,
|
||||
'take over this screen from the Hermes Desktop app') on hosts that can have a Bot Desktop; elsewhere
|
||||
the model would learn actions that can never succeed."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
|
||||
from tools.computer_use.schema import COMPUTER_USE_SCHEMA, schema_for_host
|
||||
|
||||
_HANDOFF = ("request_handoff", "wait_for_human")
|
||||
|
||||
|
||||
def _actions(schema) -> list[str]:
|
||||
return schema["parameters"]["properties"]["action"]["enum"]
|
||||
|
||||
|
||||
def test_unsupported_host_schema_has_no_handoff_vocabulary():
|
||||
schema = schema_for_host(supported=False)
|
||||
text = json.dumps(schema)
|
||||
assert not any(action in text for action in _HANDOFF), text
|
||||
assert "take over this screen" not in text
|
||||
assert "reason" not in schema["parameters"]["properties"] # request_handoff's only parameter
|
||||
assert "grace" not in schema["parameters"]["properties"] # wait_for_human's only parameter
|
||||
# Everything else survives untouched.
|
||||
assert set(_actions(schema)) == set(_actions(COMPUTER_USE_SCHEMA)) - set(_HANDOFF)
|
||||
assert schema["parameters"]["required"] == ["action"]
|
||||
|
||||
|
||||
def test_supported_host_schema_keeps_the_handoff_and_is_the_frozen_schema():
|
||||
schema = schema_for_host(supported=True)
|
||||
assert set(_HANDOFF) <= set(_actions(schema))
|
||||
assert schema is COMPUTER_USE_SCHEMA # byte-frozen value: no per-call copy on the common path
|
||||
@@ -33,34 +33,70 @@ def profile_dir() -> Path:
|
||||
|
||||
def executable() -> Optional[str]:
|
||||
"""The Chromium agent-browser launches: an explicit ``AGENT_BROWSER_EXECUTABLE_PATH``, else the newest
|
||||
Playwright Chromium it bundles, else a system Chrome/Chromium. ``None`` when there is none."""
|
||||
Playwright Chromium it bundles, else a system Chrome/Chromium. ``None`` when there is none.
|
||||
|
||||
Non-root under ``kernel.apparmor_restrict_unprivileged_userns=1`` (Ubuntu 23.10+) flips the order:
|
||||
Playwright's bundle has no setuid ``chrome_sandbox`` and dies 'FATAL: No usable sandbox!' there, while
|
||||
a distro chromium ships the helper. The bundle stays the answer when it is the only browser — a dock
|
||||
icon that fails loudly beats a non-root ``--no-sandbox``.
|
||||
"""
|
||||
explicit = os.environ.get("AGENT_BROWSER_EXECUTABLE_PATH", "").strip()
|
||||
if explicit and os.access(explicit, os.X_OK):
|
||||
return explicit
|
||||
finders = [_playwright_executable, _system_executable]
|
||||
if not _is_root() and _userns_restricted():
|
||||
finders.reverse()
|
||||
return next((exe for find in finders if (exe := find())), None)
|
||||
|
||||
|
||||
def _playwright_executable() -> Optional[str]:
|
||||
from tools.browser_tool_install import _chromium_search_roots
|
||||
candidates = sorted(
|
||||
(p for root in _chromium_search_roots() for p in glob.glob(os.path.join(root, "chromium-*", "chrome-linux*", "chrome"))),
|
||||
key=os.path.getmtime, reverse=True)
|
||||
for exe in candidates:
|
||||
if os.access(exe, os.X_OK):
|
||||
return exe
|
||||
return next((exe for exe in candidates if os.access(exe, os.X_OK)), None)
|
||||
|
||||
|
||||
def _system_executable() -> Optional[str]:
|
||||
return next((shutil.which(name) for name in _SYSTEM_BROWSERS if shutil.which(name)), None)
|
||||
|
||||
|
||||
def _is_root() -> bool:
|
||||
return hasattr(os, "geteuid") and os.geteuid() == 0
|
||||
|
||||
|
||||
def _userns_restricted() -> bool:
|
||||
from tools.browser_tool_session import apparmor_restricts_unprivileged_userns
|
||||
return apparmor_restricts_unprivileged_userns()
|
||||
|
||||
|
||||
def dock_launch() -> Optional[Tuple[str, str]]:
|
||||
"""``(executable, user_data_dir)`` for the dock's Browser icon, or ``None`` when no Chromium exists."""
|
||||
exe = executable()
|
||||
return (exe, str(profile_dir())) if exe else None
|
||||
|
||||
|
||||
def dock_command(exe: str, user_data_dir: str) -> str:
|
||||
"""Shell line the dock's Browser icon runs. ``--remote-debugging-port=0`` makes a human-started
|
||||
def dock_argv(exe: str, user_data_dir: str) -> list[str]:
|
||||
"""Command the dock's Browser icon runs. ``--remote-debugging-port=0`` makes a human-started
|
||||
instance attachable (Chromium writes the chosen port to ``<user-data-dir>/DevToolsActivePort``);
|
||||
first-run / default-browser dialogs would sit between the human and the bot's tabs."""
|
||||
# --test-type hides the "Chrome for Testing is only for automated testing" and unsupported-flag
|
||||
# (--no-sandbox as root) infobars, which otherwise sit at the top of the human's takeover view.
|
||||
return (f"{exe} --user-data-dir={user_data_dir} --remote-debugging-port=0 --no-first-run "
|
||||
f"--no-default-browser-check --test-type")
|
||||
# Root gets the same sandbox-bypass flags agent-browser starts this binary with (one policy).
|
||||
from tools.browser_tool_session import CHROMIUM_SANDBOX_BYPASS_ARGS
|
||||
return [exe, f"--user-data-dir={user_data_dir}", "--remote-debugging-port=0", "--no-first-run",
|
||||
"--no-default-browser-check", "--test-type", *(CHROMIUM_SANDBOX_BYPASS_ARGS if _is_root() else ())]
|
||||
|
||||
|
||||
def dock_exec_line(exe: str, user_data_dir: str) -> str:
|
||||
"""The ``Exec=`` line of the dock's ``.desktop`` entry. Every argument is double-quoted per the
|
||||
Desktop Entry spec (a browser under ``/opt/Google Chrome/`` or a profile under a spaced HERMES_HOME
|
||||
otherwise splits into garbage): inside the quotes ``" ` $ \\`` are backslash-escaped, and because the
|
||||
value is itself a string field, each of those backslashes is escaped once more."""
|
||||
def quote(arg: str) -> str:
|
||||
quoted = "".join("\\" + ch if ch in '"`$\\' else ch for ch in arg)
|
||||
return '"' + quoted.replace("\\", "\\\\") + '"'
|
||||
return "Exec=" + " ".join(quote(arg) for arg in dock_argv(exe, user_data_dir))
|
||||
|
||||
|
||||
def running_instance_cdp_port(user_data_dir: str, *, exclude_session: Optional[str] = None) -> Optional[int]:
|
||||
|
||||
@@ -15,10 +15,13 @@ from __future__ import annotations
|
||||
import contextlib
|
||||
import logging
|
||||
import os
|
||||
import selectors
|
||||
import shlex
|
||||
import shutil
|
||||
import signal
|
||||
import subprocess
|
||||
import threading
|
||||
import time
|
||||
from typing import Callable, Optional
|
||||
|
||||
from hermes_constants import hermes_home_key
|
||||
@@ -30,6 +33,9 @@ _install_lock = threading.Lock()
|
||||
_running: set[str] = set()
|
||||
|
||||
|
||||
NO_SUDO = -2 # install_packages: unprivileged host without sudo; the on_line stream carried the command to run as root
|
||||
|
||||
|
||||
class InstallBusy(RuntimeError):
|
||||
pass
|
||||
|
||||
@@ -60,7 +66,8 @@ def release(key: str) -> None:
|
||||
|
||||
def install_packages(*, ask_password: Callable[[], str], on_line: Callable[[str], None],
|
||||
timeout_seconds: float = 900.0, claimed: bool = False) -> int:
|
||||
"""Run the package install; returns the process exit code (0 = success, ``-1`` = cancelled).
|
||||
"""Run the package install; returns the process exit code (0 = success, ``-1`` = cancelled,
|
||||
:data:`NO_SUDO` = unprivileged host without sudo — the command to run by hand was streamed).
|
||||
``claimed=True``: the caller already holds the slot via :func:`claim`; it is released here either way."""
|
||||
key = hermes_home_key() if claimed else None
|
||||
try:
|
||||
@@ -88,38 +95,80 @@ def _sudo_nopasswd() -> bool:
|
||||
def _run(cmd: str, *, ask_password: Callable[[], str], on_line: Callable[[str], None],
|
||||
timeout_seconds: float) -> int:
|
||||
argv = shlex.split(cmd)
|
||||
assert argv[0] == "sudo", cmd
|
||||
stdin_payload: Optional[str] = None
|
||||
if not _sudo_nopasswd():
|
||||
password = ask_password() or ""
|
||||
if not password:
|
||||
on_line("install cancelled: no sudo password provided")
|
||||
return -1
|
||||
# -S: read the password from stdin; -p '': no prompt text mixed into the streamed output.
|
||||
argv = ["sudo", "-S", "-p", "", *argv[1:]]
|
||||
stdin_payload = password + "\n"
|
||||
if argv[0] == "sudo":
|
||||
if shutil.which("sudo") is None:
|
||||
# Minimal containers ship no sudo: a password card would be a dead end. Hand the human the
|
||||
# exact command for the host instead.
|
||||
on_line(f"install needs root and this host has no sudo; run on the host as root: {cmd[len('sudo '):]}")
|
||||
return NO_SUDO
|
||||
if not _sudo_nopasswd():
|
||||
password = ask_password() or ""
|
||||
if not password:
|
||||
on_line("install cancelled: no sudo password provided")
|
||||
return -1
|
||||
# -S: read the password from stdin; -p '': no prompt text mixed into the streamed output.
|
||||
argv = ["sudo", "-S", "-p", "", *argv[1:]]
|
||||
stdin_payload = password + "\n"
|
||||
on_line(f"$ {cmd}")
|
||||
env = {"DEBIAN_FRONTEND": "noninteractive", "LC_ALL": "C.UTF-8"}
|
||||
proc = subprocess.Popen( # windows-footgun: ok — Linux-only (is_supported_host)
|
||||
argv, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
|
||||
env={**os.environ, **env}, text=True, encoding="utf-8", errors="replace", start_new_session=True)
|
||||
env={**os.environ, **env}, start_new_session=True)
|
||||
try:
|
||||
if stdin_payload is not None:
|
||||
proc.stdin.write(stdin_payload) # type: ignore[union-attr]
|
||||
proc.stdin.write(stdin_payload.encode("utf-8")) # type: ignore[union-attr]
|
||||
proc.stdin.close() # type: ignore[union-attr]
|
||||
except OSError:
|
||||
pass
|
||||
# The package manager runs in its own session (start_new_session); killing only sudo would leave apt/dnf
|
||||
# running as root with the dpkg lock while the slot is released, so the whole group goes.
|
||||
def _kill_group() -> None:
|
||||
with contextlib.suppress(ProcessLookupError):
|
||||
os.killpg(proc.pid, signal.SIGKILL) # windows-footgun: ok — Linux-only (is_supported_host)
|
||||
|
||||
timer = threading.Timer(timeout_seconds, _kill_group)
|
||||
timer.start()
|
||||
try:
|
||||
for line in proc.stdout: # type: ignore[union-attr]
|
||||
on_line(line.rstrip("\n"))
|
||||
return proc.wait()
|
||||
if _drain_until(proc, on_line, time.monotonic() + timeout_seconds):
|
||||
return proc.wait()
|
||||
_kill_group(proc)
|
||||
on_line(f"install timed out after {timeout_seconds:.0f}s; the package manager may still be running as root")
|
||||
return proc.returncode if proc.returncode is not None else -9
|
||||
finally:
|
||||
timer.cancel()
|
||||
proc.stdout.close() # type: ignore[union-attr]
|
||||
|
||||
|
||||
_TERM_GRACE_SECONDS = 5.0
|
||||
|
||||
|
||||
def _drain_until(proc: subprocess.Popen, on_line: Callable[[str], None], deadline: float) -> bool:
|
||||
"""Stream ``proc.stdout`` lines to ``on_line`` until EOF (``True``) or ``deadline`` (``False``).
|
||||
|
||||
Readiness-polled rather than a blocking ``for line in proc.stdout``: from an unprivileged Hermes no
|
||||
signal reaches a root-owned apt/dnf child, and that child keeps the pipe's write end open, so a
|
||||
blocking read would never see EOF and the profile's install slot would be held forever.
|
||||
"""
|
||||
fd = proc.stdout.fileno() # type: ignore[union-attr]
|
||||
buf = b""
|
||||
with selectors.DefaultSelector() as sel:
|
||||
sel.register(fd, selectors.EVENT_READ)
|
||||
while True:
|
||||
remaining = deadline - time.monotonic()
|
||||
if remaining <= 0:
|
||||
return False
|
||||
if not sel.select(timeout=min(remaining, 1.0)):
|
||||
continue
|
||||
chunk = os.read(fd, 65536)
|
||||
if not chunk:
|
||||
if buf:
|
||||
on_line(buf.decode("utf-8", "replace"))
|
||||
return True
|
||||
*lines, buf = (buf + chunk).split(b"\n")
|
||||
for line in lines:
|
||||
on_line(line.decode("utf-8", "replace"))
|
||||
|
||||
|
||||
def _kill_group(proc: subprocess.Popen) -> None:
|
||||
"""The package manager runs in its own session (start_new_session); killing only sudo would leave
|
||||
apt/dnf running as root with the dpkg lock while the slot is released, so the whole group goes: TERM
|
||||
first so dpkg can finish its transaction, KILL after the grace. Best effort — as non-root neither
|
||||
signal reaches a root-owned child, which is why the caller never waits on EOF."""
|
||||
for sig, grace in ((signal.SIGTERM, _TERM_GRACE_SECONDS), (signal.SIGKILL, 1.0)): # windows-footgun: ok — Linux-only (is_supported_host)
|
||||
with contextlib.suppress(ProcessLookupError, PermissionError):
|
||||
os.killpg(proc.pid, sig) # windows-footgun: ok — Linux-only (is_supported_host)
|
||||
with contextlib.suppress(subprocess.TimeoutExpired):
|
||||
proc.wait(timeout=grace)
|
||||
return
|
||||
|
||||
@@ -134,18 +134,20 @@ sed -i "s|HERMES_BD_WALLPAPER_PLACEHOLDER|$HERMES_BD_WALLPAPER|" "$X/xfce4-deskt
|
||||
if [[ ! -e "$X/xfce4-panel.xml" ]]; then
|
||||
L="$XDG_CONFIG_HOME/xfce4/panel"; mkdir -p "$L"
|
||||
dock_ids=(); n=20
|
||||
add_launcher() { # name icon exec — skipped when the executable is missing
|
||||
local exe; exe=${3%% *}
|
||||
command -v "$exe" >/dev/null 2>&1 || return 0
|
||||
add_launcher() { # name icon executable [Exec= line] — skipped when the executable is missing.
|
||||
# The Exec= line is passed ready-made (spec-quoted by Python) so paths with spaces survive; a bare
|
||||
# program name is its own Exec= value.
|
||||
command -v "$3" >/dev/null 2>&1 || return 0
|
||||
n=$((n+1)); mkdir -p "$L/launcher-$n"
|
||||
printf '[Desktop Entry]\nVersion=1.0\nType=Application\nName=%s\nIcon=%s\nExec=%s\nTerminal=false\nStartupNotify=false\n' \
|
||||
"$1" "$2" "$3" > "$L/launcher-$n/hermes.desktop"
|
||||
printf '[Desktop Entry]\nVersion=1.0\nType=Application\nName=%s\nIcon=%s\n%s\nTerminal=false\nStartupNotify=false\n' \
|
||||
"$1" "$2" "${4:-Exec=$3}" > "$L/launcher-$n/hermes.desktop"
|
||||
dock_ids+=("$n")
|
||||
}
|
||||
add_launcher "Terminal" utilities-terminal "xfce4-terminal"
|
||||
# The bot's browser: runtime.py resolves the executable agent-browser drives plus the profile's
|
||||
# persistent user-data-dir, so a human taking over lands in the bot's own cookie jar.
|
||||
[[ -n "${HERMES_BD_BROWSER_EXEC:-}" ]] && add_launcher "Browser" internet-web-browser "$HERMES_BD_BROWSER_EXEC"
|
||||
[[ -n "${HERMES_BD_BROWSER_EXEC:-}" ]] && \
|
||||
add_launcher "Browser" internet-web-browser "$HERMES_BD_BROWSER_EXEC" "${HERMES_BD_BROWSER_EXEC_LINE:-}"
|
||||
add_launcher "Files" system-file-manager "thunar"
|
||||
add_launcher "Text Editor" accessories-text-editor "mousepad"
|
||||
dock_plugins=""; dock_items=""
|
||||
|
||||
@@ -91,15 +91,23 @@ def package_manager() -> Optional[str]:
|
||||
|
||||
|
||||
def install_command() -> Optional[str]:
|
||||
"""The distro command that installs the Bot Desktop packages, as the human would type it on THIS host:
|
||||
prefixed with ``sudo`` unless Hermes already runs as root (the official Docker image is uid 0 with no
|
||||
sudo binary), so it is both what the pane shows and what :mod:`tools.bot_desktop.install` runs."""
|
||||
pm = package_manager()
|
||||
if pm is None:
|
||||
return None
|
||||
pkgs = " ".join(PACKAGES[pm])
|
||||
return {
|
||||
"apt": f"sudo apt-get install -y --no-install-recommends {pkgs}",
|
||||
"dnf": f"sudo dnf install -y {pkgs}",
|
||||
"pacman": f"sudo pacman -S --needed --noconfirm {pkgs}",
|
||||
body = {
|
||||
"apt": f"apt-get install -y --no-install-recommends {pkgs}",
|
||||
"dnf": f"dnf install -y {pkgs}",
|
||||
"pacman": f"pacman -S --needed --noconfirm {pkgs}",
|
||||
}[pm]
|
||||
return body if is_root() else f"sudo {body}"
|
||||
|
||||
|
||||
def is_root() -> bool:
|
||||
return hasattr(os, "geteuid") and os.geteuid() == 0
|
||||
|
||||
|
||||
@dataclass
|
||||
@@ -114,6 +122,7 @@ class DesktopStatus:
|
||||
socket: Optional[str]
|
||||
geometry: str
|
||||
install_command: Optional[str]
|
||||
browser: Optional[str] # headed Chromium the dock's Browser icon and agent-browser share; None = no headed browser
|
||||
|
||||
def as_dict(self) -> Dict[str, object]:
|
||||
return dict(self.__dict__)
|
||||
@@ -329,6 +338,7 @@ def geometry() -> str:
|
||||
|
||||
|
||||
def status(profile: Optional[str] = None) -> DesktopStatus:
|
||||
from tools.bot_desktop import browser as _bd_browser
|
||||
missing: list[str] = missing_binaries() if is_supported_host() else list(REQUIRED_BINARIES)
|
||||
pid = _launcher_pid()
|
||||
env = published_env()
|
||||
@@ -343,6 +353,7 @@ def status(profile: Optional[str] = None) -> DesktopStatus:
|
||||
socket=str(rfb_socket_path()) if rfb_socket_path() else None,
|
||||
geometry=geometry(),
|
||||
install_command=install_command() if missing else None,
|
||||
browser=_bd_browser.executable() if is_supported_host() else None,
|
||||
)
|
||||
|
||||
|
||||
@@ -397,9 +408,11 @@ def _spawn_and_wait(sd: Path, num: int, wait_seconds: float) -> DesktopStatus:
|
||||
"HERMES_BD_CONFIG_HOME": str(sd / "xdg"),
|
||||
"HERMES_BD_GEOMETRY": geometry(),
|
||||
})
|
||||
from tools.bot_desktop.browser import dock_command, dock_launch
|
||||
from tools.bot_desktop.browser import dock_exec_line, dock_launch
|
||||
if (browser := dock_launch()) is not None:
|
||||
child_env["HERMES_BD_BROWSER_EXEC"] = dock_command(*browser)
|
||||
# The bare executable (the launcher checks it exists) and the ready-made, spec-quoted Exec= line.
|
||||
child_env["HERMES_BD_BROWSER_EXEC"] = browser[0]
|
||||
child_env["HERMES_BD_BROWSER_EXEC_LINE"] = dock_exec_line(*browser)
|
||||
# Truncated per start: the log is a diagnostic for THIS launch, and nothing rotates it otherwise.
|
||||
log = open(sd / "launcher.log", "wb") # noqa: SIM115 — handed to the child, closed by it
|
||||
proc = subprocess.Popen( # windows-footgun: ok — Linux-only runtime (is_supported_host)
|
||||
|
||||
@@ -1067,9 +1067,15 @@ def _browser_eval(expression: str, task_id: Optional[str] = None) -> str:
|
||||
if _is_camofox_mode():
|
||||
return _camofox_eval(expression, task_id)
|
||||
|
||||
fast = _eval_supervisor_fast_path(effective_task_id, expression)
|
||||
if fast is not None:
|
||||
return fast
|
||||
# The supervisor answers over its own WebSocket and never reaches _run_browser_command, so the Bot
|
||||
# Desktop lease fence has to bracket it here too — otherwise the one command that reads arbitrary
|
||||
# page state is the one a human's takeover does not stop. Same fence, same session identity.
|
||||
fenced = _session.run_fenced(_active_sessions.get(effective_task_id) or {},
|
||||
lambda: {"fast": _eval_supervisor_fast_path(effective_task_id, expression)})
|
||||
if fenced.get("code") == "human_has_control":
|
||||
return _dumps(fenced)
|
||||
if fenced["fast"] is not None:
|
||||
return fenced["fast"]
|
||||
|
||||
result = _session._run_browser_command(effective_task_id, "eval", [expression])
|
||||
if not result.get("success"):
|
||||
|
||||
@@ -11,7 +11,7 @@ import shutil
|
||||
import subprocess
|
||||
import uuid
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, List, Optional
|
||||
from typing import Any, Callable, Dict, List, Optional
|
||||
|
||||
from hermes_cli._subprocess_compat import windows_hide_flags
|
||||
from tools.browser_tool_origin import origin as _bt
|
||||
@@ -30,12 +30,15 @@ _CHROMIUM_MISSING_DOCKER_HINT = ("Chromium browser is missing. You're running in
|
||||
_CHROMIUM_MISSING_HINT = f"Chromium browser is missing. Install it with: {_CHROMIUM_INSTALL}"
|
||||
|
||||
|
||||
def _needs_chromium_sandbox_bypass() -> bool:
|
||||
"""True when Chromium needs --no-sandbox to start reliably (root, Docker, AppArmor userns)."""
|
||||
if hasattr(os, "geteuid") and os.geteuid() == 0:
|
||||
return True
|
||||
if _install._running_in_docker():
|
||||
return True
|
||||
# THE Chromium startup flags for a host where its sandbox cannot work; agent-browser gets them through
|
||||
# AGENT_BROWSER_ARGS and the Bot Desktop dock's Browser icon (same binary, same profile) through
|
||||
# ``tools.bot_desktop.browser.dock_argv`` — one list, or the human's click dies while the agent's works.
|
||||
CHROMIUM_SANDBOX_BYPASS_ARGS = ("--no-sandbox", "--disable-dev-shm-usage")
|
||||
|
||||
|
||||
def apparmor_restricts_unprivileged_userns() -> bool:
|
||||
"""Ubuntu 23.10+ default: unprivileged user namespaces are denied, so a Chromium whose
|
||||
``chrome_sandbox`` helper is not setuid (Playwright's bundle) dies with 'No usable sandbox'."""
|
||||
try:
|
||||
with open("/proc/sys/kernel/apparmor_restrict_unprivileged_userns", encoding="utf-8") as f:
|
||||
return f.read().strip() == "1"
|
||||
@@ -43,12 +46,21 @@ def _needs_chromium_sandbox_bypass() -> bool:
|
||||
return False
|
||||
|
||||
|
||||
def _needs_chromium_sandbox_bypass() -> bool:
|
||||
"""True when Chromium needs --no-sandbox to start reliably (root, Docker, AppArmor userns)."""
|
||||
if hasattr(os, "geteuid") and os.geteuid() == 0:
|
||||
return True
|
||||
if _install._running_in_docker():
|
||||
return True
|
||||
return apparmor_restricts_unprivileged_userns()
|
||||
|
||||
|
||||
def _apply_chromium_sandbox_args(browser_env: Dict[str, str]) -> None:
|
||||
"""Add required Chromium sandbox flags without overriding user settings."""
|
||||
if ("AGENT_BROWSER_ARGS" not in browser_env and "AGENT_BROWSER_CHROME_FLAGS" not in browser_env
|
||||
and _needs_chromium_sandbox_bypass()):
|
||||
_bt.logger.debug("browser: sandbox bypass needed (root/docker/AppArmor userns) — injecting --no-sandbox")
|
||||
browser_env["AGENT_BROWSER_ARGS"] = "--no-sandbox,--disable-dev-shm-usage"
|
||||
browser_env["AGENT_BROWSER_ARGS"] = ",".join(CHROMIUM_SANDBOX_BYPASS_ARGS)
|
||||
|
||||
|
||||
def _read_command_output_files(stdout_path: str, stderr_path: str) -> tuple[str, str]:
|
||||
@@ -574,23 +586,32 @@ def _run_browser_command(
|
||||
except Exception as e:
|
||||
_bt.logger.warning("Failed to create browser session for task=%s: %s", task_id, e)
|
||||
return {"success": False, "error": f"Failed to create browser session: {str(e)}"}
|
||||
# The bot's LOCAL browser lives on its Bot Desktop screen, in the same profile a human who took
|
||||
# over is typing into. While the human holds the lease every action AND read against it is
|
||||
# refused (the page may show their credential); the fence brackets the whole run so a takeover
|
||||
# mid-command also voids the result. Cloud / user-supplied CDP sessions are a different browser.
|
||||
if _shares_bot_desktop_browser(session_info):
|
||||
from tools.bot_desktop import lease as _bd_lease
|
||||
try:
|
||||
admitted = _bd_lease.assert_agent_may_act()
|
||||
except _bd_lease.HumanHasControl as e:
|
||||
return {"success": False, "error": str(e), "code": "human_has_control"}
|
||||
result = _run_browser_command_unfenced(task_id, command, args, timeout, _engine_override, browser_cmd, session_info)
|
||||
if _bd_lease.get().epoch != admitted.epoch:
|
||||
return {"success": False, "code": "human_has_control",
|
||||
"error": "A human took over the bot's screen while this browser command ran; its result was "
|
||||
"discarded. Call computer_use action='wait_for_human' to block until they hand back."}
|
||||
return result
|
||||
return _run_browser_command_unfenced(task_id, command, args, timeout, _engine_override, browser_cmd, session_info)
|
||||
return run_fenced(session_info, lambda: _run_browser_command_unfenced(
|
||||
task_id, command, args, timeout, _engine_override, browser_cmd, session_info))
|
||||
|
||||
|
||||
def run_fenced(session_info: Dict[str, Any], fn: Callable[[], Dict[str, Any]]) -> Dict[str, Any]:
|
||||
"""Run ``fn`` under the Bot Desktop lease fence when ``session_info`` is the bot's LOCAL browser.
|
||||
|
||||
That browser lives on the Bot Desktop screen, in the same profile a human who took over is typing
|
||||
into. While the human holds the lease every action AND read against it is refused (the page may show
|
||||
their credential); the fence brackets the whole run so a takeover mid-command also voids the result.
|
||||
Cloud / user-supplied CDP sessions are a different browser and run unfenced. This is THE fence: every
|
||||
path that reaches the page (agent-browser subprocess, CDP supervisor fast path) goes through here.
|
||||
"""
|
||||
if not _shares_bot_desktop_browser(session_info):
|
||||
return fn()
|
||||
from tools.bot_desktop import lease as _bd_lease
|
||||
try:
|
||||
admitted = _bd_lease.assert_agent_may_act()
|
||||
except _bd_lease.HumanHasControl as e:
|
||||
return {"success": False, "error": str(e), "code": "human_has_control"}
|
||||
result = fn()
|
||||
if _bd_lease.get().epoch != admitted.epoch:
|
||||
return {"success": False, "code": "human_has_control",
|
||||
"error": "A human took over the bot's screen while this browser command ran; its result was "
|
||||
"discarded. Call computer_use action='wait_for_human' to block until they hand back."}
|
||||
return result
|
||||
|
||||
|
||||
def _shares_bot_desktop_browser(session_info: Dict[str, Any]) -> bool:
|
||||
|
||||
@@ -10,6 +10,18 @@ from __future__ import annotations
|
||||
|
||||
from typing import Any, Dict
|
||||
|
||||
# Bot Screen handoff: the human takes the bot's headless Linux screen over from the Hermes Desktop app.
|
||||
# Only meaningful where a Bot Desktop can exist (Linux gateway hosts); `schema_for_host` strips it
|
||||
# elsewhere so macOS/Windows/seated-Linux models never learn actions that cannot succeed.
|
||||
_HANDOFF_ACTIONS = ("request_handoff", "wait_for_human")
|
||||
_HANDOFF_ONLY_PROPERTIES = ("reason", "grace")
|
||||
_HANDOFF_ACTION_HINT = (
|
||||
" When a login, 2FA, CAPTCHA or payment step needs the human, call "
|
||||
"`request_handoff` (with `reason`) so they can take over this screen from the Hermes "
|
||||
"Desktop app, then `wait_for_human`; while they hold control every other action is refused."
|
||||
)
|
||||
_HANDOFF_SECONDS_HINT = " wait_for_human: how long to block for the hand-back (default 600, max 1800)."
|
||||
|
||||
# One consolidated tool with an `action` discriminator keeps the schema compact
|
||||
# and the per-turn token cost low. Property groups: capture (mode, app, pid,
|
||||
# window_id) / targeting (element, coordinate, button, modifiers) / drag / scroll /
|
||||
@@ -39,9 +51,7 @@ _PROPERTIES: Dict[str, Any] = {
|
||||
"Which action to perform. `capture` is free (no side effects). All other actions "
|
||||
"require approval unless auto-approved. Use `set_value` for select/popup elements and "
|
||||
"sliders — it selects the matching option directly without opening the native menu (no "
|
||||
"focus steal). When a login, 2FA, CAPTCHA or payment step needs the human, call "
|
||||
"`request_handoff` (with `reason`) so they can take over this screen from the Hermes "
|
||||
"Desktop app, then `wait_for_human`; while they hold control every other action is refused."
|
||||
"focus steal)." + _HANDOFF_ACTION_HINT
|
||||
),
|
||||
},
|
||||
"mode": {
|
||||
@@ -152,7 +162,7 @@ _PROPERTIES: Dict[str, Any] = {
|
||||
"Key combo, e.g. 'cmd+s', 'ctrl+alt+t', 'return', 'escape', 'tab'. Use '+' to combine."
|
||||
),
|
||||
},
|
||||
"seconds": {"type": "number", "description": "wait: seconds to pause (max 30). wait_for_human: how long to block for the hand-back (default 600, max 1800)."},
|
||||
"seconds": {"type": "number", "description": "wait: seconds to pause (max 30)." + _HANDOFF_SECONDS_HINT},
|
||||
"grace": {"type": "number", "description": "wait_for_human: seconds to wait for someone to take over before returning no_takeover (default 60); once a human holds control the full `seconds` applies."},
|
||||
"raise_window": {
|
||||
"type": "boolean",
|
||||
@@ -212,3 +222,19 @@ COMPUTER_USE_SCHEMA: Dict[str, Any] = {
|
||||
def get_computer_use_schema() -> Dict[str, Any]:
|
||||
"""Return the generic OpenAI function-calling schema."""
|
||||
return COMPUTER_USE_SCHEMA
|
||||
|
||||
|
||||
def schema_for_host(*, supported: bool) -> Dict[str, Any]:
|
||||
"""The frozen schema on hosts that can run a Bot Desktop; elsewhere the same schema without the
|
||||
handoff actions, their parameters and their copy. Pure: the host decision is the caller's."""
|
||||
if supported:
|
||||
return COMPUTER_USE_SCHEMA
|
||||
props = dict(_PROPERTIES)
|
||||
for name in _HANDOFF_ONLY_PROPERTIES:
|
||||
props.pop(name)
|
||||
action = dict(props["action"])
|
||||
action["enum"] = [a for a in action["enum"] if a not in _HANDOFF_ACTIONS]
|
||||
action["description"] = action["description"].replace(_HANDOFF_ACTION_HINT, "")
|
||||
props["action"] = action
|
||||
props["seconds"] = {**props["seconds"], "description": props["seconds"]["description"].replace(_HANDOFF_SECONDS_HINT, "")}
|
||||
return {**COMPUTER_USE_SCHEMA, "parameters": {**COMPUTER_USE_SCHEMA["parameters"], "properties": props}}
|
||||
|
||||
Reference in New Issue
Block a user