fix(computer): keep the probe's fixture above an app's startup windows on X11

packages/agentenv-protocol/README.md

@@ -696,7 +696,9 @@ driver booted with a control token is probed with it: `--token`, by default
 
 At `full` the probe writes the Fixture Desk, a Tk window (`conformance/fixture.py`), into
 `~/.agentenv-probe` with `data/add` and starts it with `computer-exec`. It aims at the boxes the
-window's widgets report, reads back what reached them, and then removes the fixture. When the
+window's widgets report, reads back what reached them, and then removes the fixture. On X11 the
+window is maximized and kept above other windows, raising itself again when one covers it, so a
+window an app keeps on top doesn't take the clicks aimed at it; elsewhere it is full screen. When the
 fixture can't be staged, the `full` checks are unverified rather than failed.
 
 On a driver whose card says `screen: "page"`, the desk is an HTML page with the same widgets at the

packages/agentenv-protocol/src/agentenv_protocol/computer/conformance/fixture.py

@@ -2,13 +2,15 @@
 Desk, whose widgets' screen boxes are written to <state-dir>/layout.json and every effect that reaches it, with
 whatever lands in <state-dir>/inbox.txt, to <state-dir>/state.json, so the probe can aim at it and read back what
 arrived. It records the same effects under X11, macOS's Aqua and Windows, whose Tk names keys, binds select-all and
-reports the wheel differently. Run as ``python3 fixture.py <state-dir>``; it needs a 1280x800 screen or larger."""
+reports the wheel differently. On X11 it stays above an app's windows (``keep_on_top``); elsewhere it's full screen.
+Run as ``python3 fixture.py <state-dir>``; it needs a 1280x800 screen or larger."""
 from __future__ import annotations
 
 import ctypes
 import json
 import os
 import sys
+import time
 import tkinter as tk
 from pathlib import Path
 
@@ -29,6 +31,8 @@
 Y_BITS = 16
 #: The pixels of a scrolled line, macOS's for a scroll posted in lines.
 LINE_PX = 10
+#: The least time between two of the fixture's raises over other windows, so it can't fight one that raises itself.
+LIFT_GAP_S = 0.5
 
 
 def main(state_dir: Path) -> None:
@@ -37,7 +41,10 @@ def main(state_dir: Path) -> None:
     root.title("Fixture Desk")
     width, height = root.winfo_screenwidth(), root.winfo_screenheight()
     root.geometry(f"{width}x{height}+0+0")
