feat(pi): pluggable slave handler + canvas wire detection for I2C/SPI/UART

Adds the public extension points the velxio-prod overlay uses to bind
real canvas-side I2C/SPI/UART models (BME280, future MCP23017, etc.)
to a running Pi guest's protocol shims:

- qemu_manager: set_pi_slave_handler(fn) / get_pi_slave_handler() for
  pi_attach_slave + pi_detach_slave WebSocket messages. OSS image
  leaves the hook unset so the messages are silently dropped.
- simulation route: parses the two new WS message types and forwards
  them to the registered handler when present.
- RaspberryPi3Bridge: attachSlave(spec) / detachSlave(spec) frontend
  side of the protocol.
- piSlaveScanner: at simulation start walks components + wires,
  identifies I2C/SPI/UART peers wired to Pi protocol pins (40-pin
  header physical-pin numbering), and emits one attach per
  bus/address pair (deduped across SDA+SCL wires).
- RaspberryPiWorkspace: invokes the scanner once the bridge is open,
  with retries to ride out the WS-still-connecting race.
- integration test: pi3_bme280_attach.py boots the Pi, pre-attaches a
  BME280 via the slave handler, runs a host-side proto loop, runs
  guest python smbus2.read_byte_data(0x76, 0xD0) and asserts the
  console reads back CHIP=0x60.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

backend/app/api/routes/simulation.py

@@ -86,6 +86,18 @@ async def simulation_websocket(websocket: WebSocket, client_id: str):
                 state = msg_data.get('state', 0)
                 qemu_manager.set_pin_state(client_id, pin, state)
 
+            elif msg_type in ('pi_attach_slave', 'pi_detach_slave'):
+                # Pluggable hook — pro overlay registers the actual handler
+                # via qemu_manager.set_pi_slave_handler(). In the OSS image
+                # the hook is unset and the message is silently dropped.
+                handler = qemu_manager.get_pi_slave_handler()
+                if handler is not None:
+                    action = 'attach' if msg_type == 'pi_attach_slave' else 'detach'
+                    try:
+                        await handler(client_id, action, msg_data)
+                    except Exception:
+                        logger.exception('[%s] %s handler crashed', client_id, msg_type)
+
             # ── ESP32 lifecycle ──────────────────────────────────────────
             elif msg_type == 'start_esp32':
                 board        = msg_data.get('board', 'esp32')

backend/app/services/qemu_manager.py

