Pause Frametop for VR games

Frametop kept using the headset during games: with gaze mode off, our eye tracker still took
about 60% of a core, remote desktop about 2 cores while on, and KWin kept drawing hidden
screens because ft-screens sent their frame callbacks at 90 Hz. Pausing gives that back, and
resuming brings back only what pausing stopped. It's also a way to keep the gaze service and
our eye tracker off during games, which PR #13 asked for.

Paused (input/game_pause.py, run by the input relay):
- frametop-gaze stops (ft-eyegrab then idles by itself), and hand tracking and remote desktop
  stop if they run
- the desktop hides and slows down: ft-screens "pause on" hides every panel and sends KWin a
  frame callback once a second; or, with pause_desktop "close", the desktop closes and starts
  again on resume
- the relay lets go of the 3D mouse, typing goes to Steam, and mapped buttons and key
  combinations do only pause_toggle, steam_menu and commands

Toggled by both thumbsticks clicked together twice (configurable), read passively from
vrserver's web socket (input/vrws.py) so it works in games and takes nothing from them; by the
new pause_toggle action; by input/ft-pause; and, with pause_auto (default on), by a VR game
starting and ending, which the pointer helper now reports ("vrgame 1|0"). Frametop Input
Settings has a Games page for it. update-check.py checks the web socket, and doesn't count a
paused gaze service as failed.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

input/test/keys-test.py

@@ -3,8 +3,9 @@
 
 Runs the relay's main() against a fake pass-through keyboard and a fake mouse (pipes), with
 every socket it sends to renamed, no uinput devices, no grabs, and ft-steam swapped for a
