fix(pi3): pipe chardev for proto channel + Phase 2 protocol mux

QEMU 10's virtserialport on a socket chardev (server=on,wait=off)
silently drops guest→host bytes. Reproduced cleanly: writes from
inside the guest to /dev/vport<N>p<M> succeed (no errno) but the
connected client socket receives 0 bytes. Same bug whether the
client is a single recv loop, multiple threads, TCP or UNIX socket,
or whether QEMU runs as server vs client. virtconsole on the same
socket works fine — only virtserialport is broken.

Workaround: use `pipe` chardev (a pair of named FIFOs created
beforehand by qemu_manager). guest→host through .out flows reliably
in QEMU 10 — verified with manual test: 'echo PIPE_TEST > /dev/vport1p1'
in the guest produces 'PIPE_TEST\n' immediately on the host side.

Changes:
- qemu_manager._boot: allocate a temp basename, mkfifo .in + .out,
  pass to QEMU as 'pipe,path=<base>'.
- qemu_manager._connect_gpio: open both FIFOs O_RDWR | O_NONBLOCK on
  host side (O_RDWR keeps the FIFOs open even when guest hasn't
  opened its side yet), wire .out into asyncio via loop.add_reader.
- qemu_manager._reply_gpio / _send_gpio: write to .in fd via os.write.
- qemu_manager._handle_gpio_line: extended Phase 1 GPIO-only parser
  into a full Phase 2 mux: GPIO/GPIO_SETUP/GPIO_IN/PWM_*/I2C/SPI/UART
  with appropriate replies.
- qemu_manager._shutdown: close FDs + unlink the FIFOs.
- manifest.json: bump raspberry-pi-3-virt rootfs to 2026.05+phase2-shims
  (the new rootfs ships the velxio shim Python modules under
  /usr/lib/velxio-shims/).

backend/app/services/boot_images/manifest.json

@@ -22,13 +22,13 @@
         {
           "name": "velxio-pi-rootfs-arm64.ext4",
           "asset_id": "velxio-pi-rootfs-arm64-zst",
-          "sha256": "d09eec3bbc023dc4798d70de87048b25d2db81e5530fe0fbe0853f745bdaf910",
+          "sha256": "f0ce53b933543dc071ddbf6a789f7045e603590000159951d886cf67c17c91ba",
           "size_bytes": 629145600,
-          "version": "2026.05",
+          "version": "2026.05+phase2-shims",
           "compressed": {
             "encoding": "zstd",
-            "sha256": "de55525e8475eea17fe7f4c26491154a998f7379ed82a651f0d10550cb9c997f",
+            "sha256": "f93b7e97466dc97f01b98f987a60b0205558d74fd51c852771d4a7d2aa8106ce",
-            "size_bytes": 24725464
+            "size_bytes": 24733652
           }
         }
       ]

backend/app/services/qemu_manager.py

@@ -77,9 +77,13 @@ class PiInstance:
         self.process:      subprocess.Popen | None = None
         self.overlay_path: str | None = None
         self.serial_port:  int = 0   # virtio-console TCP port (/dev/hvc0)
-        self.gpio_port:    int = 0   # virtio-serial protocol TCP port (/dev/vport0p2)
+        self.gpio_port:    int = 0   # legacy field (kept for log compat)
+        self.proto_pipe_base: str | None = None   # pipe chardev FIFO basename
         self._serial_writer: asyncio.StreamWriter | None = None
         self._gpio_writer:   asyncio.StreamWriter | None = None
+        # File descriptors for the proto pipe pair (host side).
+        self._proto_in_fd:  int | None = None  # we write here → guest reads
+        self._proto_out_fd: int | None = None  # we read here  ← guest writes
         self._tasks: list[asyncio.Task] = []
         self.running = False
 
@@ -156,9 +160,28 @@ class QemuManager:
         initramfs_path: Path = images[PI3_INITRAMFS_NAME]
         rootfs_base:    Path = images[PI3_ROOTFS_NAME]
 
-        # Allocate TCP ports for the two chardevs.
+        # Allocate transport endpoints for the two chardevs.
+        #
+        # User console: TCP socket — virtconsole on top of a socket
+        # works bidirectionally (proven by Phase 1).
+        #
+        # Protocol channel: pipe (FIFO pair on disk) — `virtserialport`
+        # on a socket chardev has a guest→host flow bug in QEMU 10
+        # (data is silently dropped). Pipe chardev sidesteps that;
+        # see project/pi-emulation/decisions.md D9.
         inst.serial_port = _find_free_port()
