feat(dispatcher): stream_frame codec + Dispatcher multiplexer; OTA rides them

.github/workflows/build.yml

@@ -119,6 +119,8 @@ jobs:
           target: esp32s3
         - path: 'components/csv/example'
           target: esp32
+        - path: 'components/dispatcher/example'
+          target: esp32
         - path: 'components/display_drivers/example'
           target: esp32
         - path: 'components/dns_server/example'
@@ -285,6 +287,8 @@ jobs:
           target: esp32s3
         - path: 'components/state_machine/example'
           target: esp32
+        - path: 'components/stream_frame/example'
+          target: esp32
         - path: 'components/sx126x/example'
           target: esp32s3
         - path: 'components/t-deck/example'

.github/workflows/upload_components.yml

@@ -72,9 +72,15 @@ jobs:
             components/codec
             components/color
             components/controller
+            # stream_frame is intentionally listed here (out of alphabetical
+            # order) ahead of its first-time dependents coredump / dispatcher /
+            # ota, so it is uploaded before they try to resolve it from the
+            # registry (see the note above).
+            components/stream_frame
             components/coredump
             components/cst816
             components/csv
+            components/dispatcher
             components/display
             components/display_drivers
             components/dns_server

components/bldc_haptics/example/CMakeLists.txt

@@ -43,14 +43,15 @@ set(EXTRA_COMPONENT_DIRS
   "../../../components/ota"
   "../../../components/pid"
   "../../../components/spi"
+  "../../../components/stream_frame"
   "../../../components/task"
   "../../../components/timer"
   "../../../components/usb_device"
 )
 
 set(
   COMPONENTS
-  "main esptool_py bldc_driver bldc_haptics bldc_motor i2c motorgo-axis motorgo-mini mt6701 ota task usb_device esp_tinyusb"
+  "main esptool_py bldc_driver bldc_haptics bldc_motor i2c motorgo-axis motorgo-mini mt6701 ota stream_frame task usb_device esp_tinyusb"
   CACHE STRING
   "List of components to include"
   )

components/bldc_haptics/example/PROTOCOL.md

@@ -12,20 +12,26 @@ for the reference host implementation.
 
 ## Framing
 
-Identical to the espp `ota` component's stream framing
-(`components/ota/include/detail/ota_stream_protocol.hpp` is the authoritative
-spec). All multi-byte fields are **little-endian**:
+Uses the espp `stream_frame` v2 codec
+(`components/stream_frame/include/stream_frame.hpp` is the authoritative spec).
+The whole haptics protocol is dispatcher **module 2**. All multi-byte fields are
+**little-endian**:
 
 ```
-[magic u16 = 0x4F54 "OT"] [type u8] [len u32] [payload: len bytes] [crc32 u32]
+[magic u16 = 0x4F54 "OT"] [flags u8] [module u8] [type u8] [len u32] [payload: len bytes] [crc32 u32]
 ```
 
 - `magic`: u16 `0x4F54`; on the wire the bytes are `0x54 'T'` then `0x4F 'O'`.
+- `flags`: `bit0` = reply (`0` = host→device request, `1` = device→host
+  reply/event); `bits 4-7` = protocol version = `1`. So a request byte is
+  `0x10` and a reply/telemetry byte is `0x11`. Request types (`0x0_`/`0x1_`)
+  clear the reply bit; reply/telemetry types (`0x8_`/`0x9_`) set it.
+- `module`: `u8` dispatcher module — **2** for the entire haptics protocol.
 - `type`: message type (tables below).
 - `len`: payload length, capped at **4096** bytes per frame; receivers reject
   and resynchronize past any frame whose length field exceeds the cap.
 - `crc32`: standard zlib CRC-32 (poly `0xEDB88320` reflected, init/final xor
-  `0xFFFFFFFF`) over `magic..payload` (i.e. the 7 header bytes + payload).
+  `0xFFFFFFFF`) over `magic..payload` (i.e. the 9 header bytes + payload).
   Golden check value: `crc32("123456789") == 0xCBF43926`.
 
 Receivers parse incrementally and resynchronize on bad magic / oversized
@@ -69,7 +75,9 @@ The device suspends telemetry while an OTA session is active.
 
 Notes:
 
