velxio/backend/app/services/esp32_worker.py

#!/usr/bin/env python3
"""
esp32_worker.py — Standalone ESP32 QEMU subprocess worker.

Runs as a child process of esp32_lib_manager.  Loads libqemu-xtensa in its
own process address space so multiple instances can coexist without DLL state
conflicts.

stdin  line 1 : JSON config
               {"lib_path": "...", "firmware_b64": "...", "machine": "..."}
stdin  line 2+: JSON commands
               {"cmd": "set_pin",          "pin": N,       "value": V}
               {"cmd": "set_adc",          "channel": N,   "millivolts": V}
               {"cmd": "set_adc_raw",      "channel": N,   "raw": V}
               {"cmd": "set_adc_waveform", "channel": N,   "samples_u12_b64": "<base64-LE-uint16>", "period_ns": P}
               {"cmd": "uart_send",        "uart": N,      "data": "<base64>"}
               {"cmd": "set_i2c_response", "addr": N,      "response": V}
               {"cmd": "set_spi_response", "response": V}
               {"cmd": "stop"}

stdout        : JSON event lines (one per line, flushed immediately)
               {"type": "system",       "event": "booted"}
               {"type": "system",       "event": "crash",  "reason": "...", ...}
               {"type": "system",       "event": "reboot", "count": N}
               {"type": "gpio_change",  "pin": N,  "state": V}
               {"type": "gpio_dir",     "pin": N,  "dir": V}
               {"type": "uart_tx",      "uart": N, "byte": V}
               {"type": "ledc_duty",    "channel": N, "duty_pct": F}
               {"type": "rmt_event",    "channel": N, ...}
               {"type": "ws2812_update","channel": N, "pixels": [...]}
               {"type": "i2c_event",    "bus": N, "addr": N, "event": N, "response": N}
               {"type": "spi_event",    "bus": N, "event": N, "response": N}
               {"type": "error",        "message": "..."}

stderr        : debug logs (never part of the JSON protocol)
"""
import base64
import ctypes
import json
import os
import sys
import tempfile
import threading
import time

# I2C slave state machines — extracted to a standalone module for testability
try:
    from app.services.esp32_i2c_slaves import (
        MPU6050Slave as _MPU6050Slave,
        BMP280Slave  as _BMP280Slave,
        DS1307Slave  as _DS1307Slave,
        DS3231Slave  as _DS3231Slave,
        I2CWriteSink as _I2CWriteSink,
        ProxySlave   as _ProxySlave,
    )
except ImportError:
    # Fallback: direct import when running from backend/ directory as subprocess
    import importlib.util, pathlib, sys as _sys
    _here = pathlib.Path(__file__).parent
    _spec = importlib.util.spec_from_file_location('esp32_i2c_slaves', _here / 'esp32_i2c_slaves.py')
    _mod  = importlib.util.module_from_spec(_spec)  # type: ignore[arg-type]
    # Register in sys.modules BEFORE exec — @dataclass looks up cls.__module__
    # there, and crashes with AttributeError on None when missing.
    _sys.modules['esp32_i2c_slaves'] = _mod
    _spec.loader.exec_module(_mod)  # type: ignore[union-attr]
    _MPU6050Slave = _mod.MPU6050Slave  # type: ignore[assignment]
    _BMP280Slave  = _mod.BMP280Slave   # type: ignore[assignment]
    _DS1307Slave  = _mod.DS1307Slave   # type: ignore[assignment]
    _DS3231Slave  = _mod.DS3231Slave   # type: ignore[assignment]
    _I2CWriteSink = _mod.I2CWriteSink  # type: ignore[assignment]
    _ProxySlave   = _mod.ProxySlave    # type: ignore[assignment]

# SPI slaves (Phase 1: SSD168x ePaper). Same fallback dance — when the worker
# runs as a subprocess from backend/ the package import won't resolve.
try:
    from app.services.esp32_spi_slaves import (
        Ssd168xEpaperSlave as _Ssd168xEpaperSlave,
        Uc8159cEpaperSlave as _Uc8159cEpaperSlave,
        Uc8179EpaperSlave as _Uc8179EpaperSlave,
    )
except ImportError:
    import importlib.util, pathlib, sys as _sys
    _here = pathlib.Path(__file__).parent
    _spec = importlib.util.spec_from_file_location('esp32_spi_slaves', _here / 'esp32_spi_slaves.py')
    _mod = importlib.util.module_from_spec(_spec)  # type: ignore[arg-type]
    # Same dataclass-needs-sys.modules fix as the i2c fallback above.
    _sys.modules['esp32_spi_slaves'] = _mod
    _spec.loader.exec_module(_mod)  # type: ignore[union-attr]
    _Ssd168xEpaperSlave = _mod.Ssd168xEpaperSlave  # type: ignore[assignment]
    _Uc8159cEpaperSlave = _mod.Uc8159cEpaperSlave  # type: ignore[assignment]
    _Uc8179EpaperSlave = _mod.Uc8179EpaperSlave  # type: ignore[assignment]

# ─── stdout helpers ──────────────────────────────────────────────────────────

_stdout_lock = threading.Lock()


def _emit(obj: dict) -> None:
    """Write one JSON event line to stdout (thread-safe, always flushed)."""
    with _stdout_lock:
        sys.stdout.write(json.dumps(obj) + '\n')
        sys.stdout.flush()


def _log(msg: str) -> None:
    """Write a debug message to stderr (invisible to parent's stdout reader)."""
    sys.stderr.write(f'[esp32_worker] {msg}\n')
    sys.stderr.flush()


# ─── GPIO pinmap (identity: slot i → GPIO i-1) ──────────────────────────────
# ESP32 has 40 GPIOs (0-39), ESP32-C3 only has 22 (0-21).
# The pinmap is rebuilt after reading config (see main()), defaulting to ESP32.

_GPIO_COUNT = 40
_PINMAP = (ctypes.c_int16 * (_GPIO_COUNT + 1))(
    _GPIO_COUNT,
    *range(_GPIO_COUNT),
)


def _build_pinmap(gpio_count: int):
    """Build a pinmap array for the given GPIO count."""
    global _GPIO_COUNT, _PINMAP
    _GPIO_COUNT = gpio_count
    _PINMAP = (ctypes.c_int16 * (gpio_count + 1))(
        gpio_count,
        *range(gpio_count),
    )

# ─── ctypes callback types ───────────────────────────────────────────────────

_WRITE_PIN = ctypes.CFUNCTYPE(None,            ctypes.c_int,   ctypes.c_int)
_DIR_PIN   = ctypes.CFUNCTYPE(None,            ctypes.c_int,   ctypes.c_int)
_I2C_EVENT = ctypes.CFUNCTYPE(ctypes.c_int,    ctypes.c_uint8, ctypes.c_uint8, ctypes.c_uint16)
_SPI_EVENT = ctypes.CFUNCTYPE(ctypes.c_uint8,  ctypes.c_uint8, ctypes.c_uint16)
_UART_TX   = ctypes.CFUNCTYPE(None,            ctypes.c_uint8, ctypes.c_uint8)
_RMT_EVENT = ctypes.CFUNCTYPE(None,            ctypes.c_uint8, ctypes.c_uint32, ctypes.c_uint32)
# Synchronous GPIO Matrix routing callback. Fires on every guest write
# to GPIO_FUNCx_OUT_SEL_CFG_REG with the full 9-bit signal_id; replaces
# the 100 ms poll path in _refresh_signal_routing() once the prod
# burn-in window confirms parity. Requires libqemu-{xtensa,riscv32}
# 1.1.0+; older binaries omit the field and the placeholder runs.
_GPIO_MATRIX_CB = ctypes.CFUNCTYPE(None,        ctypes.c_int, ctypes.c_int)
# Batched write-only SPI: (id, const uint8_t *mosi, int len). Collapses the
# per-byte picsimlab_spi_event ctypes crossing for TFT-style bulk writes.
# Trailing field — older libqemu builds (without the C-side batch path) simply
# never call it and fall back to per-byte picsimlab_spi_event.
_SPI_BATCH = ctypes.CFUNCTYPE(None, ctypes.c_uint8, ctypes.POINTER(ctypes.c_uint8), ctypes.c_int)


class _CallbacksT(ctypes.Structure):
    _fields_ = [
        ('picsimlab_write_pin',         _WRITE_PIN),
        ('picsimlab_dir_pin',           _DIR_PIN),
        ('picsimlab_i2c_event',         _I2C_EVENT),
        ('picsimlab_spi_event',         _SPI_EVENT),
        ('picsimlab_uart_tx_event',     _UART_TX),
        ('pinmap',                      ctypes.c_void_p),
        ('picsimlab_rmt_event',         _RMT_EVENT),
        ('picsimlab_gpio_matrix_cb',    _GPIO_MATRIX_CB),
        ('picsimlab_spi_event_batch',   _SPI_BATCH),
    ]


# ─── RMT / WS2812 NeoPixel decoder ───────────────────────────────────────────

_WS2812_HIGH_THRESHOLD = 48  # RMT ticks; high pulse > threshold → bit 1


def _decode_rmt_item(value: int) -> tuple[int, int, int, int]:
    """Unpack a 32-bit RMT item → (level0, duration0, level1, duration1)."""
    level0    = (value >> 31) & 1
    duration0 = (value >> 16) & 0x7FFF
    level1    = (value >> 15) & 1
    duration1 =  value        & 0x7FFF
    return level0, duration0, level1, duration1


class _RmtDecoder:
    """Accumulate RMT items for one channel; flush complete WS2812 frames."""

    def __init__(self, channel: int):
        self.channel  = channel
        self._bits:   list[int] = []
        self._pixels: list[dict] = []

    @staticmethod
    def _bits_to_byte(bits: list[int], offset: int) -> int:
        val = 0
        for i in range(8):
            val = (val << 1) | bits[offset + i]
        return val

    def feed(self, value: int) -> list[dict] | None:
        """
        Process one RMT item.
        Returns a list of {r, g, b} pixel dicts on end-of-frame, else None.
        """
        level0, dur0, _, dur1 = _decode_rmt_item(value)

        # Reset pulse (both durations zero) signals end of frame
        if dur0 == 0 and dur1 == 0:
            pix = list(self._pixels)
            self._pixels.clear()
            self._bits.clear()
            return pix or None

        # Classify the high pulse → bit 1 or bit 0
        if level0 == 1 and dur0 > 0:
            self._bits.append(1 if dur0 > _WS2812_HIGH_THRESHOLD else 0)

        # Every 24 bits → one GRB pixel → convert to RGB
        while len(self._bits) >= 24:
            g = self._bits_to_byte(self._bits, 0)
            r = self._bits_to_byte(self._bits, 8)
            b = self._bits_to_byte(self._bits, 16)
            self._pixels.append({'r': r, 'g': g, 'b': b})
            self._bits = self._bits[24:]

        return None


# ─── Main ─────────────────────────────────────────────────────────────────────