-logger. Nothing reaches the live desktop, SteamVR, or the running relay, so it's safe next
-to them. Rules come from the test, not ~/.config/frametop-input.json.
+logger. Pausing (game_pause.py) is a stub: no threads, state file, or services. Nothing reaches
+the live desktop, SteamVR, or the running relay, so it's safe next to them. Rules come from the
+test, not ~/.config/frametop-input.json.
 
   input/test/keys-test.py [RELAY]   (default: input/input-relay.py next to this folder)
 """
@@ -26,8 +27,41 @@ with open(relay_path) as f:
     src = f.read().replace('"\\0frametop_relay"', f'"\\0{tag}_relay"')
 relay = importlib.util.module_from_spec(importlib.util.spec_from_loader("relay", loader=None))
 relay.__file__ = os.path.abspath(relay_path)
+sys.path.insert(0, os.path.dirname(os.path.abspath(relay_path)))  # its game_pause and vrws
 exec(compile(src, relay_path, "exec"), relay.__dict__)
 
+
+class StubPause:
+    """game_pause.GamePause without its threads, state file, or services."""
+    paused = False
+
+    def __init__(self, log, on_change, buttons):
+        self.on_change = on_change
+        stub["pause"] = self
+
+    def toggle(self, reason, now):
+        self.paused = not self.paused
+        self.on_change(self.paused)
+
+    def set(self, on, reason, now):
+        if on != self.paused:
+            self.toggle(reason, now)
+
+    def status(self):
+        return {"t": "pause", "paused": self.paused}
+
+    def timeout(self, now):
+        return 3600.0
+
+    def configure(self, *args):
+        pass
+
+    game_state = tick = helper_started = configure
+
+
+stub = {}
+relay.game_pause.GamePause = StubPause
+
 for name in ("SCREENS", "HELPER", "FLOAT", "GAZED", "KEYS"):
     setattr(relay, name, f"\0{tag}_{name.lower()}")
 OUT = tempfile.mkdtemp(prefix="ft-keys-test-")
@@ -144,6 +178,15 @@ def use(key_bindings):
     typed()
 
 
+def pause(cmd):
+    c = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
+    c.bind("")
+    c.settimeout(2)
+    c.sendto(f"pause {cmd} test".encode(), f"\0{tag}_relay")
+    c.recv(4096)
+    time.sleep(0.05)
+
+
 def lines(path, wait=0.3):
     time.sleep(wait)
     try:
@@ -163,7 +206,7 @@ def check(label, got, want):
         failures.append(label)
 
 
-META, RMETA, SHIFT, CTRL, RCTRL, A, F, J = 125, 126, 42, 29, 97, 30, 33, 36
+META, RMETA, SHIFT, CTRL, RCTRL, A, F, J, P = 125, 126, 42, 29, 97, 30, 33, 36, 25
 F24 = ["key 194 1", "key 194 0"]
 
 
@@ -207,9 +250,25 @@ def tests():
     check("right Ctrl tap runs its command, which finds Frametop's tools", lines(cmd_log), ["combo", "3"])
     check("Ctrl tap: F24 before Ctrl's release", typed(), ["key 97 1"] + F24 + ["key 97 0"])
 
+    use({"125+36": "gaze_left", "29": f"command:echo paused >> {cmd_log}", "42+125+25": "pause_toggle"})
+    pause("on")
+    check("pause on: paused", stub["pause"].paused, True)
+    key(META, 1); key(J, 1); key(J, 0); key(META, 0)
+    check("paused: Meta+J (gaze click) does nothing and is typed as is", typed(),
+          ["key 125 1", "key 36 1", "key 36 0", "key 125 0"])
+    key(RCTRL, 1); key(RCTRL, 0)
+    check("paused: a command still runs", lines(cmd_log), ["combo", "3", "paused"])
+    typed()
+    key(SHIFT, 1); key(META, 1); key(P, 1); key(P, 0); key(META, 0); key(SHIFT, 0)
+    check("paused: Meta+Shift+P (pause_toggle) resumes", stub["pause"].paused, False)
+    check("Meta+Shift+P isn't typed", [k for k in typed() if k.startswith("key 25 ")], [])
+    key(META, 1); key(J, 1); key(J, 0); key(META, 0)
+    check("resumed: Meta+J is a combination again", typed(), ["key 125 1"] + F24 + ["key 125 0"])
+
     check("an empty command isn't an action", relay.known_action("command:  "), False)
-    check("steam_menu and commands work without pointer mode",
-          (relay.needs_pointer("steam_menu"), relay.needs_pointer("command:ls")), (False, False))
+    check("steam_menu, pause_toggle and commands work without pointer mode",
+          (relay.needs_pointer("steam_menu"), relay.needs_pointer("pause_toggle"), relay.needs_pointer("command:ls")),
+          (False, False, False))
     print("FAILED: " + ", ".join(failures) if failures else "all passed", flush=True)
     shutil.rmtree(OUT, ignore_errors=True)
     os._exit(1 if failures else 0)

input/test/pause-test.py

@@ -0,0 +1,214 @@
+#!/usr/bin/env python3
+"""Offline test of pausing for VR games (input/game_pause.py): the controller gesture, and when
+Frametop pauses and resumes by itself. The worker (services, remote desktop, the desktop) and
+the controller reader are stubs, the state file is a temporary one, and ft-screens' socket is
+renamed, so nothing reaches the live Frametop, SteamVR, or the running relay.
+
+  input/test/pause-test.py
+"""
+import os
+import sys
+import tempfile
+
+sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), ".."))
+import game_pause as gp  # noqa: E402
+
+failures = []
+
+
+def check(label, got, want):
+    ok = got == want
+    print(("ok    " if ok else "FAIL  ") + label + ("" if ok else f": got {got!r}, want {want!r}"), flush=True)
+    if not ok:
+        failures.append(label)
+
+
+# ---------------------------------------------------------------- the gesture
+L, R = "left/thumbstick", "right/thumbstick"
+
+
+def presses(g, events):
+    """events: (time, button, down); how many times the gesture went off."""
+    return sum(g.feed(b, d, t) for t, b, d in events)
+
+
+def both(t, hold=0.1, gap=0.02):
+    """Both sticks clicked at t (the right one gap later), held for hold."""
+    return [(t, L, True), (t + gap, R, True), (t + hold, L, False), (t + hold + gap, R, False)]
+
+
+g = gp.Gesture((L, R), 2)
+check("both sticks twice: goes off", presses(g, both(0) + both(0.3)), 1)
+g = gp.Gesture((L, R), 2)
+check("both sticks once: nothing", presses(g, both(0)), 0)
+g = gp.Gesture((L, R), 2)
+check("both sticks twice, 0.9 s apart: nothing", presses(g, both(0) + both(0.9)), 0)
+g = gp.Gesture((L, R), 2)
+check("left held (sprint), right clicked twice: nothing",
+      presses(g, [(0, L, True), (1.0, R, True), (1.1, R, False), (1.3, R, True), (1.4, R, False)]), 0)
+g = gp.Gesture((L, R), 2)
+check("sticks 0.5 s apart, twice: nothing", presses(g, both(0, gap=0.5, hold=0.6) + both(1.2, gap=0.5, hold=0.6)), 0)
+g = gp.Gesture((L, R), 2)
+check("three double presses in a row: once", presses(g, both(0) + both(0.3) + both(0.6)), 1)
+g = gp.Gesture((L, R), 2)
+check("again after a second: twice in all", presses(g, both(0) + both(0.3) + both(2.0) + both(2.3)), 2)
+g = gp.Gesture((L, R), 2)
+check("repeated down states count once",
+      presses(g, [(0, L, True), (0.01, L, True), (0.02, R, True), (0.03, R, True), (0.1, L, False), (0.1, R, False)]
+              + both(0.3)), 1)
+g = gp.Gesture((L, R), 1)
+check("both sticks once (presses 1): goes off", presses(g, both(0)), 1)
+g = gp.Gesture(("right/b",), 2)
+check("one button double-clicked: goes off",
+      presses(g, [(0, "right/b", True), (0.1, "right/b", False), (0.3, "right/b", True)]), 1)
+g = gp.Gesture(("right/b",), 2)
+check("other buttons don't count", presses(g, [(0, "right/a", True), (0.2, "right/a", True)]), 0)
+
+VR = ("left/thumbstick", "right/thumbstick", "right/b", "right/a")
+check("no pause_gesture: both sticks twice", gp.gesture_of({}, VR), ((L, R), 2))
+check("pause_gesture null: none", gp.gesture_of({"pause_gesture": None}, VR), None)
+check("one button, presses 1: still twice", gp.gesture_of({"pause_gesture": {"buttons": ["right/b"], "presses": 1}}, VR),
+      (("right/b",), 2))
+check("unknown buttons dropped", gp.gesture_of({"pause_gesture": {"buttons": ["right/zz", "right/a"]}}, VR),
+      (("right/a",), 2))
+check("settings: defaults", gp.settings_of({}, VR)["auto"], True)
+check("settings: unknown desktop mode is hide", gp.settings_of({"pause_desktop": "x"}, VR)["desktop"], "hide")
+
+
+# ---------------------------------------------------------------- pausing by itself
+class StubWorker:
+    def __init__(self, log, save):
+        self.jobs = []
+        self.busy = False
+
+    def start(self):
+        pass
+
+    def submit(self, job, *args):
+        self.jobs.append(job.__name__)
+
+    def snapshot(self):
+        return {"units": [], "remote": False, "desktop": None}
+
+    def restore(self, stopped):
+        self.restored = stopped
+
+    def pause(self, desktop):
+        pass
+
+    def resume(self):
+        pass
+
+    def enforce(self):
+        pass
+
+
+class StubWatch:
+    connected = False
+
+    def __init__(self, log):
+        pass
+
+    def start(self):
+        pass
+
+    def set_gesture(self, spec):
+        self.spec = spec
+
+
+tmp = tempfile.mkdtemp(prefix="ft-pause-test-")
+gp.STATE_PATH = os.path.join(tmp, "pause.json")
+gp.SCREENS = f"\0ft_pause_test_{os.getpid()}_screens"
+gp.Worker, gp.ControllerWatch = StubWorker, StubWatch
+changes = []
+
+
+def new(rules=None):
+    """A relay starting afresh: not paused before."""
+    if os.path.exists(gp.STATE_PATH):
+        os.remove(gp.STATE_PATH)
+    changes.clear()
+    p = gp.GamePause(lambda *a: None, changes.append, VR)
+    p.configure(dict({"pause_sound": False}, **(rules or {})))
+    return p
+
+
+p = new()
+p.game_state(True, 0)
+check("a game starts: paused", (p.paused, p.reason, p.ends_with_game), (True, "game", True))
+check("the relay heard", changes, [True])
+p.game_state(True, 5)
+p.tick(5)
+check("the helper repeats it: still paused, nothing new", (p.paused, changes), (True, [True]))
+p.game_state(False, 10)
+p.tick(12)
+check("the game ends: not yet resumed", p.paused, True)
+p.tick(15.1)
+check("5 s later: resumed", (p.paused, changes), (False, [True, False]))
+check("the worker paused and resumed", p.worker.jobs, ["pause", "resume"])
+
+p = new()
+p.game_state(True, 0)
+p.game_state(False, 10)
+p.game_state(True, 12)
+p.tick(16)
+check("the next game starts within 5 s: stays paused", p.paused, True)
+p.game_state(False, 20)
+p.tick(25.1)
+check("and resumes 5 s after that one ends", p.paused, False)
+
+p = new()
+p.game_state(True, 0)
+p.toggle("gesture", 1)
+check("resumed during a game: running", p.paused, False)
+p.game_state(True, 6)
+p.tick(30)
+check("and stays running while that game runs", p.paused, False)
+p.game_state(False, 40)
+p.game_state(True, 50)
+check("the next game pauses again", p.paused, True)
+
+p = new()
+p.toggle("command", 0)
+check("paused outside a game: not tied to one", (p.paused, p.ends_with_game), (True, False))
+p.game_state(True, 1)
+p.game_state(False, 10)
+p.tick(20)
+check("a game comes and goes: still paused", p.paused, True)
+p.toggle("command", 21)
+check("toggled: running", p.paused, False)
+
+p = new({"pause_auto": False})
+p.game_state(True, 0)
+check("auto off: a game doesn't pause", p.paused, False)
+p.toggle("gesture", 1)
+p.game_state(False, 10)
+p.tick(20)
+check("auto off: a pause during a game outlasts it", p.paused, True)
+
+p = new()
+p.game_state(True, 0)
+p.tick(0 + gp.GAME_STALE + 0.1)
+check("the helper goes quiet: the game counts as ended", p.game, False)
+p.tick(0 + gp.GAME_STALE + gp.RESUME_DELAY + 0.2)
+check("and the pause ends", p.paused, False)
+
+p = new()
+p.toggle("gesture", 0)
+q = gp.GamePause(lambda *a: None, changes.append, VR)
+check("a relay restarted while paused stays paused", (q.paused, q.reason), (True, "gesture"))
+check("and stops the services again", q.worker.jobs, ["enforce"])
+q.toggle("gesture", 1)
+r = gp.GamePause(lambda *a: None, changes.append, VR)
+check("resumed: the next relay starts running", r.paused, False)
+
+p = new({"pause_gesture": None})
+check("pause_gesture null: the reader gets none", p.watch.spec, None)
+check("status", sorted(p.status()), sorted(["t", "paused", "reason", "since", "ends_with_game", "game", "busy",
+                                            "stopped", "gesture_reader", "auto", "desktop"]))
+
+print("FAILED: " + ", ".join(failures) if failures else "all passed", flush=True)
+for f in os.listdir(tmp):
+    os.remove(os.path.join(tmp, f))
+os.rmdir(tmp)
+sys.exit(1 if failures else 0)