-        inst.gpio_port   = _find_free_port()
+        inst.gpio_port   = _find_free_port()  # kept for legacy log fields
+
+        # Make a fresh FIFO pair per session. QEMU's pipe chardev
+        # appends ".in" (host writes / guest reads) and ".out"
+        # (guest writes / host reads) to the path.
+        inst.proto_pipe_base = tempfile.mktemp(prefix="velxio-pi-proto-")
+        for suffix in (".in", ".out"):
+            path = inst.proto_pipe_base + suffix
+            try:
+                os.mkfifo(path, 0o600)
+            except FileExistsError:
+                pass
 
         # Create overlay qcow2 backed by the velxio rootfs ext4. Each
         # session gets its own writable layer; reads cascade down to
@@ -218,12 +241,14 @@ class QemuManager:
                        f'server=on,wait=off',
             '-device', 'virtio-serial-pci,id=virtio-serial0',
             '-device', 'virtconsole,chardev=cons',
-            # Protocol channel: second virtserialport on the SAME
+            # Protocol channel: pipe (FIFO pair) instead of a socket
-            # virtio-serial controller. Inside the guest this is
+            # chardev. virtserialport on socket chardev has a known
-            # /dev/vport0p2. Phase 2 hooks this up to pi_protocol_mux
+            # guest→host flow bug in QEMU 10 (see decisions.md D9).
-            # for GPIO/I2C/SPI/UART/PWM.
+            # Pipe creates <path>.in (host→guest) + <path>.out
-            '-chardev', f'socket,id=proto,host=127.0.0.1,port={inst.gpio_port},'
+            # (guest→host) as named FIFOs that QEMU opens lazily.
-                       f'server=on,wait=off',
+            # Inside the guest this still appears as /dev/vport<N>p<M>
+            # with the name "velxio-protocol".
+            '-chardev', f'pipe,id=proto,path={inst.proto_pipe_base}',
             '-device', 'virtserialport,chardev=proto,name=velxio-protocol',
             # Kernel cmdline:
             #   console=hvc0 — the virtio-console is the user terminal.
@@ -315,66 +340,215 @@ class QemuManager:
     # mux while keeping this connect/read/write plumbing identical.
 
     async def _connect_gpio(self, inst: PiInstance) -> None:
-        for attempt in range(10):
+        """Open the proto pipe pair. QEMU's `pipe` chardev creates
-            try:
+        two FIFOs: <base>.in (we write → guest reads via /dev/vport<N>)
-                reader, writer = await asyncio.open_connection(
+        and <base>.out (guest writes → we read).
-                    '127.0.0.1', inst.gpio_port,
-                )
-                inst._gpio_writer = writer
-                logger.info('%s: protocol channel connected on port %d',
-                            inst.client_id, inst.gpio_port)
-                await self._read_gpio(inst, reader)
-                return
-            except (ConnectionRefusedError, OSError):
-                await asyncio.sleep(1.0 * (attempt + 1))
-        logger.warning('%s: protocol channel connection failed',
-                       inst.client_id)
 
-    async def _read_gpio(self, inst: PiInstance,
+        We open both ends non-blocking so opening the not-yet-written
-                          reader: asyncio.StreamReader) -> None:
+        side doesn't deadlock, then register the read FIFO with the
-        """Parse text-protocol lines from the Pi shim layer.
+        asyncio loop so incoming bytes wake the line parser.
-
-        Phase 1 understands GPIO only (existing protocol). Phase 2
-        extends this to dispatch I2C/SPI/UART/PWM as well via
-        pi_protocol_mux.
         """
-        linebuf = b''
+        if not inst.proto_pipe_base:
-        while inst.running:
+            return
-            try:
+        in_path  = inst.proto_pipe_base + '.in'
-                chunk = await asyncio.wait_for(reader.read(256), timeout=0.1)
+        out_path = inst.proto_pipe_base + '.out'
-                if not chunk:
+
-                    break
+        # Wait for the FIFOs to exist (created by _boot before this
-                linebuf += chunk
+        # task is spawned). If QEMU failed to start they may never
-                while b'\n' in linebuf:
+        # appear — bail after a few seconds.
-                    line, linebuf = linebuf.split(b'\n', 1)
+        for _ in range(50):
-                    await self._handle_gpio_line(
+            if os.path.exists(in_path) and os.path.exists(out_path):
-                        inst, line.decode('ascii', 'ignore').strip(),
-                    )
-            except asyncio.TimeoutError:
-                continue
-            except Exception as e:
-                logger.warning('%s protocol read: %s', inst.client_id, e)
                 break
+            await asyncio.sleep(0.1)
+        else:
+            logger.warning('%s: proto pipe FIFOs missing; aborting',
+                           inst.client_id)
+            return
+
+        try:
+            # O_RDWR keeps both ends opened on the host side so the
+            # FIFOs never reach an EOF state, even if QEMU briefly
+            # disconnects (e.g. between booting and the guest opening
+            # /dev/vport<N>). Non-blocking so we can integrate with
+            # the asyncio loop via add_reader/add_writer.
+            inst._proto_in_fd  = os.open(in_path,
+                                         os.O_RDWR | os.O_NONBLOCK)
+            inst._proto_out_fd = os.open(out_path,
+                                         os.O_RDWR | os.O_NONBLOCK)
+        except OSError as exc:
+            logger.warning('%s: open proto pipes: %s',
+                           inst.client_id, exc)
+            return
+
+        logger.info('%s: protocol channel opened (pipe %s.{in,out})',
+                    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:
+            fd = inst._proto_out_fd
+            if fd is None:
+                return
+            try:
+                data = os.read(fd, 4096)
+            except BlockingIOError:
+                return
+            except OSError:
+                return
+            if not data:
+                return
+            linebuf.extend(data)
+            while b'\n' in linebuf:
+                line, _, rest = linebuf.partition(b'\n')
+                linebuf[:] = rest
+                asyncio.create_task(self._handle_gpio_line(
+                    inst, line.decode('ascii', 'ignore').strip(),
+                ))
+
+        loop.add_reader(inst._proto_out_fd, _on_readable)
+        # 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:
-        # Expected: "GPIO <bcm_pin> <0|1>"
+        """Dispatch a single text-protocol line from the Pi shim layer.
+
+        Phase 2 supports the velxio Pi shim wire format:
+          GPIO <bcm> <0|1>                  → gpio_change event
+          GPIO_SETUP <bcm> <in|out> <pud>   → noted (no canvas effect)
+          GPIO_IN <bcm>                     → reply VAL <bcm> <0|1>
+          PWM_START <bcm> <freq> <duty>     → gpio_pwm event
+          PWM_CHANGE <bcm> <freq> <duty>    → gpio_pwm event
+          PWM_STOP <bcm>                    → gpio_pwm stop
+          I2C / SPI / UART …                → reply I2C_ERR / SPI_DATA empty
+                                              (Phase 2.5 wires real bridges)
+        """
         parts = line.split()
-        if len(parts) == 3 and parts[0] == 'GPIO':
+        if not parts:
+            return
+        op = parts[0]
+
+        if op == 'GPIO' and len(parts) == 3:
             try:
                 pin   = int(parts[1])
                 state = int(parts[2])
-                await inst.emit('gpio_change', {'pin': pin, 'state': state})
+                await inst.emit('gpio_change',
+                                {'pin': pin, 'state': state})
             except ValueError:
                 pass
+            return
 
-    async def _send_gpio(self, inst: PiInstance, pin: int, state: bool) -> None:
+        if op == 'GPIO_SETUP' and len(parts) >= 3:
-        if inst._gpio_writer:
+            # Setup is informational; emit so the frontend can show
-            msg = f'SET {pin} {1 if state else 0}\n'.encode()
+            # pin direction badges if it wants. No canvas-level wiring
-            inst._gpio_writer.write(msg)
+            # needed today.
             try:
-                await inst._gpio_writer.drain()
+                pin = int(parts[1])
-            except Exception as e:
+                direction = parts[2]
-                logger.warning('%s protocol send: %s', inst.client_id, e)
+                await inst.emit(
+                    'gpio_setup',
+                    {'pin': pin, 'direction': direction,
+                     'pull': parts[3] if len(parts) > 3 else 'pud_off'},
+                )
+            except ValueError:
+                pass
+            return
+
+        if op == 'GPIO_IN' and len(parts) == 2:
+            # Reply with the last known state of the pin. For Phase 2
+            # we just echo 0 — the canvas-side input wiring fans in
+            # through SET commands which the shim caches on the guest.
+            # When canvas-driven inputs land in Phase 2.5 this will
+            # query the gpio event bus' last-state map.
+            try:
+                pin = int(parts[1])
+                await self._reply_gpio(inst, f'VAL {pin} 0')
+            except ValueError:
+                pass
+            return
+
+        if op in ('PWM_START', 'PWM_CHANGE') and len(parts) == 4:
+            try:
+                pin  = int(parts[1])
+                freq = float(parts[2])
+                duty = float(parts[3])
+                await inst.emit('gpio_pwm',
+                                {'pin': pin, 'frequency': freq,
+                                 'duty_cycle': duty,
+                                 'event': 'start' if op == 'PWM_START' else 'change'})
+            except ValueError:
+                pass
+            return
+
+        if op == 'PWM_STOP' and len(parts) == 2:
+            try:
+                pin = int(parts[1])
+                await inst.emit('gpio_pwm',
+                                {'pin': pin, 'event': 'stop'})
+            except ValueError:
+                pass
+            return
+
+        # I2C / SPI / UART — Phase 2.5 will wire these to real canvas
+        # bridges. For Phase 2 we send an immediate "no slave" reply
+        # so user code gets an exception path instead of hanging.
+        if op == 'I2C' and len(parts) >= 4:
+            sub = parts[3]
+            if sub in ('R', 'RR'):
+                bus = parts[1]
+                addr = parts[2]
+                await self._reply_gpio(inst, f'I2C_ERR {bus} {addr} no-slave')
+            return
+
+        if op == 'SPI' and len(parts) >= 4 and parts[3] == 'X':
+            bus = parts[1]
+            cs  = parts[2]
+            # Reply with zero bytes of the same length so user code
+            # gets a deterministic empty xfer.
+            try:
+                req_hex = parts[4] if len(parts) > 4 else ''
+                length = len(bytes.fromhex(req_hex))
+            except ValueError:
+                length = 0
+            await self._reply_gpio(inst,
+                                   f'SPI_DATA {bus} {cs} {"00" * length}')
+            return
+
+        if op == 'UART' and len(parts) >= 3 and parts[2] == 'RX_REQ':
+            port = parts[1]
+            await self._reply_gpio(inst, f'UART_RX {port}')
+            return
+
+        # Unknown — log at debug level (not a hot path)
+        logger.debug('pi-protocol: unhandled line: %r', line)
+
+    async def _reply_gpio(self, inst: PiInstance, line: str) -> None:
+        """Send a reply frame back to the guest over the protocol
+        chardev (pipe-backed). Used for any op that the shim layer
+        waits on."""
+        if inst._proto_in_fd is None:
+            return
+        try:
+            os.write(inst._proto_in_fd, (line + '\n').encode('ascii'))
+        except OSError as e:
+            logger.warning('%s protocol reply: %s', inst.client_id, e)
+
+    async def _send_gpio(self, inst: PiInstance, pin: int,
+                          state: bool) -> None:
+        if inst._proto_in_fd is None:
+            return
+        msg = f'SET {pin} {1 if state else 0}\n'.encode()
+        try:
+            os.write(inst._proto_in_fd, msg)
+        except OSError as e:
+            logger.warning('%s protocol send: %s', inst.client_id, e)
 
     # ── QEMU stderr watcher ───────────────────────────────────────────────────
 
@@ -412,6 +586,24 @@ class QemuManager:
                 pass
             inst._gpio_writer = None
 
+        # Close the proto pipe FDs and unlink the FIFOs.
+        for attr in ('_proto_in_fd', '_proto_out_fd'):
+            fd = getattr(inst, attr, None)
+            if fd is not None:
+                try:
+                    os.close(fd)
+                except OSError:
+                    pass
+                setattr(inst, attr, None)
+        if inst.proto_pipe_base:
+            for suffix in ('.in', '.out'):
+                p = inst.proto_pipe_base + suffix
+                try:
+                    os.unlink(p)
+                except OSError:
+                    pass
+            inst.proto_pipe_base = None
+
         if inst._serial_writer:
             try:
                 inst._serial_writer.close()