-    root.attributes("-fullscreen", True)
+    if root.tk.call("tk", "windowingsystem") == "x11":
+        keep_on_top(root)
+    else:
+        root.attributes("-fullscreen", True)
     root.configure(bg="#f0f0f0")
     state: dict = {"name": "", "notes": "", "keys": [], "saved_count": 0, "subscribed": False, "selected": None,
                    "scroll_top": 0.0, "cells": {}, "right_clicks": 0, "double_clicks": 0, "drags": [],
@@ -227,6 +234,38 @@ def write_layout() -> None:
     root.mainloop()
 
 
+def keep_on_top(root: tk.Tk) -> None:
+    """Keep the fixture over the windows an app opens, which would take the probe's clicks: maximized and above other
+    windows rather than full screen, since xfwm4 keeps a window an app puts above over a full-screen one, and raised
+    again, giving the focus back to the widget that last had it, whenever another window covers some of it, at most once
+    each LIFT_GAP_S."""
+    root.attributes("-zoomed", True)
+    root.attributes("-topmost", True)
+    seen: dict = {"state": "VisibilityUnobscured", "lifted": 0.0, "pending": None}
+
+    def lift() -> None:
+        seen["pending"] = None
+        if seen["state"] == "VisibilityUnobscured":
+            return
+        seen["lifted"] = time.monotonic()
+        # Taking the layer above again restacks it on top, which a raise request alone doesn't get from xfwm4.
+        root.attributes("-topmost", False)
+        root.attributes("-topmost", True)
+        root.lift()
+        # The widget that last had the focus gets it back, so keys the probe types into a text box still reach it.
+        (root.focus_lastfor() or root).focus_force()
+
+    def visibility(event: tk.Event) -> None:
+        if event.widget is not root:
+            return
+        seen["state"] = event.state
+        if event.state != "VisibilityUnobscured" and seen["pending"] is None:
+            wait = max(0.0, seen["lifted"] + LIFT_GAP_S - time.monotonic())
+            seen["pending"] = root.after(round(wait * 1000), lift)
+
+    root.bind("<Visibility>", visibility, add="+")
+
+
 def wheel_steps(delta: int) -> int:
     """The wheel steps a <MouseWheel> event's ``delta`` holds, up positive: whole notches of 120 where Tk scales them
     so (Windows, and Tk 9 everywhere), else one a unit (Aqua's Tk 8.6)."""

packages/agentenv-protocol/tests/test_computer_conformance.py

@@ -323,6 +323,77 @@ def test_the_fixture_counts_a_scroll_tk_9_reports_in_pixels_by_the_lines_it_move
     assert _fixture(monkeypatch).touchpad_steps(delta) == steps
 
 
+class _Root:
+    """A Tk root stood in for: the attributes set on it in order, its bindings, the calls made to it, and what
+    ``after`` was given to run, with its delay."""
+
+    def __init__(self):
+        self.set, self.calls, self.bound, self.timers = [], [], {}, []
+        self.last = self
+
+    def focus_lastfor(self):
+        return self.last
+
+    def attributes(self, name, value):
+        self.set.append((name, value))
+
+    def bind(self, sequence, handler, add=None):
+        self.bound[sequence] = handler
+
+    def after(self, ms, callback):
+        self.timers.append((ms, callback))
+        return f"after#{len(self.timers)}"
+
+    def lift(self):
+        self.calls.append("lift")
+
+    def focus_force(self):
+        self.calls.append("focus_force")
+
+
+def test_the_fixture_on_x11_stays_maximized_above_other_windows_and_takes_the_top_back_once_one_covers_it(monkeypatch):
+    """A window an app keeps above, as Typora's licence dialog is for its first seconds, otherwise takes the clicks the
+    probe aims at the fixture."""
+    fixture = _fixture(monkeypatch)
+    now = [100.0]
+    monkeypatch.setattr(fixture.time, "monotonic", lambda: now[0])
+    root = _Root()
+    fixture.keep_on_top(root)
+    seen = root.bound["<Visibility>"]
+
+    seen(SimpleNamespace(widget="a child", state="VisibilityFullyObscured"))
+    for_child = list(root.timers)
+    seen(SimpleNamespace(widget=root, state="VisibilityPartiallyObscured"))
+    seen(SimpleNamespace(widget=root, state="VisibilityPartiallyObscured"))
+    (first, lift), = root.timers
+    lift()
+    lifted = (root.set[2:], list(root.calls))
+    now[0] += 0.1
+    seen(SimpleNamespace(widget=root, state="VisibilityFullyObscured"))
+    second, again = root.timers[1]
+    seen(SimpleNamespace(widget=root, state="VisibilityUnobscured"))
+    again()
+
+    assert root.set[:2] == [("-zoomed", True), ("-topmost", True)] and for_child == []
+    assert first == 0 and lifted == ([("-topmost", False), ("-topmost", True)], ["lift", "focus_force"])
+    assert second == round((fixture.LIFT_GAP_S - 0.1) * 1000) and root.calls == ["lift", "focus_force"]
+
+
+def test_the_fixture_gives_the_focus_back_to_the_text_box_that_had_it_when_it_takes_the_top_back(monkeypatch):
+    """A window an app opens during a typing check would otherwise leave the probe's keys with the fixture's root."""
+    fixture = _fixture(monkeypatch)
+    monkeypatch.setattr(fixture.time, "monotonic", lambda: 100.0)
+    root, name = _Root(), _Root()
+    root.last = name
+    fixture.keep_on_top(root)
+
+    root.bound["<Visibility>"](SimpleNamespace(widget=root, state="VisibilityPartiallyObscured"))
+    (_, lift), = root.timers
+    lift()
+
+    assert root.calls == ["lift"] and name.calls == ["focus_force"]
+
+
 def _fixture(monkeypatch):
     """The fixture module, which runs in the computer, read with Tk stood in for, as this Python may have none."""
     monkeypatch.setitem(sys.modules, "tkinter", ModuleType("tkinter"))

tst/integration/computer/covered/envs/covered/app.py

@@ -0,0 +1,10 @@
+"""An app for the probe test whose window, launched at boot, stays above every other window over the middle of the
+screen, where the conformance fixture's grid is, as a dialog an app keeps on top does."""
+from agentenv_protocol.computer import ComputerAppEnvironment, Launch, Ready, computer_app, linux
+
+
+@computer_app(supports=[{"os": {"type": "linux", "display": ["x11"]}, "arch": ["amd64", "arm64"]}])
+@linux(launch=Launch(at="boot", command=["venv/bin/python", "window.py"], window={"class": "Covered"},
+                     ready=Ready(window=True)))
+class Covered(ComputerAppEnvironment):
+    pass

tst/integration/computer/covered/envs/covered/env.toml

@@ -0,0 +1,4 @@
+type = "computer_app"
+module = "app.py:Covered"
+computer = "ubuntu-24.04"
+driver = "reference"

tst/integration/computer/covered/envs/covered/window.py

@@ -0,0 +1,10 @@
+"""The covered app's window, of class Covered: half the screen's size over its middle, kept above other windows."""
+import tkinter as tk
+
+root = tk.Tk(className="covered")
+root.title("Covered")
+width, height = root.winfo_screenwidth(), root.winfo_screenheight()
+root.geometry(f"{width // 2}x{height // 2}+{width // 4}+{height // 4}")
+root.attributes("-topmost", True)
+tk.Label(root, text="Kept above").pack(expand=True)
+root.mainloop()

tst/integration/computer/covered/tasks/covered.json

@@ -0,0 +1,3 @@
+[
+  {"id": "env", "type": "deploy_env", "env_id": "covered"}
+]

tst/integration/computer/covered_test.py

@@ -0,0 +1,32 @@
+"""A computer app whose window stays above every other window over the middle of the screen, as a dialog an app keeps
+on top does, run through ``agent-env run``: gate 3's probe passes every check, its clicks reaching the conformance
+fixture rather than that window, and the computer deploys. Needs Docker and the local registry on localhost:5000."""
+
+import platform
+from pathlib import Path
+
+import pytest
+from click.testing import CliRunner
+
+from agent_env.cli import cli
+from agent_env.env import Env
+from agent_env.store.routing import namespace_routing
+from tst.integration.computer.conftest import run_context
+
+pytestmark = [pytest.mark.integration, pytest.mark.int_test_slow]
+
+COVERED = Path(__file__).parent / "covered"
+ARCH = {"x86_64": "amd64", "amd64": "amd64", "aarch64": "arm64", "arm64": "arm64"}[platform.machine().lower()]
+
+
+def test_gate_3_passes_a_computer_whose_app_keeps_a_window_above_the_fixtures_grid(state):
+    ran = CliRunner().invoke(cli, ["run", str(COVERED), "--task", "covered", "--no-emulate"])
+
+    assert ran.exit_code == 0, ran.output
+    (deployed,) = run_context(ran.output)["deployed_envs"]
+    with namespace_routing():
+        build = Env.get(deployed["env_id"]).metadata["composed"]["container"][ARCH]
+    assert build["status"] == "ready", build.get("reason")
+    checks = build["probe"]["checks"]
+    # A fixture that never started leaves its checks unverified rather than failed, so they must have run and passed.
+    assert build["probe"]["passed"] and checks["failed"] == 0 and checks["unverified"] == 0 and checks["passed"] > 0