Merge origin/main into pad-gamescope

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

README.md

@@ -73,7 +73,7 @@ Drag files onto the window to send them. Drop a game's .zip, folder or .exe to a
 Browse the shots you take in the headset and save them to your Pictures folder.
 
 **⌨️ Keyboard and trackpad**<br>
-Type and point in the Frame's apps from your computer or phone, through KDE Connect, which Frame Control brings along and sets up on the Frame. Nothing else to install, anywhere.
+Type and point in the headset from your computer or phone, through Valve's own input path, with nothing to install. Accents and emoji use KDE Connect, which Frame Control brings along and sets up the first time you type one.
 
 </td>
 </tr>

docs/streaming.md

@@ -123,10 +123,13 @@ two fingers to scroll, two-finger tap to right-click) plus a text field that
 types on the Frame. On a computer, clicking the pad passes your mouse and
 keyboard through to the Frame until you press Esc (⌘ is sent as Ctrl on a Mac).
 
-It goes through **KDE Connect**, the first-party route (KDE makes the Frame's
+Since Control (below), the pad goes through **gamescope's own input socket**
-desktop): Frame Control's server runs [`ui/frame_input_agent.py`](../ui/frame_input_agent.py)
+([`ui/frame_touch.py`](../ui/frame_touch.py)), so it reaches whichever panel has
-on the Frame, which talks KDE Connect's own LAN protocol to the Frame's
+focus, Steam's own menus included, and needs nothing installed on the Frame.
-`kdeconnectd` as if it were a phone. KDE Connect does the typing and clicking.
+Only characters that aren't on a US keyboard (accents, emoji) go through
+**KDE Connect** as below; it starts the first time you type one. (Before
+Control, everything went through KDE Connect, which is why the notes below
+describe it in detail.)
 
 **Verified 2026-09-28** (SteamOS 0.4.1, build 20260925.6191901):
 

tests/test_input.py

@@ -630,3 +630,194 @@ class PageQueue(unittest.TestCase):
 
 if __name__ == "__main__":
     unittest.main()