-- **OTA** semantics are identical to the espp `ota` example: `OTA_BEGIN` erases
+- **OTA** semantics match the espp `ota` example, but the frames are **not**
+  byte-compatible: the haptics OTA subset rides dispatcher module 2, whereas the
+  `ota` example / `ota_console.html` use module 0. `OTA_BEGIN` erases
   the next OTA app partition (can take several seconds — use a generous
   timeout), `OTA_DATA` streams image bytes, `OTA_FINISH` validates the complete
   image (structure + appended SHA-256) and sets it as the boot partition, then

components/bldc_haptics/example/README.md

@@ -100,9 +100,9 @@ otadata.
    image crashes before that, the bootloader automatically rolls back to the
    previous slot on the next reset.
 
-The same OTA transfer can also be driven from the generic espp OTA console
-(`components/ota/web/ota_console.html`), since the OTA subset of the protocol
-is byte-compatible with the espp `ota` example.
+The OTA subset is part of the haptics protocol on **dispatcher module 2**, so it
+is *not* interchangeable with the generic espp `ota` example (which is module 0)
+— use this example's own web console for OTA here.
 
 ## Example Behaviors
 
@@ -159,8 +159,8 @@ components:
 * `espp::UsbDevice` — native USB vendor interface with WebUSB + MS OS 2.0
   descriptors (driverless browser access)
 * `espp::Ota` — transport-agnostic OTA engine fed from the USB protocol
-* The `ota_stream` framing (`components/ota/include/detail/ota_stream_protocol.hpp`)
-  reused as the framing layer for the haptics protocol
+* The `stream_frame` codec (`components/stream_frame/include/stream_frame.hpp`)
+  as the framing layer for the haptics protocol (module 2)
   (see [PROTOCOL.md](./PROTOCOL.md))
 
 You combine the `Mt6701` and `BldcDriver` together when creating the `BldcMotor`

components/bldc_haptics/example/main/CMakeLists.txt

@@ -1,4 +1,4 @@
 idf_component_register(SRC_DIRS "."
                        INCLUDE_DIRS "."
                        REQUIRES bldc_driver bldc_haptics bldc_motor i2c motorgo-axis motorgo-mini
-                                mt6701 ota task usb_device esp_tinyusb esp_timer espcoredump)
+                                mt6701 ota stream_frame task usb_device esp_tinyusb esp_timer espcoredump)

components/bldc_haptics/example/main/bldc_haptics_example.cpp