def main() -> None:  # noqa: C901  (complexity OK for inline worker)
    # ── 1. Read config from stdin ─────────────────────────────────────────────
    raw_cfg = sys.stdin.readline()
    if not raw_cfg.strip():
        _log('No config received on stdin — exiting')
        os._exit(1)
    try:
        cfg = json.loads(raw_cfg)
    except Exception as exc:
        _log(f'Bad config JSON: {exc}')
        os._exit(1)

    lib_path          = cfg['lib_path']
    firmware_b64      = cfg['firmware_b64']
    machine           = cfg.get('machine', 'esp32-picsimlab')
    initial_sensors   = cfg.get('sensors', [])
    wifi_enabled      = cfg.get('wifi_enabled', False)
    wifi_hostfwd_port = cfg.get('wifi_hostfwd_port', 0)

    # Adjust GPIO pinmap based on chip: ESP32-C3 has only 22 GPIOs
    if 'c3' in machine:
        _build_pinmap(22)

    # ── 2. Load DLL ───────────────────────────────────────────────────────────
    _MINGW64_BIN = r'C:\msys64\mingw64\bin'
    if os.name == 'nt' and os.path.isdir(_MINGW64_BIN):
        os.add_dll_directory(_MINGW64_BIN)
    try:
        lib_size = os.path.getsize(lib_path) if os.path.isfile(lib_path) else 0
        _log(f'Loading library: {lib_path} ({lib_size} bytes)')
        lib = ctypes.CDLL(lib_path)
    except Exception as exc:
        _emit({'type': 'error', 'message': f'Cannot load DLL: {exc}'})
        os._exit(1)
    lib.qemu_picsimlab_get_internals.restype = ctypes.c_void_p

    # qemu_picsimlab_uart_receive() injects a UART-RX interrupt into the guest CPU.
    # QEMU asserts qemu_mutex_iothread_locked() at that point, so the caller MUST
    # hold the IO-thread lock.  Acquire it before every uart_receive call and
    # release immediately after.  The functions are exported as:
    #   qemu_mutex_lock_iothread_impl(const char *file, int line)
    #   qemu_mutex_unlock_iothread()
    try:
        _lock_iothread   = lib.qemu_mutex_lock_iothread_impl
        _lock_iothread.restype  = None
        _lock_iothread.argtypes = [ctypes.c_char_p, ctypes.c_int]
        _unlock_iothread = lib.qemu_mutex_unlock_iothread
        _unlock_iothread.restype  = None
        _unlock_iothread.argtypes = []
    except AttributeError:
        _lock_iothread   = None
        _unlock_iothread = None

    # Predicate: is the iothread lock currently held by this thread?
    # Used to avoid re-acquiring when we're already inside a QEMU callback
    # (e.g., chip's vx_uart_write fired from inside _on_uart_tx).
    try:
        _iothread_locked = lib.qemu_mutex_iothread_locked
        _iothread_locked.restype  = ctypes.c_bool
        _iothread_locked.argtypes = []
    except AttributeError:
        _iothread_locked = None

    # qemu_system_shutdown_request() schedules a clean shutdown from inside
    # the QEMU main-loop thread (which owns the AIO context).  Calling
    # qemu_cleanup() directly from a Python thread (the command loop) triggers
    # the "blk_exp_close_all_type: in_aio_context_home_thread" assertion
    # because the block device teardown happens on the wrong thread.
    # SHUTDOWN_CAUSE_HOST_SIGNAL = 3 (matches the constant in qapi/run-state.json)
    try:
        _shutdown_request = lib.qemu_system_shutdown_request
        _shutdown_request.restype  = None
        _shutdown_request.argtypes = [ctypes.c_int]
    except AttributeError:
        _shutdown_request = None

    # ── ESP32-CAM frame injection ─────────────────────────────────────────
    # Exported by hw/misc/esp32_i2s_cam.c (the OV2640+I²S patch). When the
    # symbol is absent (= stock library, no camera patch yet), we keep a
    # no-op so the worker stays compatible with un-patched libraries.
    try:
        _push_camera_frame_c = lib.velxio_push_camera_frame
        _push_camera_frame_c.restype  = None
        _push_camera_frame_c.argtypes = [ctypes.c_char_p, ctypes.c_size_t]
        def _push_camera_frame(payload: bytes) -> None:
            buf = ctypes.c_char_p(payload) if payload else None
            n = len(payload) if payload else 0
            if _lock_iothread:
                _lock_iothread(b'esp32_worker.py:camera', 0)
            try:
                _push_camera_frame_c(buf, n)
            finally:
                if _unlock_iothread:
                    _unlock_iothread()
    except AttributeError:
        def _push_camera_frame(payload: bytes) -> None:
            # Stock library — no camera support compiled in. The first
            # time we hit this path we emit a warning so the user
            # understands why fb_get returns nothing; subsequent calls
            # are silent.
            if not getattr(_push_camera_frame, '_warned', False):
                _log('camera_frame: velxio_push_camera_frame symbol '
                     'missing — rebuild libqemu-xtensa with the '
                     'OV2640+I²S patch (test/test-esp32-cam/autosearch).')
                _push_camera_frame._warned = True  # type: ignore[attr-defined]

    # ── 3. Write firmware to a temp file ──────────────────────────────────────
    try:
        # The compiler trims trailing 0xFF padding before serializing (issue
        # #101 — full 4 MB images blew nginx buffers). Re-pad here so QEMU's
        # MTD layer sees a valid power-of-2 flash size.
        # Imported via fallback because this file runs as a subprocess and
        # `app.*` is not on sys.path; mirrors the esp32_i2c_slaves pattern
        # at the top of the file.
        try:
            from app.services.esp32_flash_image import pad_to_flash_size  # type: ignore[import-not-found]
        except ImportError:
            import importlib.util as _ilu, pathlib as _pl
            _spec = _ilu.spec_from_file_location(
                'esp32_flash_image',
                _pl.Path(__file__).parent / 'esp32_flash_image.py',
            )
            _mod = _ilu.module_from_spec(_spec)  # type: ignore[arg-type]
            _spec.loader.exec_module(_mod)  # type: ignore[union-attr]
            pad_to_flash_size = _mod.pad_to_flash_size
        fw_bytes = pad_to_flash_size(base64.b64decode(firmware_b64))
        tmp = tempfile.NamedTemporaryFile(suffix='.bin', delete=False)
        tmp.write(fw_bytes)
        tmp.close()
        firmware_path: str | None = tmp.name
    except Exception as exc:
        _emit({'type': 'error', 'message': f'Firmware decode error: {exc}'})
        os._exit(1)

    rom_dir   = os.path.dirname(lib_path).encode()
    args_list = [
        b'qemu',
        b'-M', machine.encode(),
        b'-nographic',
        b'-L', rom_dir,
        b'-drive', f'file={firmware_path},if=mtd,format=raw'.encode(),
    ]

    # Deterministic instruction counting for stable timers.
    # Required for ESP32-C3 boot (RISC-V needs deterministic timing).
    # For ESP32 (Xtensa), -icount is NOT used: the WiFi AP beacon timer
    # runs on QEMU_CLOCK_REALTIME, so decoupling virtual time from real
    # time can cause beacon delivery issues on slow/virtualized hosts.
    if 'c3' in machine:
        args_list.extend([b'-icount', b'3'])

    # ── WiFi NIC (slirp user-mode networking) ──────────────────────────────
    if wifi_enabled:
        nic_model = 'esp32c3_wifi' if 'c3' in machine else 'esp32_wifi'
        nic_arg = f'user,model={nic_model},net=192.168.4.0/24'
        if wifi_hostfwd_port:
            nic_arg += f',hostfwd=tcp::{wifi_hostfwd_port}-192.168.4.15:80'
        args_list.extend([b'-nic', nic_arg.encode()])
        _log(f'WiFi enabled: -nic {nic_arg}')

    argc = len(args_list)
    argv = (ctypes.c_char_p * argc)(*args_list)

    # ── 4. Shared mutable state ───────────────────────────────────────────────
    _stopped       = threading.Event()      # set on "stop" command
    _init_done     = threading.Event()      # set when qemu_init() returns
    _sensors_ready = threading.Event()      # set after pre-registering initial sensors
    _i2c_responses: dict[int, int] = {}     # 7-bit addr → response byte (simple)
    _i2c_slaves:    dict = {}               # 7-bit addr → I2C slave/sink instance
    _spi_response   = [0xFF]                # MISO byte for SPI transfers (default)

    # Custom-chip runtimes that registered their respective protocols at chip_setup.
    # Mutated when sensor_type=='custom-chip' is processed in initial_sensors.
    _chip_uart_runtimes: list = []          # runtimes that called vx_uart_attach
    _chip_spi_runtimes:  list = []          # runtimes that called vx_spi_attach
    _chip_timer_runtimes: list = []         # runtimes with active timers
    _chip_pin_watch_runtimes: list = []     # runtimes that called vx_pin_watch

    # ePaper SSD168x slaves keyed by frontend component_id. The slave decodes
    # SPI bytes; on MASTER_ACTIVATION it emits an `epaper_update` WS frame.
    # `dc_pin` / `cs_pin` / `rst_pin` (gpio numbers) are tracked via
    # `_on_pin_change`; an active slave is one whose `cs_low` is True.
    _epaper_slaves: dict = {}
    # Per-slave runtime state keyed identically: dict with keys
    #   'slave', 'dc_pin', 'cs_pin', 'rst_pin', 'busy_pin', 'cs_low',
    #   'dc_high', 'refresh_ms'.
    _epaper_state: dict = {}

    # Live GPIO state tracked from QEMU's _on_pin_change callback. Custom-chip
    # runtimes' vx_pin_read consults this to see what the firmware just drove.
    _pin_state: dict[int, int] = {}
    _rmt_decoders:  dict[int, _RmtDecoder] = {}
    _uart0_buf      = bytearray()           # accumulate UART0 for crash detection
    _reboot_count   = [0]
    _crashed        = [False]
    _camera_frame_count = [0]               # ESP32-CAM frame trace counter
    _CRASH_STR      = b'Cache disabled but cached memory region accessed'
    _REBOOT_STR     = b'Rebooting...'

    # ── Signal routing (GPIO Matrix mirror) ───────────────────────────────
    # The SignalRouter owns the per-GPIO routing table that the firmware
    # writes through `GPIO_FUNCx_OUT_SEL_CFG_REG[x]`. We currently fill it
    # by polling `gpio_out_sel[40]` once every 100 ms in the LEDC poll
    # thread (and diffing); the C-side plugin will gain a synchronous
    # callback in a future bump that turns the poll into a push without
    # touching this code path.
    #
    # The old `_ledc_gpio_map: dict[int, int]` (channel → gpio) has been
    # subsumed by the router's reverse index — call
    # `_signal_router.pins_for_signal(SIG_LEDC_*+channel)` instead.
    #
    # `app.*` is not on sys.path inside this subprocess; mirror the same
    # importlib fallback pattern used for esp32_flash_image (further down
    # this file) so the worker can find its sibling modules without
    # depending on the backend's package layout.
    try:
        from app.services.signal_router import SignalRouter  # type: ignore[import-not-found] # noqa: E402
        from app.services.esp32_signals import (  # type: ignore[import-not-found] # noqa: E402
            ledc_signal_for_channel,
            SIG_LEDC_HS_CH0_OUT_IDX,
            SIG_LEDC_LS_CH_LAST,
        )
    except ImportError:
        import importlib.util as _ilu, pathlib as _pl
        _here = _pl.Path(__file__).parent
        for _name in ('signal_router', 'esp32_signals'):
            _spec = _ilu.spec_from_file_location(_name, _here / f'{_name}.py')
            _mod = _ilu.module_from_spec(_spec)  # type: ignore[arg-type]
            _spec.loader.exec_module(_mod)        # type: ignore[union-attr]
            sys.modules[_name] = _mod
        SignalRouter = sys.modules['signal_router'].SignalRouter
        ledc_signal_for_channel = sys.modules['esp32_signals'].ledc_signal_for_channel
        SIG_LEDC_HS_CH0_OUT_IDX = sys.modules['esp32_signals'].SIG_LEDC_HS_CH0_OUT_IDX
        SIG_LEDC_LS_CH_LAST = sys.modules['esp32_signals'].SIG_LEDC_LS_CH_LAST
    _signal_router = SignalRouter()

    def _refresh_signal_routing() -> None:
        """Scan `gpio_out_sel[40]` and reconcile the SignalRouter.

        Emits `gpio_routing` for every routing that changed since the
        last scan and `gpio_routing_clear` for routings that disappeared,
        so the frontend's mirror stays in lock-step without re-sending
        the whole table.  Idempotent: a scan with no changes emits no
        events.

        Called every 100 ms from the LEDC poll thread.  Also called
        eagerly from the 0x5000 LEDC duty callback so the first duty
        write after `ledcAttachPin` doesn't race the periodic poll.
        """
        try:
            out_sel_ptr = lib.qemu_picsimlab_get_internals(2)
            if not out_sel_ptr:
                return
            out_sel = (ctypes.c_uint32 * 40).from_address(out_sel_ptr)
            snapshot: dict[int, int] = {}
            for gpio_pin in range(40):
                signal_id = int(out_sel[gpio_pin]) & 0xFF
                # 0..71 / 88..255 are signal sources velxio doesn't
                # model yet; include them in the snapshot only if the
                # firmware actively routed them so future peripherals
                # can opt in without code changes here.
                if SIG_LEDC_HS_CH0_OUT_IDX <= signal_id <= SIG_LEDC_LS_CH_LAST:
                    snapshot[gpio_pin] = signal_id
            changed, cleared = _signal_router.replace_snapshot(snapshot)
            for gpio_pin, signal_id in changed:
                _emit({'type': 'gpio_routing',
                       'gpio': gpio_pin,
                       'signal_id': signal_id})
            for gpio_pin in cleared:
                _emit({'type': 'gpio_routing_clear', 'gpio': gpio_pin})
        except Exception:
            pass

    # Sensor state: gpio_pin → {type, properties..., saw_low, responding}
    _sensors: dict[int, dict] = {}
    _sensors_lock = threading.Lock()

    # ── Generic sync-handler registry ────────────────────────────────────────
    # Each entry implements step() -> bool.  step() is called once per
    # GPIO_IN read sync (every digitalRead() / pulseIn() iteration in firmware).
    # Returning True signals completion; the dispatcher removes the handler.
    # All mutations happen exclusively on the QEMU thread — no locks needed.
    #
    # To add a new GPIO-timed sensor:
    #   1. Write a class with a step() -> bool method
    #   2. Append an instance to _sync_handlers from _on_pin_change or _on_dir_change
    #   3. No changes to the dispatcher are needed
    _sync_handlers: list = []

    def _dht22_build_payload(temperature: float, humidity: float) -> list[int]:
        """Build 5-byte DHT22 data payload: [hum_H, hum_L, temp_H, temp_L, checksum]."""
        hum = round(humidity * 10)
        tmp = round(temperature * 10)
        h_H = (hum >> 8) & 0xFF
        h_L = hum & 0xFF
        raw_t = ((-tmp) & 0x7FFF) | 0x8000 if tmp < 0 else tmp & 0x7FFF
        t_H = (raw_t >> 8) & 0xFF
        t_L = raw_t & 0xFF
        chk = (h_H + h_L + t_H + t_L) & 0xFF
        return [h_H, h_L, t_H, t_L, chk]

    def _dht22_build_sync_phases(payload: list[int]) -> list[tuple[int, int]]:
        """Build list of (sync_count, pin_value) phase transitions for DHT22.

        Each entry means: after sync_count digitalRead() calls in this phase,
        drive the pin to pin_value and advance to the next phase.

        The Adafruit DHT library decodes bits by comparing
        highCycles > lowCycles — only RATIOS matter, not absolute values.
        We use the raw µs values as sync counts to preserve correct ratios.

        After the last data bit (40th bit HIGH→LOW), the firmware's
        expectPulse() loop ends — no more syncs will arrive.  So we do
        NOT add a trailing phase; cleanup happens immediately after the
        last phase transition fires.
        """
        phases: list[tuple[int, int]] = []
        # Preamble: LOW 80 syncs → drive HIGH
        phases.append((80, 1))
        # Preamble: HIGH 80 syncs → drive LOW
        phases.append((80, 0))
        # 40 data bits: LOW 50 syncs → HIGH, then HIGH (26 or 70) → LOW
        for byte_val in payload:
            for b in range(7, -1, -1):
                bit = (byte_val >> b) & 1
                phases.append((50, 1))              # LOW phase → drive HIGH
                phases.append((70 if bit else 26, 0))  # HIGH phase → drive LOW
        return phases

    class DHT22SyncHandler:
        """Drives the DHT22 waveform synchronously, one GPIO_IN read sync at a time.

        Uses phase-based counting: each phase defines how many syncs to wait before
        driving the pin to a new value.  The Adafruit DHT library decodes bits by
        comparing highCycles vs lowCycles — only RATIOS matter, so raw µs values used
        as sync counts preserve the correct bit decoding.
        """
        def __init__(self, gpio: int, slot: int, phases: list[tuple[int, int]]) -> None:
            self._gpio        = gpio
            self._slot        = slot
            self._phases      = phases
            self._phase_idx   = 0
            self._count       = 0
            self._total_syncs = 0

        def step(self) -> bool:
            """Advance one sync tick.  Returns True when the handler is done."""
            self._count += 1
            if self._phase_idx >= len(self._phases):
                return self._finish()
            target, pin_value = self._phases[self._phase_idx]
            if self._count >= target:
                lib.qemu_picsimlab_set_pin(self._slot, pin_value)
                self._total_syncs += self._count
                self._count       = 0
                self._phase_idx  += 1
                if self._phase_idx >= len(self._phases):
                    return self._finish()
            return False

        def _finish(self) -> bool:
            with _sensors_lock:
                sensor = _sensors.get(self._gpio)
                if sensor:
                    sensor['responding'] = False
            _log(f'DHT22 sync respond done gpio={self._gpio} '
                 f'total_syncs={self._total_syncs} phases={len(self._phases)}')
            _emit({'type': 'system', 'event': 'dht22_diag', 'gpio': self._gpio,
                   'status': 'ok', 'total_syncs': self._total_syncs})
            return True

    class HCSR04SyncHandler:
        """Drives HC-SR04 ECHO pin synchronously from the QEMU GPIO_IN read callback.

        _on_dir_change(-1, -1) fires for EVERY gpio_get_level() call in the
        firmware — including pulseIn()'s busy-wait loops.  The state machine:

          Phase 1 of pulseIn() (wait for !HIGH = wait for LOW):
            ECHO is LOW so the condition is immediately false.  One
            gpio_get_level() call fires.  We skip it.

          Phase 2 of pulseIn() (wait for HIGH):
            After skipping _SKIP_COUNT pre-phase2 callbacks, the next
            gpio_get_level() fires.  We set ECHO HIGH here.  pulseIn() sees HIGH
            immediately (qemu_picsimlab_set_pin is synchronous) and exits phase 2.

          Phase 3 of pulseIn() (measure HIGH duration):
            Subsequent gpio_get_level() calls fire step().  We hold ECHO HIGH
            until echo_us wall-clock µs have elapsed (perf_counter_ns), then
            set LOW.  Virtual time ≈ wall-clock time (confirmed: 30 000 µs
            pulseIn timeout = 30 ms wall-clock), so pulseIn() measures ≈ echo_us
            virtual µs → correct distance.

        Guard: wall-clock timeouts replace step-count limits.  A step-count
        guard is wrong because steps fire at rates that vary with QEMU load;
        using a fixed count would cut the pulse short for longer distances
        (100 cm = 5 800 µs, 200 cm = 11 600 µs) before elapsed_us is reached.
        """
        _SKIP_COUNT       = 2         # pre-phase2 callbacks to skip
        _ARMED_TIMEOUT_US = 40_000    # µs; give up if we never enter 'high'
        _HIGH_TIMEOUT_US  = 32_000    # µs; pulseIn() timeout is 30 000 µs

        def __init__(self, trig_gpio: int, echo_slot: int, echo_us: int) -> None:
            self._trig_gpio     = trig_gpio
            self._echo_slot     = echo_slot
            self._echo_us       = echo_us
            self._state         = 'armed'
            self._total_steps   = 0
            self._arm_start_ns  = time.perf_counter_ns()
            self._echo_start_ns = 0

        def step(self) -> bool:
            self._total_steps += 1

            if self._state == 'armed':
                if self._total_steps <= self._SKIP_COUNT:
                    return False
                # Check armed timeout (handler never entered 'high')
                arm_us = (time.perf_counter_ns() - self._arm_start_ns) // 1000
                if arm_us > self._ARMED_TIMEOUT_US:
                    _log(f'HCSR04 armed timeout trig={self._trig_gpio} '
                         f'arm_us={arm_us} steps={self._total_steps} — releasing')
                    with _sensors_lock:
                        sensor = _sensors.get(self._trig_gpio)
                        if sensor:
                            sensor['responding'] = False
                    return True
                # pulseIn() is now in phase 2 (waiting for HIGH) → raise ECHO
                lib.qemu_picsimlab_set_pin(self._echo_slot, 1)
                _emit({'type': 'system', 'event': 'hcsr04_echo_high',
                       'gpio': self._trig_gpio, 'echo_us': self._echo_us})
                _log(f'HCSR04 ECHO HIGH (sync) trig={self._trig_gpio} '
                     f'slot={self._echo_slot} echo_us={self._echo_us} '
                     f'armed_us={arm_us} skip={self._total_steps - 1}')
                self._echo_start_ns = time.perf_counter_ns()
                self._state = 'high'
                return False

            elif self._state == 'high':
                elapsed_us = (time.perf_counter_ns() - self._echo_start_ns) // 1000
                if elapsed_us >= self._echo_us:
                    return self._finish(elapsed_us)
                # Safety: don't hold ECHO past pulseIn() timeout
                if elapsed_us >= self._HIGH_TIMEOUT_US:
                    _log(f'HCSR04 high timeout trig={self._trig_gpio} '
                         f'elapsed_us={elapsed_us} echo_us={self._echo_us}')
                    lib.qemu_picsimlab_set_pin(self._echo_slot, 0)
                    with _sensors_lock:
                        sensor = _sensors.get(self._trig_gpio)
                        if sensor:
                            sensor['responding'] = False
                    return True

            return False

        def _finish(self, elapsed_us: int) -> bool:
            lib.qemu_picsimlab_set_pin(self._echo_slot, 0)
            _emit({'type': 'system', 'event': 'hcsr04_echo_low',
                   'gpio': self._trig_gpio})
            _log(f'HCSR04 ECHO LOW (sync) trig={self._trig_gpio} '
                 f'elapsed_us={elapsed_us} echo_us={self._echo_us} '
                 f'steps={self._total_steps}')
            with _sensors_lock:
                sensor = _sensors.get(self._trig_gpio)
                if sensor:
                    sensor['responding'] = False
            return True

    # ── 5. ctypes callbacks (called from QEMU thread) ─────────────────────────

    def _on_pin_change(slot: int, value: int) -> None:
        if _stopped.is_set():
            return
        gpio = int(_PINMAP[slot]) if 1 <= slot <= _GPIO_COUNT else slot
        _pin_state[gpio] = value & 1
        # Flush pending SPI bytes BEFORE announcing this pin change so the
        # frontend processes them under the pin state (e.g. the ILI9341 DC line)
        # that was in effect when they were sent. With CS events gated off for
        # pure-display sims, this — plus the buffer cap / timer — is what keeps
        # the SPI byte stream correctly ordered against the DC gpio_change on the
        # single WS channel (the per-CS flush used to do it).
        with _spi_buf_lock:
            _flush_spi_batch_locked()
        _emit({'type': 'gpio_change', 'pin': gpio, 'state': value})

        # Dispatch to any custom-chip runtime that has a vx_pin_watch on this
        # GPIO. We're called from QEMU's GPIO state-change path which holds the
        # IO-thread lock, so the chip's callback can safely call vx_pin_write
        # (which goes back into picsimlab and requires the same lock).
        if _chip_pin_watch_runtimes:
            for rt in _chip_pin_watch_runtimes:
                try:
                    rt.notify_pin_change(gpio, value)
                except Exception as e:
                    _log(f'[custom-chip pin_watch] error: {e!r}')

        # ePaper SSD168x: track DC / CS / RST pin states for every slave.
        # CS rising re-arms the next byte; CS falling activates the slave.
        # RST falling clears the controller's RAM (active LOW).
        if _epaper_state:
            for st in _epaper_state.values():
                if gpio == st['dc_pin']:
                    st['dc_high'] = bool(value & 1)
                elif gpio == st['cs_pin']:
                    st['cs_low'] = (value & 1) == 0
                elif gpio == st['rst_pin']:
                    if (value & 1) == 0:
                        st['slave'].reset()

        # Sensor protocol dispatch by type
        with _sensors_lock:
            sensor = _sensors.get(gpio)
        if sensor is None:
            return

        stype = sensor.get('type', '')

        if stype == 'dht22':
            # Record that the firmware drove the pin LOW (start signal).
            # The actual response is triggered from _on_dir_change when the
            # firmware switches the pin to INPUT mode.
            if value == 0 and not sensor.get('responding', False):
                sensor['saw_low'] = True

        elif stype == 'hc-sr04':
            # HC-SR04 trigger:
            #   TRIG HIGH → arm: save echo params
            #   TRIG LOW  → add HCSR04SyncHandler to _sync_handlers
            #
            # The sync handler drives ECHO from within the QEMU GPIO_IN read
            # callback (_on_dir_change slot=-1 direction=-1), which fires for
            # every gpio_get_level() call in the firmware — including pulseIn().
            # This is 100% synchronous with the QEMU thread, eliminating the
            # non-deterministic visibility issue that plagued the background-thread
            # approach (only ~33% success rate due to cross-thread pin propagation).
            if value == 1 and not sensor.get('responding', False):
                echo_pin = int(sensor.get('echo_pin', gpio + 1))
                distance = float(sensor.get('distance', 40.0))
                echo_us  = max(100, int(distance * 58))
                sensor['_trig_armed'] = {'echo_slot': echo_pin + 1, 'echo_us': echo_us}
                _log(f'HCSR04 TRIG HIGH (armed) gpio={gpio} echo_slot={echo_pin + 1} '
                     f'echo_us={echo_us} dist={distance}cm')

            elif value == 0 and sensor.get('_trig_armed') and not sensor.get('responding', False):
                armed     = sensor.pop('_trig_armed')
                echo_slot = armed['echo_slot']
                echo_us   = armed['echo_us']
                sensor['responding'] = True
                _sync_handlers.append(HCSR04SyncHandler(gpio, echo_slot, echo_us))
                _log(f'HCSR04 TRIG LOW → sync handler armed gpio={gpio} '
                     f'echo_slot={echo_slot} echo_us={echo_us}')

    def _on_dir_change(slot: int, direction: int) -> None:
        if _stopped.is_set():
            return

        # ── GPIO_IN read sync (slot == -1, direction == -1) ──────────────
        # Every digitalRead() in the firmware triggers this sync.  We use
        # it to drive DHT22 pin transitions synchronously on the QEMU
        # thread, perfectly synchronized with the firmware's expectPulse()
        # loop iterations.
        if slot == -1:
            if direction == -1:
                # GPIO_IN read sync — advance all active sync handlers.
                # step() returns True when done; list-comp removes finished handlers.
                if _sync_handlers:
                    _sync_handlers[:] = [h for h in _sync_handlers if not h.step()]
                return  # always return for GPIO_IN syncs (fast path)
            marker = direction & 0xF000
            if marker == 0x5000:  # LEDC duty change (from esp32_ledc.c)
                ledc_ch = (direction >> 8) & 0x0F
                intensity = direction & 0xFF  # 0-100 percentage

                # Refresh the GPIO Matrix snapshot first so the routing
                # is current — emits any gpio_routing events the
                # frontend needs to update its SignalRouter mirror
                # BEFORE the duty arrives.
                _refresh_signal_routing()

                # New canonical event (SignalRouter consumer): channel +
                # duty only, no gpio.  The frontend resolves channel →
                # signal_id → pins via its mirror.
                _emit({'type': 'ledc_duty',
                       'channel': ledc_ch,
                       'duty_pct': intensity})
            return

        # ── DHT22: track direction changes + trigger sync response ───────
        if slot >= 1:
            gpio = int(_PINMAP[slot]) if slot <= _GPIO_COUNT else slot
            with _sensors_lock:
                sensor = _sensors.get(gpio)
            if sensor is not None and sensor.get('type') == 'dht22':
                if direction == 1:
                    # OUTPUT mode — record timestamp for diagnostics
                    sensor['dir_out_ns'] = time.perf_counter_ns()
                elif direction == 0:
                    # INPUT mode — trigger DHT22 sync-based response
                    if sensor.get('saw_low', False) and not sensor.get('responding', False):
                        sensor['saw_low'] = False
                        sensor['responding'] = True

                        # Build the response waveform phases
                        temp = sensor.get('temperature', 25.0)
                        hum = sensor.get('humidity', 50.0)
                        payload = _dht22_build_payload(temp, hum)
                        phases = _dht22_build_sync_phases(payload)

                        # Drive pin LOW synchronously — firmware sees LOW
                        # at its first digitalRead() in expectPulse().
                        lib.qemu_picsimlab_set_pin(slot, 0)

                        # Arm the sync-based response state machine
                        _sync_handlers.append(DHT22SyncHandler(gpio, slot, phases))
                        _log(f'DHT22 sync armed gpio={gpio} '
                             f'temp={temp} hum={hum} '
                             f'phases={len(phases)} payload={payload}')
        gpio = int(_PINMAP[slot]) if 1 <= slot <= _GPIO_COUNT else slot
        _emit({'type': 'gpio_dir', 'pin': gpio, 'dir': direction})

    def _on_uart_tx(uart_id: int, byte_val: int) -> None:
        if _stopped.is_set():
            return
        _emit({'type': 'uart_tx', 'uart': uart_id, 'byte': byte_val})
        # Dispatch to any custom-chip runtimes that declared a UART.
        # The chip's on_rx_byte callback runs synchronously in this thread.
        for rt in _chip_uart_runtimes:
            try:
                rt.feed_uart_byte(byte_val)
            except Exception as e:
                _log(f'[custom-chip uart_tx] error: {e!r}')
        # Crash / reboot detection on UART0 only
        if uart_id == 0:
            _uart0_buf.append(byte_val)
            if byte_val == ord('\n') or len(_uart0_buf) >= 512:
                chunk = bytes(_uart0_buf)
                _uart0_buf.clear()
                if _CRASH_STR in chunk and not _crashed[0]:
                    _crashed[0] = True
                    _emit({'type': 'system', 'event': 'crash',
                           'reason': 'cache_error', 'reboot': _reboot_count[0]})
                if _REBOOT_STR in chunk:
                    _crashed[0] = False
                    _reboot_count[0] += 1
                    _emit({'type': 'system', 'event': 'reboot',
                           'count': _reboot_count[0]})
                # WiFi progress logging (only in debug — helps diagnose prod issues)
                if wifi_enabled:
                    line = chunk.decode('utf-8', errors='replace').strip()
                    if any(kw in line.lower() for kw in (
                        'wifi', 'connect', 'ip address', 'wl_connected',
                        'dhcp', 'sta_start', 'sta_got_ip', 'sta_disconnect',
                    )):
                        _log(f'[wifi-uart] {line}')

    def _on_rmt_event(channel: int, config0: int, value: int) -> None:
        if _stopped.is_set():
            return
        level0, dur0, level1, dur1 = _decode_rmt_item(value)
        _emit({'type': 'rmt_event', 'channel': channel, 'config0': config0,
               'value': value, 'level0': level0, 'dur0': dur0,
               'level1': level1, 'dur1': dur1})
        if channel not in _rmt_decoders:
            _rmt_decoders[channel] = _RmtDecoder(channel)
        pixels = _rmt_decoders[channel].feed(value)
        if pixels:
            _emit({'type': 'ws2812_update', 'channel': channel, 'pixels': pixels})

    def _on_gpio_matrix(gpio: int, signal_id: int) -> None:
        """Synchronous GPIO Matrix routing event from libqemu 1.1.0+.

        Critical: this fires on QEMU's iothread, hundreds of times during
        early boot (bootloader + IDF init configure every GPIO Matrix
        slot). It MUST NOT do anything that can block the iothread —
        most importantly NOT _emit() over the stdout pipe, because if
        the manager's reader is even briefly stalled, the pipe fills,
        write() blocks, the iothread freezes, and the entire guest
        stops (symptom: ESP32 boot stops at "entry 0x400805e4" with no
        Arduino setup() output).

        So this callback ONLY mutates the in-memory SignalRouter
        snapshot. The 10 Hz poll thread (_refresh_signal_routing) is
        the sole emitter of gpio_routing / gpio_routing_clear events.
        The callback's only benefit over the poll alone is reducing
        the worst-case routing-to-emit latency from ~100 ms to one
        poll tick, AND keeping the snapshot dict warm so the next
        poll's diff is cheaper.
        """
        if _stopped.is_set():
            return
        try:
            sid_lo = signal_id & 0xFF
            if signal_id == 0x100:
                _signal_router.clear_routing(gpio)
            elif SIG_LEDC_HS_CH0_OUT_IDX <= sid_lo <= SIG_LEDC_LS_CH_LAST:
                _signal_router.update_routing(gpio, sid_lo)
            # Other signal_id values fall outside what the frontend
            # SignalRouter currently cares about; future peripherals
            # extend the range above.
        except Exception:
            # Iothread callback — never raise, never block.
            pass

    # ── Per-slave I2C event counter (for logging) ─────────────────────────────
    _i2c_event_seq: dict = {}   # addr → event count

    _I2C_OP_NAME = {0x00: 'START_RECV', 0x01: 'START_SEND', 0x02: 'START_ASYNC',
                    0x03: 'FINISH',    0x04: 'NACK',
                    0x05: 'WRITE',     0x06: 'READ'}
    _MPU_REG_NAME = {
        0x19: 'SMPRT_DIV', 0x1A: 'CONFIG', 0x1B: 'GYRO_CFG', 0x1C: 'ACCEL_CFG',
        0x3B: 'AX_H', 0x3C: 'AX_L', 0x3D: 'AY_H', 0x3E: 'AY_L',
        0x3F: 'AZ_H', 0x40: 'AZ_L', 0x41: 'T_H',  0x42: 'T_L',
        0x43: 'GX_H', 0x44: 'GX_L', 0x45: 'GY_H', 0x46: 'GY_L',
        0x47: 'GZ_H', 0x48: 'GZ_L',
        0x6B: 'PWR_MGMT1', 0x68: 'SIG_RST', 0x75: 'WHO_AM_I',
    }

    def _on_i2c_event(bus_id: int, addr: int, event: int) -> int:
        """Synchronous — must return immediately; called from QEMU thread."""
        slave = _i2c_slaves.get(addr)
        op    = event & 0xFF
        data  = (event >> 8) & 0xFF
        op_name = _I2C_OP_NAME.get(op, f'0x{op:02x}')

        if slave is not None:
            result  = slave.handle_event(event)
            reg_ptr = getattr(slave, 'reg_ptr', 0)

            # Build descriptive annotation
            if op in (0x00, 0x01):   # START_RECV / START_SEND
                note = f'→ reg_ptr=0x{reg_ptr:02x}'
            elif op == 0x06:  # READ byte (actual data delivery to firmware)
                reg_nm = _MPU_REG_NAME.get((reg_ptr - 1) & 0xFF, f'0x{(reg_ptr-1)&0xFF:02x}')
                note = f'→ {reg_nm}=0x{result:02x}'
            elif op == 0x05:  # WRITE byte
                note = f'byte=0x{data:02x} → reg_ptr=0x{reg_ptr:02x}'
            else:
                note = ''

            slave_type_name = type(slave).__name__
            if slave_type_name == 'MPU6050Slave':
                seq = _i2c_event_seq
                n   = seq[addr] = seq.get(addr, 0) + 1
                _log(f'I2C #{n:03d} bus={bus_id} addr=0x{addr:02x} {op_name} {note}')
            elif slave_type_name != 'I2CWriteSink':
                # I2CWriteSink fires per-byte for display drivers (SSD1306,
                # PCF8574). A 1024-byte writevto (oled.show()) generates
                # ~1025 events. Logging each one + emitting a WS message
                # blocks the QEMU thread long enough that the firmware's
                # ESP-IDF I2C ISR re-enters and trips IWDT on the SECOND
                # consecutive show() call. Skip the verbose per-event log
                # and WS trace for write-only sinks — the user-visible
                # OLED render is what matters, not byte-level tracing.
                _log(f'I2C bus={bus_id} addr=0x{addr:02x} event=0x{event:04x} '
                     f'op={op_name} result=0x{result:02x} slave={slave_type_name}')
            # Emit trace event to WebSocket so JS test can observe I2C traffic.
            # Skip for I2CWriteSink (display data dumps) — see comment above.
            if not _stopped.is_set() and slave_type_name != 'I2CWriteSink':
                _emit({'type': 'i2c_trace', 'bus': bus_id, 'addr': addr,
                       'event': event, 'op': op_name, 'result': result,
                       'reg_ptr': reg_ptr})
            return result

        _log(f'I2C bus={bus_id} addr=0x{addr:02x} event=0x{event:04x} op={op_name} '
             f'NO_SLAVE registered={list(_i2c_slaves.keys())}')
        resp = _i2c_responses.get(addr, 0)
        if not _stopped.is_set():
            _emit({'type': 'i2c_event', 'bus': bus_id, 'addr': addr,
                   'event': event, 'response': resp})

        # NACK on START_SEND / START_RECV when no slave responds. Real
        # I²C hardware NACKs by leaving SDA high during the ack slot;
        # the picsimlab_i2c bridge has been claiming every address and
        # returning ACK by default, which fooled drivers (notably the
        # esp32-camera SCCB auto-probe — it kept thinking 0x21 was a
        # valid OV7725 sensor and never advanced to 0x30 / OV2640).
        # Returning non-zero from the I2CSlave.event callback is the
        # QEMU convention for "I don't recognise this address".
        # An explicit override via _i2c_responses still wins so test
        # harnesses that register a fake response keep working.
        if op in (0x00, 0x01) and resp == 0 and addr not in _i2c_responses:
            return 1
        return resp

    # SPI byte batching — emitting one WS message per byte saturates the
    # uvicorn → frontend pipe and caps tft.drawRGBBitmap at < 1 fps even
    # for tiny previews. Buffer the MOSI bytes here and flush as a single
    # base64-encoded `spi_batch` message when:
    #   1. CS goes HIGH (transaction ended) — only fires when the firmware
    #      uses the SPI peripheral's hardware CS line. If CS is bit-banged
    #      via digitalWrite (the default for many Adafruit-style drivers
    #      on ESP32), this trigger never fires and we fall back to (2)+(3).
    #   2. Buffer crosses _SPI_BATCH_FLUSH_AT bytes (safety cap for big
    #      transactions).
    #   3. _spi_flush_timer fires every _SPI_BATCH_PERIOD_MS regardless —
    #      catches the GPIO-CS case so partial batches don't sit in the
    #      buffer forever between transactions. Without this, after a few
    #      drawRGBBitmap calls the firmware advances faster than the
    #      buffer fills, and frames stop appearing on the screen.
    # MISO is still returned synchronously per byte from _spi_response[0]
    # because the QEMU master writes can't wait. Frontend Esp32Bridge
    # unpacks the batch and replays each byte through onSpiByte.
    _spi_byte_buf       = bytearray()
    _spi_buf_lock       = threading.Lock()
    _SPI_BATCH_FLUSH_AT = 4096
    _SPI_BATCH_PERIOD_S = 0.05   # 50 ms → 20 fps cadence ceiling

    def _flush_spi_batch_locked():
        if _spi_byte_buf and not _stopped.is_set():
            b64 = base64.b64encode(bytes(_spi_byte_buf)).decode('ascii')
            _emit({'type': 'spi_batch', 'b64': b64})
            _spi_byte_buf.clear()

    def _sync_cs_events():
        """Tell QEMU whether to forward SPI chip-select toggles to us. Only
        ePaper / custom-chip SPI slaves consume CS; pure-display sims (DC pin +
        batched data) do not, so turning CS off there removes ~9k C->Python
        crossings/sec for a TFT redraw. No-op on older libqemu builds without
        the symbol (CS events stay on, as before)."""
        try:
            lib.qemu_picsimlab_enable_spi_cs_events(
                1 if (_epaper_state or _chip_spi_runtimes) else 0)
        except Exception:
            pass

    def _spi_flush_timer_loop():
        """Background thread: flushes any pending SPI bytes every
        _SPI_BATCH_PERIOD_S so partial transactions reach the frontend
        even when the firmware drives CS via GPIO and we never see a
        SPI peripheral CS-high event."""
        while not _stopped.is_set():
            _stopped.wait(_SPI_BATCH_PERIOD_S)
            if _stopped.is_set():
                break
            with _spi_buf_lock:
                _flush_spi_batch_locked()

    threading.Thread(
        target=_spi_flush_timer_loop, daemon=True,
        name='esp32-spi-batch-flush',
    ).start()

    def _on_spi_event(bus_id: int, event: int) -> int:
        """Synchronous — must return immediately; called from QEMU thread.

        Event encoding (picsimlab — see hw/ssi/picsimlab_spi.c and the CS irq
        handler in esp32_picsimlab.c):
            event = data << 8                                  → SPI byte transfer
                                                                 (op = low byte = 0x00,
                                                                  MOSI = high byte)
            event = ((((cs_idx & 3) << 1) | level) << 8) | 0x01 → CS line change
                                                                  (op = 0x01,
                                                                   ignored by chips
                                                                   that drive their own
                                                                   CS via pin_watch)
        """
        # Custom-chip SPI runtimes get first dibs on byte transfers. The chip's
        # pre-armed buffer holds the next MISO byte; the runtime overwrites it
        # with the master's MOSI byte and advances. on_done fires when count is
        # reached.
        op   = event & 0xFF
        mosi = (event >> 8) & 0xFF
        if _chip_spi_runtimes and op == 0x00:
            for rt in _chip_spi_runtimes:
                try:
                    return rt.spi_transfer_byte(mosi) & 0xFF
                except Exception as e:
                    _log(f'[custom-chip spi_event] error: {e!r}')

        # ePaper SSD168x panels — feed every byte to the active slave (CS LOW).
        # ePaper is write-only on MOSI; the panel uses BUSY for status, so we
        # always respond 0xFF on MISO. Multiple panels on the same bus would
        # both receive the byte, but the user's wiring + CS gating decide
        # which slave's `cs_low` is True.
        if _epaper_state and op == 0x00:
            any_active = False
            for st in _epaper_state.values():
                if st['cs_low']:
                    any_active = True
                    try:
                        st['slave'].feed(mosi, st['dc_high'])
                    except Exception as e:
                        _log(f'[epaper spi_event] error: {e!r}')
            if any_active:
                return 0xFF
        resp = _spi_response[0]
        if _stopped.is_set():
            return resp
        # ── Batching path (replaces the per-byte _emit) ─────────────────
        if op == 0x00:
            # Byte transfer — append to buffer, flush if oversized.
            with _spi_buf_lock:
                _spi_byte_buf.append(mosi)
                if len(_spi_byte_buf) >= _SPI_BATCH_FLUSH_AT:
                    _flush_spi_batch_locked()
        else:
            # CS-line change. Flush any pending bytes from the previous
            # transaction so the frontend processes them before the
            # (rare) CS-state event itself. Then forward the CS event
            # via the legacy spi_event channel for chips that observe
            # CS state (e.g. ePaper, custom chips that subscribe to it).
            with _spi_buf_lock:
                _flush_spi_batch_locked()
            _emit({'type': 'spi_event', 'bus': bus_id, 'event': event, 'response': resp})
        return resp

    def _on_spi_batch(bus_id: int, mosi_ptr, length: int) -> None:
        """Batched write-only SPI transfer — the whole MOSI buffer arrives in a
        single call instead of one picsimlab_spi_event per byte. libqemu only
        invokes this for rx==0 (MISO-ignored) transfers on the host SPI shim, so
        nothing is returned. Mirrors the per-byte _on_spi_event side effects in
        bulk: custom-chip runtimes first, then ePaper, then the spi_batch buffer.
        This is the path that removes ~150k C->Python crossings/frame for TFTs."""
        if length <= 0 or _stopped.is_set():
            return
        try:
            data = ctypes.string_at(mosi_ptr, length)
        except Exception:
            return
        # Custom-chip SPI runtimes get first dibs (replay per byte; the chip's
        # MISO return is discarded because this transfer is write-only).
        if _chip_spi_runtimes:
            rt = _chip_spi_runtimes[0]
            for mb in data:
                try:
                    rt.spi_transfer_byte(mb)
                except Exception as e:
                    _log(f'[custom-chip spi_batch] error: {e!r}')
            return
        # ePaper SSD168x: feed each byte under the current DC to every active
        # slave (DC is constant for a write-only transaction).
        if _epaper_state:
            any_active = False
            for st in _epaper_state.values():
                if st['cs_low']:
                    any_active = True
                    slave = st['slave']
                    dc = st['dc_high']
                    for mb in data:
                        try:
                            slave.feed(mb, dc)
                        except Exception as e:
                            _log(f'[epaper spi_batch] error: {e!r}')
            if any_active:
                return
        # Fast path: bulk-append to the spi_batch buffer (same buffer/flush the
        # per-byte path uses, so frontend ordering is unchanged).
        with _spi_buf_lock:
            _spi_byte_buf.extend(data)
            if len(_spi_byte_buf) >= _SPI_BATCH_FLUSH_AT:
                _flush_spi_batch_locked()

    # Keep callback struct alive (prevent GC from freeing ctypes closures)
    _cbs_ref = _CallbacksT(
        picsimlab_write_pin      = _WRITE_PIN(_on_pin_change),
        picsimlab_dir_pin        = _DIR_PIN(_on_dir_change),
        picsimlab_i2c_event      = _I2C_EVENT(_on_i2c_event),
        picsimlab_spi_event      = _SPI_EVENT(_on_spi_event),
        picsimlab_uart_tx_event  = _UART_TX(_on_uart_tx),
        pinmap                   = ctypes.cast(_PINMAP, ctypes.c_void_p).value,
        picsimlab_rmt_event      = _RMT_EVENT(_on_rmt_event),
        picsimlab_gpio_matrix_cb = _GPIO_MATRIX_CB(_on_gpio_matrix),
        picsimlab_spi_event_batch = _SPI_BATCH(_on_spi_batch),
    )
    lib.qemu_picsimlab_register_callbacks(ctypes.byref(_cbs_ref))
    # Log whether the new symbol is present in this libqemu build.
    # Older binaries (pre-1.1.0) silently fall back to the 100 ms
    # poll path; the WS event shape is identical either way.
    try:
        if hasattr(lib, 'picsimlab_gpio_matrix_cb'):
            _log('[gpio-matrix] libqemu 1.1.0+ detected; callback path active '
                 '(poll thread runs as a safety net during burn-in)')
        else:
            _log('[gpio-matrix] libqemu <1.1.0; using poll path only')
    except Exception:
        pass

    # ── 6. QEMU thread ────────────────────────────────────────────────────────

    def _qemu_thread() -> None:
        try:
            lib.qemu_init(argc, argv, None)
        except Exception as exc:
            _emit({'type': 'error', 'message': f'qemu_init failed: {exc}'})
        finally:
            _init_done.set()
        # Wait for initial sensors to be pre-registered before executing firmware.
        # This prevents race conditions where the firmware tries to read a sensor
        # (e.g. DHT22 pulseIn) before the sensor handler is registered.
        _sensors_ready.wait(timeout=5.0)
        lib.qemu_main_loop()

    # With -nographic, qemu_init registers the stdio mux chardev which reads
    # from fd 0.  If we leave fd 0 as the JSON-command pipe from the parent,
    # QEMU's mux will consume those bytes and forward them to UART0 RX,
    # corrupting user-sent serial data.  Redirect fd 0 to /dev/null before
    # qemu_init runs so the mux gets EOF and leaves our command pipe alone.
    # Save the original pipe fd for the command loop below.
    _orig_stdin_fd = os.dup(0)
    _nul = os.open(os.devnull, os.O_RDONLY)
    os.dup2(_nul, 0)
    os.close(_nul)

    # Also redirect fd 1 (stdout) to /dev/null so QEMU's -nographic UART mux
    # doesn't write raw UART bytes onto our JSON event pipe.  Without this:
    #   1. Raw UART bytes prefix each JSON line, corrupting the protocol.
    #   2. On a busy host the pipe fills up, causing _on_uart_tx (called
    #      synchronously from qemu_main_loop) to block inside sys.stdout.flush(),
    #      which stalls qemu_main_loop() and prevents QEMU_CLOCK_REALTIME timers
    #      (including Esp32_WLAN_beacon_timer) from firing → WiFi never connects.
    # Save the real pipe fd and rebind sys.stdout so _emit() keeps working.
    import io as _io
    _orig_stdout_fd = os.dup(1)
    _nul_w = os.open(os.devnull, os.O_WRONLY)
    os.dup2(_nul_w, 1)
    os.close(_nul_w)
    sys.stdout = _io.TextIOWrapper(
        _io.FileIO(_orig_stdout_fd, mode='w', closefd=True),
        line_buffering=True,
        write_through=True,
    )

    qemu_t = threading.Thread(target=_qemu_thread, daemon=True, name=f'qemu-{machine}')
    qemu_t.start()

    if not _init_done.wait(timeout=30.0):
        _emit({'type': 'error', 'message': 'qemu_init timed out after 30 s'})
        os._exit(1)

    # Pre-register initial sensors before letting QEMU execute firmware.
    for s in initial_sensors:
        gpio = int(s.get('pin', 0))
        sensor_type = s.get('sensor_type', '')
        with _sensors_lock:
            sensor_data: dict = {
                'type': sensor_type,
                **{k: v for k, v in s.items() if k not in ('sensor_type', 'pin')},
                'saw_low': False,
                'responding': False,
            }
            # For I2C sensors, also create the slave state machine immediately
            # so _on_i2c_event can find it when the firmware's Wire.begin() runs.
            if sensor_type == 'mpu6050':
                i2c_addr = int(s.get('addr', 0x68))
                slave = _MPU6050Slave(i2c_addr)
                _i2c_slaves[i2c_addr] = slave
                sensor_data['i2c_addr'] = i2c_addr
                sensor_data['slave'] = slave
            elif sensor_type == 'bmp280':
                i2c_addr = int(s.get('addr', 0x76))
                slave = _BMP280Slave(i2c_addr)
                if 'temperature' in s: slave.update(float(s['temperature']), slave._press_hpa)
                if 'pressure'    in s: slave.update(slave._temp_c, float(s['pressure']))
                _i2c_slaves[i2c_addr] = slave
                sensor_data['i2c_addr'] = i2c_addr
                sensor_data['slave'] = slave
            elif sensor_type in ('ds1307', 'ds3231'):
                i2c_addr = int(s.get('addr', 0x68))
                slave = _DS3231Slave() if sensor_type == 'ds3231' else _DS1307Slave()
                _i2c_slaves[i2c_addr] = slave
                sensor_data['i2c_addr'] = i2c_addr
                sensor_data['slave'] = slave
            elif sensor_type == 'epaper-ssd168x':
                # ePaper panel: backend decodes SPI traffic and emits
                # `epaper_update` events with the latched framebuffer.  The
                # `controller_family` payload field selects the decoder
                # ('ssd168x' or 'uc8159c') and ALSO determines the BUSY
                # polarity, because the two controller families use opposite
                # active levels in GxEPD2:
                #
                #   SSD168x family (1.54 / 2.13 / 2.9 / 4.2"):
                #     `_busy_level = HIGH` → BUSY=HIGH means "busy",
                #                            BUSY=LOW means "ready".
                #
                #   UltraChip family — UC8159c (5.65" ACeP) and UC8179/GD7965
                #   (7.5" 800x480): `_busy_level = LOW` → BUSY=LOW means "busy",
                #                            BUSY=HIGH means "ready".
                #
                # Pick the IDLE level per family and (a) seed the pin to IDLE
                # at registration so the firmware's first `_waitBusy()` —
                # which runs inside `_PowerOn()` / `_InitDisplay()` BEFORE any
                # frame is sent — sees "ready" and proceeds, and (b) use that
                # polarity when pulsing on frame flush below.
                comp_id = str(s.get('component_id', f'epaper-{gpio}'))
                width = int(s.get('width', 200))
                height = int(s.get('height', 200))
                refresh_ms = int(s.get('refresh_ms', 50))
                busy_pin = int(s.get('busy_pin', -1))
                # Read controller_family early; default to ssd168x for
                # back-compat with old frontends that didn't send it.
                ctl_family_early = str(s.get('controller_family', 'ssd168x'))
                # UltraChip controllers (uc8159c, uc8179) idle BUSY HIGH; the
                # SSD168x family idles BUSY LOW.
                busy_idle_level = 1 if ctl_family_early in ('uc8159c', 'uc8179') else 0
                busy_busy_level = 1 - busy_idle_level
                if busy_pin is not None and busy_pin >= 0:
                    try:
                        lib.qemu_picsimlab_set_pin(busy_pin + 1, busy_idle_level)
                    except Exception:
                        pass

                def _flush_factory(_comp_id=comp_id,
                                   _w=width, _h=height,
                                   _refresh=refresh_ms,
                                   _busy=busy_pin,
                                   _busy_busy=busy_busy_level,
                                   _busy_idle=busy_idle_level,
                                   _lib=lib):
                    """Build an on_flush callback bound to this slave's
                    component_id so the WS event can route to the right panel.
                    Pulses BUSY to its "busy" level for refresh_ms, then back
                    to "ready" — polarity per controller family (see above)."""
                    def _on_flush(frame):
                        try:
                            frame_b64 = base64.b64encode(frame.pixels).decode('ascii')
                        except Exception:
                            return
                        # NOTE: emit FLAT (fields at top level), like every other
                        # worker event. The backend's qemu_callback re-wraps the
                        # post-'type' payload under 'data' (simulation.py), so a
                        # nested 'data' here would double-wrap and the frontend's
                        # msg.data.component_id would be undefined (panel never
                        # renders). This was the long-standing "ESP32 ePaper is
                        # blank" bug.
                        _emit({
                            'type': 'epaper_update',
                            'component_id': _comp_id,
                            'width': _w,
                            'height': _h,
                            'frame_b64': frame_b64,
                            'refresh_ms': _refresh,
                        })
                        if _busy is not None and _busy >= 0:
                            try:
                                _lib.qemu_picsimlab_set_pin(_busy + 1, _busy_busy)

                                def _busy_idle_cb(_b=_busy, _lvl=_busy_idle):
                                    try:
                                        _lib.qemu_picsimlab_set_pin(_b + 1, _lvl)
                                    except Exception:
                                        pass

                                threading.Timer(_refresh / 1000.0, _busy_idle_cb).start()
                            except Exception:
                                pass
                    return _on_flush

                # Pick the decoder family from the payload. Defaults to
                # SSD168x for backward compatibility (initial frontends only
                # sent SSD168x); the UC8159c value is sent for ACeP panels.
                ctl_family = str(s.get('controller_family', 'ssd168x'))
                if ctl_family == 'uc8159c':
                    slave = _Uc8159cEpaperSlave(
                        component_id=comp_id, width=width, height=height,
                        on_flush=_flush_factory(),
                    )
                elif ctl_family == 'uc8179':
                    slave = _Uc8179EpaperSlave(
                        component_id=comp_id, width=width, height=height,
                        on_flush=_flush_factory(),
                    )
                else:
                    _is_bwr = 'bwr' in str(s.get('panel_kind', '')).lower()
                    slave = _Ssd168xEpaperSlave(
                        component_id=comp_id, width=width, height=height,
                        on_flush=_flush_factory(), is_bwr=_is_bwr,
                    )
                state = {
                    'slave': slave,
                    'dc_pin': int(s.get('dc_pin', -1)),
                    'cs_pin': int(s.get('cs_pin', -1)),
                    'rst_pin': int(s.get('rst_pin', -1)),
                    'busy_pin': busy_pin,
                    'cs_low': False,
                    'dc_high': False,
                    'refresh_ms': refresh_ms,
                    'controller_family': ctl_family,
                }
                _epaper_slaves[comp_id] = slave
                _epaper_state[comp_id] = state
                _sync_cs_events()
                sensor_data['epaper_component_id'] = comp_id
                _log(f"[epaper:{ctl_family}] registered '{comp_id}' "
                     f"({width}x{height}) "
                     f"DC={state['dc_pin']} CS={state['cs_pin']} "
                     f"RST={state['rst_pin']} BUSY={state['busy_pin']}")
            elif sensor_type in ('ssd1306', 'pcf8574'):
                default_addr = 0x3C if sensor_type == 'ssd1306' else 0x27
                i2c_addr = int(s.get('addr', default_addr))
                sink = _I2CWriteSink(i2c_addr, _emit)
                _i2c_slaves[i2c_addr] = sink
                sensor_data['i2c_addr'] = i2c_addr
                sensor_data['slave'] = sink
            elif sensor_type == 'custom-chip':
                # User-supplied chip compiled to WASM. The runtime loads the
                # binary in this same Python process so I2C callbacks fire
                # synchronously when QEMU calls _on_i2c_event — same fidelity
                # as the hardcoded slaves above.
                # See docs/wiki/custom-chips-esp32-backend-runtime.md
                try:
                    from app.services.wasm_chip_runtime import WasmChipRuntime
                    from app.services.wasm_chip_slave   import WasmChipI2CSlave
                except ImportError:
                    # Fallback: same pattern as esp32_i2c_slaves at the top of
                    # this file. The worker subprocess may run from a cwd that
                    # doesn't have `app.services` on sys.path.
                    import importlib.util, pathlib as _pl
                    _here = _pl.Path(__file__).parent
                    _spec_rt = importlib.util.spec_from_file_location(
                        'wasm_chip_runtime', _here / 'wasm_chip_runtime.py'
                    )
                    _mod_rt = importlib.util.module_from_spec(_spec_rt)
                    _spec_rt.loader.exec_module(_mod_rt)
                    WasmChipRuntime = _mod_rt.WasmChipRuntime
                    _spec_sl = importlib.util.spec_from_file_location(
                        'wasm_chip_slave', _here / 'wasm_chip_slave.py'
                    )
                    _mod_sl = importlib.util.module_from_spec(_spec_sl)
                    # The slave module imports from app.services.wasm_chip_runtime;
                    # patch sys.modules so that import resolves to our loaded module.
                    sys.modules['app.services.wasm_chip_runtime'] = _mod_rt
                    _spec_sl.loader.exec_module(_mod_sl)
                    WasmChipI2CSlave = _mod_sl.WasmChipI2CSlave
                wasm_b64 = s.get('wasm_b64', '')
                if not wasm_b64:
                    _log("[custom-chip] missing wasm_b64 in sensor payload")
                else:
                    try:
                        wasm_bytes = base64.b64decode(wasm_b64)
                        attrs      = s.get('attrs', {}) or {}
                        pin_map    = s.get('pin_map', {}) or {}

                        # ── Plumbing: hook the runtime to QEMU's live peripherals ──
                        # GPIO output: chip's vx_pin_write → qemu_picsimlab_set_pin
                        def _chip_pin_writer(gpio: int, value: int, _lib=lib):
                            _lib.qemu_picsimlab_set_pin(gpio + 1, value)

                        # GPIO input: chip reads current QEMU pin state.
                        def _chip_pin_reader(gpio: int, _store=_pin_state):
                            return int(_store.get(gpio, 0)) & 1

                        # UART RX: chip's vx_uart_write → inject bytes into firmware UART.
                        # Acquire the iothread lock ONLY if we don't already hold it
                        # (typical case: the chip's vx_uart_write is fired from inside
                        # _on_uart_tx, which is already in the QEMU thread holding the
                        # lock — re-acquiring there triggers an assertion).
                        def _chip_uart_writer(uart_id: int, data: bytes,
                                              _lib=lib,
                                              _lock=_lock_iothread,
                                              _unlock=_unlock_iothread,
                                              _is_locked=_iothread_locked):
                            buf = (ctypes.c_uint8 * len(data))(*data)
                            need_lock = bool(_lock) and (not _is_locked or not _is_locked())
                            if need_lock:
                                _lock(b'esp32_worker.py:custom-chip', 0)
                            try:
                                _lib.qemu_picsimlab_uart_receive(int(uart_id), buf, len(data))
                            finally:
                                if need_lock and _unlock:
                                    _unlock()

                        # Timer arm: just track this runtime so the scheduler thread
                        # picks up the new deadline on its next iteration.
                        def _chip_timer_scheduler(rt):
                            if rt not in _chip_timer_runtimes:
                                _chip_timer_runtimes.append(rt)

                        runtime = WasmChipRuntime(
                            wasm_bytes, attrs, _emit,
                            pin_map=pin_map,
                            pin_writer=_chip_pin_writer,
                            pin_reader=_chip_pin_reader,
                            uart_writer=_chip_uart_writer,
                            timer_scheduler=_chip_timer_scheduler,
                        )
                        runtime.run_chip_setup()

                        if runtime.i2c_address is not None:
                            slave = WasmChipI2CSlave(runtime.i2c_address, runtime)
                            _i2c_slaves[runtime.i2c_address] = slave
                            sensor_data['i2c_addr'] = runtime.i2c_address
                            sensor_data['slave']    = slave
                            _log(f"[custom-chip] I2C slave registered at 0x{runtime.i2c_address:02x}")
                        if runtime.uart_config is not None:
                            _chip_uart_runtimes.append(runtime)
                            _log("[custom-chip] UART chip registered on UART0")
                        if runtime.spi_config is not None:
                            _chip_spi_runtimes.append(runtime)
                            _sync_cs_events()
                            _log("[custom-chip] SPI chip registered")
                        if runtime.has_pin_watches():
                            _chip_pin_watch_runtimes.append(runtime)
                            _log(f"[custom-chip] pin watches registered: {list(runtime._pin_watches.keys())}")
                        if (runtime.i2c_address is None and runtime.uart_config is None
                                and runtime.spi_config is None):
                            _log("[custom-chip] WASM loaded but no I2C/UART/SPI peripherals declared "
                                 "— chip is GPIO-only")
                        sensor_data['runtime'] = runtime
                    except Exception as e:
                        _log(f"[custom-chip] failed to load: {e!r}")
            _sensors[gpio] = sensor_data
    _sensors_ready.set()
    _log(f'_i2c_slaves registered: {list(_i2c_slaves.keys())}')

    _emit({'type': 'system', 'event': 'booted'})
    # Now that the initial components are registered, tell QEMU whether to
    # forward SPI CS toggles (only ePaper/custom-chip need them).
    _sync_cs_events()
    _log(f'QEMU started: machine={machine} firmware={firmware_path}')
    _log(f'QEMU args: {[a.decode() for a in args_list]}')

    # ── 6.5 Custom-chip timer thread ──────────────────────────────────────────
    # Wakes on each chip's next_timer_deadline. Acquires the QEMU IO-thread lock
    # before firing callbacks because vx_pin_write inside a timer would touch
    # picsimlab_set_pin which requires the lock.
    def _chip_timer_thread() -> None:
        while not _stopped.is_set():
            # Find the soonest deadline across all chips with active timers.
            soonest_ns: int | None = None
            for rt in list(_chip_timer_runtimes):
                d = rt.next_timer_deadline()
                if d is not None and (soonest_ns is None or d < soonest_ns):
                    soonest_ns = d
            if soonest_ns is None:
                # No active timers — sleep a bit and re-check.
                _stopped.wait(0.050)
                continue
            now_ns = time.monotonic_ns() - _t0_ref[0]
            wait_ns = max(0, soonest_ns - now_ns)
            if wait_ns > 0:
                _stopped.wait(wait_ns / 1e9)
                if _stopped.is_set():
                    break
            # Fire under the IO-thread lock so any pin_write the timer triggers is safe.
            if _lock_iothread:
                _lock_iothread(b'esp32_worker.py:chip_timer', 0)
            try:
                for rt in list(_chip_timer_runtimes):
                    try:
                        rt.fire_due_timers()
                    except Exception as e:
                        _log(f'[custom-chip timer] error: {e!r}')
            finally:
                if _unlock_iothread:
                    _unlock_iothread()

    _t0_ref = [time.monotonic_ns()]   # used so the timer thread can compute "now"
    _timer_t = threading.Thread(target=_chip_timer_thread, daemon=True, name='chip-timer')
    _timer_t.start()

    # ── 7. LEDC polling thread (100 ms interval) ──────────────────────────────

    def _ledc_poll_thread() -> None:
        # Track last-emitted duty to avoid flooding identical updates
        _last_duty = [0.0] * 16
        while not _stopped.wait(0.1):
            try:
                ptr = lib.qemu_picsimlab_get_internals(6)  # LEDC_CHANNEL_DUTY
                if ptr is None or ptr == 0:
                    continue
                arr = (ctypes.c_float * 16).from_address(ptr)
                # Reconcile the GPIO Matrix mirror first; any routing
                # changes since the last poll are emitted as
                # `gpio_routing` events so frontend's SignalRouter is
                # in sync before duty updates land.
                _refresh_signal_routing()
                for ch in range(16):
                    duty_pct = float(arr[ch])
                    if abs(duty_pct - _last_duty[ch]) < 0.01:
                        continue
                    _last_duty[ch] = duty_pct
                    if duty_pct > 0:
                        rounded = round(duty_pct, 2)
                        _emit({'type': 'ledc_duty',
                               'channel': ch,
                               'duty_pct': rounded})
            except Exception:
                pass

    threading.Thread(target=_ledc_poll_thread, daemon=True, name='ledc-poll').start()

    # ── 8. Command loop (main thread reads original stdin pipe) ───────────────

    for raw_line in os.fdopen(_orig_stdin_fd, 'r'):
        raw_line = raw_line.strip()
        if not raw_line:
            continue
        try:
            cmd = json.loads(raw_line)
        except Exception:
            continue

        c = cmd.get('cmd', '')

        if c == 'set_pin':
            # Identity pinmap: slot = gpio_num + 1
            lib.qemu_picsimlab_set_pin(int(cmd['pin']) + 1, int(cmd['value']))

        elif c == 'set_adc':
            raw_v = int(int(cmd['millivolts']) * 4095 / 3300)
            ch = int(cmd['channel'])
            clamped = max(0, min(4095, raw_v))
            lib.qemu_picsimlab_set_apin(ch, clamped)

        elif c == 'set_adc_raw':
            lib.qemu_picsimlab_set_apin(int(cmd['channel']),
                                        max(0, min(4095, int(cmd['raw']))))

        elif c == 'set_adc_waveform':
            # Push a periodic 12-bit waveform LUT to QEMU so the SAR ADC
            # peripheral can interpolate against its virtual clock on every
            # MMIO read. Matches the per-read fidelity of AVR/RP2040.
            #
            # libqemu-xtensa must export `qemu_picsimlab_set_apin_waveform`;
            # if not (older binary), silently downgrade to a single-sample
            # `set_apin` so circuits with waveforms still produce *something*.
            try:
                ch = int(cmd['channel'])
                b64 = cmd.get('samples_u12_b64', '') or ''
                period_ns = int(cmd.get('period_ns', 0))
                if b64 and period_ns > 0 and hasattr(lib, 'qemu_picsimlab_set_apin_waveform'):
                    raw = base64.b64decode(b64)
                    # Samples are little-endian uint16. Allocate a C buffer and
                    # hand QEMU a pointer; the waveform setter copies the data
                    # internally.
                    n = len(raw) // 2
                    if n > 0:
                        arr_type = ctypes.c_uint16 * n
                        arr = arr_type.from_buffer_copy(raw[:n * 2])
                        lib.qemu_picsimlab_set_apin_waveform(
                            ch,
                            arr,
                            ctypes.c_int(n),
                            ctypes.c_uint64(period_ns),
                        )
                else:
                    # Clear waveform: fall back to last-known DC value (no-op if
                    # the API isn't available — QEMU just keeps whatever was
                    # last written via `set_apin`).
                    if hasattr(lib, 'qemu_picsimlab_set_apin_waveform'):
                        lib.qemu_picsimlab_set_apin_waveform(
                            ch, None, ctypes.c_int(0), ctypes.c_uint64(0)
                        )
            except Exception as err:
                # Never let an ADC-waveform failure kill the worker — log to
                # stderr and keep the guest running with its last DC sample.
                print(f'[esp32_worker] set_adc_waveform failed: {err}',
                      file=sys.stderr, flush=True)

        elif c == 'uart_send':
            data = base64.b64decode(cmd['data'])
            buf  = (ctypes.c_uint8 * len(data))(*data)
            # Must hold the QEMU IO-thread lock: uart_receive injects a UART-RX
            # interrupt into the guest CPU and QEMU asserts the lock is held.
            if _lock_iothread:
                _lock_iothread(b'esp32_worker.py', 0)
            try:
                lib.qemu_picsimlab_uart_receive(
                    int(cmd.get('uart', 0)), buf, len(data)
                )
            finally:
                if _unlock_iothread:
                    _unlock_iothread()

        elif c == 'set_i2c_response':
            _i2c_responses[int(cmd['addr'])] = int(cmd['response']) & 0xFF

        elif c == 'set_spi_response':
            _spi_response[0] = int(cmd['response']) & 0xFF

        elif c == 'sensor_attach':
            gpio = int(cmd['pin'])
            sensor_type = cmd.get('sensor_type', '')
            with _sensors_lock:
                sensor_data: dict = {
                    'type': sensor_type,
                    **{k: v for k, v in cmd.items()
                       if k not in ('cmd', 'pin', 'sensor_type')},
                    'saw_low': False,
                    'responding': False,
                }
                if sensor_type == 'mpu6050':
                    i2c_addr = int(cmd.get('addr', 0x68))
                    slave = _MPU6050Slave(i2c_addr)
                    _i2c_slaves[i2c_addr] = slave
                    sensor_data['i2c_addr'] = i2c_addr
                    sensor_data['slave'] = slave
                elif sensor_type == 'bmp280':
                    i2c_addr = int(cmd.get('addr', 0x76))
                    slave = _BMP280Slave(i2c_addr)
                    _i2c_slaves[i2c_addr] = slave
                    sensor_data['i2c_addr'] = i2c_addr
                    sensor_data['slave'] = slave
                elif sensor_type in ('ds1307', 'ds3231'):
                    i2c_addr = int(cmd.get('addr', 0x68))
                    slave = _DS3231Slave() if sensor_type == 'ds3231' else _DS1307Slave()
                    _i2c_slaves[i2c_addr] = slave
                    sensor_data['i2c_addr'] = i2c_addr
                    sensor_data['slave'] = slave
                elif sensor_type in ('ssd1306', 'pcf8574'):
                    default_addr = 0x3C if sensor_type == 'ssd1306' else 0x27
                    i2c_addr = int(cmd.get('addr', default_addr))
                    sink = _I2CWriteSink(i2c_addr, _emit)
                    _i2c_slaves[i2c_addr] = sink
                    sensor_data['i2c_addr'] = i2c_addr
                    sensor_data['slave'] = sink
                elif sensor_type == 'epaper-ssd168x':
                    # Runtime registration of an SSD168x ePaper panel. Mirrors
                    # the `_init_sensors` branch above. Component-id keyed so
                    # multiple panels on the same board route correctly.
                    comp_id = str(cmd.get('component_id', f'epaper-{gpio}'))
                    width = int(cmd.get('width', 200))
                    height = int(cmd.get('height', 200))
                    refresh_ms = int(cmd.get('refresh_ms', 50))
                    busy_pin = int(cmd.get('busy_pin', -1))

                    def _flush_factory_rt(_comp_id=comp_id,
                                          _w=width, _h=height,
                                          _refresh=refresh_ms,
                                          _busy=busy_pin,
                                          _lib=lib):
                        def _on_flush(frame):
                            try:
                                frame_b64 = base64.b64encode(frame.pixels).decode('ascii')
                            except Exception:
                                return
                            # Emit FLAT (see the _init_sensors path) — the backend
                            # re-wraps under 'data', so a nested 'data' here would
                            # double-wrap and the frontend would never render.
                            _emit({
                                'type': 'epaper_update',
                                'component_id': _comp_id,
                                'width': _w,
                                'height': _h,
                                'frame_b64': frame_b64,
                                'refresh_ms': _refresh,
                            })
                            if _busy is not None and _busy >= 0:
                                try:
                                    _lib.qemu_picsimlab_set_pin(_busy + 1, 1)

                                    def _busy_low(_b=_busy):
                                        try:
                                            _lib.qemu_picsimlab_set_pin(_b + 1, 0)
                                        except Exception:
                                            pass

                                    threading.Timer(_refresh / 1000.0, _busy_low).start()
                                except Exception:
                                    pass
                        return _on_flush

                    ctl_family = str(cmd.get('controller_family', 'ssd168x'))
                    if ctl_family == 'uc8159c':
                        slave = _Uc8159cEpaperSlave(
                            component_id=comp_id, width=width, height=height,
                            on_flush=_flush_factory_rt(),
                        )
                    elif ctl_family == 'uc8179':
                        slave = _Uc8179EpaperSlave(
                            component_id=comp_id, width=width, height=height,
                            on_flush=_flush_factory_rt(),
                        )
                    else:
                        _is_bwr = 'bwr' in str(cmd.get('panel_kind', '')).lower()
                        slave = _Ssd168xEpaperSlave(
                            component_id=comp_id, width=width, height=height,
                            on_flush=_flush_factory_rt(), is_bwr=_is_bwr,
                        )
                    state = {
                        'slave': slave,
                        'dc_pin': int(cmd.get('dc_pin', -1)),
                        'cs_pin': int(cmd.get('cs_pin', -1)),
                        'rst_pin': int(cmd.get('rst_pin', -1)),
                        'busy_pin': busy_pin,
                        'cs_low': False,
                        'dc_high': False,
                        'refresh_ms': refresh_ms,
                        'controller_family': ctl_family,
                    }
                    _epaper_slaves[comp_id] = slave
                    _epaper_state[comp_id] = state
                    _sync_cs_events()
                    sensor_data['epaper_component_id'] = comp_id
                _sensors[gpio] = sensor_data
            _log(f'Sensor {sensor_type} attached on GPIO {gpio}')

        elif c == 'sensor_update':
            gpio = int(cmd['pin'])
            with _sensors_lock:
                sensor = _sensors.get(gpio)
                if sensor:
                    for k, v in cmd.items():
                        if k not in ('cmd', 'pin'):
                            sensor[k] = v
                    stype = sensor.get('type')
                    slave = sensor.get('slave')
                    if stype == 'mpu6050' and slave is not None:
                        slave.update(
                            accel_x=float(sensor.get('accelX', 0)),
                            accel_y=float(sensor.get('accelY', 0)),
                            accel_z=float(sensor.get('accelZ', 1)),
                            gyro_x =float(sensor.get('gyroX',  0)),
                            gyro_y =float(sensor.get('gyroY',  0)),
                            gyro_z =float(sensor.get('gyroZ',  0)),
                            temp   =float(sensor.get('temp',   25.0)),
                        )
                    elif stype == 'bmp280' and slave is not None:
                        slave.update(
                            temperature_c =float(sensor.get('temperature', 25.0)),
                            pressure_hpa  =float(sensor.get('pressure', 1013.25)),
                        )
                    elif stype == 'ds3231' and slave is not None:
                        slave.temperatureC = float(sensor.get('temperature', 25.0))

        elif c == 'sensor_detach':
            gpio = int(cmd['pin'])
            with _sensors_lock:
                sensor = _sensors.pop(gpio, None)
                if sensor and 'i2c_addr' in sensor:
                    _i2c_slaves.pop(sensor['i2c_addr'], None)
                if sensor and 'epaper_component_id' in sensor:
                    cid = sensor['epaper_component_id']
                    _epaper_slaves.pop(cid, None)
                    _epaper_state.pop(cid, None)
            _log(f'Sensor detached from GPIO {gpio}')

        # ── Cross-board I2C proxy slave ──────────────────────────────────
        # Installed by the frontend when an ESP32 board is wired across
        # the I2C bus to a peer board (Uno, Pico, …) that owns a virtual
        # device.  The frontend snapshots the device's register state and
        # pushes it here; we install a ProxySlave at the address so the
        # ESP32 firmware's Wire master reads succeed inside QEMU.
        elif c == 'proxy_i2c_register':
            i2c_addr = int(cmd.get('addr', 0)) & 0x7F
            try:
                regs = base64.b64decode(cmd.get('regs_b64', ''))
            except Exception as exc:
                _log(f'proxy_i2c_register: bad base64: {exc}')
                regs = b''
            # Pass _emit so writes from the ESP32 firmware get forwarded
            # back to the frontend as `proxy_i2c_complete` events.  The
            # frontend then replays the byte sequence on the actual
            # peer I2CDevice so its state (PCF8574 latch, SSD1306
            # GDDRAM, memory device registers …) stays in sync.
            _i2c_slaves[i2c_addr] = _ProxySlave(i2c_addr, regs, emit_fn=_emit)
            _log(f'proxy_i2c registered at 0x{i2c_addr:02x} ({len(regs)} bytes)')

        elif c == 'proxy_i2c_update':
            i2c_addr = int(cmd.get('addr', 0)) & 0x7F
            try:
                regs = base64.b64decode(cmd.get('regs_b64', ''))
            except Exception as exc:
                _log(f'proxy_i2c_update: bad base64: {exc}')
                regs = b''
            slave = _i2c_slaves.get(i2c_addr)
            if slave is not None and hasattr(slave, 'update_registers'):
                slave.update_registers(regs)
                _log(f'proxy_i2c updated at 0x{i2c_addr:02x} ({len(regs)} bytes)')

        elif c == 'proxy_i2c_unregister':
            i2c_addr = int(cmd.get('addr', 0)) & 0x7F
            popped = _i2c_slaves.pop(i2c_addr, None)
            if popped is not None:
                _log(f'proxy_i2c unregistered at 0x{i2c_addr:02x}')

        # ── ESP32-CAM frame injection ────────────────────────────────────
        # Pushes a JPEG (or other format) into the QEMU OV2640 device's
        # frame buffer via the velxio_push_camera_frame() symbol exported
        # by the rebuilt libqemu-xtensa. Feature-detected at runtime so
        # this branch is a no-op on a stock library.
        elif c == 'camera_attach':
            _log('camera_attach received (frame source ready)')

        elif c == 'camera_frame':
            try:
                payload = base64.b64decode(cmd.get('b64', ''))
            except Exception as exc:
                _log(f'camera_frame: bad base64: {exc}')
                payload = b''
            # Throttled trace — log every 30th frame so noisy streaming
            # leaves a footprint in the lib_manager log without spamming.
            _camera_frame_count[0] += 1
            n = _camera_frame_count[0]
            if n == 1 or n % 30 == 0:
                _log(f'camera_frame #{n} received ({len(payload)} bytes payload)')
            if payload:
                _push_camera_frame(payload)

        elif c == 'camera_detach':
            _push_camera_frame(b'')   # NULL/0 detaches in the C side

        elif c == 'stop':
            _stopped.set()
            # Request a clean shutdown via the QEMU main-loop thread.
            # qemu_system_shutdown_request() is safe to call from any thread:
            # it posts an event to the main loop which then tears down block
            # devices in the correct AIO context, avoiding the
            # "blk_exp_close_all_type: in_aio_context_home_thread" assertion
            # that fires when qemu_cleanup() is called directly from here.
            if _shutdown_request:
                try:
                    _shutdown_request(3)   # SHUTDOWN_CAUSE_HOST_SIGNAL = 3
                except Exception:
                    pass
            qemu_t.join(timeout=5.0)
            # Clean up temp firmware file
            if firmware_path:
                try:
                    os.unlink(firmware_path)
                except OSError:
                    pass
            os._exit(0)


if __name__ == '__main__':
    main()