+
+
+class PadEvents(unittest.TestCase):
+    """The trackpad and key row go through gamescope; accents and emoji go the KDE Connect way."""
+
+    @classmethod
+    def setUpClass(cls):
+        import server
+        cls.pad = staticmethod(server.pad_events)
+
+    def test_moves_clicks_and_scroll(self):
+        out, extra = self.pad([{"dx": 3, "dy": -2}, {"singleclick": True}, {"singlehold": True},
+                               {"singlerelease": True}, {"scroll": True, "dy": 1}, {"rightclick": True}])
+        self.assertEqual(out[0], {"dx": 3, "dy": -2})
+        self.assertEqual(out[1:3], [{"button": "left", "down": True}, {"button": "left", "down": False}])
+        self.assertEqual(out[3:5], [{"button": "left", "down": True}, {"button": "left", "down": False}])
+        self.assertEqual(out[5], {"scroll": [0, -15]})  # up = content follows the finger
+        self.assertEqual(out[6:], [{"button": "right", "down": True}, {"button": "right", "down": False}])
+        self.assertEqual(extra, [])
+
+    def test_keys_and_modifiers(self):
+        out, _ = self.pad([{"specialKey": 12}, {"specialKey": 1, "ctrl": True}, {"key": "c", "ctrl": True}])
+        self.assertEqual(out[:2], [{"key": 28, "down": True}, {"key": 28, "down": False}])
+        self.assertEqual([e["key"] for e in out[2:6]], [29, 14, 14, 29])  # Ctrl+Backspace
+        self.assertEqual([(e["key"], e["down"]) for e in out[6:]],
+                         [(29, True), (46, True), (46, False), (29, False)])  # Ctrl+C
+
+    def test_text_stays_ascii_through_gamescope(self):
+        out, slow = self.pad([{"key": "Hi "}, {"key": "there"}])
+        self.assertEqual(out, [{"text": "Hi there"}])
+        self.assertEqual(slow, [])
+
+    def test_keyboard_goes_to_kde_whole_and_in_order(self):
+        events = [{"key": "a"}, {"key": "é"}, {"specialKey": 12}, {"key": "c", "ctrl": True}, {"dx": 2, "dy": 1}]
+        out, slow = self.pad(events, kde=True)
+        self.assertEqual(out, [{"dx": 2, "dy": 1}])  # the pointer still goes through gamescope
+        self.assertEqual(slow, events[:4])  # the keyboard as sent, in order
+
+    def test_shifted_key_with_modifier(self):
+        out, _ = self.pad([{"key": "A", "ctrl": True}])
+        self.assertEqual([e["key"] for e in out], [29, 42, 30, 30, 42, 29])
+
+
+class PadDelivery(unittest.TestCase):
+    """remote_input's promises: "sent" means the whole batch is taken; accents go once, in order."""
+
+    class Agent:
+        def __init__(self, sent=True, state="ready"):
+            self.sent, self.state, self.got = sent, state, []
+
+        def send(self, events):
+            if events and self.sent:
+                self.got.append(events)
+            return {"state": self.state, "sent": self.sent and bool(events)}
+
+    def setUp(self):
+        import server
+        self.s = server
+        self.saved = (server._touch, server._input, server.KDE_WAIT)
+        server._touch, server._input = self.Agent(), self.Agent()
+        self.reset()
+
+    def tearDown(self):
+        self.s._touch, self.s._input, self.s.KDE_WAIT = self.saved
+        self.settle()
+        self.reset()
+
+    def reset(self):
+        self.s._kde_queue.clear()
+        self.s._kde_until = self.s._typed_until = 0.0
+        self.s._kde_worker[0] = None
+
+    def settle(self):
+        worker = self.s._kde_worker[0]
+        if worker:
+            worker.join(5)
+
+    def typed(self):
+        return [e for batch in self.s._input.got for e in batch]
+
+    def test_accent_only_batch_is_acknowledged_and_typed_once(self):
+        status = self.s.remote_input({"events": [{"key": "é"}]})
+        self.assertTrue(status["sent"])  # so the page doesn't send it again
+        self.settle()
+        self.assertEqual(self.typed(), [{"key": "é"}])
+
+    def test_mixed_text_stays_in_one_ordered_stream(self):
+        self.s.remote_input({"events": [{"key": "aéb"}, {"specialKey": 12}]})
+        self.settle()
+        self.assertEqual(self.typed(), [{"key": "aéb"}, {"specialKey": 12}])
+        self.assertEqual(self.s._touch.got, [])  # none of it split off through gamescope
+
+    def test_batch_gamescope_didnt_take_is_not_queued(self):
+        self.s._touch = self.Agent(sent=False, state="starting")
+        status = self.s.remote_input({"events": [{"dx": 1, "dy": 1}, {"key": "é"}]})
+        self.assertFalse(status["sent"])  # the page tries the whole batch again...
+        self.assertEqual(self.s._kde_queue, [])  # ...so the accent isn't queued twice
+
+    def test_waits_for_kde_connect_then_sends_without_more_input(self):
+        self.s._input = self.Agent(sent=False, state="starting")
+        self.s.remote_input({"events": [{"key": "é"}]})
+        time.sleep(0.2)
+        self.assertEqual(self.s._kde_queue, [{"key": "é"}])
+        self.s._input.sent = True
+        self.settle()
+        self.assertEqual(self.typed(), [{"key": "é"}])
+        self.assertEqual(self.s._kde_queue, [])
+
+    def test_later_ascii_waits_behind_a_pending_accent(self):
+        self.s._input = self.Agent(sent=False, state="starting")
+        self.s.remote_input({"events": [{"key": "é"}]})
+        self.s.remote_input({"events": [{"key": "x"}]})
+        self.assertEqual(self.s._touch.got, [])  # "x" can't overtake the "é"
+        self.assertEqual(self.s._kde_queue, [{"key": "é"}, {"key": "x"}])
+        self.s._input.sent = True
+        self.settle()
+        self.assertEqual(self.typed(), [{"key": "é"}, {"key": "x"}])
+
+    def test_queue_is_bounded_and_dropped_on_error(self):
+        self.s._input = self.Agent(sent=False, state="error")
+        self.s.remote_input({"events": [{"key": "é"}] * 150})
+        self.s.remote_input({"events": [{"key": "é"}] * 150})
+        self.settle()
+        self.assertEqual(self.s._kde_queue, [])
+        self.s._input = self.Agent(sent=False, state="starting")
+        self.s.KDE_WAIT = 0.3
+        self.s.remote_input({"events": [{"key": "é"}] * 150})
+        self.s.remote_input({"events": [{"key": "é"}] * 150})
+        self.assertLessEqual(len(self.s._kde_queue), self.s.KDE_QUEUE_LIMIT)
+        self.settle()
+        self.assertEqual(self.s._kde_queue, [])
+
+    def test_ascii_just_after_an_accent_stays_behind_it(self):
+        self.s.remote_input({"events": [{"key": "é"}]})
+        self.settle()
+        self.s.remote_input({"events": [{"key": "x"}]})  # KDE Connect may still be typing the é
+        self.settle()
+        self.assertEqual(self.s._touch.got, [])
+        self.assertEqual(self.typed(), [{"key": "é"}, {"key": "x"}])
+
+    def test_accent_just_after_gamescope_text_waits_for_it(self):
+        self.s.remote_input({"events": [{"key": "x" * 50}]})
+        self.assertEqual(self.s._touch.got, [[{"text": "x" * 50}]])
+        start = time.time()
+        self.s.remote_input({"events": [{"key": "é"}]})
+        self.settle()
+        self.assertGreater(time.time() - start, 0.6)  # 50 keys at 16 ms, plus the margin
+        self.assertEqual(self.typed(), [{"key": "é"}])
+
+    def test_typing_time_counts_every_key_transition(self):
+        t = self.s.typing_seconds
+        self.assertAlmostEqual(t([{"text": "ab"}]), 4 * 0.008)
+        self.assertAlmostEqual(t([{"text": "A"}]), 4 * 0.008)  # shift down, key down, key up, shift up
+        self.assertAlmostEqual(t([{"text": "x" * 500}]), 8.0)  # a long paste
+        self.assertAlmostEqual(t([{"key": 29, "down": True}, {"dx": 3, "dy": 0}]), 0.008)
+
+    def test_the_margin_is_added_once_not_per_request(self):
+        for _ in range(10):
+            self.s.remote_input({"events": [{"key": "ab"}]})
+        # ten requests of 4 transitions each: the work queues up, with no margin added per request
+        self.assertLess(self.s._typed_until - time.time(), 10 * 4 * 0.008 + 0.05)
+
+    def test_trimming_while_sending_doesnt_drop_unsent_events(self):
+        slow = self.Agent()
+        gate = threading.Event()
+        send = slow.send
+
+        def held(events):
+            gate.wait(5)
+            return send(events)
+        slow.send = held
+        self.s._input = slow
+        self.s.remote_input({"events": [{"key": "é"}] * 150})
+        time.sleep(0.2)  # the worker has taken its 150 and is mid-send
+        self.s.remote_input({"events": [{"key": "ü"}] * 150})
+        gate.set()
+        self.settle()
+        sent = self.typed()
+        self.assertEqual(sent[:150], [{"key": "é"}] * 150)
+        self.assertEqual(sent[150:], [{"key": "ü"}] * 150)  # none of the new ones went missing
+
+    def test_input_queued_as_the_worker_finishes_is_still_sent(self):
+        for _ in range(40):
+            self.reset()
+            self.s._input.got.clear()
+            self.s.remote_input({"events": [{"key": "é"}]})
+            time.sleep(0.002)
+            self.s.remote_input({"events": [{"key": "ü"}]})
+            self.settle()
+            self.assertEqual(self.s._kde_queue, [])
+            self.assertEqual(self.typed(), [{"key": "é"}, {"key": "ü"}])