@@ -454,10 +454,19 @@ extern "C" void app_main(void) {
   proto::stream::StreamParser parser;
   bool restart_pending = false;
 
-  auto reply_ok = [&](uint32_t value) { usb_send(proto::stream::make_ok(value)); };
+  // Build replies via proto::build so they carry the haptics module (2) + reply
+  // flag — NOT the OTA make_ok/make_error (those are OTA module 0).
+  auto reply_ok = [&](uint32_t value) {
+    std::vector<uint8_t> payload;
+    proto::put_u32(payload, value);
+    usb_send(proto::build(proto::Msg::Ok, payload));
+  };
   auto reply_error = [&](const std::error_code &err, const std::string &context) {
-    usb_send(proto::stream::make_error(static_cast<uint32_t>(err.value()),
-                                       context + ": " + err.message()));
+    std::vector<uint8_t> payload;
+    proto::put_u32(payload, static_cast<uint32_t>(err.value()));
+    const std::string message = context + ": " + err.message();
+    payload.insert(payload.end(), message.begin(), message.end());
+    usb_send(proto::build(proto::Msg::Error, payload));
   };
   auto reply_errc = [&](std::errc errc, const std::string &context) {
     reply_error(std::make_error_code(errc), context);
@@ -688,14 +697,17 @@ extern "C" void app_main(void) {
            std::error_code abort_ec;
            ota.abort(abort_ec);
            parser.reset();
-           usb_send(proto::stream::make_error(
-               static_cast<uint32_t>(std::make_error_code(std::errc::no_buffer_space).value()),
-               "RX overflow: frames dropped -- wait for OK replies between frames"));
+           reply_errc(std::errc::no_buffer_space,
+                      "RX overflow: frames dropped -- wait for OK replies between frames");
            return false; // dropped chunks are gone; skip parse
          }
          for (const auto &chunk : chunks)
            for (const auto &frame : parser.feed(chunk))
-             handle_frame(frame);
+             // this protocol's REQUESTS only: ignore other modules and
+             // reply-flagged frames (the device answers requests; a reply-typed
+             // echo must not re-enter the request handler)
+             if (frame.module == proto::kModule && !frame.is_reply())
+               handle_frame(frame);
          if (restart_pending) {
            // give the final OK reply time to reach the host
            std::this_thread::sleep_for(750ms);

components/bldc_haptics/example/main/haptics_usb_protocol.hpp

@@ -1,13 +1,16 @@
 #pragma once
 
 // espp BLDC haptics USB protocol — message ids + payload helpers layered on the
-// espp `ota_stream` framing (magic "OT" + type u8 + len u32 + payload + CRC-32,
-// all little-endian; see components/ota/include/detail/ota_stream_protocol.hpp
-// for the authoritative framing spec and ../PROTOCOL.md next to this example
-// for the full haptics wire protocol).
+// espp `stream_frame` codec (magic "OT" + flags u8 + module u8 + type u8 + len
+// u32 + payload + CRC-32, all little-endian; see
+// components/stream_frame/include/stream_frame.hpp for the authoritative framing
+// spec and ../PROTOCOL.md next to this example for the full haptics wire
+// protocol).
 //
-// The message-type space is partitioned so the OTA subset stays byte-compatible
-// with the espp `ota` example / ota_console.html web app:
+// The whole protocol occupies dispatcher MODULE 2. The `type` byte carries the
+// message id below; request types (host->device) clear the frame reply flag and
+// reply/telemetry types (0x8_/0x9_, host<-device) set it (build() derives it
+// from the type's high bit):
 //   0x01..0x04  host -> device  OTA (BEGIN / DATA / FINISH / ABORT)
 //   0x10..0x2F  host -> device  haptics commands
 //   0x81..0x8F  device -> host  generic + OTA replies (OK / ERROR / PROGRESS)
@@ -22,15 +25,23 @@
 #include <vector>
 
 #include "detail/ota_stream_protocol.hpp"
+#include "stream_frame.hpp"
 
 namespace haptics_proto {
 
+// The ota_stream facade re-exports the stream_frame codec (StreamParser / Frame
+// / put_* / get_* / parse_u32_payload); build() below uses the generic
+// stream_frame builder directly so it can set this protocol's module + reply
+// flag.
 namespace stream = espp::detail::ota_stream;
 
+/// Dispatcher module id owned by the haptics protocol (the frame `module` byte).
+static constexpr uint8_t kModule = 2;
+
 /// Protocol version reported in the INFO reply.
 static constexpr uint8_t kProtocolVersion = 1;
 
-/// Message types carried in the ota_stream frame `type` byte.
+/// Message types carried in the frame `type` byte (within module 2).
 enum class Msg : uint8_t {
   // --- OTA subset (identical semantics to the espp ota example) -------------
   OtaBegin = 0x01,  ///< host->dev: u32 image_size (0 = unknown / streaming)
@@ -100,9 +111,11 @@ inline std::optional<float> get_f32_at(std::span<const uint8_t> bytes, size_t of
   return std::bit_cast<float>(get_u32(bytes.subspan(offset)));
 }
 
-/// Build a frame for any haptics-protocol message type.
+/// Build a frame for any haptics-protocol message type (module 2; the reply flag
+/// is set for reply/telemetry types, whose ids have the high bit set).
 inline std::vector<uint8_t> build(Msg type, std::span<const uint8_t> payload = {}) {
-  return stream::build_frame(static_cast<stream::MessageType>(type), payload);
+  const bool reply = (static_cast<uint8_t>(type) & 0x80) != 0;
+  return espp::stream_frame::build_frame(reply, kModule, static_cast<uint8_t>(type), payload);
 }
 
 /// Status flag bits (Status + Telemetry `flags` byte).

components/bldc_haptics/example/webapp/index.html

@@ -10,13 +10,18 @@
     ==================================
     Single-file, fully offline, dependency-free browser console for the espp
     bldc_haptics USB example. Speaks the framed vendor-interface protocol
-    documented in ../PROTOCOL.md:
+    documented in ../PROTOCOL.md (dispatcher framing v2):
 
-      [magic u16 = 0x4F54 "OT"][type u8][len u32][payload...][crc32 u32]
+      [magic u16 = 0x4F54 "OT"][flags u8][module u8][type u8][len u32][payload...][crc32 u32]
 
+    - flags u8: bit0 = reply (0 host->device request, 1 device->host
+      reply/event); bits 4-7 = protocol version = 1. Request byte = 0x10,
+      reply = 0x11.
+    - module u8: the whole haptics protocol is dispatcher module 2.
     - CRC-32 is the standard zlib CRC (poly 0xEDB88320 reflected, init/final
-      0xFFFFFFFF) over magic..payload; check value crc32("123456789")
-      == 0xCBF43926. Payloads are capped at 4096 bytes per frame.
+      0xFFFFFFFF) over the 9-byte header + payload; check value
+      crc32("123456789") == 0xCBF43926. Payloads are capped at 4096 bytes
+      per frame.
     - Commands are serialized (one in flight, wait for the reply); TELEMETRY
       (0x93) and OTA_PROGRESS (0x83) frames arrive unsolicited and are
       dispatched out-of-band by the RX pump.
@@ -288,7 +293,8 @@ <h2>Log</h2>
 
     <footer>
       Speaks the espp bldc_haptics example's vendor-interface protocol (frame:
-      magic "OT" + type + len + payload + CRC-32; see PROTOCOL.md). After an
+      magic "OT" + flags + module + type + len + payload + CRC-32, dispatcher
+      module 2; see PROTOCOL.md). After an
       OTA FINISH the device validates the image (SHA-256), sets the boot
       partition and restarts; with bootloader rollback enabled the new app must
       mark itself valid or the device rolls back.
@@ -304,9 +310,15 @@ <h2>Log</h2>
     const DEFAULT_VID = 0x1209, DEFAULT_PID = 0x0d34;
     const VENDOR_CLASS = 0xFF;
     const MAGIC0 = 0x54, MAGIC1 = 0x4F;      // u16 0x4F54 "OT", little-endian on the wire
-    const HEADER_SIZE = 7, CRC_SIZE = 4;
+    // v2 framing: [magic u16][flags u8][module u8][type u8][len u32] => 9-byte header.
+    const HEADER_SIZE = 9, CRC_SIZE = 4;
     const MAX_PAYLOAD = 4096;
     const MAX_FRAME = HEADER_SIZE + MAX_PAYLOAD + CRC_SIZE;
+    const MODULE_HAPTICS = 2;                 // the whole haptics protocol is dispatcher module 2
+    const FLAGS_VERSION = 1;                  // protocol version lives in flags bits 4-7
+    const FLAGS_REQUEST = (FLAGS_VERSION << 4) | 0; // host->device request: 0x10 (reply bit 0)
+    const FLAGS_REPLY_BIT = 0x01;            // bit0 set on device->host replies/telemetry
+    const FLAG_CORRELATION = 0x02;           // bit1: optional u16 correlation id present after `type`
     const TYPE = {
       OTA_BEGIN: 0x01, OTA_DATA: 0x02, OTA_FINISH: 0x03, OTA_ABORT: 0x04,
       GET_INFO: 0x10, GET_STATUS: 0x11, GET_MODES: 0x12, SET_MODE: 0x13,
@@ -390,14 +402,18 @@ <h2>Log</h2>
     // ===================================================================
     // Frame building & incremental parsing (mirrors ota_stream_protocol.hpp)
     // ===================================================================
+    // This app only ever sends haptics requests, so module is hardcoded to
+    // MODULE_HAPTICS and flags to FLAGS_REQUEST (reply bit clear).
     function buildFrame(type, payload) {
       payload = payload || new Uint8Array(0);
       if (payload.length > MAX_PAYLOAD) throw new Error("payload exceeds " + MAX_PAYLOAD + " bytes");
       const frame = new Uint8Array(HEADER_SIZE + payload.length + CRC_SIZE);
       const view = new DataView(frame.buffer);
-      view.setUint16(0, 0x4F54, true);
-      frame[2] = type;
-      view.setUint32(3, payload.length, true);
+      view.setUint16(0, 0x4F54, true);   // magic 'T''O' on the wire
+      frame[2] = FLAGS_REQUEST;          // flags: version 1, request (reply bit 0)
+      frame[3] = MODULE_HAPTICS;         // module: haptics = 2
+      frame[4] = type;                   // message type
+      view.setUint32(5, payload.length, true);
       frame.set(payload, HEADER_SIZE);
       view.setUint32(HEADER_SIZE + payload.length,
                      crc32(frame.subarray(0, HEADER_SIZE + payload.length)), true);
@@ -417,17 +433,27 @@ <h2>Log</h2>
                  !(merged[pos] === MAGIC0 && (pos + 1 >= merged.length || merged[pos + 1] === MAGIC1))) {
             pos++;
           }
-          if (merged.length - pos < HEADER_SIZE) break;
+          if (merged.length - pos < 3) break; // need magic+flags before header size is known
+          const flags = merged[pos + 2];
+          // v2: flags bit1 => optional u16 correlation id sits after `type`
+          const ext = (flags & FLAG_CORRELATION) ? 2 : 0;
+          const headerSize = HEADER_SIZE + ext;
+          if (merged.length - pos < headerSize) break;
           const view = new DataView(merged.buffer, merged.byteOffset + pos);
-          const len = view.getUint32(3, true);
+          const len = view.getUint32(5 + ext, true); // v2: len u32 follows magic+flags+module+type(+correlation)
           if (len > MAX_PAYLOAD) { pos++; continue; }
-          const total = HEADER_SIZE + len + CRC_SIZE;
+          const total = headerSize + len + CRC_SIZE;
           if (merged.length - pos < total) break;
-          const expected = view.getUint32(HEADER_SIZE + len, true);
-          const actual = crc32(merged.subarray(pos, pos + HEADER_SIZE + len));
+          const expected = view.getUint32(headerSize + len, true);
+          const actual = crc32(merged.subarray(pos, pos + headerSize + len));
           if (actual !== expected) { pos++; continue; }
-          frames.push({ type: merged[pos + 2],
-                        payload: merged.slice(pos + HEADER_SIZE, pos + HEADER_SIZE + len) });
+          const correlation = ext ? (merged[pos + 5] | (merged[pos + 6] << 8)) : null;
+          frames.push({ flags,
+                        reply: (flags & FLAGS_REPLY_BIT) !== 0,
+                        correlation,
+                        module: merged[pos + 3],
+                        type: merged[pos + 4],
+                        payload: merged.slice(pos + headerSize, pos + headerSize + len) });
           pos += total;
         }
         this.buf = merged.slice(pos);
@@ -652,6 +678,16 @@ <h2>Log</h2>
     }
 
     function dispatchFrame(frame) {
+      // Every device->host haptics frame (replies + telemetry) is dispatcher
+      // module 2 with the reply flag set; ignore anything else that shares the
+      // vendor pipe (other modules, or a request-flagged frame echoed back).
+      if (frame.module !== MODULE_HAPTICS || !frame.reply) {
+        if (els.logFrames.checked) {
+          logLine("err", "Ignoring non-haptics-reply frame (module 0x" + frame.module.toString(16) +
+                  ", flags 0x" + frame.flags.toString(16) + ", type 0x" + frame.type.toString(16) + ")");
+        }
+        return;
+      }
       if (els.logFrames.checked && frame.type !== TYPE.TELEMETRY) {
         logLine("rx", (TYPE_NAME[frame.type] || ("0x" + frame.type.toString(16))) + " " + hex(frame.payload));
       }

components/coredump/CMakeLists.txt

@@ -5,9 +5,9 @@
 #   esp_partition — esp_partition.h (raw image reads)
 #   esp_system    — esp_system.h (esp_reset_reason)
 #   format        — format.hpp (fmt::format in the report formatting)
-#   ota           — detail/ota_stream_protocol.hpp (the shared stream framing
-#                   used by coredump_service.hpp; header-only, ESP-free)
+#   stream_frame  — stream_frame.hpp (the shared frame codec used by
+#                   coredump_service.hpp; header-only, ESP-free)
 idf_component_register(
   INCLUDE_DIRS "include"
-  REQUIRES base_component espcoredump esp_partition esp_system format ota
+  REQUIRES base_component espcoredump esp_partition esp_system format stream_frame
 )