@@ -95,6 +95,55 @@ PI_CONFIGS: dict[str, dict] = {
 DEFAULT_PI_BOARD = 'raspberry-pi-3'
 
 
+# ── Pluggable I2C/SPI/UART dispatcher ────────────────────────────────────
+#
+# When the pro overlay loads, it can call
+# ``set_pi_protocol_dispatcher(fn)`` to register a coroutine that
+# receives the raw protocol tokens for I2C/SPI/UART frames. If the
+# dispatcher returns a non-None string, that string is written back
+# to the guest as a reply line. Returning None falls through to the
+# default no-slave stubs.
+#
+# Signature: async def(client_id: str, tokens: list[str]) -> str | None
+#
+# This keeps the upstream OSS code unaware of the pro slave models
+# (BME280, MCP23017, ...) while letting the overlay attach real
+# behaviour at register_pro() time.
+import typing as _typing
+_ProtocolDispatcher = _typing.Callable[
+    [str, list[str]],
+    _typing.Awaitable[_typing.Optional[str]],
+]
+_PI_PROTOCOL_DISPATCHER: _ProtocolDispatcher | None = None
+
+
+def set_pi_protocol_dispatcher(fn: _ProtocolDispatcher | None) -> None:
+    """Install (or clear) the pro overlay's I2C/SPI/UART dispatcher."""
+    global _PI_PROTOCOL_DISPATCHER
+    _PI_PROTOCOL_DISPATCHER = fn
+
+
+# Pro overlay can also register a handler for attach/detach WebSocket
+# messages from the canvas (e.g. when the user wires a BME280 to the Pi).
+# Signature: async def(client_id: str, action: str, data: dict) -> None
+# where action is 'attach' or 'detach'.
+_SlaveHandler = _typing.Callable[
+    [str, str, dict],
+    _typing.Awaitable[None],
+]
+_PI_SLAVE_HANDLER: _SlaveHandler | None = None
+
+
+def set_pi_slave_handler(fn: _SlaveHandler | None) -> None:
+    """Install (or clear) the pro overlay's slave attach/detach handler."""
+    global _PI_SLAVE_HANDLER
+    _PI_SLAVE_HANDLER = fn
+
+
+def get_pi_slave_handler() -> _SlaveHandler | None:
+    return _PI_SLAVE_HANDLER
+
+
 def _find_free_port() -> int:
     with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
         s.bind(('127.0.0.1', 0))
@@ -550,34 +599,39 @@ class QemuManager:
                 pass
             return
 
-        # I2C / SPI / UART — Phase 2.5 will wire these to real canvas
+        # I2C / SPI / UART — if a pro overlay has registered a slave
-        # bridges. For Phase 2 we send an immediate "no slave" reply
+        # dispatcher, route the frame to it; otherwise reply with
-        # so user code gets an exception path instead of hanging.
+        # stubs (Phase 2 behaviour) so user code gets a deterministic
-        if op == 'I2C' and len(parts) >= 4:
+        # answer instead of hanging.
-            sub = parts[3]
+        if op in ('I2C', 'SPI', 'UART'):
-            if sub in ('R', 'RR'):
+            disp = _PI_PROTOCOL_DISPATCHER
-                bus = parts[1]
+            if disp is not None:
-                addr = parts[2]
+                try:
-                await self._reply_gpio(inst, f'I2C_ERR {bus} {addr} no-slave')
+                    reply = await disp(inst.client_id, parts)
-            return
+                except Exception:
-
+                    logger.exception('pi-protocol dispatcher crashed')
-        if op == 'SPI' and len(parts) >= 4 and parts[3] == 'X':
+                    reply = None
-            bus = parts[1]
+                if reply is not None:
-            cs  = parts[2]
+                    await self._reply_gpio(inst, reply)
-            # Reply with zero bytes of the same length so user code
+                    return
-            # gets a deterministic empty xfer.
+            # Fall through to default stubs
-            try:
+            if op == 'I2C' and len(parts) >= 4:
-                req_hex = parts[4] if len(parts) > 4 else ''
+                sub = parts[3]
-                length = len(bytes.fromhex(req_hex))
+                if sub in ('R', 'RR'):
-            except ValueError:
+                    await self._reply_gpio(
-                length = 0
+                        inst, f'I2C_ERR {parts[1]} {parts[2]} no-slave')
-            await self._reply_gpio(inst,
+                return
-                                   f'SPI_DATA {bus} {cs} {"00" * length}')
+            if op == 'SPI' and len(parts) >= 4 and parts[3] == 'X':
-            return
+                try:
-
+                    req_hex = parts[4] if len(parts) > 4 else ''
-        if op == 'UART' and len(parts) >= 3 and parts[2] == 'RX_REQ':
+                    length = len(bytes.fromhex(req_hex))
-            port = parts[1]
+                except ValueError:
-            await self._reply_gpio(inst, f'UART_RX {port}')
+                    length = 0
+                await self._reply_gpio(
+                    inst, f'SPI_DATA {parts[1]} {parts[2]} {"00" * length}')
+                return
+            if op == 'UART' and len(parts) >= 3 and parts[2] == 'RX_REQ':
+                await self._reply_gpio(inst, f'UART_RX {parts[1]}')
             return
 
         # Unknown — log at debug level (not a hot path)
@@ -688,6 +742,14 @@ class QemuManager:
                 pass
             inst.overlay_path = None
 
+        # Notify pro overlay (if any) so it can drop its slave registry
+        # entry for this client. OSS image has no handler → no-op.
+        if _PI_SLAVE_HANDLER is not None:
+            try:
+                await _PI_SLAVE_HANDLER(inst.client_id, 'shutdown', {})
+            except Exception:
+                logger.exception('pi-slave shutdown hook crashed')
+
         logger.info('PiInstance %s shut down', inst.client_id)
 
     # ── Helpers ───────────────────────────────────────────────────────────────

frontend/src/__tests__/piSlaveScanner.test.ts

@@ -0,0 +1,155 @@
+/**
+ * Tests for piSlaveScanner — verifies that canvas wires from a Pi
+ * board's protocol pins to a known I2C/SPI/UART component result in
+ * the right pi_attach_slave frames.
+ */
+import { describe, it, expect, vi } from 'vitest';
+
+import { attachSlavesFromCanvas } from '../simulation/piSlaveScanner';
+
+function makeBridge() {
+  return { attachSlave: vi.fn() };
+}
+
+const PI_ID = 'raspberry-pi-3';
+
+describe('attachSlavesFromCanvas', () => {
+  it('attaches a BMP280 wired via SDA/SCL on bus 1', () => {
+    const bridge = makeBridge();
+    const components = [
+      { id: 'bmp1', metadataId: 'wokwi-bmp280', properties: {} },
+    ];
+    const wires = [
+      { start: { componentId: PI_ID,  pinName: '3' },
+        end:   { componentId: 'bmp1', pinName: 'SDA' } },
+      { start: { componentId: PI_ID,  pinName: '5' },
+        end:   { componentId: 'bmp1', pinName: 'SCL' } },
+    ];
+
+    const emitted = attachSlavesFromCanvas(PI_ID, bridge, components, wires);
+    expect(emitted).toHaveLength(1);
+    expect(bridge.attachSlave).toHaveBeenCalledTimes(1);
+    expect(emitted[0]).toMatchObject({
+      bus_kind: 'i2c',
+      bus_num: 1,
+      address: 0x76,
+      model_id: 'bme280',
+    });
+  });
+
+  it('uses the address property when set', () => {
+    const bridge = makeBridge();
+    const components = [
+      { id: 'bmp1', metadataId: 'wokwi-bmp280',
+        properties: { address: 0x77 } },
+    ];
+    const wires = [
+      { start: { componentId: PI_ID, pinName: '3' },
+        end:   { componentId: 'bmp1', pinName: 'SDA' } },
+    ];
+    attachSlavesFromCanvas(PI_ID, bridge, components, wires);
+    expect(bridge.attachSlave).toHaveBeenCalledWith(
+      expect.objectContaining({ address: 0x77 }),
+    );
+  });
+
+  it('skips unknown components silently', () => {
+    const bridge = makeBridge();
+    const components = [
+      { id: 'led1', metadataId: 'wokwi-led', properties: {} },
+    ];
+    const wires = [
+      { start: { componentId: PI_ID, pinName: '3' },
+        end:   { componentId: 'led1', pinName: 'A' } },
+    ];
+    attachSlavesFromCanvas(PI_ID, bridge, components, wires);
+    expect(bridge.attachSlave).not.toHaveBeenCalled();
+  });
+
+  it('skips wires that do not touch the Pi', () => {
+    const bridge = makeBridge();
+    const components = [
+      { id: 'bmp1', metadataId: 'wokwi-bmp280', properties: {} },
+      { id: 'arduino', metadataId: 'wokwi-arduino-uno', properties: {} },
+    ];
+    const wires = [
+      { start: { componentId: 'arduino', pinName: 'A4' },
+        end:   { componentId: 'bmp1',    pinName: 'SDA' } },
+    ];
+    attachSlavesFromCanvas(PI_ID, bridge, components, wires);
+    expect(bridge.attachSlave).not.toHaveBeenCalled();
+  });
+
+  it('dedupes when multiple wires hit the same slave', () => {
+    const bridge = makeBridge();
+    const components = [
+      { id: 'bmp1', metadataId: 'wokwi-bmp280', properties: {} },
+    ];
+    // SDA + SCL both wired — should still produce a single attach.
+    const wires = [
+      { start: { componentId: PI_ID, pinName: '3' },
+        end:   { componentId: 'bmp1', pinName: 'SDA' } },
+      { start: { componentId: 'bmp1', pinName: 'SCL' },
+        end:   { componentId: PI_ID,  pinName: '5' } },
+    ];
+    attachSlavesFromCanvas(PI_ID, bridge, components, wires);
+    expect(bridge.attachSlave).toHaveBeenCalledTimes(1);
+  });
+
+  it('forwards temperature/humidity/pressure props as config', () => {
+    const bridge = makeBridge();
+    const components = [
+      { id: 'bmp1', metadataId: 'wokwi-bmp280',
+        properties: {
+          temperature_c: 21.5,
+          humidity_pct:  55,
+          pressure_pa:   99000,
+          color:         'red',   // unrelated key, dropped
+        } },
+    ];
+    const wires = [
+      { start: { componentId: PI_ID, pinName: '3' },
+        end:   { componentId: 'bmp1', pinName: 'SDA' } },
+    ];
+    attachSlavesFromCanvas(PI_ID, bridge, components, wires);
+    expect(bridge.attachSlave).toHaveBeenCalledWith(
+      expect.objectContaining({
+        config: { temperature_c: 21.5, humidity_pct: 55, pressure_pa: 99000 },
+      }),
+    );
+  });
+
+  it('attaches SPI peer on CE0 when present', () => {
+    const bridge = makeBridge();
+    // Only CE0 wire is enough to identify the slave; MOSI/MISO/SCLK
+    // wires are informational. (Re-uses BMP280 mapping since we
+    // don't have a real SPI model in the registry yet — adapt this
+    // test when an actual SPI slave model is added.)
+    const components = [
+      { id: 'flash', metadataId: 'wokwi-bmp280', properties: {} },
+    ];
+    const wires = [
+      { start: { componentId: PI_ID,    pinName: '24' },   // CE0
+        end:   { componentId: 'flash',  pinName: 'CS' } },
+    ];
+    attachSlavesFromCanvas(PI_ID, bridge, components, wires);
+    expect(bridge.attachSlave).toHaveBeenCalledTimes(1);
+    expect(bridge.attachSlave).toHaveBeenCalledWith(
+      expect.objectContaining({ bus_kind: 'spi', bus_num: 0, cs: 0 }),
+    );
+  });
+
+  it('does not attach SPI peer when only data lines (no CE) are wired', () => {
+    const bridge = makeBridge();
+    const components = [
+      { id: 'flash', metadataId: 'wokwi-bmp280', properties: {} },
+    ];
+    const wires = [
+      // MOSI only — no CE — should not attach.
+      { start: { componentId: PI_ID,   pinName: '19' },
+        end:   { componentId: 'flash', pinName: 'MOSI' } },
+    ];
+    attachSlavesFromCanvas(PI_ID, bridge, components, wires);
+    expect(bridge.attachSlave).not.toHaveBeenCalled();
+  });
+});

frontend/src/components/raspberry-pi/RaspberryPiWorkspace.tsx

@@ -12,6 +12,7 @@ import Editor from '@monaco-editor/react';
 import { VirtualFileSystem } from './VirtualFileSystem';
 import { useVfsStore } from '../../store/useVfsStore';
 import { getBoardBridge, useSimulatorStore } from '../../store/useSimulatorStore';
+import { attachSlavesFromCanvas } from '../../simulation/piSlaveScanner';
 
 // Lazy-load PiTerminal so @xterm/xterm is only bundled when needed
 const PiTerminal = lazy(() => import('./PiTerminal').then((m) => ({ default: m.PiTerminal })));
@@ -49,6 +50,25 @@ export const RaspberryPiWorkspace: React.FC<RaspberryPiWorkspaceProps> = ({ boar
         bridge.connect();
       }
       setBridgeConnected(bridge?.connected ?? false);
+
+      // After the WS is open, scan the canvas for I2C/SPI/UART
+      // peripherals wired to this Pi and tell the backend to attach
+      // their slave models. We retry up to ~3s in case attachSlave
+      // calls race the WS open.
+      const attachOnce = (): boolean => {
+        const b = getBoardBridge(boardId);
+        if (!b?.connected) return false;
+        const { components, wires } = useSimulatorStore.getState();
+        attachSlavesFromCanvas(boardId, b, components, wires);
+        return true;
+      };
+      if (!attachOnce()) {
+        let attempts = 0;
+        const interval = setInterval(() => {
+          attempts++;
+          if (attachOnce() || attempts >= 6) clearInterval(interval);
+        }, 500);
+      }
     }, 800);
     return () => clearTimeout(timer);
   }, [board?.running, boardId]);

