fix(qemu): pi protocol reader — executor pump instead of loop.add_reader

add_reader on the proto FIFO armed epoll but never delivered callbacks
under uvloop when the fd number recycled a just-closed socket fd
(nondeterministic per instance): the guest's GPIO lines sat unread in
the pipe and canvas LEDs stayed dark while serial kept flowing. Pump
the FIFO from a worker thread (select + os.read, like _watch_stderr)
and schedule _handle_gpio_line back onto the loop.

backend/app/services/qemu_manager.py

@@ -587,38 +587,48 @@ class QemuManager:
                     inst.client_id, inst.proto_pipe_base)
 
         loop = asyncio.get_running_loop()
-        # Use add_reader to integrate the readable FIFO with the loop.
-        linebuf = bytearray()
 
-        def _on_readable() -> None:
+        # Thread-based blocking pump instead of loop.add_reader. add_reader
-            fd = inst._proto_out_fd
+        # on the FIFO fd was observed (staging, 2026-07-28) to arm epoll but
-            if fd is None:
+        # never deliver callbacks under uvloop when the fd number recycles a
-                return
+        # just-closed socket fd (e.g. _connect_serial's retry sockets) —
-            try:
+        # nondeterministic per instance, and the guest's GPIO lines then sat
-                data = os.read(fd, 4096)
+        # unread in the pipe while LEDs stayed dark. A worker thread doing
-            except BlockingIOError:
+        # kernel select() + os.read() has no fd-reuse hazard; handler
-                return
+        # coroutines are scheduled back onto the loop. Mirrors the executor
-            except OSError:
+        # pattern _watch_stderr already uses.
-                return
+        def _pump() -> None:
-            if not data:
+            import select as _select
-                return
+            buf = bytearray()
-            linebuf.extend(data)
+            while inst.running:
-            while b'\n' in linebuf:
+                fd = inst._proto_out_fd
-                line, _, rest = linebuf.partition(b'\n')
+                if fd is None:
-                linebuf[:] = rest
+                    return
-                asyncio.create_task(self._handle_gpio_line(
+                try:
-                    inst, line.decode('ascii', 'ignore').strip(),
+                    ready, _, _ = _select.select([fd], [], [], 0.5)
-                ))
+                except (OSError, ValueError):
+                    return  # fd closed by _shutdown
+                if not ready:
+                    continue
+                try:
+                    data = os.read(fd, 4096)
+                except BlockingIOError:
+                    continue
+                except OSError:
+                    return
+                if not data:
+                    continue
+                buf.extend(data)
+                while b'\n' in buf:
+                    line, _, rest = buf.partition(b'\n')
+                    buf[:] = rest
+                    asyncio.run_coroutine_threadsafe(
+                        self._handle_gpio_line(
+                            inst, line.decode('ascii', 'ignore').strip()),
+                        loop,
+                    )
 
-        loop.add_reader(inst._proto_out_fd, _on_readable)
+        await loop.run_in_executor(None, _pump)
-        # Keep this coroutine alive while inst is running so the loop
-        # doesn't garbage-collect the reader registration.
-        while inst.running:
-            await asyncio.sleep(1.0)
-        try:
-            loop.remove_reader(inst._proto_out_fd)
-        except Exception:
-            pass
 
     async def _handle_gpio_line(self, inst: PiInstance, line: str) -> None:
         """Dispatch a single text-protocol line from the Pi shim layer.