Reserve USB KISS framing before normal parsers consume it

.github/workflows/run-unit-tests.yml

@@ -582,6 +582,8 @@ jobs:
           python3 -B test/test_host_kiss_infrastructure.py -v
           python3 -B test/test_host_kiss.py -v
           python3 -B test/test_host_kiss_usb.py -v
+          python3 -B test/test_host_kiss_ingress.py -v
+          python3 -B test/test_host_kiss_capabilities.py -v
           python3 -B test/test_ethernet_kiss.py -v
           python3 -B test/test_ethernet_radio.py -v
           python3 -B test/test_companion_ethernet.py -v

build_legacy.sh

@@ -4809,6 +4809,25 @@ sys.exit(0 if required else 3)
   esac
 }
 
+requires_host_kiss_code() {
+  local env_name=$1
+  local pio_env_name=$2
+  local env_platform=$3
+  local decision_status=0
+  local -a arguments=(--required --env "$pio_env_name" --platform "$env_platform")
+  is_kiss_modem_target "$env_name" && return 1
+  if is_companion_radio_full_target "$env_name"; then
+    arguments+=(--full-companion)
+  fi
+  python3 scripts/host_kiss_capabilities.py "${arguments[@]}" \
+    <<<"$PIO_CONFIG_JSON" || decision_status=$?
+  case "$decision_status" in
+    0) return 0 ;;
+    3) return 1 ;;
+    *) return 2 ;;
+  esac
+}
+
 write_build_capability_manifest() {
   local env_name=$1
   local env_platform=$2
@@ -4846,6 +4865,14 @@ write_build_capability_manifest() {
     echo "Cannot determine the Full Companion Meshtext capability contract." >&2
     return "$meshtext_status"
   fi
+  local host_kiss_status=0
+  requires_host_kiss_code "$env_name" "$pio_env_name" "$env_platform" || host_kiss_status=$?
+  if [ "$host_kiss_status" -eq 0 ]; then
+    checker_args+=(--require-host-kiss)
+  elif [ "$host_kiss_status" -ne 1 ]; then
+    echo "Cannot determine the host KISS capability contract." >&2
+    return "$host_kiss_status"
+  fi
   if [ "$env_platform" = "ESP32_PLATFORM" ]; then
     checker_args+=(--firmware-bin "${build_output_dir}/firmware.bin"
                    --partitions "${build_output_dir}/partitions.bin")

docs/kiss_modem_protocol.md

@@ -36,7 +36,9 @@ If the initial radio2 change is still pending, the connection waits and sends
 READY when radio2 becomes available; no reconnect is needed. DATA received
 while waiting is discarded, so hosts should wait for the initial READY.
 
-USB remains the normal CLI until an initial KISS FEND (`C0`). During the raw
+USB remains the normal CLI until an initial KISS FEND (`C0`). A delimiter
+inside an unfinished Companion frame or an active mOTA session keeps its
+original protocol ownership. During the raw
 session, USB Companion framing and text/debug output are parked. Send the
 standard KISS RETURN frame (`C0 FF C0`) to leave the session, then issue CLI
 commands normally. A detected USB disconnect/reset also ends the session.

examples/companion_radio/MyMesh.cpp

@@ -11182,6 +11182,7 @@ void MyMesh::checkCLIRescueCmd() {
   }
   unsigned budget = sizeof(cli_command);
   while (budget-- && input.available()) {
+    if (mesh::hostkiss::usbInputReserved(input.peek())) return;
     char c = input.read();
     output.print(c);  // echo
     if (c == '\r' || c == '\n') {

examples/companion_radio/main.cpp

@@ -1113,6 +1113,15 @@ static void resetUsbTerminalHostSession() {
 // Park this one Companion transport; BLE/WiFi and the primary mesh stay live.
 static bool serviceCompanionHostKissUsb() {
 #if MESH_ENABLE_HOST_KISS
+  // A FEND inside an already-started Companion frame is payload, not a
+  // request to change protocols. The idle ingress gate handles the next one.
+  if (!mesh::hostkiss::usbActive()) {
+    if (usb_serial_interface.isReceivingFrame()) return false;
+#if COMPANION_FEATURE_USB_MOTA_SOURCE
+    if (usb_mota_mode
+        || mesh::ota::OtaContext::serialFolderSource().hasPendingResponse()) return false;
+#endif
+  }
   const bool claimed = mesh::hostkiss::serviceUsb();
   if (claimed && mesh::hostkiss::usbActive() && !host_kiss_usb_parked) {
     host_kiss_usb_restore_enabled = usb_serial_interface.isEnabled();
@@ -1320,6 +1329,7 @@ static void serviceUsbTerminal() {
   bool blank_input = false;
   int pending = usb_input.available();
   while (pending-- > 0) {
+    if (mesh::hostkiss::usbInputReserved(usb_input.peek())) return;
     // A leftover CR/LF (for example after mOTA shutdown) must not hide a
     // following frame or cause a banner to poison its first binary reply.
     if (usb_terminal_line_len == 0 && !usb_terminal_discard_line) {
@@ -1488,7 +1498,10 @@ static void serviceUsbLoggingOwnership(bool logging_enabled) {
     // the stream quarantined forever. Let its bounded consumer drain instead.
     Stream& input = mesh::usbCompanionPort();
     int pending = response_pending ? 0 : input.available();
-    while (pending-- > 0) input.read();
+    while (pending-- > 0) {
+      if (mesh::hostkiss::usbInputReserved(input.peek())) return;
+      input.read();
+    }
     usb_logging_network_parked = false;
     usb_serial_interface.setPassthroughMode(
         the_mesh.isTerminalMode() || response_pending);
@@ -1610,7 +1623,10 @@ static void expireUsbBinaryStartupProbeBeforeDispatch() {
   // already queued for that expired attempt so they cannot begin a new probe.
   Stream& usb_input = mesh::usbCompanionPort();
   int pending = usb_input.available();
-  while (pending-- > 0) usb_input.read();
+  while (pending-- > 0) {
+    if (mesh::hostkiss::usbInputReserved(usb_input.peek())) return;
+    usb_input.read();
+  }
   enterUsbTerminalMode();
 }
 #endif
@@ -4141,6 +4157,11 @@ void setup() {
   // keep frames intact and pace the contact stream when the host is slow
   usb_serial_interface.enableFlowControl(true);
   usb_serial_interface.setReceiveFrameCheck(acceptUsbBinaryStartupFrame);
+#if MESH_ENABLE_HOST_KISS
+  usb_serial_interface.setIdleReceiveCheck([](int next_byte) {
+    return !mesh::hostkiss::usbInputReserved(next_byte);
+  });
+#endif
 #if defined(ESP32) && defined(ARDUINO_USB_MODE) && ARDUINO_USB_MODE == 1 \
     && defined(ARDUINO_USB_CDC_ON_BOOT) && ARDUINO_USB_CDC_ON_BOOT
   // The ESP32 USB-Serial-JTAG peripheral (HWCDC) has no DTR concept at all.

examples/simple_repeater/main.cpp

@@ -526,6 +526,10 @@ static void __attribute__((noinline)) serviceCommandInterfaces() {
   bool overlong_line_complete = false;
   size_t read_budget = 256;
   while (usb_ready && read_budget-- > 0 && console.available()) {
+#if MESH_ENABLE_HOST_KISS
+    // Bytes may arrive after the loop-top KISS poll. Preserve its first FEND.
+    if (mesh::hostkiss::usbInputReserved(console.peek())) break;
+#endif
     char c = console.read();
     if (c == '\n') continue;
     console.print(c);

examples/simple_room_server/main.cpp

@@ -392,6 +392,10 @@ void loop() {
     len = 0;
   }
   while (usb_ready && console.available() && len < sizeof(command)-1) {
+#if MESH_ENABLE_HOST_KISS
+    // Bytes may arrive after the loop-top KISS poll. Preserve its first FEND.
+    if (mesh::hostkiss::usbInputReserved(console.peek())) break;
+#endif
     char c = console.read();
     if (c != '\n') {
       command[len++] = c;

examples/simple_sensor/main.cpp

@@ -375,6 +375,10 @@ void loop() {
     len = 0;
   }
   while (usb_ready && console.available() && len < sizeof(command)-1) {
+#if MESH_ENABLE_HOST_KISS
+    // Bytes may arrive after the loop-top KISS poll. Preserve its first FEND.
+    if (mesh::hostkiss::usbInputReserved(console.peek())) break;
+#endif
     char c = console.read();
     if (c != '\n') {
       command[len++] = c;

scripts/check_firmware_capabilities.py

@@ -12,6 +12,10 @@ import struct
 import zipfile
 import zlib
 from rp2040_application import read_application as read_rp2040_application
+from host_kiss_capabilities import (CAPABILITY as HOST_KISS_CAPABILITY,
+    USB_CAPABILITY as HOST_KISS_USB_CAPABILITY, TCP_CAPABILITY as HOST_KISS_TCP_CAPABILITY,
+    COMMAND_MARKERS as HOST_KISS_COMMAND_MARKERS, TCP_MARKER as HOST_KISS_TCP_MARKER,
+    evidence as host_kiss_evidence)
 from package_meshtext_license import (CAPABILITY as MESHTEXT_CAPABILITY,
     meshtext_evidence, notice_bytes as meshtext_notice_bytes,
     notice_record as meshtext_notice_record)
@@ -81,6 +85,10 @@ def parse_args() -> argparse.Namespace:
         "--require-meshtext", action="store_true",
         help="Require linked Meshtext encode/decode and Companion raw-frame transport implementations.",
     )
+    parser.add_argument(
+        "--require-host-kiss", action="store_true",
+        help="Require linked KISS framing/Dispatcher transport and packaged USB/TCP service commands.",
+    )
     parser.add_argument("--firmware-bin", type=Path)
     parser.add_argument("--partitions", type=Path)
     parser.add_argument("--dfu-package", type=Path)
@@ -231,12 +239,34 @@ def main() -> int:
                        "evidence": "Meshtext encode/decode, Companion send/receive, and Dispatcher raw-frame queue implementations",
                        "present": False, "source": "linked ELF symbols"})
     application = b""
-    if args.expect_application:
+    if args.expect_application or args.require_host_kiss:
         try:
             application = read_application(args)
         except (OSError, ValueError, KeyError, zipfile.BadZipFile) as exc:
             print(f"capability check: cannot read packaged application: {exc}", file=sys.stderr)
             malformed = True
+    host_kiss_symbols = []
+    host_kiss_verified = False
+    if args.require_host_kiss:
+        try:
+            host_kiss_symbols = host_kiss_evidence(args.image)
+        except (OSError, ValueError) as exc:
+            print(f"capability check: cannot qualify linked host KISS code: {exc}", file=sys.stderr)
+            malformed = True
+        checks.append({"capability": HOST_KISS_CAPABILITY,
+                       "evidence": "; ".join(host_kiss_symbols) if host_kiss_symbols else
+                           "executable KISS frame decoder/encoder and Dispatcher raw-frame queue",
+                       "present": bool(host_kiss_symbols), "source": "linked ELF symbols"})
+        usb_present = all(marker in application for marker in HOST_KISS_COMMAND_MARKERS)
+        checks.append({"capability": HOST_KISS_USB_CAPABILITY,
+                       "evidence": "; ".join(marker.decode("ascii") for marker in HOST_KISS_COMMAND_MARKERS),
+                       "present": usb_present, "source": "packaged application"})
+        tcp_present = args.platform != "ESP32_PLATFORM" or HOST_KISS_TCP_MARKER in application
+        if args.platform == "ESP32_PLATFORM":
+            checks.append({"capability": HOST_KISS_TCP_CAPABILITY,
+                           "evidence": HOST_KISS_TCP_MARKER.decode("ascii"),
+                           "present": tcp_present, "source": "packaged application"})
+        host_kiss_verified = bool(host_kiss_symbols) and usb_present and tcp_present
     expectations = [(value, image, "linked image") for value in args.expect]
     expectations += [(value, application, "packaged application") for value in args.expect_application]
     gzip_assets = linked_gzip_assets(image) if args.expect_gzip else []
@@ -269,6 +299,11 @@ def main() -> int:
         # Neither a requested capability nor CLI strings prove the optional
         # codec and its complete Companion transport survived linking.
         capabilities = [value for value in capabilities if value != MESHTEXT_CAPABILITY]
+    if not host_kiss_verified:
+        capabilities = [value for value in capabilities if value not in
+                        (HOST_KISS_CAPABILITY, HOST_KISS_USB_CAPABILITY, HOST_KISS_TCP_CAPABILITY)]
+    if args.platform != "ESP32_PLATFORM":
+        capabilities = [value for value in capabilities if value != HOST_KISS_TCP_CAPABILITY]
     missing = [check for check in checks if not check["present"]]
     ota_methods = stable_unique([
         check["capability"].removeprefix("ota.update.")

scripts/host_kiss_capabilities.py

@@ -0,0 +1,111 @@
+#!/usr/bin/env python3
+"""Qualify the optional host KISS service from effective flags and linked code."""
+from __future__ import annotations
+
+import argparse
+import json
+import os
+from pathlib import Path
+import re
+import shlex
+import sys
+
+from package_meshtext_license import linked_symbols
+
+CAPABILITY = "radio2.kiss.host"
+USB_CAPABILITY = "radio2.kiss.usb"
+TCP_CAPABILITY = "radio2.kiss.tcp"
+COMMAND_MARKERS = (b"Error: use set kiss on|off",
+                   b"Error: use set kiss.transport usb|tcp",
+                   b"USB: send KISS FEND to start")
+TCP_MARKER = b"station TCP 5004"
+
+
+def evidence(path):
+    # LTO can inline the transport callbacks. Require executable framing and
+    # the bounded Dispatcher queue shared by the actual RX/TX implementation.
+    # A declaration, an undefined reference, or a CLI help string is not code.
+    symbols = linked_symbols(path)
+    required = ("_ZN4mesh8hostkiss5Codec4feedE",
+                "_ZN4mesh8hostkiss5Codec11queueOutputE",
+                "_ZN4mesh10Dispatcher22queueForeignRadioFrameE")
+    selected = []
+    for operation in required:
+        matches = [name for name in symbols if name.startswith(operation)]
+        if not matches:
+            return []
+        selected.append(matches[0])
+    return selected
+
+
+def operations(value):
+    words = iter(shlex.split(value if isinstance(value, str) else "\n".join(map(str, value))))
+    result = []
+    for word in words:
+        if word in ("-D", "-U"):
+            word += next(words, "")
+        match = re.fullmatch(r"-([DU])(\w+)(?:=(.*))?", word)
+        if match:
+            result.append(match.groups())
+    return result
+
+
+def required(config, env, platform, full_companion, overlay="", unflags=""):
+    if platform not in ("ESP32_PLATFORM", "NRF52_PLATFORM"):
+        return False
+    rows = {name: dict(options) for name, options in config}
+    options = rows.get("env:" + env, {})
+    flags = operations(options.get("build_flags", [])) + operations(overlay)
+    removed = set(operations(options.get("build_unflags", [])) + operations(unflags))
+    flags = [flag for flag in flags if flag not in removed
+             and not (flag[0] == "D" and ("D", flag[1], None) in removed)]
+    defines = {name: "1" if value is None else value
+               for kind, name, value in flags if kind == "D"}
+    # PlatformIO emits surviving -U after CPPDEFINES, so -U always wins.
+    for kind, name, _value in flags:
+        if kind == "U":
+            defines.pop(name, None)
+
+    def enabled(name, default=False, seen=()):
+        if name not in defines:
+            return default
+        if name in seen:
+            return None
+        value = defines[name].strip().strip("() ")
+        if value in ("false", "true"):
+            return value == "true"
+        if re.fullmatch(r"[A-Za-z_]\w*", value):
+            return enabled(value, False, (*seen, name))
+        number = re.fullmatch(r"([+-]?(?:0[xX][0-9a-fA-F]+|[0-9]+))[uUlL]*", value)
+        if number:
+            value = number[1]
+            base = 16 if "x" in value.lower() else 8 if re.fullmatch(r"[+-]?0[0-7]+", value) else 10
+            return int(value, base) != 0
+        return None
+
+    # Match HostKissConfig.h's hard exclusion even with an explicit enable.
+    if enabled("MESH_PRIMARY_ESPNOW") is True:
+        return False
+    if "MESH_ENABLE_HOST_KISS" in defines:
+        return enabled("MESH_ENABLE_HOST_KISS") is not False
+    return enabled("COMPANION_RADIO_FULL", full_companion) is not False
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--required", action="store_true")
+    parser.add_argument("--env", required=True)
+    parser.add_argument("--platform", required=True)
+    parser.add_argument("--full-companion", action="store_true")
+    args = parser.parse_args()
+    try:
+        selected = required(json.load(sys.stdin), args.env, args.platform, args.full_companion,
+                            os.environ.get("PLATFORMIO_BUILD_FLAGS", ""),
+                            os.environ.get("PLATFORMIO_BUILD_UNFLAGS", ""))
+    except (OSError, ValueError, TypeError) as error:
+        parser.exit(2, f"Cannot determine host KISS capability contract: {error}\n")
+    return 0 if selected else 3
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

src/helpers/ArduinoSerialInterface.cpp

@@ -307,6 +307,10 @@ size_t ArduinoSerialInterface::checkRecvFrame(uint8_t dest[]) {
   // host must not keep the radio/watchdog loop trapped in this parser.
   int pending = _serial->available();
   while (pending-- > 0) {
+    // Input can arrive after the role's initial protocol-owner poll. Leave a
+    // reserved first byte for that owner without consuming its framing.
+    if (_state == RECV_STATE_IDLE && _idle_receive_check != nullptr
+        && !_idle_receive_check(_serial->peek())) return 0;
     int c = _serial->read();
     if (c < 0) break;
     _last_rx_byte_ms = millis();

src/helpers/ArduinoSerialInterface.h

@@ -10,6 +10,9 @@ public:
   // Optional protocol-owner gate, checked before a completed frame is exposed
   // or recorded as client activity. Other transports retain their old behavior.
   typedef bool (*ReceiveFrameCheck)(uint32_t completed_at);
+  // Before reading an idle byte, another protocol may reserve it in-place.
+  // Length-prefixed frame bodies bypass this gate, including opaque 0xC0.
+  typedef bool (*IdleReceiveCheck)(int next_byte);
 
 private:
   static constexpr uint8_t TX_QUEUE_SIZE = 4;
@@ -46,6 +49,7 @@ private:
   const char* _secondaryControlSequence;
   ConnectedCheck _conn_check;
   ReceiveFrameCheck _receive_frame_check;
+  IdleReceiveCheck _idle_receive_check;
   uint8_t rx_buf[MAX_FRAME_SIZE];
   TxFrame _tx_queue[TX_QUEUE_SIZE];
 
@@ -71,7 +75,7 @@ public:
         _has_received_frame(false),
         _serial(nullptr), _controlSequence(nullptr),
         _secondaryControlSequence(nullptr), _conn_check(nullptr),
-        _receive_frame_check(nullptr) {}
+        _receive_frame_check(nullptr), _idle_receive_check(nullptr) {}
 
   void begin(Stream& serial, const char* controlSequence = nullptr,
              const char* secondaryControlSequence = nullptr) {
@@ -105,6 +109,8 @@ public:
   // Without it isConnected() assumes true, as a plain UART has no way of knowing.
   void setConnectedCheck(ConnectedCheck fn) { _conn_check = fn; }
   void setReceiveFrameCheck(ReceiveFrameCheck fn) { _receive_frame_check = fn; }
+  void setIdleReceiveCheck(IdleReceiveCheck fn) { _idle_receive_check = fn; }
+  bool isReceivingFrame() const { return _state != 0; }
 
   // Clear all protocol state owned by one host session while retaining the
   // monotonic completed-frame counter used by startup handoff arbitration.

src/helpers/HostKiss.cpp

@@ -522,6 +522,10 @@ bool serviceUsb() {
 }
 
 bool usbActive() { return usb_session; }
+bool usbInputReserved(int next_byte) {
+  return usb_session || (settings.selection().enabled
+      && settings.selection().transport == Transport::Usb && next_byte == Codec::Fend);
+}
 bool radioActive() { return current(); }
 void resetUsbSession() { if (usb_session) closeSession(); }
 

src/helpers/HostKiss.h

@@ -16,6 +16,9 @@ bool handleCommand(const char* command, char* reply, size_t capacity, bool local
 void loop();
 bool serviceUsb();
 bool usbActive();
+// Normal USB consumers must leave this byte for serviceUsb(), even when it
+// arrives after their loop-top owner poll. Never call inside another frame.
+bool usbInputReserved(int next_byte);
 bool radioActive();
 void resetUsbSession();
 void receive(const uint8_t* bytes, size_t length, uint8_t profile,
@@ -29,6 +32,7 @@ inline bool handleCommand(const char*, char*, size_t, bool) { return false; }
 inline void loop() {}
 inline bool serviceUsb() { return false; }
 inline bool usbActive() { return false; }
+inline bool usbInputReserved(int) { return false; }
 inline bool radioActive() { return false; }
 inline void resetUsbSession() {}
 inline void receive(const uint8_t*, size_t, uint8_t, uint32_t, float, float) {}

test/test_host_kiss.py

@@ -343,10 +343,12 @@ int main() {
   GuestMemoryFS fs;Radio radio;RadioProfileCLI profiles(radio);Dispatcher dispatcher;
   hostkiss::begin(&fs,radio,profiles,dispatcher);
   command("get kiss");assert(strstr(reply,"> off"));
+  assert(!hostkiss::usbInputReserved(0xc0));
   input({0xc0,0,1,0xc0});assert(!hostkiss::serviceUsb()&&serial.available()==4);
   serial.input.clear();
   command("set kiss on",false);assert(strstr(reply,"local connection"));
   command("set kiss on");assert(strstr(reply,"OK"));
+  assert(hostkiss::usbInputReserved(0xc0) && !hostkiss::usbInputReserved(0));
   hostkiss::Settings saved;assert(saved.begin(&fs)&&saved.selection().enabled);
   input({'g','e','t',' '});assert(!hostkiss::serviceUsb()&&serial.available()==4);
   serial.input.clear();
@@ -354,6 +356,7 @@ int main() {
   // same connection becomes ready automatically once the tuple is stable.
   profiles.busy=true;input({0xc0});poll();
   assert(hostkiss::usbActive()&&!hostkiss::radioActive()&&serial.output.empty());
+  assert(hostkiss::usbInputReserved(0) && hostkiss::usbInputReserved(-1));
   command("get kiss");assert(strstr(reply,"waiting-radio")&&strstr(reply,"pending radio settings"));
   fake_millis=1999;poll();assert(serial.output.empty());
   profiles.busy=false;fake_millis=2000;poll();

test/test_host_kiss_capabilities.py

@@ -0,0 +1,175 @@
+#!/usr/bin/env python3
+"""Require executable KISS transport and commands in uploadable Full images."""
+import json
+import os
+from pathlib import Path
+import subprocess
+import sys
+import tempfile
+import unittest
+import zipfile
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT / "scripts"))
+import host_kiss_capabilities as kiss
+from test_meshtext_license import elf32
+
+SYMBOLS = ("_ZN4mesh8hostkiss5Codec4feedEh",
+           "_ZN4mesh8hostkiss5Codec11queueOutputEhPKhj",
+           "_ZN4mesh10Dispatcher22queueForeignRadioFrameEPKhjhmNS_20RadioProfileProtocolE")
+COMMANDS = b"\0".join(kiss.COMMAND_MARKERS)
+
+
+class HostKissCapabilitiesTest(unittest.TestCase):
+    def test_effective_recipe_matches_header_overrides_and_primary_radio(self):
+        def required(*flags, platform="ESP32_PLATFORM", companion=True, overlay="", unflags="", config_unflags=()):
+            config = [["env:base", [["build_flags", list(flags)], ["build_unflags", list(config_unflags)]]]]
+            return kiss.required(config, "base", platform, companion, overlay, unflags)
+        self.assertTrue(required())
+        self.assertTrue(required(platform="NRF52_PLATFORM"))
+        self.assertFalse(required(companion=False))
+        self.assertTrue(required("-DMESH_ENABLE_HOST_KISS=1", companion=False))
+        self.assertFalse(required("-DMESH_ENABLE_HOST_KISS=0"))
+        self.assertFalse(required("-DPOLICY=0", "-DMESH_ENABLE_HOST_KISS=POLICY"))
+        self.assertFalse(required("-DMESH_ENABLE_HOST_KISS=(0UL)"))
+        self.assertTrue(required("-DMESH_ENABLE_HOST_KISS=0", overlay="-DMESH_ENABLE_HOST_KISS=1"))
+        self.assertFalse(required("-DMESH_ENABLE_HOST_KISS=1", overlay="-DMESH_ENABLE_HOST_KISS=0"))
+        self.assertTrue(required("-DMESH_ENABLE_HOST_KISS=0", overlay="-UMESH_ENABLE_HOST_KISS"))
+        self.assertTrue(required("-DMESH_ENABLE_HOST_KISS=0", config_unflags=["-DMESH_ENABLE_HOST_KISS"]))
+        self.assertTrue(required("-DMESH_ENABLE_HOST_KISS=0", unflags="-DMESH_ENABLE_HOST_KISS=0"))
+        self.assertFalse(required("-DMESH_PRIMARY_ESPNOW=1", "-DMESH_ENABLE_HOST_KISS=1"))
+        self.assertTrue(required("-DMESH_PRIMARY_ESPNOW=1", "-UMESH_PRIMARY_ESPNOW"))
+        self.assertFalse(required("-DMESH_PRIMARY_ESPNOW=1", "-UMESH_PRIMARY_ESPNOW",
+                                  config_unflags=["-UMESH_PRIMARY_ESPNOW"]))
+        self.assertTrue(required("-DMESH_PRIMARY_ESPNOW=0"))
+        self.assertFalse(required("-DCOMPANION_RADIO_FULL=0"))
+        self.assertFalse(required("-DMESH_ENABLE_HOST_KISS=1", platform="STM32_PLATFORM"))
+        self.assertFalse(required("-DMESH_ENABLE_HOST_KISS=1", platform="RP2040_PLATFORM"))
+
+    def qualify(self, *, symbols=SYMBOLS, flags=6, undefined=(), zero=(),
+                commands=COMMANDS, platform="ESP32_PLATFORM", tcp=True, require=True):
+        with tempfile.TemporaryDirectory(prefix="mesh-kiss-capabilities-") as directory:
+            root = Path(directory)
+            elf32(root / "firmware.elf", symbols=symbols, flags=flags, undefined=undefined, zero=zero)
+            app = commands + (b"\0" + kiss.TCP_MARKER if tcp else b"")
+            args = []
+            if platform == "NRF52_PLATFORM":
+                with zipfile.ZipFile(root / "firmware.zip", "w") as package:
+                    package.writestr("manifest.json", json.dumps({"manifest": {"application": {"bin_file": "app.bin"}}}))
+                    package.writestr("app.bin", app)
+                args += ["--dfu-package", str(root / "firmware.zip")]
+            else:
+                (root / "firmware.bin").write_bytes(app)
+                args += ["--firmware-bin", str(root / "firmware.bin")]
+            if require:
+                args.append("--require-host-kiss")
+            result = subprocess.run([sys.executable, str(ROOT / "scripts/check_firmware_capabilities.py"),
+                "--image", str(root / "firmware.elf"), "--output", str(root / "caps.json"),
+                "--target", "fixture", "--platform", platform, "--build-profile", "full",
+                "--capability", kiss.CAPABILITY, "--capability", kiss.USB_CAPABILITY,
+                "--capability", kiss.TCP_CAPABILITY, *args], capture_output=True, text=True)
+            return result, json.loads((root / "caps.json").read_text())
+
+    def test_packaged_usb_and_platform_specific_tcp_capabilities(self):
+        for platform in ("ESP32_PLATFORM", "NRF52_PLATFORM"):
+            with self.subTest(platform=platform):
+                result, manifest = self.qualify(platform=platform, tcp=platform=="ESP32_PLATFORM")
+                self.assertEqual(result.returncode, 0, result.stderr)
+                self.assertTrue(manifest["verified"])
+                self.assertIn(kiss.CAPABILITY, manifest["capabilities"])
+                self.assertIn(kiss.USB_CAPABILITY, manifest["capabilities"])
+                # Requested labels alone must not advertise unsupported TCP.
+                self.assertEqual(kiss.TCP_CAPABILITY in manifest["capabilities"], platform=="ESP32_PLATFORM")
+
+    def test_rejects_each_missing_or_nonexecutable_method_and_command(self):
+        cases = [dict(symbols=("unrelated_function",)), dict(flags=4), dict(flags=2), dict(tcp=False)]
+        for symbol in SYMBOLS:
+            cases.extend((dict(symbols=tuple(item for item in SYMBOLS if item!=symbol)),
+                          dict(undefined=(symbol,)), dict(zero=(symbol,))))
+        cases += [dict(commands=b"\0".join(marker for marker in kiss.COMMAND_MARKERS if marker!=missing))
+                  for missing in kiss.COMMAND_MARKERS]
+        for options in cases:
+            with self.subTest(options=options):
+                result, manifest = self.qualify(**options)
+                self.assertEqual(result.returncode, 1, result.stderr)
+                self.assertFalse(manifest["verified"])
+                self.assertFalse(set(manifest["capabilities"]) &
+                                 {kiss.CAPABILITY,kiss.USB_CAPABILITY,kiss.TCP_CAPABILITY})
+        result, manifest = self.qualify(require=False)
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertFalse(set(manifest["capabilities"]) &
+                         {kiss.CAPABILITY,kiss.USB_CAPABILITY,kiss.TCP_CAPABILITY})
+
+    def test_actual_builder_checks_linked_code_and_packaged_application(self):
+        script = r'''
+set -euo pipefail
+pio() { printf 'unexpected PlatformIO invocation\n' >&2; return 99; }
+source build.sh
+PIO_CONFIG_JSON=$MESH_TEST_CONFIG
+OUTPUT_DIR=$MESH_TEST_DIRECTORY/out
+PIO_BUILD_DIR_OVERRIDE=$MESH_TEST_DIRECTORY/build
+BUILD_PROFILE_FOR_TARGET=full
+BUILD_RECIPE_SHA256=test
+BUILD_CAPABILITIES=(); BUILD_EXPECTATIONS=()
+BUILD_APPLICATION_EXPECTATIONS=(); BUILD_REDUCTIONS=()
+requires_wireless_self_update() { return 1; }
+is_xiao_qspi_canonical_repeater_build() { return 1; }
+is_nrf52_sensor_ota_pair_target() { return 1; }
+embedded_bootloader_artifact_target() { printf '%s\n' "$1"; }
+python3() {
+  if [ "$1" = scripts/firmware_build_recipe.py ]; then return 0; fi
+  command "$MESH_TEST_PYTHON" "$@"
+}
+write_build_capability_manifest fixture_repeater "$MESH_TEST_PLATFORM" base fixture
+'''
+        for platform in ("ESP32_PLATFORM", "NRF52_PLATFORM"):
+            for complete in (True, False):
+                with self.subTest(platform=platform, complete=complete), tempfile.TemporaryDirectory() as folder:
+                    root=Path(folder);build=root/"build/base";build.mkdir(parents=True)
+                    elf32(build/"firmware.elf", symbols=SYMBOLS if complete else ("unrelated_function",))
+                    app=COMMANDS+b"\0"+kiss.TCP_MARKER
+                    (build/"firmware.bin").write_bytes(app)
+                    with zipfile.ZipFile(build/"firmware.zip", "w") as package:
+                        package.writestr("manifest.json", json.dumps({"manifest":{"application":{"bin_file":"app.bin"}}}))
+                        package.writestr("app.bin", app)
+                    config=[["env:base", [["build_flags", ["-DMESH_ENABLE_HOST_KISS=1"]]]]]
+                    result=subprocess.run(["bash","-c",script],cwd=ROOT,capture_output=True,text=True,
+                        env={**os.environ,"MESH_TEST_CONFIG":json.dumps(config),"MESH_TEST_PLATFORM":platform,
+                             "MESH_TEST_DIRECTORY":str(root),"MESH_TEST_PYTHON":sys.executable,
+                             "PLATFORMIO_BUILD_FLAGS":"","PLATFORMIO_BUILD_UNFLAGS":""})
+                    self.assertEqual(result.returncode,0 if complete else 1,result.stdout+result.stderr)
+                    manifest=json.loads((root/"out/fixture.capabilities.json").read_text())
+                    self.assertEqual(manifest["verified"],complete)
+                    self.assertEqual(kiss.CAPABILITY in manifest["capabilities"],complete)
+
+    def test_builder_requires_service_only_when_effectively_enabled(self):
+        script = r'''
+set -euo pipefail
+pio() { printf 'unexpected PlatformIO invocation\n' >&2; return 99; }
+source build.sh
+PIO_CONFIG_JSON=$MESH_TEST_CONFIG
+is_companion_radio_full_target() { [ "$MESH_TEST_COMPANION" = 1 ]; }
+is_kiss_modem_target() { [ "$MESH_TEST_MODEM" = 1 ]; }
+requires_host_kiss_code fixture base "$MESH_TEST_PLATFORM"
+'''
+        for platform, companion, modem, flags, expected in (
+                ("ESP32_PLATFORM", True, False, [], 0),
+                ("NRF52_PLATFORM", True, False, [], 0),
+                ("ESP32_PLATFORM", False, False, ["-DMESH_ENABLE_HOST_KISS=1"], 0),
+                ("NRF52_PLATFORM", False, False, ["-DMESH_ENABLE_HOST_KISS=1"], 0),
+                ("ESP32_PLATFORM", True, False, ["-DMESH_ENABLE_HOST_KISS=0"], 1),
+                ("ESP32_PLATFORM", True, False, ["-DMESH_ENABLE_HOST_KISS=1", "-DMESH_PRIMARY_ESPNOW=1"], 1),
+                ("ESP32_PLATFORM", True, True, ["-DMESH_ENABLE_HOST_KISS=1"], 1),
+                ("STM32_PLATFORM", True, False, ["-DMESH_ENABLE_HOST_KISS=1"], 1),
+                ("ESP32_PLATFORM", False, False, [], 1)):
+            with self.subTest(platform=platform,companion=companion,modem=modem,flags=flags):
+                config = [["env:base", [["build_flags",flags]]]]
+                result = subprocess.run(["bash","-c",script],cwd=ROOT,capture_output=True,text=True,
+                    env={**os.environ,"MESH_TEST_CONFIG":json.dumps(config),"MESH_TEST_PLATFORM":platform,
+                         "MESH_TEST_COMPANION":str(int(companion)),"MESH_TEST_MODEM":str(int(modem)),
+                         "PLATFORMIO_BUILD_FLAGS":"","PLATFORMIO_BUILD_UNFLAGS":""})
+                self.assertEqual(result.returncode,expected,result.stdout+result.stderr)
+
+
+if __name__ == "__main__":
+    unittest.main()

test/test_host_kiss_infrastructure.py

@@ -24,11 +24,16 @@ FIXTURE = r'''
 #include <string>
 static std::deque<unsigned char> input;
 static bool kiss_active = false, reset_pending = false, network_pending = false;
+static bool inject_late_fend=false;
 static unsigned reset_reads = 0, kiss_resets = 0, ascii_commands = 0;
 static unsigned board_loops = 0, ethernet_loops = 0, network_commands = 0;
 static unsigned serial_cancels = 0, terminal_drains = 0;
 struct Stream {
-  int available() const { return int(input.size()); }
+  int available() const {
+    if(inject_late_fend){inject_late_fend=false;input={0xc0,0x00,0x41,0xc0};}
+    return int(input.size());
+  }
+  int peek()const{return input.empty()?-1:input.front();}
   int read() {
     assert(!input.empty() && input.front() != 0xc0);
     int value = input.front(); input.pop_front(); return value;
@@ -49,6 +54,7 @@ namespace power { void serviceRegulatorSettings() {} }
 #if MESH_ENABLE_HOST_KISS
 namespace hostkiss {
 void resetUsbSession() { ++kiss_resets; kiss_active=false; }
+bool usbInputReserved(int next){return kiss_active || next==0xc0;}
 bool serviceUsb() {
   if (!input.empty() && input.front()==0xc0) {
     input.pop_front(); kiss_active=true;
@@ -127,6 +133,11 @@ int main() {
   assert(command[0]==0 && !command_overflow);
   ascii(); serviceCommandInterfaces(); assert(ascii_commands==3);
 #endif
+  // The first FEND arrives after serviceUsb() returned false this poll.
+  inject_late_fend=true;const unsigned prior_commands=ascii_commands;
+  serviceCommandInterfaces();
+  assert(!kiss_active && !input.empty() && input.front()==0xc0 && ascii_commands==prior_commands);
+  serviceCommandInterfaces();assert(kiss_active && input.empty());
 #endif
 }
 '''

test/test_host_kiss_ingress.py

@@ -0,0 +1,185 @@
+#!/usr/bin/env python3
+"""Exercise late KISS delimiters through the real Companion USB parser."""
+import os
+from pathlib import Path
+import shutil
+import subprocess
+import sys
+import tempfile
+import unittest
+
+from cpp_source import body
+import test_companion_usb_input as usb
+from test_replay_reset_integration import extract_braced
+
+ROOT = Path(__file__).resolve().parents[1]
+FIXTURE = r'''
+#include <helpers/ArduinoSerialInterface.h>
+#include <cassert>
+#include <deque>
+#include <vector>
+static bool enabled=true, active=false;
+static unsigned gate_calls=0;
+static bool receiveIdle(int next) { ++gate_calls; return !enabled || (next!=0xc0 && !active); }
+struct Port:Stream {
+  std::deque<uint8_t> input;
+  unsigned reads=0;
+  int available()override{return int(input.size());}
+  int read()override{if(input.empty())return -1;auto c=input.front();input.pop_front();++reads;return c;}
+  int peek()override{return input.empty()?-1:input.front();}
+  int availableForWrite()override{return 65536;}
+  size_t write(uint8_t)override{return 1;}
+  size_t write(const uint8_t*,size_t n)override{return n;}
+  void push(const std::vector<uint8_t>& bytes){input.insert(input.end(),bytes.begin(),bytes.end());}
+};
+int main(){
+  Port port;ArduinoSerialInterface usb;uint8_t out[MAX_FRAME_SIZE]={};
+  usb.begin(port);usb.enable();usb.setIdleReceiveCheck(receiveIdle);
+  assert(!usb.isReceivingFrame());
+  // The top-of-loop KISS owner saw an empty port. A complete KISS frame then
+  // arrives during mesh dispatch; the idle Companion parser must leave it all.
+  assert(!usb.checkRecvFrame(out));
+  const std::vector<uint8_t> kiss={0xc0,0,0x41,0xdb,0xdc,0xc0};
+  port.push(kiss);assert(!usb.checkRecvFrame(out));
+  assert(port.reads==0 && std::vector<uint8_t>(port.input.begin(),port.input.end())==kiss);
+  port.input.clear();
+  // Every possible split of an existing length-prefixed frame preserves its
+  // opaque payload delimiter, including one arriving just after the split.
+  const std::vector<uint8_t> binary={'<',5,0,0x42,0xc0,0xdb,0,0xff};
+  for(size_t split=1;split<binary.size();++split){
+    usb.resetSessionState();port.input.clear();gate_calls=0;
+    port.push(std::vector<uint8_t>(binary.begin(),binary.begin()+split));
+    assert(!usb.checkRecvFrame(out) && usb.isReceivingFrame());
+    const unsigned before=gate_calls;
+    port.push(std::vector<uint8_t>(binary.begin()+split,binary.end()));
+    assert(usb.checkRecvFrame(out)==5 && !usb.isReceivingFrame());
+    assert(std::vector<uint8_t>(out,out+5)==std::vector<uint8_t>(binary.begin()+3,binary.end()));
+    assert(gate_calls==before); // no idle gate anywhere inside the frame
+  }
+  // A completed CLI RPC followed immediately by FEND dispatches the RPC once
+  // and leaves the complete next protocol frame untouched on the same stream.
+  port.push(binary);port.push(kiss);
+  assert(usb.checkRecvFrame(out)==5);const unsigned before=port.reads;
+  assert(!usb.checkRecvFrame(out) && port.reads==before && port.peek()==0xc0);
+  port.input.clear();
+  // Off and non-USB callers keep the original framing behavior.
+  enabled=false;port.push(kiss);port.push(binary);
+  assert(usb.checkRecvFrame(out)==5 && port.input.empty());
+  usb.setIdleReceiveCheck(nullptr);enabled=true;port.push(kiss);port.push(binary);
+  assert(usb.checkRecvFrame(out)==5 && port.input.empty());
+}
+'''
+
+
+class HostKissIngressTest(unittest.TestCase):
+    def test_late_fend_is_reserved_and_binary_payload_is_opaque(self):
+        compiler = os.environ.get('CXX') or shutil.which('g++') or shutil.which('clang++')
+        self.assertIsNotNone(compiler)
+        source = (ROOT / 'src/helpers/ArduinoSerialInterface.cpp').read_text()
+        gate = ('    if (_state == RECV_STATE_IDLE && _idle_receive_check != nullptr\n'
+                '        && !_idle_receive_check(_serial->peek())) return 0;\n')
+        self.assertEqual(source.count(gate), 1)
+        for negative in (False, True):
+            with self.subTest(negative=negative), tempfile.TemporaryDirectory(prefix='host-kiss-ingress-') as directory:
+                directory = Path(directory)
+                (directory / 'fixture.cpp').write_text(FIXTURE, encoding='ascii')
+                (directory / 'transport.cpp').write_text(source.replace(gate, '') if negative else source, encoding='ascii')
+                flags = (['-fsanitize=address,undefined', '-fno-sanitize-recover=all',
+                          '-fno-pie', '-no-pie'] if sys.platform.startswith('linux') else [])
+                result = subprocess.run([compiler, '-std=c++17', '-Wall', '-Wextra', *flags,
+                    '-I', str(ROOT / 'test/mocks'), '-I', str(ROOT / 'src'),
+                    '-I', str(ROOT / 'src/helpers'), str(directory / 'fixture.cpp'),
+                    str(directory / 'transport.cpp'), '-o', str(directory / 'test')],
+                    text=True, capture_output=True, timeout=60)
+                self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
+                result = subprocess.run([str(directory / 'test')], text=True, capture_output=True, timeout=10)
+                if negative:
+                    self.assertNotEqual(result.returncode, 0, 'Old parser unexpectedly preserved late KISS input')
+                else:
+                    self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
+
+    def test_actual_ascii_pump_leaves_late_fend_even_with_partial_input(self):
+        main = (ROOT / 'examples/companion_radio/main.cpp').read_text()
+        signatures = (
+            'static void clearUsbTerminalLine()',
+            'static bool acceptUsbBinaryStartupFrame(uint32_t completed_at)',
+            'static void enterUsbTerminalMode(bool show_banner',
+            'static void leaveUsbTerminalMode(bool acknowledge)',
+            'static void resetUsbMotaMode()', 'static void leaveUsbMotaMode(bool acknowledge)',
+            'static bool enterUsbMotaMode(mesh::UsbMotaEntryOrigin origin)',
+            'static void serviceUsbMota()', 'static bool serviceUsbMotaResponseBoundary()',
+            'static void resetUsbTerminalHostSession()', 'static void serviceUsbTerminal()',
+            'static void expireUsbBinaryStartupProbeBeforeDispatch()',
+        )
+        functions = '\n'.join(extract_braced(main, signature) for signature in signatures)
+        functions = functions.replace('static void enterUsbTerminalMode(bool show_banner)',
+                                      'static void enterUsbTerminalMode(bool show_banner = true)')
+        functions = functions.replace('mesh::hostkiss::usbInputReserved(', 'reserveTestInput(')
+        accessor = extract_braced(usb.CONTEXT, 'static SerialMotaSource& serialFolderSource()')
+        declarations = ('static bool reserve_kiss=false;\n'
+                        'static bool reserveTestInput(int next){return reserve_kiss && next==0xc0;}\n')
+        checks = r"""
+  // An expired Binary probe must not drain FEND or print its ASCII banner.
+  reset();reserve_kiss=true;leaveUsbTerminalMode(false);
+  usb_binary_startup_probe.start(g_mock_millis,0);g_mock_millis+=1001;
+  input.bytes={0xc0,0,0x41,0xc0};const auto expired_output=input.output;
+  expireUsbBinaryStartupProbeBeforeDispatch();
+  assert(input.bytes.size()==4 && input.peek()==0xc0 && input.output==expired_output);
+  assert(!the_mesh.terminal && usb_binary_startup_probe.isActive());
+  for(unsigned mode=0;mode<3;++mode){
+    reset();reserve_kiss=true;
+    if(mode==1){input.push("prefix");serviceUsbTerminal();}
+    if(mode==2)usb_terminal_discard_line=true;
+    const auto output_before=input.output;
+    const std::vector<uint8_t> wire={0xc0,0,'v','e','r','\r',0xc0};
+    for(auto byte:wire)input.bytes.push_back(byte);
+    serviceUsbTerminal();
+    assert(std::vector<uint8_t>(input.bytes.begin(),input.bytes.end())==wire);
+    assert(input.output==output_before && the_mesh.commands.empty());
+  }
+  reserve_kiss=false;
+"""
+        template = usb.HARNESS.replace('@FUNCTIONS@', declarations + functions)
+        template = template.replace('@SOURCE_ACCESSOR@', accessor).replace('int main() {', 'int main() {' + checks)
+        template = usb.bind_production_input_constants(template)
+        guard = '    if (reserveTestInput(usb_input.peek())) return;\n'
+        self.assertEqual(template.count(guard), 2)
+        for negative in (False, True):
+            with self.subTest(negative=negative), tempfile.TemporaryDirectory(prefix='host-kiss-ascii-') as directory:
+                directory = Path(directory)
+                (directory / 'fixture.cpp').write_text(template.replace(guard, '') if negative else template, encoding='ascii')
+                result = subprocess.run(['g++', '-std=c++17', '-fsanitize=address,undefined',
+                    '-fno-sanitize-recover=all', '-fno-pie', '-no-pie',
+                    '-I', str(ROOT / 'test/mocks'), '-I', str(ROOT / 'src'),
+                    str(directory / 'fixture.cpp'), str(ROOT / 'src/helpers/ArduinoSerialInterface.cpp'),
+                    str(ROOT / 'src/helpers/ota/MotaSourceSerial.cpp'), '-o', str(directory / 'test')],
+                    text=True, capture_output=True, timeout=60)
+                self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
+                result = subprocess.run([str(directory / 'test')], text=True, capture_output=True, timeout=15)
+                if negative:
+                    self.assertNotEqual(result.returncode, 0, 'Old ASCII pump unexpectedly preserved late FEND')
+                else:
+                    self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
+
+    def test_every_raw_usb_consumer_defers_reserved_input(self):
+        main = (ROOT / 'examples/companion_radio/main.cpp').read_text()
+        setup = body(main, 'void setup()')
+        self.assertIn('usb_serial_interface.setIdleReceiveCheck', setup)
+        self.assertIn('return !mesh::hostkiss::usbInputReserved(next_byte)', setup)
+        for signature in ('static void serviceUsbTerminal()',
+                          'static void serviceUsbLoggingOwnership(bool logging_enabled)',
+                          'static void expireUsbBinaryStartupProbeBeforeDispatch()'):
+            self.assertIn('mesh::hostkiss::usbInputReserved(', body(main, signature))
+        rescue = body((ROOT / 'examples/companion_radio/MyMesh.cpp').read_text(), 'void MyMesh::checkCLIRescueCmd()')
+        self.assertLess(rescue.index('mesh::hostkiss::usbInputReserved(input.peek())'), rescue.index('input.read()'))
+        for role in ('simple_repeater', 'simple_room_server', 'simple_sensor'):
+            source = (ROOT / 'examples' / role / 'main.cpp').read_text()
+            signature = ('static void __attribute__((noinline)) serviceCommandInterfaces()'
+                         if role == 'simple_repeater' else 'void loop()')
+            pump = body(source, signature)
+            self.assertLess(pump.index('mesh::hostkiss::usbInputReserved(console.peek())'),
+                            pump.index('console.read()'))
+
+
+if __name__ == '__main__':
+    unittest.main()

test/test_host_kiss_usb.py

@@ -113,7 +113,8 @@ static unsigned ascii_calls=0,usb_binary_calls=0,network_calls=0,kiss_resets=0;
 static std::deque<uint8_t> input;
 struct Port {} terminal_port;
 struct Usb {
-  bool enabled=true,passthrough=false;
+  bool enabled=true,passthrough=false,receiving=false;
+  bool isReceivingFrame()const{return receiving;}
   unsigned resets=0,frames=19,enables=0,disables=0;
   bool isEnabled() const { return enabled; }
   void enable(){enabled=true;++enables;}
@@ -150,7 +151,7 @@ void cancelReaderCommand(Port* p){assert(p==&terminal_port);++reader_cancels;}
 void discardUsbTerminalOutput(){++text_discards;}
 void setUsbBinaryStreamOwned(bool value){binary_owned=value;}
 namespace ota {
-struct Source {void resetSessionState(){++mota_resets;}} source;
+struct Source {bool pending=false;bool hasPendingResponse()const{return pending;}void resetSessionState(){++mota_resets;}} source;
 struct OtaContext {static Source& serialFolderSource(){return source;}};
 }
 namespace hostkiss {
@@ -179,7 +180,17 @@ static bool poll(){
 int main(){
   input={'v','e','r','\r'};assert(!poll() && ascii_calls==1 && !serial_cancels);
 #if MESH_ENABLE_HOST_KISS
-  terminal_mode=true;usb_mota_mode=true;usb_terminal_discard_line=usb_logging_terminal_mode=true;
+  // Existing binary ownership cannot be stolen by a FEND payload byte.
+  usb_serial_interface.receiving=true; input={0xc0};
+  assert(!serviceCompanionHostKissUsb() && !kiss_active && input.size()==1);
+  usb_serial_interface.receiving=false;
+#if COMPANION_FEATURE_USB_MOTA_SOURCE
+  usb_mota_mode=true;assert(!serviceCompanionHostKissUsb() && !kiss_active && input.size()==1);
+  usb_mota_mode=false;mesh::ota::source.pending=true;
+  assert(!serviceCompanionHostKissUsb() && !kiss_active && input.size()==1);
+  mesh::ota::source.pending=false;
+#endif
+  terminal_mode=true;usb_terminal_discard_line=usb_logging_terminal_mode=true;
   input={0xc0,0x00,0x41};assert(poll());
   assert(kiss_active && host_kiss_usb_parked && host_kiss_usb_restore_enabled && !usb_serial_interface.enabled);
   assert(serial_cancels==1 && route_cancels==1 && forgotten_routes==1);

test/test_meshtext_license.py

@@ -50,7 +50,7 @@ class MeshtextLicenseTest(unittest.TestCase):
             build = root / "build/base"
             build.mkdir(parents=True)
             elf32(build / "firmware.elf", symbols=symbols)
-            config = [["env:base", [["build_flags", ["-DCOMPANION_RADIO_FULL=1", *flags]],
+            config = [["env:base", [["build_flags", ["-DCOMPANION_RADIO_FULL=1", "-DMESH_ENABLE_HOST_KISS=0", *flags]],
                                     ["build_unflags", list(config_unflags)]]]]
             script = r'''
 set -euo pipefail