tests/test_touch.py

@@ -147,6 +147,14 @@ class Apply(unittest.TestCase):
         self.assertFalse(self.t.STALE[0])
         self.assertEqual(self.said[-1][1].get("stale"), None)
 
+    def test_stale_survives_a_bare_release(self):
+        gs = FakeGamescope()
+        panel = self.t.apply(gs, {"button": "left", "down": True, "window": 99, "display": ":1"}, None)
+        self.assertTrue(self.t.STALE[0])
+        self.t.apply(gs, {"button": "left", "down": False}, panel)  # the page's release names no panel
+        self.assertTrue(self.t.STALE[0])
+        self.assertEqual(self.said[-1][1].get("stale"), True)
+
     def test_same_window_id_on_the_other_display_is_another_panel(self):
         gs = FakeGamescope()
         self.t.apply(gs, {"fx": 0.5, "fy": 0.5, "window": 7, "display": ":0"}, None)

ui/frame_touch.py

@@ -369,8 +369,10 @@ def apply(gs, event, panel):
         say("ready", focus=panel.get("window"), display=panel.get("display"), stale=True)  # the page re-syncs
         STALE[0] = True
         return panel
-    if STALE[0] and not stale:
+    release = event.get("down") is False and ("button" in event or "key" in event)
+    if STALE[0] and not stale and not (release and "window" not in event):
         # Anything that goes through (a trackpad move names no panel) means caught up: stop re-syncing.
