Run Harbor computer-1 on a desktop its image already runs

examples/experimental/harbor/harbor_agent_function.py

@@ -53,6 +53,13 @@
                          per-sandbox disk, Harbor's ``override_storage_mb``
   HARBOR_CUA_DESKTOP_WIDTH / HARBOR_CUA_DESKTOP_HEIGHT
                          computer-1's desktop size (Harbor's default 1024x900)
+  HARBOR_AGENT_IMPORT_PATH
+                         ``module:Class`` Harbor builds instead of the binding's agent
+  HARBOR_AGENT_KWARGS    JSON object added to ``AgentConfig.kwargs``; a key the
+                         binding sets is refused
+  HARBOR_VERIFIER_IMPORT_PATH / HARBOR_VERIFIER_KWARGS
+                         a Harbor verifier class, and its JSON kwargs, run in
+                         place of the task's ``tests/``
 
 Failure semantics: a verdict is returned as-is. A trial the policy cannot have
 failed is discarded with ``InfraAbort``: one that failed before its agent
@@ -112,6 +119,13 @@ def _allowed_hosts(var: str) -> list[str]:
     return [host for host in re.split(r"[,\s]+", os.getenv(var, "")) if host]
 
 
+def _env_json_object(var: str) -> dict[str, Any]:
+    value = json.loads(os.getenv(var, "") or "{}")
+    if not isinstance(value, dict):
+        raise ValueError(f"{var} must be a JSON object, got {type(value).__name__}")
+    return value
+
+
 # --- harness bindings -------------------------------------------------------
 #
 # What Miles must know per harness to hand it the session URL: which env vars /
@@ -415,10 +429,24 @@ def build_trial_config(metadata: dict[str, Any], session_url: str, request_kwarg
     }
     if max_seq_len is not None and binding.accepts_max_seq_len:
         agent_kwargs["max_seq_len"] = max_seq_len
+    extra_agent_kwargs = _env_json_object("HARBOR_AGENT_KWARGS")
+    if clash := sorted(set(extra_agent_kwargs) & set(agent_kwargs)):
+        raise ValueError(f"HARBOR_AGENT_KWARGS sets {clash}, which the {agent_name} binding owns")
+    agent_kwargs.update(extra_agent_kwargs)
+    import_path = os.getenv("HARBOR_AGENT_IMPORT_PATH", "").strip() or binding.import_path
 
     extra: dict[str, Any] = {}
-    if verifier_timeout := os.getenv("HARBOR_VERIFIER_TIMEOUT_SEC"):
-        extra["verifier"] = VerifierConfig(override_timeout_sec=float(verifier_timeout))
+    verifier_timeout = os.getenv("HARBOR_VERIFIER_TIMEOUT_SEC")
+    verifier_import_path = os.getenv("HARBOR_VERIFIER_IMPORT_PATH", "").strip() or None
+    verifier_kwargs = _env_json_object("HARBOR_VERIFIER_KWARGS")
+    if verifier_kwargs and verifier_import_path is None:
+        raise ValueError("HARBOR_VERIFIER_KWARGS needs HARBOR_VERIFIER_IMPORT_PATH")
+    if verifier_timeout or verifier_import_path:
+        extra["verifier"] = VerifierConfig(
+            override_timeout_sec=float(verifier_timeout) if verifier_timeout else None,
+            import_path=verifier_import_path,
+            kwargs=verifier_kwargs,
+        )
     if build_mult := os.getenv("HARBOR_ENV_BUILD_TIMEOUT_MULTIPLIER"):
         extra["environment_build_timeout_multiplier"] = float(build_mult)
 