frontend/src/simulation/RaspberryPi3Bridge.ts

@@ -10,6 +10,15 @@
  *     { type: 'stop_pi' }
  *     { type: 'serial_input', data: { bytes: number[] } }
  *     { type: 'gpio_in', data: { pin: number, state: 0 | 1 } }
+ *     { type: 'pi_attach_slave', data: {
+ *         bus_kind: 'i2c'|'spi'|'uart',
+ *         bus_num:  number,
+ *         address?: number,   // i2c
+ *         cs?:      number,   // spi
+ *         model_id: string,   // e.g. 'bme280'
+ *         config?:  Record<string, unknown>,
+ *     }}
+ *     { type: 'pi_detach_slave', data: { bus_kind, bus_num, address?|cs? } }
  *
  *   Backend → Frontend
  *     { type: 'serial_output', data: { data: string } }
@@ -134,6 +143,32 @@ export class RaspberryPi3Bridge {
     this._send({ type: 'gpio_in', data: { pin: gpioPin, state: state ? 1 : 0 } });
   }
 
+  /**
+   * Attach an I2C/SPI/UART slave model to the running Pi. The backend
+   * pro overlay turns this into a PiSlaveRegistry entry that the
+   * protocol dispatcher consults on each guest read. OSS images
+   * silently drop the message.
+   */
+  attachSlave(spec: {
+    bus_kind: 'i2c' | 'spi' | 'uart';
+    bus_num: number;
+    address?: number;
+    cs?: number;
+    model_id: string;
+    config?: Record<string, unknown>;
+  }): void {
+    this._send({ type: 'pi_attach_slave', data: spec });
+  }
+
+  detachSlave(spec: {
+    bus_kind: 'i2c' | 'spi' | 'uart';
+    bus_num: number;
+    address?: number;
+    cs?: number;
+  }): void {
+    this._send({ type: 'pi_detach_slave', data: spec });
+  }
+
   private _send(payload: unknown): void {
     if (this.socket && this.socket.readyState === WebSocket.OPEN) {
       this.socket.send(JSON.stringify(payload));

frontend/src/simulation/piSlaveScanner.ts

@@ -0,0 +1,177 @@
+/**
+ * piSlaveScanner — turn canvas wires into pi_attach_slave commands.
+ *
+ * When the user wires a virtual I2C/SPI/UART component (e.g. BMP280)
+ * to a Raspberry Pi's protocol pins on the canvas, this helper detects
+ * the connection and emits the backend WebSocket frame that tells the
+ * pro overlay to instantiate the corresponding slave model in the
+ * PiSlaveRegistry.
+ *
+ * The mapping from a wokwi component metadata ID to a backend
+ * model_id lives in COMPONENT_TO_MODEL. New devices are added there
+ * plus a new file under pro/backend/app/pro/services/pi_slaves/.
+ */
+import type { RaspberryPi3Bridge } from './RaspberryPi3Bridge';
+
+type CanvasWire = {
+  start: { componentId: string; pinName: string };
+  end:   { componentId: string; pinName: string };
+};
+
+/** Shape we depend on from useSimulatorStore.Component. */
+type CanvasComponent = {
+  id: string;
+  metadataId: string;
+  properties?: Record<string, unknown>;
+};
+
+// Raspberry Pi 40-pin header → bus assignment. Keys are physical pin
+// numbers as strings (matching the wire endpoint pinName format).
+//   - SDA1/SCL1 → I2C bus 1
+//   - MOSI/MISO/SCLK → SPI bus 0 (CE0/CE1 distinguish slaves)
+//   - TXD/RXD → primary UART (port 0)
+const I2C_PINS = new Set(['3', '5']);
+const SPI_DATA_PINS = new Set(['19', '21', '23']);
+const SPI_CE0_PIN = '24';
+const SPI_CE1_PIN = '26';
+const UART_PINS = new Set(['8', '10']);
+
+// Map wokwi component metadata IDs / element types → backend model_id.
+// Lowercase. Components not in this table get skipped silently.
+const COMPONENT_TO_MODEL: Record<string, string> = {
+  'wokwi-bmp280':  'bme280',   // same chip family, treat as superset
+  'velxio-bmp280': 'bme280',
+  'wokwi-bme280':  'bme280',
+};
+
+const DEFAULT_I2C_ADDRESS: Record<string, number> = {
+  bme280: 0x76,
+};
+
+function modelIdFor(component: CanvasComponent): string | null {
+  const t = component.metadataId?.toLowerCase();
+  return t ? COMPONENT_TO_MODEL[t] ?? null : null;
+}
+
+function i2cAddressFor(component: CanvasComponent, modelId: string): number {
+  const propAddr = component.properties?.address;
+  if (typeof propAddr === 'number') return propAddr;
+  if (typeof propAddr === 'string' && propAddr.length > 0) {
+    const parsed = Number(propAddr);
+    if (!Number.isNaN(parsed)) return parsed;
+  }
+  return DEFAULT_I2C_ADDRESS[modelId] ?? 0x76;
+}
+
+function configFor(component: CanvasComponent): Record<string, unknown> {
+  // Initial pass: only forward known keys so backend ctor surface is
+  // narrow. Each model declares its own kwargs.
+  const out: Record<string, unknown> = {};
+  const p = component.properties ?? {};
+  for (const key of ['temperature_c', 'humidity_pct', 'pressure_pa']) {
+    if (typeof p[key] === 'number') out[key] = p[key];
+  }
+  return out;
+}
+
+/**
+ * Returns the (bus_kind, identifying_pin) describing how `pinName`
+ * on the Pi is being used, or null if it isn't a protocol pin.
+ */
+function classifyPiPin(pinName: string): {
+  bus_kind: 'i2c' | 'spi' | 'uart';
+  bus_num: number;
+} | null {
+  if (I2C_PINS.has(pinName)) return { bus_kind: 'i2c', bus_num: 1 };
+  if (SPI_DATA_PINS.has(pinName) || pinName === SPI_CE0_PIN || pinName === SPI_CE1_PIN)
+    return { bus_kind: 'spi', bus_num: 0 };
+  if (UART_PINS.has(pinName)) return { bus_kind: 'uart', bus_num: 0 };
+  return null;
+}
+
+/**
+ * Iterate every wire; if one side is a Pi protocol pin and the other
+ * side is a known I2C/SPI/UART component, emit pi_attach_slave.
+ *
+ * Returns the list of attach specs emitted (for tests).
+ */
+export function attachSlavesFromCanvas(
+  piBoardId: string,
+  bridge: Pick<RaspberryPi3Bridge, 'attachSlave'>,
+  components: CanvasComponent[],
+  wires: CanvasWire[],
+): Array<Parameters<RaspberryPi3Bridge['attachSlave']>[0]> {
+  const componentsById = new Map(components.map((c) => [c.id, c]));
+  // dedupe attaches by (bus_kind, bus_num, address_or_cs)
+  const seen = new Set<string>();
+  const emitted: Array<Parameters<RaspberryPi3Bridge['attachSlave']>[0]> = [];
+
+  for (const wire of wires) {
+    let piEndpoint: { pinName: string } | null = null;
+    let otherEndpoint: { componentId: string } | null = null;
+    if (wire.start.componentId === piBoardId) {
+      piEndpoint = wire.start;
+      otherEndpoint = wire.end;
+    } else if (wire.end.componentId === piBoardId) {
+      piEndpoint = wire.end;
+      otherEndpoint = wire.start;
+    } else {
+      continue;
+    }
+
+    const bus = classifyPiPin(piEndpoint.pinName);
+    if (!bus) continue;
+
+    const peer = componentsById.get(otherEndpoint.componentId);
+    if (!peer) continue;
+    const modelId = modelIdFor(peer);
+    if (!modelId) continue;
+
+    let spec: Parameters<RaspberryPi3Bridge['attachSlave']>[0];
+    if (bus.bus_kind === 'i2c') {
+      const address = i2cAddressFor(peer, modelId);
+      spec = {
+        bus_kind: 'i2c',
+        bus_num:  bus.bus_num,
+        address,
+        model_id: modelId,
+        config:   configFor(peer),
+      };
+    } else if (bus.bus_kind === 'spi') {
+      // CS line determines logical slave index. Treat MOSI/MISO/SCLK
+      // wires as informational only — the CE wire is the one that
+      // pins down which slave gets attached.
+      let cs: number;
+      if (piEndpoint.pinName === SPI_CE0_PIN) cs = 0;
+      else if (piEndpoint.pinName === SPI_CE1_PIN) cs = 1;
+      else continue;
+      spec = {
+        bus_kind: 'spi',
+        bus_num:  bus.bus_num,
+        cs,
+        model_id: modelId,
+        config:   configFor(peer),
+      };
+    } else {
+      spec = {
+        bus_kind: 'uart',
+        bus_num:  bus.bus_num,
+        model_id: modelId,
+        config:   configFor(peer),
+      };
+    }
+
+    const key =
+      spec.bus_kind === 'i2c'
+        ? `i2c:${spec.bus_num}:${spec.address}`
+        : spec.bus_kind === 'spi'
+          ? `spi:${spec.bus_num}:${spec.cs}`
+          : `uart:${spec.bus_num}`;
+    if (seen.has(key)) continue;
+    seen.add(key);
+
+    bridge.attachSlave(spec);
+    emitted.push(spec);
+  }
+  return emitted;
+}

test/pi3_protocols/test_pi3_bme280_attach.py

@@ -0,0 +1,279 @@
+#!/usr/bin/env python3
+"""
+Phase 2.5 end-to-end: BME280 attach over I2C
+============================================
+
+Boots the Pi 3 (-M virt) the exact same way ``qemu_manager.py`` does
+(pipe chardev for the protocol channel) and then runs a small
+host-side loop that mirrors what ``QemuManager._handle_gpio_line``
+does for I2C frames: it forwards the frame to the
+``pi_protocol_dispatcher`` in the pro overlay, then writes the reply
+back over the proto pipe.
+
+We pre-attach a BME280 to the per-client ``PiSlaveRegistry`` via
+``pi_slave_handler.handle('attach', ...)`` before booting, then run
+guest Python:
+
+    import smbus2
+    bus = smbus2.SMBus(1)
+    chip = bus.read_byte_data(0x76, 0xD0)
+    print(f'CHIP=0x{chip:02x}')
+
+Assertion: console reads back ``CHIP=0x60`` (the real BME280's chip ID).
+
+What this catches (above and beyond test_pi3_protocols.py):
+  - dispatcher → registry → model lookup chain
+  - shim I2C wire format (RR with hex register / length tokens)
+  - reply line emitted as ``I2C_DATA <bus> <addr> <hex>``
+  - shim correctly parsing the reply hex back into an int
+
+Run inside the velxio-app container:
+
+    docker exec velxio-app python3 /tmp/test_pi3_bme280_attach.py
+"""
+from __future__ import annotations
+
+import asyncio
+import base64
+import os
+import socket
+import subprocess
+import sys
+import tempfile
+import threading
+import time
+from pathlib import Path
+
+# Make `app.*` resolvable (mirrors conftest in pro/backend/tests).
+sys.path.insert(0, '/app/backend')
+
+BOOT_IMAGES = Path('/var/cache/velxio/boot-images/raspberry-pi-3-virt')
+KERNEL    = BOOT_IMAGES / 'velxio-kernel-arm64'
+INITRAMFS = BOOT_IMAGES / 'velxio-initramfs-arm64.cpio.gz'
+ROOTFS    = BOOT_IMAGES / 'velxio-pi-rootfs-arm64.ext4'
+
+CLIENT_ID = 'phase2.5-bme280-test'
+BOOT_TIMEOUT_S = 90
+
+
+def _find_free_port() -> int:
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
+        s.bind(('127.0.0.1', 0))
+        return s.getsockname()[1]
+
+
+def _make_overlay() -> str:
+    overlay = tempfile.NamedTemporaryFile(suffix='.qcow2', delete=False)
+    overlay.close()
+    subprocess.run(
+        ['qemu-img', 'create', '-f', 'qcow2',
+         '-b', str(ROOTFS), '-F', 'raw', overlay.name],
+        check=True, capture_output=True,
+    )
+    return overlay.name
+
+
+def _mk_proto_pipe() -> str:
+    base = tempfile.mktemp(prefix='velxio-pi-bme280-test-')
+    for suffix in ('.in', '.out'):
+        os.mkfifo(base + suffix, 0o600)
+    return base
+
+
+def _qemu_argv(overlay: str, cons_port: int, proto_base: str) -> list[str]:
+    return [
+        'qemu-system-aarch64',
+        '-M', 'virt', '-cpu', 'cortex-a53', '-smp', '4', '-m', '1G',
+        '-kernel', str(KERNEL), '-initrd', str(INITRAMFS),
+        '-drive', f'if=none,file={overlay},format=qcow2,id=rootfs',
+        '-device', 'virtio-blk-pci,drive=rootfs',
+        '-nic', 'none', '-display', 'none', '-monitor', 'none', '-serial', 'none',
+        '-chardev',
+        f'socket,id=cons,host=127.0.0.1,port={cons_port},server=on,wait=off',
+        '-device', 'virtio-serial-pci,id=virtio-serial0',
+        '-device', 'virtconsole,chardev=cons',
+        '-chardev', f'pipe,id=proto,path={proto_base}',
+        '-device', 'virtserialport,chardev=proto,name=velxio-protocol',
+        '-append', 'console=hvc0 root=/dev/vda rw panic=10',
+    ]
+
+
+def _proto_router_thread(proto_in: int, proto_out: int, stop: threading.Event) -> None:
+    """Mirror QemuManager's _handle_gpio_line loop on the host side.
+
+    We poll ``proto_out`` (guest → host) for newline-terminated frames,
+    feed each one to the pro overlay's protocol dispatcher, and write
+    the reply (if any) back to ``proto_in`` (host → guest).
+    """
+    from app.pro.services.pi_protocol_dispatcher import dispatch
+
+    buf = bytearray()
+    loop = asyncio.new_event_loop()
+    asyncio.set_event_loop(loop)
+    while not stop.is_set():
+        try:
+            data = os.read(proto_out, 4096)
+        except BlockingIOError:
+            time.sleep(0.02)
+            continue
+        except OSError:
+            return
+        if not data:
+            time.sleep(0.02)
+            continue
+        buf.extend(data)
+        while b'\n' in buf:
+            line, _, rest = buf.partition(b'\n')
+            buf = bytearray(rest)
+            tokens = line.decode('ascii', 'replace').strip().split()
+            if not tokens or tokens[0] not in ('I2C', 'SPI', 'UART'):
+                continue
+            print(f'[proto] >>> {tokens}')
+            reply = loop.run_until_complete(dispatch(CLIENT_ID, tokens))
+            if reply:
+                print(f'[proto] <<< {reply}')
+                os.write(proto_in, (reply + '\n').encode('ascii'))
+
+
+def run() -> int:
+    for p in (KERNEL, INITRAMFS, ROOTFS):
+        if not p.exists():
+            print(f'FAIL: missing boot image: {p}', file=sys.stderr)
+            return 2
+
+    # Pre-register the dispatcher + slave handler the same way
+    # register_pro() does, then attach a BME280 to the registry.
+    from app.pro.services import pi_slave_handler
+
+    asyncio.run(pi_slave_handler.handle(CLIENT_ID, 'attach', {
+        'bus_kind': 'i2c',
+        'bus_num': 1,
+        'address': 0x76,
+        'model_id': 'bme280',
+        'config': {'temperature_c': 22.0, 'humidity_pct': 47.0,
+                   'pressure_pa': 101000.0},
+    }))
+
+    overlay = _make_overlay()
+    proto_base = _mk_proto_pipe()
+    cons_port = _find_free_port()
+    argv = _qemu_argv(overlay, cons_port, proto_base)
+    print('[test] launching:', ' '.join(argv))
+
+    proto_in  = os.open(proto_base + '.in',  os.O_RDWR | os.O_NONBLOCK)
+    proto_out = os.open(proto_base + '.out', os.O_RDWR | os.O_NONBLOCK)
+
+    stop = threading.Event()
+    router = threading.Thread(
+        target=_proto_router_thread,
+        args=(proto_in, proto_out, stop),
+        daemon=True,
+    )
+    router.start()
+
+    qemu = subprocess.Popen(
+        argv, stdout=subprocess.DEVNULL,
+        stderr=subprocess.PIPE, stdin=subprocess.DEVNULL,
+    )
+    try:
+        sock: socket.socket | None = None
+        deadline = time.monotonic() + BOOT_TIMEOUT_S
+        while time.monotonic() < deadline:
+            try:
+                sock = socket.create_connection(('127.0.0.1', cons_port),
+                                                timeout=5)
+                break
+            except (ConnectionRefusedError, OSError):
+                time.sleep(0.3)
+        if not sock:
+            print('FAIL: console TCP connection refused', file=sys.stderr)
+            return 1
+        sock.settimeout(2)
+
+        buf = bytearray()
+        saw_prompt = False
+        sent_test = False
+
+        while time.monotonic() < deadline:
+            try:
+                chunk = sock.recv(4096)
+            except socket.timeout:
+                continue
+            if not chunk:
+                break
+            buf.extend(chunk)
+
+            if not saw_prompt and b':~#' in buf:
+                saw_prompt = True
+                print('[test] bash prompt reached, sending Python BME280 read')
+                time.sleep(3)
+                try:
+                    while True:
+                        more = sock.recv(4096)
+                        if not more:
+                            break
+                        buf.extend(more)
+                except socket.timeout:
+                    pass
+                py = (
+                    'import smbus2\n'
+                    'bus = smbus2.SMBus(1)\n'
+                    'chip = bus.read_byte_data(0x76, 0xD0)\n'
+                    "print(f'CHIP=0x{chip:02x}')\n"
+                    'data = bus.read_i2c_block_data(0x76, 0xF7, 8)\n'
+                    "print('BLOCK=' + ''.join(f'{b:02x}' for b in data))\n"
+                )
+                b64 = base64.b64encode(py.encode()).decode()
+                cmd = (
+                    f'echo {b64} | base64 -d > /tmp/bme280_test.py && '
+                    f'python3 /tmp/bme280_test.py\n'
+                ).encode()
+                sock.sendall(cmd)
+                sent_test = True
+
+            if sent_test and b'CHIP=0x60' in buf:
+                print('[test] OK — guest read chip ID = 0x60')
+                # Also check that BLOCK= came back as 16 hex chars (8 bytes)
+                txt = buf.decode('utf-8', 'replace')
+                for ln in txt.splitlines():
+                    if ln.startswith('BLOCK='):
+                        if len(ln) - len('BLOCK=') == 16:
+                            print(f'[test] OK — block read {ln}')
+                            return 0
+                        print(f'FAIL: block length wrong: {ln}',
+                              file=sys.stderr)
+                        return 1
+                print('FAIL: chip id OK but no block readback observed',
+                      file=sys.stderr)
+                return 1
+
+            if sent_test and (b'Traceback' in buf or b'ModuleNotFoundError' in buf):
+                print('FAIL: guest python raised:', file=sys.stderr)
+                print(buf[-1500:].decode('utf-8', 'replace'))
+                return 1
+
+        print(f'FAIL: timeout. saw_prompt={saw_prompt} '
+              f'sent_test={sent_test} buf_len={len(buf)}',
+              file=sys.stderr)
+        print(buf[-1500:].decode('utf-8', 'replace'), file=sys.stderr)
+        return 1
+    finally:
+        stop.set()
+        try:
+            qemu.terminate()
+            qemu.wait(timeout=5)
+        except subprocess.TimeoutExpired:
+            qemu.kill()
+        for fd in (proto_in, proto_out):
+            try: os.close(fd)
+            except OSError: pass
+        for suffix in ('.in', '.out'):
+            try: os.unlink(proto_base + suffix)
+            except OSError: pass
+        try: os.unlink(overlay)
+        except OSError: pass
+        asyncio.run(pi_slave_handler.handle(CLIENT_ID, 'shutdown', {}))
+
+
+if __name__ == '__main__':
+    sys.exit(run())