+        # A bare release doesn't: the page leaves the panel off releases, so it says nothing about focus.
         STALE[0] = False
         say("ready", focus=(panel or {}).get("window"), display=(panel or {}).get("display"))
     if "fx" in event and panel and panel.get("window"):

ui/index.html

@@ -876,10 +876,9 @@
         <button data-pad-key="12" title="Enter">Enter</button>
       </span>
     </div>
-    <div class="hint">Types and points in apps on the Frame, such as Chromium or the desktop's Linux apps; Steam's own VR menus
+    <div class="hint">Types and points in whichever panel has focus in the headset, through Valve's own input path: nothing to
-      don't take it. It works through KDE Connect, which comes with Frame Control: the first time, it copies Valve's build
+      install and no internet needed. Accents and emoji go through KDE Connect, which comes with Frame Control and starts
-      for the Frame (3.6 MB, 18 MB unpacked) into your home folder there and pairs with it. No internet needed, and nothing
+      the first time you type one (it copies Valve's build for the Frame, 3.6 MB, into your home folder there). <a href="#" data-about>Licences</a></div>
-      else to install. <a href="#" data-about>Licences</a></div>
   </section>
 
   <section id="shots">
@@ -3901,10 +3900,10 @@ const typesText = el => el?.matches?.('textarea, input:not([type=range], [type=c
 document.addEventListener("focusin", e => document.body.classList.toggle("typing", !!typesText(e.target)));
 document.addEventListener("focusout", () => document.body.classList.remove("typing"));
 
-// ---- keyboard and trackpad: events go to KDE Connect on the Frame (see ui/frame_input_agent.py) ----
+// ---- keyboard and trackpad: events go to the Frame through gamescope (ui/frame_touch.py); accents and emoji through KDE Connect ----
 const pad = { state: "off", queue: [], sending: false, poll: null, captured: false };
 const PAD_STATES = {
-  off: "", starting: "Connecting…", installing: "Setting up KDE Connect on the Frame (first time only)…",
+  off: "", starting: "Connecting…", installing: "Setting up (first time only)…",
   pairing: "Pairing with KDE Connect…", ready: "On",
 };
 // KDE Connect's specialKey numbers (plugins/mousepad in its source).
@@ -3922,7 +3921,7 @@ function padShow(status) {
   $("padOn").hidden = ready || busy;
   $("padOn").textContent = pad.state === "error" ? "Try again" : "Turn on";
   $("padArea").classList.toggle("off", !ready);
-  $("padHint").textContent = pad.state === "error" ? status.message || "Couldn't reach KDE Connect on the Frame."
+  $("padHint").textContent = pad.state === "error" ? status.message || "Couldn't reach the Frame."
     : !ready ? (busy ? PAD_STATES[pad.state] : "Turn on to use this as a trackpad for the Frame.")
     : TOUCH ? "Drag to move the pointer · tap to click · two fingers to scroll · two-finger tap to right-click"
     : pad.captured ? "Your mouse and keyboard now control the Frame. Press Esc to stop."

ui/server.py

@@ -60,6 +60,7 @@ import frame_report  # noqa: E402
 import frame_store  # noqa: E402
 import frame_telemetry  # noqa: E402
 import frame_titles  # noqa: E402
+import frame_touch  # noqa: E402
 import frame_webinstall  # noqa: E402
 import frame_vr  # noqa: E402
 import frame_utilities  # noqa: E402
@@ -1005,12 +1006,274 @@ def licenses():
     return notices
 
 
+# KDE Connect's specialKey numbers -> Linux key codes (KEY_BACKSPACE, KEY_TAB, ...).
+PAD_KEYS = {1: 14, 2: 15, 4: 105, 5: 103, 6: 106, 7: 108, 8: 104, 9: 109, 10: 102, 11: 107, 12: 28, 13: 111, 14: 1,
+            **{21 + i: code for i, code in enumerate([59, 60, 61, 62, 63, 64, 65, 66, 67, 68, 87, 88])}}
+PAD_MODS = (("ctrl", 29), ("alt", 56), ("shift", 42), ("super", 125))
+PAD_CLICKS = {"singleclick": "left", "rightclick": "right", "middleclick": "middle"}
+PAD_NOTCH = 15  # scroll units for one wheel notch
+
+
+def typed_in_kde(events):
+    """Whether these events must go the KDE Connect way: they hold an accent or emoji, or
+    earlier ones are queued, being sent or still being typed there (typing must stay in
+    order, so it can't be split between the two ways)."""
+    with _kde_lock:
+        waiting = bool(_kde_queue) or time.time() < _kde_until
+    return waiting or any(ch not in frame_touch.ASCII for e in events for ch in e.get("key", ""))
+
+
+def pad_events(events, kde=False):
+    """The trackpad's and key row's events as (gamescope events, KDE Connect events).
+
+    Gamescope's own input reaches every panel and needs nothing installed, but types only
+    US-keyboard characters. With kde, the keyboard goes to KDE Connect as it was sent, so
+    text stays in order; the pointer always goes through gamescope.
+    """
+    ascii_keys = frame_touch.ASCII
+    out, slow = [], []
+
+    def tap(code, mods):
+        down = [{"key": c, "down": True} for c in mods]
+        out.extend(down + [{"key": code, "down": True}, {"key": code, "down": False}]
+                   + [{"key": c, "down": False} for c in reversed(mods)])
+    for e in events:
+        mods = [code for name, code in PAD_MODS if e.get(name)]
+        if "dx" in e or "dy" in e:
+            if not e.get("scroll"):
+                out.append({k: e[k] for k in ("dx", "dy") if k in e})
+        if e.get("scroll"):
+            out.append({"scroll": [0, -(e.get("dy") or 0) * PAD_NOTCH]})
+        for flag, button in PAD_CLICKS.items():
+            if e.get(flag):
+                out += [{"button": button, "down": True}, {"button": button, "down": False}]
+        if e.get("doubleclick"):
+            out += [{"button": "left", "down": d} for d in (True, False, True, False)]
+        if e.get("singlehold"):
+            out.append({"button": "left", "down": True})
+        if e.get("singlerelease"):
+            out.append({"button": "left", "down": False})
+        if kde and ("key" in e or "specialKey" in e):
+            slow.append({k: e[k] for k in ("key", "specialKey", "ctrl", "alt", "shift", "super") if k in e})
+            continue
+        if "specialKey" in e and e["specialKey"] in PAD_KEYS:
+            tap(PAD_KEYS[e["specialKey"]], mods)
+        if "key" in e:
+            for ch in e["key"]:
+                if ch not in ascii_keys:
+                    continue  # only reached when kde is wrong; the caller checks typed_in_kde first
+                elif mods:
+                    code, shifted = ascii_keys[ch]
+                    tap(code, mods + ([42] if shifted and 42 not in mods else []))
+                else:
+                    if out and "text" in out[-1] and len(out[-1]["text"]) < 100:
+                        out[-1]["text"] += ch
+                    else:
+                        out.append({"text": ch})
+    return out, slow
+
+
 def remote_input(body):
-    """{"events": [...]} sends keyboard and pointer events; {} (or none yet) just starts the agent."""
+    """{"events": [...]} sends keyboard and pointer events; {} (or none yet) just starts the agent.
+
+    "sent" in the answer means the whole batch is taken (the page keeps it and tries again
+    if not), so the keyboard's share is queued only once that's so.
+    """
     events = body.get("events", [])
     if not isinstance(events, list) or len(events) > INPUT_BATCH_LIMIT:
         raise Failure(f"events must be a list of at most {INPUT_BATCH_LIMIT}", 400)
-    return _input.send([input_event(e) for e in events])
+    events = [input_event(e) for e in events]
+    touch, slow = pad_events(events, typed_in_kde(events))
+    status = _touch.send(touch)
+    if touch and not status.get("sent"):
+        return status
+    if touch:
+        _note_typed(touch)
+    if slow:
+        _queue_for_kde(slow)
+    return {**status, "sent": True}
+
+
+# Typing KDE Connect is to do: it starts the first time it's needed, so this waits for it.
+# Nothing on the Frame says when typed text has landed, so each way holds the other back
+# for as long as what it was given should take (a short margin on a time estimate).
+_kde_queue = []
+_kde_lock = threading.Lock()
+_kde_worker = [None]
+_kde_until = 0.0     # keys typed through KDE Connect should have landed by then
+_typed_until = 0.0   # when gamescope's agent will have typed everything it's been given
+KDE_QUEUE_LIMIT = 200
+KDE_WAIT = 90  # seconds to wait for KDE Connect before giving up on what's queued
+TYPING_SETTLE = 0.6  # seconds past the estimate, once
+
+
+def typing_seconds(touch):
+    """How long gamescope's agent takes over these: it sleeps 8 ms after every key transition."""
+    transitions = 0
+    for e in touch:
+        if "text" in e:
+            transitions += sum(2 + 2 * frame_touch.ASCII[ch][1] for ch in e["text"] if ch in frame_touch.ASCII)
+        elif "key" in e:
+            transitions += 1
+    return transitions * 0.008
+
+
+def _note_typed(touch):
+    global _typed_until
+    seconds = typing_seconds(touch)
+    if seconds:
+        with _kde_lock:  # the end of the work, which queues up; the margin is added once, where it's checked
+            _typed_until = max(_typed_until, time.time()) + seconds
+
+
+def _queue_for_kde(events):
+    with _kde_lock:
+        _kde_queue.extend(events)
+        del _kde_queue[:-KDE_QUEUE_LIMIT]
+        if _kde_worker[0] is None:
+            _kde_worker[0] = threading.Thread(target=_drain_kde, daemon=True)
+            _kde_worker[0].start()
+
+
+def _drain_kde():
+    """The only sender, in order. A batch leaves the queue while it's being sent (the queue
+    can be trimmed meanwhile) and goes back to the front if it didn't go."""
+    global _kde_until
+    deadline = time.time() + KDE_WAIT
+    while True:
+        with _kde_lock:
+            batch = _kde_queue[:INPUT_BATCH_LIMIT]
+            del _kde_queue[:len(batch)]
+            if not batch:
+                _kde_worker[0] = None  # retired under the lock, so a new request starts another
+                return
+            wait = _typed_until + TYPING_SETTLE - time.time()
+            _kde_until = time.time() + 3600  # held while in flight
+        if wait > 0:
+            time.sleep(wait)  # gamescope's typing, from just before, goes first
+        status = _input.send(batch)
+        with _kde_lock:
+            if status.get("sent"):
+                _kde_until = time.time() + len(batch) * 0.03 + TYPING_SETTLE
+                deadline = time.time() + KDE_WAIT
+                continue
+            _kde_until = 0.0
+            if status.get("state") == "error" or time.time() > deadline:
+                _kde_queue.clear()  # not coming; don't hold it forever
+                _kde_worker[0] = None
+                return
+            _kde_queue[:0] = batch
+            del _kde_queue[:-KDE_QUEUE_LIMIT]  # as when queuing: the oldest go first
+        time.sleep(0.4)
+
+
+# ---- touch: the headset's panels, through gamescope's own input (frame_touch.py) ----
+
+PANEL_WINDOW = re.compile(r"^\d{1,10}$")
+PANEL_DISPLAY = re.compile(r"^:\d{1,2}$")
+TOUCH_BUTTONS = ("left", "right", "middle")
+
+
+def panels():
+    """The headset's app panels (window, display, name, size) and which has focus."""
+    return json.loads(ssh("python3 - panels", stdin=(HERE / "frame_touch.py").read_text(), timeout=20))
+
+
+def panel_target(q):
+    window, display = (q.get("window") or [""])[0], (q.get("display") or [""])[0]
+    if not PANEL_WINDOW.match(window) or not PANEL_DISPLAY.match(display):
+        raise Failure("window must be a window id and display an X display such as :1", 400)
+    return window, display
+
+
+def panel_capture(query):
+    """One frame of a panel's own window, as PNG (its pixels, without the room around it)."""
+    window, display = panel_target(parse_qs(query))
+    return ssh(f"DISPLAY={display} timeout 10 ffmpeg -hide_banner -loglevel error -nostdin -f x11grab "
+               f"-window_id {window} -i {display} -frames:v 1 -f image2pipe -c:v png -", timeout=20, text=False)
+
+
+def touch_event(event):
+    """A frame_touch.py event with only the fields it knows, in range."""
+    if not isinstance(event, dict):
+        raise Failure("each touch event must be an object", 400)
+
+    def num(name, limit):
+        value = event.get(name)
+        if isinstance(value, bool) or not isinstance(value, (int, float)) or value != value:
+            raise Failure(f"{name} must be a number", 400)
+        return max(-limit, min(limit, round(float(value), 4)))
+    out = {}
+    if "fx" in event or "fy" in event:
+        if "window" not in event:
+            raise Failure("a position needs the panel's window and display", 400)
+        out.update(fx=num("fx", 1), fy=num("fy", 1))
+    # Any event can name the panel it's meant for; the Frame drops it if another has focus.
+    if "window" in event:
+        window, display = event.get("window"), event.get("display")
+        if isinstance(window, bool) or not isinstance(window, int) or window <= 0:
+            raise Failure("window must be the panel's window id", 400)
+        if not isinstance(display, str) or not PANEL_DISPLAY.match(display):
+            raise Failure("display must be an X display such as :1", 400)
+        out.update(window=window, display=display)
+    for name in ("dx", "dy"):
+        if name in event:
+            out[name] = num(name, INPUT_MOVE_LIMIT)
+    if "button" in event:
+        if event["button"] not in TOUCH_BUTTONS:
+            raise Failure(f"button must be one of {', '.join(TOUCH_BUTTONS)}", 400)
+        out["button"], out["down"] = event["button"], event.get("down") is not False
+    if "scroll" in event:
+        sc = event["scroll"]
+        if not isinstance(sc, list) or len(sc) != 2:
+            raise Failure("scroll must be [dx, dy]", 400)
+        out["scroll"] = [num_value(v, 5000) for v in sc]
+    if "key" in event:
+        key = event["key"]
+        if isinstance(key, bool) or not isinstance(key, int) or not 0 < key < 768:
+            raise Failure("key must be a Linux key code", 400)
+        out["key"], out["down"] = key, event.get("down") is not False
+    if "text" in event:
+        text = event["text"]
+        if not isinstance(text, str) or not 0 < len(text) <= INPUT_TEXT_LIMIT:
+            raise Failure(f"text must be 1 to {INPUT_TEXT_LIMIT} characters", 400)
+        out["text"] = text
+    if not set(out) - {"window", "display"}:
+        raise Failure("touch event has nothing to do", 400)
+    return out
+
+
+def num_value(value, limit):
+    if isinstance(value, bool) or not isinstance(value, (int, float)) or value != value:
+        raise Failure("scroll values must be numbers", 400)
+    return max(-limit, min(limit, round(float(value), 2)))
+
+
+class TouchAgent(InputAgent):
+    """frame_touch.py on the Frame, fed events over one long-lived ssh. Nothing to install:
+    it uses gamescope's own input socket and the libei that's on the image."""
+
+    def __init__(self):
+        super().__init__(source=HERE / "frame_touch.py", packages=[])
+
+    def deliver(self, report, force=False):
+        return ""
+
+    def command(self, folder=""):
+        code = base64.b64encode(self.source.read_bytes()).decode()
+        return "python3 -u -c " + shlex.quote(
+            f"import base64;exec(compile(base64.b64decode('{code}'),'frame_touch','exec'))")
+
+
+_touch = TouchAgent()
+
+
+def remote_touch(body):
+    """{"events": [...]} points, clicks, scrolls and types into the focused panel."""
+    events = body.get("events", [])
+    if not isinstance(events, list) or len(events) > INPUT_BATCH_LIMIT:
+        raise Failure(f"events must be a list of at most {INPUT_BATCH_LIMIT}", 400)
+    return _touch.send([touch_event(e) for e in events])
 
 
 # ---- touch: the headset's panels, through gamescope's own input (frame_touch.py) ----
@@ -2372,7 +2635,7 @@ class Handler(BaseHTTPRequestHandler):
             elif path == "/api/touch":
                 self.send_json(_touch.send([]) if parse_qs(url.query).get("start") == ["1"] else dict(_touch.status))
             elif path == "/api/input":
-                self.send_json(_input.send([]) if parse_qs(url.query).get("start") == ["1"] else dict(_input.status))
+                self.send_json(_touch.send([]) if parse_qs(url.query).get("start") == ["1"] else dict(_touch.status))
             elif path == "/api/job":
                 self.send_json(job_status(url.query))
             elif path == "/api/android/displays":