@@ -427,8 +455,8 @@ def build_trial_config(metadata: dict[str, Any], session_url: str, request_kwarg
         task=TaskConfig(path=_task_path(tasks_dir, str(metadata.get("instance_id", "")))),
         agent=AgentConfig(
             # Harbor builds the registered class whenever the name is a registered one.
-            name=None if binding.import_path else agent_name,
-            import_path=binding.import_path,
+            name=None if import_path else agent_name,
+            import_path=import_path,
             model_name=binding.model_name(model),
             override_timeout_sec=float(agent_timeout) if agent_timeout else None,
             env=binding.env(session_url, api_key),

examples/experimental/harbor/launch_common.py

@@ -121,6 +121,10 @@ def harbor_env_vars(args) -> dict[str, str]:
         "HARBOR_VERIFIER_TIMEOUT_SEC",
         "HARBOR_ENV_BUILD_TIMEOUT_MULTIPLIER",
         "HARBOR_AGENT_ALLOWED_HOSTS",
+        "HARBOR_AGENT_IMPORT_PATH",
+        "HARBOR_AGENT_KWARGS",
+        "HARBOR_VERIFIER_IMPORT_PATH",
+        "HARBOR_VERIFIER_KWARGS",
     ):
         if value := os.environ.get(var, "").strip():
             env[var] = value

examples/experimental/harbor_cua/desktop_computer.py

@@ -0,0 +1,187 @@
+"""``Qwen3VLComputer1`` on a desktop its sandbox image already runs, such as CUA-Lite's GNOME template.
+
+Harbor's computer-1 brings up its own Xvfb, XFCE and Chromium and treats Chromium's death as
+the session's; here the image's display is used as it is, and only the X server's socket says
+the session is alive.
+"""
+
+import asyncio
+import shlex
+from pathlib import PurePosixPath
+from types import SimpleNamespace
+from typing import Any
+
+from examples.experimental.harbor_cua.qwen_computer import Qwen3VLComputer1, Qwen3VLJsonProvider
+from harbor.agents.computer_1.runtime import (
+    _DEFAULT_DISPLAY,
+    _RUNTIME_DIR,
+    _SCREENSHOT_DIR,
+    TERMINAL_ACTION_TYPES,
+    Computer1Session,
+    ComputerAction,
+    RuntimeRequestError,
+    _bash_inline,
+)
+from harbor.environments.base import BaseEnvironment
+
+# zoom crops the next frame, against which Qwen3-VL's grid clicks are then mis-scaled.
+UNAVAILABLE_ACTIONS = frozenset({"zoom", "navigate", "go_back", "go_forward"})
+_UNCHANGED_SCREEN_ACTIONS = frozenset({"wait", "sleep", *TERMINAL_ACTION_TYPES})
+
+# Each passage is removed from Harbor's computer-1-json.txt, and a missing one fails the build.
+_PROMPT_CUTS = (
+    " zoom, navigate,",
+    '\n    "url": <string, optional, used by navigate>,',
+    '\n    "zoom_region": <[x0, y0, x1, y1] in desktop pixels, optional, used by zoom>,',
+    '\n- For "zoom": provide "zoom_region" as [x0, y0, x1, y1] in desktop pixels.'
+    "\n  The NEXT screenshot is cropped (no resize) to that region, then auto-resets."
+    "\n  Use this to inspect a small UI area at native pixel density.",
+    '\n- For "navigate": provide the destination URL in "url".',
+)
+
+
+def _images_in(prompt: Any) -> int:
+    turns = [prompt] if isinstance(prompt, dict) else prompt if isinstance(prompt, list) else []
+    return sum(
+        isinstance(part, dict) and part.get("type") == "image_url"
+        for turn in turns
+        for part in (turn.get("content") if isinstance(turn, dict) and isinstance(turn.get("content"), list) else [])
+    )
+
+
+def desktop_prompt_template(template: str) -> str:
+    for cut in _PROMPT_CUTS:
+        if template.count(cut) != 1:
+            raise ValueError(f"computer-1's prompt template no longer holds {cut!r} exactly once")
+        template = template.replace(cut, "")
+    return template
+
+
+class DesktopJsonProvider(Qwen3VLJsonProvider):
+    screenshot_format = "png"
+
+    @classmethod
+    def from_agent(cls, agent: Qwen3VLComputer1) -> "DesktopJsonProvider":
+        provider = super().from_agent(agent)
+        provider._prompt_template = desktop_prompt_template(provider._prompt_template)
+        return provider
+
+
+class DesktopSession(Computer1Session):
+    def __init__(self, *args: Any, settle_sec: float, screenshot_size: tuple[int, int] | None, **kwargs: Any) -> None:
+        super().__init__(*args, **kwargs)
+        self._settle_sec = settle_sec
+        self._screenshot_size = screenshot_size
+
+    async def start(self) -> None:
+        if self._started:
+            return
+        dirs = " ".join(shlex.quote(d) for d in (_RUNTIME_DIR, _SCREENSHOT_DIR, str(self._agent_dir)))
+        await self._exec(_bash_inline(f"mkdir -p {dirs}"), timeout_sec=60, label="mkdir runtime dirs", retries=2)
+        await self._wait_for_x11()
+        self._started = True
+
+    async def is_session_alive(self) -> bool:
+        try:
+            result = await self._raw_exec("test -S /tmp/.X11-unix/X1 && echo ok || echo down", timeout_sec=5)
+        except Exception:
+            return False
+        return "ok" in (result.stdout or "")
+
+    async def reset(self) -> None:
+        raise RuntimeError("a desktop session has no browser to relaunch")
+
+    async def _recover_chromium_if_needed(self, action_type: str, exc: Exception) -> dict[str, Any] | None:
+        return None
+
+    async def execute(self, action: ComputerAction) -> dict[str, Any]:
+        if action.type in UNAVAILABLE_ACTIONS:
+            raise RuntimeRequestError(action.type, 400, f"{action.type!r} is not available on this desktop")
+        result = await super().execute(action)
+        if action.type not in _UNCHANGED_SCREEN_ACTIONS and self._settle_sec > 0:
+            await asyncio.sleep(self._settle_sec)
+        return result
+
+    def _capture_script(self, env_png: str) -> str:
+        """Shell that writes the desktop to ``env_png`` as a PNG, at ``screenshot_size`` when one is set."""
+        resize = ""
+        if self._screenshot_size is not None:
+            width, height = self._screenshot_size
+            resize = f" && convert {shlex.quote(env_png)} -resize {width}x{height}! {shlex.quote(env_png)}"
+        return f"import -window root {shlex.quote(env_png)}{resize}"
+
+    async def fetch_screenshot(self, env_path: PurePosixPath | str) -> str:
+        target = str(env_path)
+        if PurePosixPath(target).suffix.lower() != ".png":
+            raise ValueError(f"a desktop session writes PNG screenshots, not {target!r}")
+        env_png = f"{_SCREENSHOT_DIR}/latest.png"
+        script = (
+            f"set -e; export DISPLAY={_DEFAULT_DISPLAY}; "
+            f"mkdir -p {shlex.quote(_SCREENSHOT_DIR)} {shlex.quote(str(PurePosixPath(target).parent))}; "
+            f"{self._capture_script(env_png)}; "
+            f"cp {shlex.quote(env_png)} {shlex.quote(target)}"
+        )
+        await self._exec(_bash_inline(script), timeout_sec=max(30, self._request_timeout_sec), label="screenshot")
+        return target
+
+
+class DesktopQwen3VLComputer1(Qwen3VLComputer1):
+    """``settle_sec`` waits after every screen-changing action; ``keep_screenshots`` caps the images a request carries."""
+
+    session_class: type[DesktopSession] = DesktopSession
+
+    def __init__(
+        self,
+        *args: Any,
+        settle_sec: float = 2.0,
+        screenshot_width: int | None = None,
+        screenshot_height: int | None = None,
+        keep_screenshots: int | None = None,
+        **kwargs: Any,
+    ) -> None:
+        if (screenshot_width is None) != (screenshot_height is None):
+            raise ValueError("set screenshot_width and screenshot_height together")
+        if keep_screenshots is not None and keep_screenshots < 1:
+            raise ValueError("keep_screenshots must be at least 1: the request needs the current screen")
+        super().__init__(*args, **kwargs)
+        self._settle_sec = settle_sec
+        self._screenshot_size = (screenshot_width, screenshot_height) if screenshot_width is not None else None
+        self._keep_screenshots = keep_screenshots
+
+    def _build_provider(self) -> DesktopJsonProvider:
+        return DesktopJsonProvider.from_agent(self)
+
+    async def setup(self, environment: BaseEnvironment) -> None:
+        geometry = self._desktop_geometry
+        self._session = self.session_class(
+            environment=environment,
+            agent_dir=self._env_io_dir,
+            desktop_width=geometry.desktop_width,
+            desktop_height=geometry.desktop_height,
+            window_width=geometry.desktop_width,
+            window_height=geometry.desktop_height,
+            readiness_timeout_sec=self._runtime_readiness_timeout_sec,
+            request_timeout_sec=self._runtime_request_timeout_sec,
+            extra_env=self._extra_env,
+            user=environment.default_user,
+            enable_bash=self._enable_bash,
+            settle_sec=self._settle_sec,
+            screenshot_size=self._screenshot_size,
+        )
+        await self._session.start()
+
+    def _trim_history(self, messages: Any, prompt_images: int) -> None:
+        if self._keep_screenshots is not None:
+            keep = max(0, self._keep_screenshots - prompt_images)
+            self._compactor._trim_old_screenshots(SimpleNamespace(messages=messages), keep_last=keep)
+
+    async def _query_litellm(self, chat: Any, prompt: Any, *args: Any, **kwargs: Any) -> Any:
+        # A parse-error retry carries no screen of its own, so the history keeps the current one.
+        self._trim_history(chat.messages, _images_in(prompt))
+        return await super()._query_litellm(chat, prompt, *args, **kwargs)
+
+    def _fallback_message_history(self) -> list[Any]:
+        # The turn-cap fallback sends this history with the latest screen attached whenever images are on.
+        history = super()._fallback_message_history()
+        self._trim_history(history, int(self._enable_images and self._latest_screenshot_path is not None))
+        return history

tests/fast/examples/experimental/harbor/test_harbor_agent_function.py

@@ -95,6 +95,13 @@ def tasks_dir(tmp_path, monkeypatch):
     monkeypatch.delenv("HARBOR_ENV_KWARGS", raising=False)
     monkeypatch.delenv("HARBOR_CPU_ENFORCEMENT_POLICY", raising=False)
     monkeypatch.delenv("HARBOR_MEMORY_ENFORCEMENT_POLICY", raising=False)
+    for var in (
+        "HARBOR_AGENT_IMPORT_PATH",
+        "HARBOR_AGENT_KWARGS",
+        "HARBOR_VERIFIER_IMPORT_PATH",
+        "HARBOR_VERIFIER_KWARGS",
+    ):
+        monkeypatch.delenv(var, raising=False)
     return tmp_path
 
 
@@ -349,6 +356,67 @@ def test_other_bindings_keep_the_registered_class(tasks_dir):
     assert cfg.agent.import_path is None
 
 
+@pytest.mark.parametrize("agent_name", ["computer-1", "terminus-2"])
+def test_an_agent_import_path_replaces_the_bindings_class(tasks_dir, monkeypatch, agent_name):
+    monkeypatch.setenv("HARBOR_AGENT_IMPORT_PATH", "pkg.agent:DesktopAgent")
+    cfg = haf.build_trial_config({"instance_id": "task-1", "agent_name": agent_name}, "http://s/v1", {})
+
+    assert cfg.agent.import_path == "pkg.agent:DesktopAgent"
+    assert cfg.agent.name is None
+
+
+def test_agent_kwargs_join_the_bindings_own(tasks_dir, monkeypatch):
+    monkeypatch.setenv("HARBOR_AGENT_KWARGS", '{"settle_sec": 2.0, "keep_screenshots": 1}')
+    kwargs = _computer_1_config().agent.kwargs
+
+    assert (kwargs["settle_sec"], kwargs["keep_screenshots"]) == (2.0, 1)
+    assert kwargs["api_base"] == "http://s/sessions/s1/v1"
+
+
+@pytest.mark.parametrize("owned", ["temperature", "llm_kwargs", "model_info"])
+def test_agent_kwargs_cannot_replace_what_the_binding_sets(tasks_dir, monkeypatch, owned):
+    monkeypatch.setenv("HARBOR_AGENT_KWARGS", f'{{"{owned}": 0.7}}')
+
+    with pytest.raises(ValueError, match=owned):
+        _computer_1_config()
+
+
+def test_a_verifier_class_reaches_harbor_with_its_kwargs(tasks_dir, monkeypatch):
+    monkeypatch.setenv("HARBOR_VERIFIER_IMPORT_PATH", "pkg.verify:HostVerifier")
+    monkeypatch.setenv("HARBOR_VERIFIER_KWARGS", '{"read_limit_mb": 64}')
+    monkeypatch.setenv("HARBOR_VERIFIER_TIMEOUT_SEC", "600")
+    verifier = _computer_1_config().verifier
+
+    assert (verifier.import_path, verifier.kwargs, verifier.override_timeout_sec) == (
+        "pkg.verify:HostVerifier",
+        {"read_limit_mb": 64},
+        600.0,
+    )
+
+
+def test_a_task_keeps_its_own_tests_without_a_verifier_class(tasks_dir, monkeypatch):
+    monkeypatch.delenv("HARBOR_VERIFIER_TIMEOUT_SEC", raising=False)
+    assert not hasattr(_computer_1_config(), "verifier")
+
+    monkeypatch.setenv("HARBOR_VERIFIER_TIMEOUT_SEC", "600")
+    assert _computer_1_config().verifier.import_path is None
+
+
+@pytest.mark.parametrize(
+    ("env", "match"),
+    [
+        ({"HARBOR_VERIFIER_KWARGS": '{"x": 1}'}, "HARBOR_VERIFIER_IMPORT_PATH"),
+        ({"HARBOR_AGENT_KWARGS": "[1]"}, "JSON object"),
+    ],
+)
+def test_malformed_agent_or_verifier_settings_stop_the_run(tasks_dir, monkeypatch, env, match):
+    for var, value in env.items():
+        monkeypatch.setenv(var, value)
+
+    with pytest.raises(ValueError, match=match):
+        _computer_1_config()
+
+
 @pytest.mark.parametrize(
     ("model", "expected"),
     [

tests/fast/examples/experimental/harbor/test_launch_common.py

@@ -90,6 +90,22 @@ def test_env_kwargs_and_server_knobs_are_forwarded_when_set(monkeypatch):
     assert "HARBOR_MAX_SEQ_LEN" not in env
 
 
+def test_agent_and_verifier_classes_are_forwarded_with_their_kwargs(monkeypatch):
+    monkeypatch.setenv("E2B_API_KEY", "e2b_x")
+    knobs = {
+        "HARBOR_AGENT_IMPORT_PATH": "pkg.agent:DesktopAgent",
+        "HARBOR_AGENT_KWARGS": '{"settle_sec": 2.0}',
+        "HARBOR_VERIFIER_IMPORT_PATH": "pkg.verify:HostVerifier",
+        "HARBOR_VERIFIER_KWARGS": '{"read_limit_mb": 64}',
+    }
+    for var, value in knobs.items():
+        monkeypatch.setenv(var, value)
+
+    env = launch_common.harbor_env_vars(_args())
+
+    assert {var: env[var] for var in knobs} == knobs
+
+
 def test_agentic_wiring_is_the_shipped_wiring():
     """One shared copy: the flags the GPU e2e exercises are the flags the recipes ship."""
     args = launch_common.agentic_train_args(tito_model="qwen3", session_server_workers=4)