hermes-agent/tests/tools/test_computer_use_backend_lifecycle.py

"""Backend admission during Bot Screen rebind/release; extracted from #114565 for #108914."""

from __future__ import annotations

import contextvars
import json
import threading
from concurrent.futures import Future
from contextlib import contextmanager

import pytest

from hermes_constants import reset_hermes_home_override, set_hermes_home_override
from tools.bot_desktop import browser, lease, runtime as desktop
from tools.computer_use import tool as cu
from tools.computer_use_tool import registry


@contextmanager
def _profile(home):
    token = set_hermes_home_override(str(home))
    try:
        yield
    finally:
        reset_hermes_home_override(token)


def _spawn(fn, name):
    future = Future()
    context = contextvars.copy_context()

    def run():
        try:
            future.set_result(context.run(fn))
        except BaseException as exc:
            future.set_exception(exc)

    thread = threading.Thread(target=run, name=name, daemon=True)
    thread.start()
    return thread, future


def _call():
    return json.loads(registry.dispatch("computer_use", {"action": "list_apps"}, session_id="shared"))


@pytest.fixture
def runtime(tmp_path, monkeypatch):
    homes = [tmp_path / "a", tmp_path / "b"]
    for home, display in zip(homes, (":37", ":38")):
        (home / "bot-desktop").mkdir(parents=True)
        (home / "bot-desktop" / "env").write_text(f"DISPLAY={display}\n", encoding="utf-8")
        (home / "config.yaml").write_text("computer_use:\n  permission_mode: standard\n", encoding="utf-8")
    monkeypatch.setenv("HERMES_HOME", str(homes[0]))
    # Fake the device boundary only. Profile env publication, spawn identity,
    # config, backend caches, dispatch and persisted human leases stay real.
    monkeypatch.setattr(desktop, "_launcher_pid", lambda: 1)
    monkeypatch.setattr(browser, "executable", lambda: None)
    created = []

    class Backend(cu._NoopBackend):
        def __init__(self, mode):
            super().__init__()
            self.stopped = False
            self.count = 0
            self.before_action = lambda: None
            created.append(self)

        def stop(self):
            self.stopped = True

        def list_apps(self):
            assert not self.stopped, "dispatch used a retired backend"
            self.count += 1
            self.before_action()
            assert not self.stopped, "teardown ran during dispatch"
            return [{"name": str(created.index(self))}]

    cu.reset_backend_for_tests()
    monkeypatch.setattr(cu, "_new_backend", Backend)
    yield homes, created
    cu.reset_backend_for_tests()
    lease._reset_for_tests()


@pytest.mark.parametrize("pause_at", ["after_lookup", "before_acquire"])
@pytest.mark.parametrize("change", ["display", "mode", "release", "queued_display"])
def test_dispatch_rechecks_admission(runtime, monkeypatch, pause_at, change):
    homes, created = runtime
    paused, resume = threading.Event(), threading.Event()
    real_get = cu._get_backend
    old = real_get("shared")
    old_lock = cu._backend_call_locks["shared"]

    def pause():
        if threading.current_thread().name == "waiting-call" and not paused.is_set():
            paused.set()
            assert resume.wait(10), "test did not release the waiting call"

    def get(session_id=""):
        backend = real_get(session_id)
        if pause_at == "after_lookup":
            pause()
        return backend

    class PausingLock:
        def __enter__(self):
            pause()
            old_lock.acquire()
            return self

        def __exit__(self, *_):
            old_lock.release()

    monkeypatch.setattr(cu, "_get_backend", get)
    if pause_at == "before_acquire":
        cu._backend_call_locks["shared"] = PausingLock()
    worker, result = _spawn(_call, "waiting-call")
    try:
        assert paused.wait(10)
        if change == "release":
            assert cu.release_computer_use_session("shared")
        else:
            if change == "mode":
                (homes[0] / "config.yaml").write_text("computer_use:\n  permission_mode: bounded\n", encoding="utf-8")
            else:
                (homes[0] / "bot-desktop" / "env").write_text("DISPLAY=:39\n", encoding="utf-8")
            if change != "queued_display":
                assert real_get("shared") is not old
        if change != "queued_display":
            assert old.stopped
        resume.set()
        value = result.result(timeout=10)
        assert "error" not in value, value
        assert old.count == 0
        assert sum(backend.count for backend in created) == 1
        assert value["apps"] == [{"name": str(created.index(real_get("shared")))}]
    finally:
        resume.set()
        worker.join(timeout=10)
        assert not worker.is_alive()


@pytest.mark.parametrize("fail_action", [False, True])
def test_release_waits_without_blocking_another_profile_or_replaying(runtime, monkeypatch, fail_action):
    homes, created = runtime
    action_entered, finish_action, stop_attempted = (threading.Event() for _ in range(3))
    threads = []
    with _profile(homes[0]):
        old = cu._get_backend("shared")

        def action():
            action_entered.set()
            assert finish_action.wait(10)
            if fail_action:
                raise RuntimeError("action outcome is unknown; do not replay")

        old.before_action = action
        real_stop = cu._stop_backend

        def stop(backend, lock, on_error):
            if backend is old:
                stop_attempted.set()
            return real_stop(backend, lock, on_error)

        monkeypatch.setattr(cu, "_stop_backend", stop)
        try:
            thread, result = _spawn(_call, "admitted-call")
            threads.append(thread)
            assert action_entered.wait(10)
            thread, released = _spawn(lambda: cu.release_computer_use_session("shared"), "release")
            threads.append(thread)
            assert stop_attempted.wait(10)
            assert not old.stopped and not released.done()
            with _profile(homes[1]):
                thread, other = _spawn(_call, "other-profile")
                threads.append(thread)
                assert "error" not in other.result(timeout=10)
                assert not old.stopped
            finish_action.set()
            value = result.result(timeout=10)
            assert ("error" in value) is fail_action
            assert released.result(timeout=10) is True
            assert old.stopped and old.count == 1
            # A -> B -> A: the released profile gets its own fresh backend.
            assert "error" not in _call()
            assert cu._get_backend("shared") is not old
            with _profile(homes[1]):
                assert not cu._get_backend("shared").stopped
        finally:
            finish_action.set()
            for thread in threads:
                thread.join(timeout=10)
                assert not thread.is_alive()


@pytest.mark.parametrize("hand_back", [False, True])
def test_backend_retry_preserves_the_original_human_lease_epoch(runtime, monkeypatch, hand_back):
    _, created = runtime
    paused, resume = threading.Event(), threading.Event()
    real_get = cu._get_backend
    old = real_get("shared")

    def get(session_id=""):
        backend = real_get(session_id)
        if threading.current_thread().name == "waiting-call" and not paused.is_set():
            paused.set()
            assert resume.wait(10)
        return backend

    monkeypatch.setattr(cu, "_get_backend", get)
    worker, result = _spawn(_call, "waiting-call")
    try:
        assert paused.wait(10)
        assert cu.release_computer_use_session("shared")
        assert old.stopped
        lease.acquire("human")
        if hand_back:
            lease.release("human")
        resume.set()
        assert result.result(timeout=10)["code"] == "human_has_control"
        assert all(backend.count == 0 for backend in created)
    finally:
        resume.set()
        worker.join(timeout=10)
        assert not worker.is_alive()