Support atomic two-packet signed fleet commands

Automatically split longer signed envelopes into two bounded group-data packets, reserve both before queueing, and roll back exact packet copies and retries on admission failure. Collect fragments in either order with a fixed deadline and verify the complete command before durable replay reservation and dispatch.

Raise whole-fleet protocol capacity to 230 bytes, preserve target-specific budgets and radio acknowledgement barriers, and reject rule-token and USB rescue-line truncation. Document existing app input limits and add real-signature, sanitizer, queue failure, timeout, replay, and compact-board regressions.

test/fixtures/companion_fleet_dispatch.h

@@ -23,7 +23,7 @@
 namespace mesh {
 struct FleetCommand {
   static constexpr uint16_t DataType = 0xFF01;
-  static constexpr size_t KeySize = 16, TargetSize = 16, MaxPayloadLength = 165;
+  static constexpr size_t KeySize = 16, TargetSize = 16, MaxPayloadLength = 165, MaxEnvelopeLength = 308;
   static constexpr uint32_t MinEpoch = 1735689600UL, MaxLifetime = 600;
   struct Targets { uint8_t length = 0, count = 0, data[86] = {}; };
   static bool parseTargets(const char*, size_t, Targets&) { return false; }
@@ -52,4 +52,8 @@ struct CompanionFleetFixture {
     assert(false && "fleet packet admission belongs in its dedicated suite");
     return false;
   }
+  bool sendFleetCommandData(FleetFixtureChannel&, const uint8_t*, size_t) {
+    assert(false && "fleet packet admission belongs in its dedicated suite");
+    return false;
+  }
 };

test/test_companion_extrafs_hil_guard.py

@@ -155,7 +155,7 @@ class CompanionExtraFsHilGuardTest(unittest.TestCase):
         self.assertNotIn("handleExtraFsHilCommand", generic)
         self.assertNotIn("hil extrafs", generic)
         self.assertNotIn("handleExtraFsHilCommand", on_air)
-        self.assertIn("handleCommand(text, sender_timestamp, reply)", on_air)
+        self.assertIn("handleCommand(text, sender_timestamp ? sender_timestamp : 1, reply)", on_air)
 
     def test_filler_reports_only_read_verified_durable_bytes(self):
         body = function_body(

test/test_companion_fleet_control.py

@@ -7,6 +7,7 @@ supplies real Ed25519 signing and verification through the Identity interface.
 """
 
 from pathlib import Path
+import os
 import shutil
 import subprocess
 import tempfile
@@ -24,10 +25,23 @@ HARNESS = r'''
 #include <cstdio>
 #include <cstring>
 #include <string>
+#include <vector>
 #include <helpers/FleetCommand.h>
+#include <helpers/CompanionFrameLimits.h>
 #include <openssl/evp.h>
 
+struct RescueStream : Stream {
+  std::string input, output;
+  size_t position = 0, reads = 0;
+  int available() override { return position < input.size(); }
+  int read() override { ++reads; return input[position++]; }
+  size_t write(uint8_t value) override { output += char(value); return 1; }
+};
+static RescueStream rescue_input, rescue_output;
+
 namespace mesh {
+Stream& usbTerminalPort() { return rescue_output; }
+Stream& usbCompanionPort() { return rescue_input; }
 Identity::Identity() { memset(pub_key, 0, sizeof(pub_key)); }
 LocalIdentity::LocalIdentity() : Identity() {
   memset(prv_key, 0xAA, 32);
@@ -62,23 +76,78 @@ bool Identity::verify(const uint8_t* signature, const uint8_t* data, int length)
 
 #define MAX_GROUP_CHANNELS 4
 #define OUT_PATH_UNKNOWN 255
+namespace mesh {
 struct GroupChannel {
   uint8_t secret[32] = {};
   uint8_t tx_radio = 0;
 };
+struct Packet {
+  uint8_t payload[168] = {};
+  size_t payload_len = 0;
+  uint8_t tx_radio = 0;
+  uint16_t scope = 0;
+  bool isRouteDirect() const { return false; }
+  uint8_t getPayloadType() const { return 6; }
+};
+struct Mesh {
+  unsigned base_copy_hooks = 0;
+  void onRadioProfileCopyQueued(Packet*, const Packet*, uint8_t) { ++base_copy_hooks; }
+};
+}
+using GroupChannel = mesh::GroupChannel;
+#define PAYLOAD_TYPE_GRP_DATA 6
+#define PAYLOAD_TYPE_TRACE 9
+
+struct PacketManager {
+  mesh::Packet pool[64];
+  bool used[64] = {};
+  std::vector<mesh::Packet*> outbound;
+  unsigned total = 0, allocations = 0, released = 0;
+  unsigned fail_allocation = 0;
+  mesh::Packet* allocNew() {
+    if (++allocations == fail_allocation) return nullptr;
+    for (unsigned i = 0; i < 64; ++i) if (!used[i]) {
+      used[i] = true; pool[i] = {}; return &pool[i];
+    }
+    return nullptr;
+  }
+  void free(mesh::Packet* packet) {
+    const ptrdiff_t index = packet - pool;
+    assert(index >= 0 && index < 64 && used[index]);
+    used[index] = false; ++released;
+  }
+  unsigned freeCount() const { unsigned count = 0; for (bool value : used) count += !value; return count; }
+  int getOutboundTotal() const { return int(total); }
+  mesh::Packet* getOutboundByIdx(int index) { return outbound.at(index); }
+  mesh::Packet* removeOutboundByIdx(int index) {
+    auto* packet = outbound.at(index); outbound.erase(outbound.begin() + index); --total;
+    return packet;
+  }
+  void queue(mesh::Packet* packet) { outbound.push_back(packet); ++total; }
+};
 struct ChannelDetails { GroupChannel channel; char name[32] = {}; };
 struct Clock { uint32_t now = mesh::FleetCommand::MinEpoch + 1000;
   uint32_t getCurrentTime() const { return now; } };
-class MyMesh {
+class MyMesh : public mesh::Mesh {
  public:
   mesh::LocalIdentity self_id;
   Clock clock;
   ChannelDetails channels[MAX_GROUP_CHANNELS];
   bool queue_ok = true;
-  unsigned queued = 0;
+  PacketManager manager;
+  PacketManager* _mgr = &manager;
+  unsigned& queued = manager.total;
+  unsigned queue_calls = 0, fail_queue = 0, failed_hooks = 0, radio_cancellations = 0;
+  bool radio_copy = false;
+  mesh::Packet* _fleet_admission_original = nullptr;
+  mesh::Packet* _fleet_admission_copy = nullptr;
+  char cli_command[80] = {};
+  bool _cli_line_overflow = false;
+  bool _cli_rescue = false;
+  std::vector<std::string> rescue_commands;
   uint8_t sent_policy = 0;
   uint16_t sent_type = 0;
-  uint8_t sent[mesh::FleetCommand::MaxPayloadLength] = {};
+  uint8_t sent[mesh::FleetCommand::MaxEnvelopeLength] = {};
   size_t sent_length = 0;
   MyMesh() {
     strcpy(channels[1].name, "private fleet");
@@ -90,18 +159,64 @@ class MyMesh {
     out = channels[index]; return true;
   }
   Clock* getRTCClock() { return &clock; }
+  mesh::Packet* createGroupDatagram(uint8_t type, const GroupChannel& channel,
+                                  const uint8_t* data, size_t length) {
+    assert(type == PAYLOAD_TYPE_GRP_DATA && length <= sizeof(mesh::Packet::payload));
+    auto* packet = manager.allocNew();
+    if (!packet) return nullptr;
+    packet->tx_radio = channel.tx_radio; packet->payload_len = length;
+    memcpy(packet->payload, data, length); return packet;
+  }
+  void releasePacket(mesh::Packet* packet) { manager.free(packet); }
+  void onSendFail(mesh::Packet*) { ++failed_hooks; }
+  void cancelOutboundRadioRetry(const mesh::Packet*) { ++radio_cancellations; }
+  void onTracePacketQueuedForSend(mesh::Packet*) { assert(false); }
+  bool queueOutboundPacket(mesh::Packet* packet, uint8_t, uint32_t) {
+    if (++queue_calls == fail_queue || !queue_ok) return false;
+    manager.queue(packet);
+    sent_policy = packet->tx_radio;
+    sent_type = uint16_t(packet->payload[0]) | uint16_t(packet->payload[1]) << 8;
+    const uint8_t* blob = packet->payload + 3;
+    const size_t length = packet->payload[2];
+    assert(length + 3 == packet->payload_len && length <= mesh::FleetCommand::MaxPayloadLength);
+    mesh::FleetCommand::Fragment part;
+    if (mesh::FleetCommand::parseFragment(blob, length, part)) {
+      memcpy(sent + part.index * mesh::FleetCommand::FragmentDataLength, part.data, part.length);
+      sent_length = part.total_length;
+    } else {
+      memcpy(sent, blob, length); sent_length = length;
+    }
+    if (radio_copy) {
+      auto* copy = manager.allocNew(); assert(copy); *copy = *packet;
+      manager.queue(copy); onRadioProfileCopyQueued(copy, packet, 1);
+    }
+    return true;
+  }
+  bool sendPacket(mesh::Packet*, uint8_t, uint32_t = 0);
+  bool sendFloodScoped(const GroupChannel& channel, mesh::Packet* packet, uint32_t = 0) {
+    assert(packet->tx_radio == channel.tx_radio);
+    packet->scope = 0x4321; // Scope adapter; production helper preserves this channel route.
+    return sendPacket(packet, 1);
+  }
   bool sendGroupData(GroupChannel& channel, uint8_t* path, uint8_t path_length,
                      uint16_t type, const uint8_t* data, int length) {
     assert(path == nullptr && path_length == OUT_PATH_UNKNOWN);
-    assert(length > 0 && size_t(length) <= sizeof(sent));
-    if (!queue_ok) return false;
-    sent_policy = channel.tx_radio; sent_type = type;
-    memcpy(sent, data, length); sent_length = length; ++queued;
-    return true;
+    assert(length > 0 && size_t(length) <= mesh::FleetCommand::MaxPayloadLength);
+    uint8_t group[168] = {uint8_t(type), uint8_t(type >> 8), uint8_t(length)};
+    memcpy(group + 3, data, length);
+    auto* packet = createGroupDatagram(PAYLOAD_TYPE_GRP_DATA, channel, group, 3 + length);
+    return packet && sendFloodScoped(channel, packet);
   }
+  bool sendFleetCommandData(GroupChannel&, const uint8_t*, size_t);
+  void onRadioProfileCopyQueued(mesh::Packet*, const mesh::Packet*, uint8_t);
   bool handleCommand(const char* command, uint32_t sender_timestamp, char* reply);
+  void checkCLIRescueCmd();
+  void enterCLIRescue();
+  void resetUsbHostSessionInput();
 };
 @METHOD@
+@SEND_HELPERS@
+@RESCUE_READER@
 
 static void rejected(MyMesh& node, const char* text, uint32_t remote = 0) {
   char reply[160] = {};
@@ -118,6 +233,37 @@ static std::string fullKey(uint8_t value) {
   return hex;
 }
 
+static void checkedFragments(MyMesh& node, size_t first_index) {
+  assert(node.sent_length > mesh::FleetCommand::MaxPayloadLength);
+  uint8_t assembled[mesh::FleetCommand::MaxEnvelopeLength] = {};
+  // Read in reverse order; the immutable part index determines placement.
+  for (uint8_t index : {1, 0}) {
+    auto* packet = node.manager.outbound.at(first_index + index);
+    assert(packet->scope == 0x4321 && packet->tx_radio == 2);
+    assert(packet->payload[0] == 1 && packet->payload[1] == 0xff);
+    const size_t bytes = packet->payload[2];
+    mesh::FleetCommand::Fragment part;
+    assert(mesh::FleetCommand::parseFragment(packet->payload + 3, bytes, part));
+    assert(part.index == index && part.total_length == node.sent_length);
+    assert(part.sequence == node.clock.now && bytes <= mesh::FleetCommand::MaxPayloadLength);
+    assert(part.length == (index == 0 ? 154 : node.sent_length - 154));
+    assert(bytes == mesh::FleetCommand::FragmentHeaderSize + part.length);
+    memcpy(assembled + index * 154, part.data, part.length);
+    uint8_t recipient[32]; memset(recipient, 0x42, sizeof(recipient));
+    mesh::FleetCommand::Decoded decoded;
+    assert(!mesh::FleetCommand::decode(node.self_id, node.channels[1].channel.secret,
+        packet->payload + 3, bytes, node.clock.now, recipient, decoded));
+  }
+  assert(!memcmp(assembled, node.sent, node.sent_length));
+  uint8_t recipient[32]; memset(recipient, 0x42, sizeof(recipient));
+  mesh::FleetCommand::Decoded decoded;
+  assert(mesh::FleetCommand::decode(node.self_id, node.channels[1].channel.secret,
+      assembled, node.sent_length, node.clock.now, recipient, decoded));
+  assembled[node.sent_length - 1] ^= 1;
+  assert(!mesh::FleetCommand::decode(node.self_id, node.channels[1].channel.secret,
+      assembled, node.sent_length, node.clock.now, recipient, decoded));
+}
+
 static void compactBroadcast(MyMesh& node, const char* command) {
   assert(!memcmp(node.sent, "FMC2", 4));
   assert(node.sent[12] == 0); // No target records means every authorized receiver.
@@ -125,6 +271,8 @@ static void compactBroadcast(MyMesh& node, const char* command) {
   assert(mesh::FleetCommand::MinHeaderSize == 14);
   assert(node.sent_length == 14 + strlen(command) + mesh::FleetCommand::SignatureSize);
   assert(!memcmp(node.sent + 14, command, strlen(command)));
+  if (node.sent_length > mesh::FleetCommand::MaxPayloadLength)
+    checkedFragments(node, node.manager.outbound.size() - 2);
   for (uint8_t value : {0x42, 0x77}) {
     uint8_t recipient[32]; memset(recipient, value, sizeof(recipient));
     mesh::FleetCommand::Decoded decoded;
@@ -225,12 +373,210 @@ int main(int argc, char** argv) {
     }
     const std::string prefix = "set flood.filter ";
     const std::string maximum = prefix + std::string(mesh::FleetCommand::MaxCommandLength - prefix.size(), '1');
-    assert(maximum.size() == 72 && mesh::FleetCommand::commandAllowed(maximum.c_str()));
+    assert(maximum.size() == 230 && mesh::FleetCommand::commandAllowed(maximum.c_str()));
     rejected(node, ("fleet send 1 " + maximum + "1").c_str());
     rejected(node, ("fleet send 1 all " + maximum + "1").c_str());
     assert(node.handleCommand(("fleet send 1 " + maximum).c_str(), 0, reply));
     compactBroadcast(node, maximum.c_str());
-    assert(node.sent_length == 150 && node.queued == 1); // Compact header keeps the 72-byte cap.
+    assert(node.sent_length == 308 && node.queued == 2);
+    ++node.clock.now;
+    assert(node.handleCommand(("fleet send 1 all " + maximum).c_str(), 0, reply));
+    compactBroadcast(node, maximum.c_str());
+    assert(node.sent_length == 308 && node.queued == 4);
+  } else if (scenario == "target_command_capacity") {
+    struct Boundary { std::string targets; size_t command_length; };
+    const Boundary boundaries[] = {
+      {"42424242", 225}, {"424242424242", 223}, {fullKey(0x42), 215},
+      {"42424242,51515151", 220}, {"42424242,515151515151", 218},
+      {fullKey(0x42) + "," + fullKey(0x51), 196},
+    };
+    const std::string prefix = "set flood.filter ";
+    for (const Boundary& boundary : boundaries) {
+      const std::string fitting = prefix + std::string(boundary.command_length - prefix.size(), '1');
+      const std::string oversize = fitting + '1';
+      assert(mesh::FleetCommand::commandAllowed(oversize.c_str()));
+      const std::string cli_prefix = "fleet send 1 " + boundary.targets + " ";
+      const unsigned before = node.queued;
+      rejected(node, (cli_prefix + oversize).c_str());
+      // A valid target plus a long command must not become a broadcast, a
+      // partial send, or a consumed sequence. A queue refusal is retryable too.
+      assert(node.handleCommand((cli_prefix + oversize).c_str(), 0, reply));
+      assert(strstr(reply, "exceed two packets") && node.queued == before);
+      node.queue_ok = false;
+      rejected(node, (cli_prefix + fitting).c_str());
+      node.queue_ok = true;
+      assert(node.handleCommand((cli_prefix + fitting).c_str(), 0, reply));
+      assert(node.queued == before + 2 && node.sent_length == 308);
+      assert(node.sent_policy == 2);
+      uint8_t recipient[32]; memset(recipient, 0x42, sizeof(recipient));
+      mesh::FleetCommand::Decoded decoded;
+      assert(mesh::FleetCommand::decode(node.self_id, node.channels[1].channel.secret,
+          node.sent, node.sent_length, node.clock.now, recipient, decoded));
+      assert(decoded.sequence == node.clock.now && !strcmp(decoded.command, fitting.c_str()));
+      assert(!memcmp(node.sent, boundary.targets.size() == 64 ? "FMC1" : "FMC2", 4));
+      memset(recipient, 0x77, sizeof(recipient));
+      assert(!mesh::FleetCommand::decode(node.self_id, node.channels[1].channel.secret,
+          node.sent, node.sent_length, node.clock.now, recipient, decoded));
+      rejected(node, (cli_prefix + fitting).c_str()); // Successful send consumes its second.
+      ++node.clock.now;
+    }
+    assert(node.queued == 12);
+  } else if (scenario == "maximum_correlated_app_command") {
+    const std::string prefix = "set flood.filter ";
+    const std::string maximum = prefix + std::string(158 - prefix.size(), '1');
+    const std::string local_cli = "AB|fleet send 1 " + maximum;
+    // The app's local CLI frame adds opcode 66 and a trailing NUL. Its request
+    // correlation prefix is already inside local_cli; all are ASCII bytes.
+    assert(local_cli.size() + 2 == MAX_FRAME_SIZE);
+    assert((local_cli + '1').size() + 2 > MAX_FRAME_SIZE);
+    rejected(node, local_cli.c_str(), 1);
+    node.queue_ok = false;
+    rejected(node, local_cli.c_str());
+    node.queue_ok = true;
+    assert(node.handleCommand(local_cli.c_str(), 0, reply));
+    assert(!strncmp(reply, "AB|OK - fleet command queued; seq=", 32));
+    compactBroadcast(node, maximum.c_str());
+    assert(node.sent_length == 236 && node.queued == 2);
+    assert(node.sent_policy == 2);
+    rejected(node, local_cli.c_str());
+    ++node.clock.now;
+    const std::string unprefixed = prefix + std::string(161 - prefix.size(), '1');
+    const std::string implicit = "fleet send 1 " + unprefixed;
+    assert(implicit.size() + 2 == MAX_FRAME_SIZE);
+    assert((implicit + '1').size() + 2 > MAX_FRAME_SIZE);
+    assert(node.handleCommand(implicit.c_str(), 0, reply));
+    compactBroadcast(node, unprefixed.c_str());
+    assert(node.sent_length == 239 && node.queued == 4);
+  } else if (scenario == "fragment_boundaries") {
+    struct Boundary { std::string targets; size_t single_length; };
+    const Boundary boundaries[] = {
+      {"", 87}, {"42424242 ", 82}, {"424242424242 ", 80}, {fullKey(0x42) + " ", 72},
+    };
+    const std::string prefix = "set flood.filter ";
+    for (const Boundary& boundary : boundaries) {
+      std::string command = prefix + std::string(boundary.single_length - prefix.size(), '1');
+      const std::string cli_prefix = "fleet send 1 " + boundary.targets;
+      const size_t before = node.manager.outbound.size();
+      assert(node.handleCommand((cli_prefix + command).c_str(), 0, reply));
+      assert(node.queued == before + 1 && node.sent_length == 165);
+      assert(!memcmp(node.manager.outbound.back()->payload + 3,
+          boundary.targets.size() == 65 ? "FMC1" : "FMC2", 4));
+      ++node.clock.now; command += '1';
+      assert(node.handleCommand((cli_prefix + command).c_str(), 0, reply));
+      assert(node.queued == before + 3 && node.sent_length == 166);
+      checkedFragments(node, before + 1);
+      ++node.clock.now;
+    }
+    assert(node.queued == 12);
+  } else if (scenario == "fragment_atomic_admission") {
+    unsigned scenario_index = 0;
+    for (const char* failure : {"pool_first", "pool_second", "queue_first", "queue_second", "queue_second_copy"}) {
+      MyMesh trial; trial.clock.now += ++scenario_index * 10;
+      const std::string prefix = "set flood.filter ";
+      const std::string command = prefix + std::string(161 - prefix.size(), '1');
+      const std::string cli = "fleet send 1 " + command;
+      auto* unrelated = trial.manager.allocNew(); assert(unrelated);
+      unrelated->payload[0] = 99; trial.manager.queue(unrelated);
+      const unsigned free_before = trial.manager.freeCount();
+      if (!strcmp(failure, "pool_first")) trial.manager.fail_allocation = trial.manager.allocations + 1;
+      if (!strcmp(failure, "pool_second")) trial.manager.fail_allocation = trial.manager.allocations + 2;
+      if (!strcmp(failure, "queue_first")) trial.fail_queue = trial.queue_calls + 1;
+      if (!strcmp(failure, "queue_second")) trial.fail_queue = trial.queue_calls + 2;
+      if (!strcmp(failure, "queue_second_copy")) {
+        trial.fail_queue = trial.queue_calls + 2; trial.radio_copy = true;
+      }
+      rejected(trial, cli.c_str());
+      assert(trial.queued == 1 && trial.manager.outbound[0] == unrelated);
+      assert(trial.manager.freeCount() == free_before);
+      assert(!trial._fleet_admission_original && !trial._fleet_admission_copy);
+      if (!strcmp(failure, "queue_first")) assert(trial.failed_hooks == 1);
+      if (!strcmp(failure, "queue_second"))
+        assert(trial.failed_hooks == 2 && trial.radio_cancellations == 1);
+      if (!strcmp(failure, "queue_second_copy")) {
+        assert(trial.failed_hooks == 3 && trial.radio_cancellations == 2);
+        assert(trial.base_copy_hooks == 1); // The production override chains Mesh.
+      }
+      trial.manager.fail_allocation = 0; trial.fail_queue = 0; trial.radio_copy = false;
+      assert(trial.handleCommand(cli.c_str(), 0, reply));
+      assert(trial.queued == 3 && trial.manager.outbound[0] == unrelated);
+      compactBroadcast(trial, command.c_str());
+      assert(trial.manager.freeCount() == free_before - 2);
+    }
+  } else if (scenario == "rescue_overflow") {
+    const std::string unsafe = "fleet send 1 set flood.rule.1 type=grp_txt drop hops=all in=any priority=00001 suspend=tempradio";
+    assert(unsafe.size() > sizeof(node.cli_command));
+    // The old forced-completion behavior would execute this valid prefix,
+    // silently omitting the requested suspend=tempradio condition.
+    const std::string truncated = unsafe.substr(0, 78);
+    assert(mesh::FleetCommand::commandAllowed(truncated.c_str() + 13));
+    rescue_input.input = unsafe + "\r\nfleet send 1 clock\r\n";
+    while (rescue_input.available()) {
+      const size_t before = rescue_input.reads;
+      node.checkCLIRescueCmd();
+      assert(rescue_input.reads - before <= sizeof(node.cli_command));
+    }
+    assert(node.rescue_commands.size() == 1 && node.rescue_commands[0] == "fleet send 1 clock");
+    assert(node.queued == 1 && !node._cli_line_overflow && node.cli_command[0] == 0);
+    assert(rescue_output.output.find("ERR: too long") != std::string::npos);
+    compactBroadcast(node, "clock");
+  } else if (scenario == "rescue_split_overflow") {
+    rescue_input.input = std::string(400, 'x');
+    node.checkCLIRescueCmd();
+    assert(rescue_input.reads == sizeof(node.cli_command) && node._cli_line_overflow);
+    assert(node.rescue_commands.empty() && node.queued == 0);
+    while (rescue_input.available()) node.checkCLIRescueCmd();
+    assert(node._cli_line_overflow && node.rescue_commands.empty());
+    // Even a valid fleet suffix is discarded until the actual line ending.
+    rescue_input.input += "fleet send 1 set radio2 off\nfleet send 1 clock\n";
+    while (rescue_input.available()) node.checkCLIRescueCmd();
+    assert(node.rescue_commands.size() == 1 && node.rescue_commands[0] == "fleet send 1 clock");
+    assert(node.queued == 1 && !node._cli_line_overflow);
+  } else if (scenario == "rescue_boundaries") {
+    const std::string exact(79, 'x');
+    rescue_input.input = exact;
+    node.checkCLIRescueCmd();
+    assert(node.rescue_commands.empty() && strlen(node.cli_command) == 79);
+    rescue_input.input += "\r\nfirst\r\nsecond\nthird\r";
+    while (rescue_input.available()) node.checkCLIRescueCmd();
+    const std::vector<std::string> expected = {exact, "first", "second", "third"};
+    assert(node.rescue_commands == expected && !node._cli_line_overflow);
+    assert(node.cli_command[0] == 0);
+  } else if (scenario == "rescue_unterminated") {
+    memset(node.cli_command, 'x', sizeof(node.cli_command));
+    rescue_input.input = "fleet send 1 set radio2 off\r\nfleet send 1 clock\r";
+    while (rescue_input.available()) node.checkCLIRescueCmd();
+    assert(node.rescue_commands.size() == 1 && node.rescue_commands[0] == "fleet send 1 clock");
+    assert(node.queued == 1 && !node._cli_line_overflow);
+  } else if (scenario == "rescue_session_reset") {
+    node._cli_line_overflow = true; strcpy(node.cli_command, "old host prefix");
+    node.enterCLIRescue();
+    assert(node._cli_rescue && !node._cli_line_overflow && node.cli_command[0] == 0);
+    rescue_input.input = "fleet send 1 clock\r";
+    while (rescue_input.available()) node.checkCLIRescueCmd();
+    assert(node.queued == 1);
+    ++node.clock.now;
+    node._cli_line_overflow = true; strcpy(node.cli_command, "old host prefix");
+    node.resetUsbHostSessionInput();
+    assert(!node._cli_line_overflow);
+    for (char value : node.cli_command) assert(value == 0);
+    rescue_input.input += "fleet send 1 clock\n";
+    while (rescue_input.available()) node.checkCLIRescueCmd();
+    assert(node.queued == 2 && node.rescue_commands.size() == 2);
+  } else if (scenario == "stm32_rescue") {
+    rescue_input.input = std::string(400, 'x') + "fleet send 1 clock\r\nnext\r\n";
+    while (rescue_input.available()) {
+      const size_t before = rescue_input.reads;
+      node.checkCLIRescueCmd();
+      assert(rescue_input.reads - before <= sizeof(node.cli_command));
+    }
+    assert(node.rescue_commands == std::vector<std::string>{"next"});
+    assert(rescue_output.output.find("  Error: unknown command") != std::string::npos);
+    assert(rescue_output.output.find("ERR: too long") == std::string::npos);
+    assert(node.queued == 0 && !node._cli_line_overflow);
+    memset(node.cli_command, 'x', sizeof(node.cli_command));
+    rescue_input.input += "suffix\nlast\n";
+    while (rescue_input.available()) node.checkCLIRescueCmd();
+    assert(node.rescue_commands == (std::vector<std::string>{"next", "last"}));
   } else if (scenario == "explicit_target_forms") {
     for (const std::string& targets : {std::string("42424242"), std::string("424242424242"),
         fullKey(0x42), std::string("42424242,515151515151,") + fullKey(0x63)}) {
@@ -258,7 +604,7 @@ int main(int argc, char** argv) {
         "fleet send 1 11111111111111 set radio2 off", "fleet send 1 all get password",
         "fleet send 1 all set radio2 off\nreboot"}) rejected(node, command);
     rejected(node, ("fleet send 1 " + std::string(65, '1') + " set radio2 off").c_str());
-    rejected(node, ("fleet send 1 all set flood.filter " + std::string(150, '1')).c_str());
+    rejected(node, ("fleet send 1 all set flood.filter " + std::string(mesh::FleetCommand::MaxCommandLength, '1')).c_str());
     assert(node.handleCommand("fleet\tsend\t1\tall\tset radio2 off", 0, reply));
     assert(!strncmp(reply, "OK - fleet command queued", 25));
   } else if (scenario == "private_channel") {
@@ -378,14 +724,16 @@ int main(int argc, char** argv) {
     }
     mesh::FleetCommand::Targets parsed;
     assert(mesh::FleetCommand::parseTargets(targets.c_str(), targets.size(), parsed));
+    const std::string prefix = "set flood.filter ";
+    const std::string too_long = prefix + std::string(mesh::FleetCommand::MaxCommandLength - prefix.size(), '1');
+    assert(node.handleCommand(("fleet send 1 " + targets + " " + too_long).c_str(), 0, reply));
+    assert(strstr(reply, "exceed two packets") && node.queued == 0);
     assert(node.handleCommand(("fleet send 1 " + targets + " set radio2 off").c_str(), 0, reply));
-    assert(strstr(reply, "exceed one packet") && node.queued == 0);
-    assert(node.handleCommand("fleet send 1 00112233 set radio2 off", 0, reply));
-    assert(node.queued == 1);
+    assert(node.queued == 2); // The valid list now fits one atomic two-packet envelope.
     ++node.clock.now;
     targets += ",44556677"; // More than the entire target-record capacity.
     rejected(node, ("fleet send 1 " + targets + " set radio2 off").c_str());
-    assert(node.queued == 1);
+    assert(node.queued == 2);
   } else if (scenario == "scheduled_sets") {
     const char* relative[] = {
       "set radioat 910.5,500,5,5,+1,auto",
@@ -462,9 +810,15 @@ int main(int argc, char** argv) {
     assert(mesh::FleetCommand::parseTargets(targets.c_str(), targets.size(), parsed));
     assert(mesh::FleetCommand::commandAllowed(scheduled.c_str()));
     assert(node.handleCommand(("fleet send 1 " + targets + " " + scheduled).c_str(), 0, reply));
-    assert(strstr(reply, "exceed one packet") && node.queued == 0);
+    assert(!strncmp(reply, "OK - fleet command queued", 25) && node.queued == 2);
+    uint8_t recipient[32]; memset(recipient, 0x42, sizeof(recipient));
+    mesh::FleetCommand::Decoded decoded;
+    assert(mesh::FleetCommand::decode(node.self_id, node.channels[1].channel.secret,
+        node.sent, node.sent_length, node.clock.now, recipient, decoded));
+    assert(!strcmp(decoded.command, scheduled.c_str()));
+    ++node.clock.now;
     assert(node.handleCommand(("fleet send 1 42424242 " + scheduled).c_str(), 0, reply));
-    assert(node.queued == 1 && node.sent_policy == 2);
+    assert(node.queued == 3 && node.sent_policy == 2);
   } else if (scenario == "clock_controls") {
     uint8_t recipient[32]; memset(recipient, 0x42, sizeof(recipient));
     for (const char* clock_command : {"clock", "clock sync", "time 1800000000"}) {
@@ -510,10 +864,45 @@ class CompanionFleetControlTests(unittest.TestCase):
         work = Path(cls.work.name)
         cls.binary = work / "companion-fleet"
         fixture = work / "fixture.cpp"
-        fixture.write_text(HARNESS.replace("@METHOD@", cls.production))
+        helpers = "\n".join(body(source, signature) for signature in (
+            "void MyMesh::onRadioProfileCopyQueued(", "bool MyMesh::sendFleetCommandData(",
+        ))
+        send_packet = body((ROOT / "src/Dispatcher.cpp").read_text(), "bool Dispatcher::sendPacket(")
+        send_packet = send_packet.replace("bool Dispatcher::sendPacket(Packet*", "bool MyMesh::sendPacket(mesh::Packet*", 1)
+        rescue = body(source, "void MyMesh::checkCLIRescueCmd()")
+        # Keep the entire production reader/line-admission logic. Replace only
+        # downstream rescue commands (filesystem/UI peripherals) with a record
+        # and the real Companion fleet dispatcher used throughout this suite.
+        rescue = rescue[:rescue.index("    reply_buf[0] = 0;")]
+        # Print::println(text) is absent from the shared host Stream adapter.
+        # Preserve STM32's actual shared error text and adapt only printing.
+        rescue = rescue.replace('output.println("  Error: unknown command");',
+                                'output.print("  Error: unknown command"); output.println();')
+        rescue += '''
+    rescue_commands.emplace_back(cli_command);
+    char reply[160] = {};
+    handleCommand(cli_command, 0, reply);
+    cli_command[0] = 0;
+  }
+}
+'''
+        reset = body(source, "void MyMesh::resetUsbHostSessionInput()")
+        enter = body(source, "void MyMesh::enterCLIRescue()")
+        # The shared Stream mock lacks Print::println(text); preserve the
+        # actual entry-state reset and adapt only this banner's formatting.
+        banner = 'mesh::usbTerminalPort().println("========= CLI Rescue =========");'
+        assert enter.count(banner) == 1
+        enter = enter.replace(banner, 'mesh::usbTerminalPort().print("========= CLI Rescue =========\\n");')
+        cls.fixture = fixture
+        fixture.write_text(HARNESS.replace("@METHOD@", cls.production)
+                           .replace("@SEND_HELPERS@", send_packet + "\n" + helpers)
+                           .replace("@RESCUE_READER@", rescue + "\n" + reset + "\n" + enter))
+        sanitizer_flags = (["-fsanitize=address,undefined", "-fno-omit-frame-pointer", "-no-pie"]
+                           if os.environ.get("MESHCORE_FLEET_SANITIZERS") == "1" else [])
+        cls.sanitizer_flags = sanitizer_flags
         result = subprocess.run([
             compiler, "-std=c++17", "-Wall", "-Wextra", "-Werror",
-            "-Wno-unused-parameter", "-O2",
+            "-Wno-unused-parameter", "-O2", *sanitizer_flags,
             "-fstack-usage", "-isystem", str(ROOT / "test/mocks"),
             "-I", str(ROOT / "src"), str(fixture),
             str(ROOT / "src/helpers/FleetCommand.cpp"), "-lcrypto", "-o", str(cls.binary),
@@ -546,9 +935,50 @@ class CompanionFleetControlTests(unittest.TestCase):
     def test_unknown_or_malformed_explicit_targets_never_fall_back_to_broadcast(self):
         self.scenario("implicit_target_ambiguity")
 
-    def test_omitted_target_still_requires_canonical_allowlisted_command_and_72_byte_cap(self):
+    def test_omitted_target_requires_canonical_allowlisted_command_and_230_byte_cap(self):
         self.scenario("implicit_command_controls")
 
+    def test_target_specific_exact_packet_limits_reject_atomically_and_allow_retry(self):
+        self.scenario("target_command_capacity")
+
+    def test_maximum_correlated_implicit_command_fits_app_frame_and_remains_local_only(self):
+        self.scenario("maximum_correlated_app_command")
+
+    def test_one_packet_boundary_automatically_becomes_two_signed_fragments(self):
+        self.scenario("fragment_boundaries")
+
+    def test_two_part_admission_reserves_pool_and_rolls_back_exact_packets_and_radio_copy(self):
+        self.scenario("fragment_atomic_admission")
+
+    def test_rescue_overflow_discards_full_fleet_rule_and_preserves_next_crlf_command(self):
+        self.scenario("rescue_overflow")
+
+    def test_rescue_overflow_drains_bounded_chunks_and_never_executes_valid_suffix(self):
+        self.scenario("rescue_split_overflow")
+
+    def test_rescue_exact_capacity_waits_for_real_terminator_and_keeps_lines_separate(self):
+        self.scenario("rescue_boundaries")
+
+    def test_rescue_unterminated_buffer_is_bounded_and_discards_remainder(self):
+        self.scenario("rescue_unterminated")
+
+    def test_rescue_entry_and_usb_host_recovery_reset_discard_state(self):
+        self.scenario("rescue_session_reset")
+
+    def test_stm32_rescue_shared_error_still_drains_and_rejects_entire_overflow(self):
+        binary = Path(self.work.name) / "companion-stm32-rescue"
+        built = subprocess.run([
+            self.compiler, "-std=c++17", "-Wall", "-Wextra", "-Werror",
+            "-Wno-unused-parameter", "-DSTM32_PLATFORM=1", "-O2", *self.sanitizer_flags,
+            "-isystem", str(ROOT / "test/mocks"), "-I", str(ROOT / "src"),
+            str(self.fixture), str(ROOT / "src/helpers/FleetCommand.cpp"),
+            "-lcrypto", "-o", str(binary),
+        ], capture_output=True, text=True, timeout=60)
+        self.assertEqual(built.returncode, 0, built.stdout + built.stderr)
+        result = subprocess.run([str(binary), "stm32_rescue"],
+                                capture_output=True, text=True, timeout=10)
+        self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
+
     def test_explicit_8_12_64_hex_and_list_targets_remain_restricted(self):
         self.scenario("explicit_target_forms")
 
@@ -576,7 +1006,7 @@ class CompanionFleetControlTests(unittest.TestCase):
     def test_invalid_lists_reject_atomically_without_consuming_sequence(self):
         self.scenario("invalid_target_lists")
 
-    def test_valid_list_exceeding_packet_capacity_never_partially_sends(self):
+    def test_valid_list_exceeding_two_packet_capacity_never_partially_sends(self):
         self.scenario("target_capacity")
 
     def test_four_schedule_set_forms_are_signed_without_rewriting(self):

test/test_fleet_channel_runtime.py

@@ -279,13 +279,16 @@ static void enroll(FleetChannel& fleet,const LocalIdentity& publisher){
  assert(config(fleet,std::string("set fleet.controller ")+hex).rfind("OK",0)==0);
 }
 static Packet raw(const uint8_t* bytes,size_t length){
- Packet packet;Utils::sha256(packet.payload,1,channel_key,16);
+ Packet packet;
+ assert(1+CIPHER_MAC_SIZE+(length+15)/16*16<=sizeof(packet.payload));
+ Utils::sha256(packet.payload,1,channel_key,16);
  packet.payload_len=1+Utils::encryptThenMAC(channel_key,packet.payload+1,bytes,int(length));
  packet.radio_profile=1;packet.radio_generation=17;packet.path_hash_size=2;
  return packet;
 }
 static Packet command(const LocalIdentity& publisher,uint32_t sequence,const char* text="set radio2 off",
                       const char* target="all"){
+ assert(strlen(text)<=FleetCommand::MaxPayloadLength-FleetCommand::HeaderSize-FleetCommand::SignatureSize);
  uint8_t payload[184]={},destination[16];assert(FleetCommand::parseTarget(target,destination));
  size_t length=FleetCommand::encode(publisher,channel_key,sequence,sequence+300,destination,text,payload+3,sizeof(payload)-3);
  assert(length);payload[0]=uint8_t(FleetCommand::DataType);payload[1]=uint8_t(FleetCommand::DataType>>8);payload[2]=uint8_t(length);
@@ -321,7 +324,7 @@ static std::string publicHex(const Identity& identity){
 static Packet signedUnchecked(const LocalIdentity& publisher,uint32_t sequence,uint32_t expires,
                               const char* text){
  uint8_t data[184]={},message[160]={};size_t text_size=strlen(text);
- assert(text_size<=FleetCommand::MaxCommandLength);
+ assert(text_size<=FleetCommand::MaxPayloadLength-FleetCommand::HeaderSize-FleetCommand::SignatureSize);
  uint8_t* payload=data+3;memcpy(payload,"FMC1",4);
  storage::writeLE32(payload+4,sequence);storage::writeLE32(payload+8,expires);
  payload[28]=uint8_t(text_size);memcpy(payload+29,text,text_size);
@@ -332,6 +335,20 @@ static Packet signedUnchecked(const LocalIdentity& publisher,uint32_t sequence,u
  data[0]=uint8_t(FleetCommand::DataType);data[1]=uint8_t(FleetCommand::DataType>>8);
  data[2]=uint8_t(29+text_size+64);return raw(data,3+data[2]);
 }
+static Packet compactUnchecked(const LocalIdentity& publisher,uint32_t sequence,const char* text){
+ uint8_t data[184]={},message[160]={};const size_t text_size=strlen(text);
+ // One byte over the valid limit still fits the cipher's final padded block.
+ assert(text_size<=FleetCommand::SinglePacketMaxCommandLength+1);
+ uint8_t* payload=data+3;memcpy(payload,"FMC2",4);
+ storage::writeLE32(payload+4,sequence);storage::writeLE32(payload+8,sequence+300);
+ payload[12]=0;payload[13]=uint8_t(text_size);memcpy(payload+14,text,text_size);
+ constexpr const char* domain="MeshCoreFleet1";const size_t domain_size=strlen(domain);
+ memcpy(message,domain,domain_size);memcpy(message+domain_size,channel_key,16);
+ memcpy(message+domain_size+16,payload,14+text_size);
+ publisher.sign(payload+14+text_size,message,int(domain_size+16+14+text_size));
+ data[0]=uint8_t(FleetCommand::DataType);data[1]=uint8_t(FleetCommand::DataType>>8);
+ data[2]=uint8_t(14+text_size+64);return raw(data,3+data[2]);
+}
 
 static void bind(FleetChannel& fleet,Mesh& mesh,RadioProfileCLI& profiles){
  mesh.copying=[&](Packet* second,const Packet* first){fleet.copy(second,first);};
@@ -351,8 +368,213 @@ static unsigned apply(FleetChannel& fleet,fs::FS& fs,Mesh& mesh,RadioProfileCLI&
 static void reseal(std::vector<uint8_t>& image){
  storage::writeLE32(image.data()+60,storage::updateCRC32(0xffffffff,image.data(),60));
 }
+static std::string longRule(){
+ const std::string prefix="set flood.rule.1 type=grp_txt channel="+std::string(64,'1')
+  +" prefix=AABBCC,DDEEFF,112233 hops=6+ priority=255 suspend=tempradio mode=radio hashbytes=3 in=any";
+ const std::string action=" action=drop";
+ assert(prefix.size()+action.size()<=FleetCommand::MaxCommandLength);
+ return prefix+std::string(FleetCommand::MaxCommandLength-prefix.size()-action.size(),' ')+action;
+}
+static std::vector<uint8_t> envelope(const LocalIdentity& publisher,uint32_t sequence,
+                                   const std::string& text,const std::string& target="all"){
+ FleetCommand::Targets targets;assert(FleetCommand::parseTargets(target.c_str(),target.size(),targets));
+ std::vector<uint8_t> bytes(FleetCommand::MaxEnvelopeLength);
+ size_t length=FleetCommand::encode(publisher,channel_key,sequence,sequence+300,targets,
+                                  text.c_str(),bytes.data(),bytes.size());
+ assert(length);bytes.resize(length);return bytes;
+}
+static Packet part(const std::vector<uint8_t>& bytes,uint8_t index){
+ uint8_t data[168]={};size_t length=FleetCommand::fragment(bytes.data(),bytes.size(),index,
+                                                     data+3,sizeof(data)-3);
+ assert(length&&length<=FleetCommand::MaxPayloadLength);data[0]=uint8_t(FleetCommand::DataType);
+ data[1]=uint8_t(FleetCommand::DataType>>8);data[2]=uint8_t(length);
+ Packet result=raw(data,3+length);assert(result.payload_len<=MAX_PACKET_PAYLOAD);return result;
+}
+template<typename Change> static Packet changePart(const Packet& packet,Change change){
+ uint8_t data[184]={};int length=Utils::MACThenDecrypt(channel_key,data,packet.payload+1,packet.payload_len-1);
+ assert(length>0);change(data+3);return raw(data,size_t(length));
+}
+struct FragmentFixture {
+ fs::FS disk;Mesh mesh;RadioProfileCLI profiles;FleetChannel fleet{&disk};LocalIdentity publisher;
+ std::string expected=longRule();unsigned calls=0;bool mutation=false;
+ FragmentFixture(){enroll(fleet,publisher);bind(fleet,mesh,profiles);}
+ uint32_t reserved()const{return storage::readLE32(disk.files.at(Path).data()+56);}
+ std::string stats(){return config(fleet,"get fleet.stats");}
+ unsigned feed(Packet& packet,const TransportKey* scope=nullptr,uint32_t advance_ms=1000){
+  fake_ms+=advance_ms;const unsigned before=calls;
+  const auto image=disk.files.at(Path);const unsigned writes=disk.writes,verifies=verify_calls;
+  fleet.receive(&packet,mesh,scope);
+  assert(disk.files.at(Path)==image&&disk.writes==writes&&verify_calls==verifies);
+  fleet.service(mesh,profiles,"long:rule",[&](uint32_t seq,const char* text,char* reply){
+   ++calls;assert(reserved()==seq&&profiles.command);
+   assert(mesh.scoped_profile==packet.radio_profile&&mesh.scoped_generation==packet.radio_generation);
+   assert(strlen(text)==expected.size()&&!strcmp(text,expected.c_str()));
+   assert(text[expected.size()]==0);if(mutation)profiles.stage();strcpy(reply,"OK");
+  });
+  assert(!profiles.command&&mesh.scoped_profile==0&&mesh.scoped_generation==0);return calls-before;
+ }
+ void tick(uint32_t advance_ms){
+  fake_ms+=advance_ms;fleet.service(mesh,profiles,"node",[](uint32_t,const char*,char*){assert(false);});
+ }
+ void partial(){assert(calls==0&&reserved()==0&&mesh.queued==0&&!profiles.mutation);
+  assert(stats().find("accepted=0")!=std::string::npos&&stats().find("parts=1")!=std::string::npos);}
+};
+static int fragmentScenario(const std::string& scenario){
+ FragmentFixture f;const uint32_t seq=f.mesh.clock.now;
+ auto bytes=envelope(f.publisher,seq,f.expected);assert(bytes.size()==308&&f.expected.size()==230);
+ Packet a=part(bytes,0),b=part(bytes,1);assert(a.payload_len==179&&b.payload_len==179);
+ if(scenario=="fragments_orders"){
+  for(unsigned reverse=0;reverse<2;++reverse){
+   FragmentFixture x;auto encoded=envelope(x.publisher,seq,x.expected);
+   Packet first=part(encoded,reverse?1:0),second=part(encoded,reverse?0:1);
+   const auto old=x.disk.files.at(Path);const unsigned verifies=verify_calls,writes=x.disk.writes;
+   assert(x.feed(first)==0);x.partial();assert(x.disk.files.at(Path)==old&&x.disk.writes==writes);
+   assert(verify_calls==verifies&&x.stats().find("rejected=0")!=std::string::npos);
+   assert(x.feed(second)==1&&verify_calls==verifies+1&&x.reserved()==seq&&x.mesh.queued==1);
+   assert(x.stats().find("parts=0")!=std::string::npos&&last_jitter_max==60500);
+   assert(x.feed(first)==0&&x.feed(second)==0&&x.calls==1);
+   FleetChannel rebooted(&x.disk);unsigned calls=0;
+   for(Packet* packet:{&first,&second}){fake_ms+=1000;rebooted.receive(packet,x.mesh);
+    rebooted.service(x.mesh,x.profiles,"node",[&](uint32_t,const char*,char*){++calls;});}
+   assert(calls==0&&x.reserved()==seq);
+  }
+ }else if(scenario=="fragments_duplicates"){
+  const unsigned verifies=verify_calls,writes=f.disk.writes;
+  assert(f.feed(b)==0&&f.feed(b)==0&&f.feed(b)==0);f.partial();
+  assert(verify_calls==verifies&&f.disk.writes==writes&&f.stats().find("rejected=0")!=std::string::npos);
+  assert(f.feed(a)==1&&f.feed(a)==0&&f.feed(b)==0&&f.calls==1&&f.mesh.queued==1);
+ }else if(scenario=="fragments_minimum_size"){
+  f.expected="set flood.rule.1 05 c="+std::string(64,'1')+" d";
+  assert(f.expected.size()==88);auto small=envelope(f.publisher,seq,f.expected);
+  assert(small.size()==166);Packet first=part(small,0),second=part(small,1);
+  assert(first.payload_len==179&&second.payload_len==35);
+  assert(f.feed(second)==0);f.partial();assert(f.feed(first)==1&&f.reserved()==seq);
+  FleetCommand::Targets all;assert(FleetCommand::parseTargets("all",3,all));
+  uint8_t output[308];memset(output,0xA5,sizeof(output));
+  const std::string too_long=longRule()+" ";assert(too_long.size()==231);
+  assert(!FleetCommand::encode(f.publisher,channel_key,seq+1,seq+301,all,too_long.c_str(),output,sizeof(output)));
+  for(uint8_t byte:output)assert(byte==0xA5);
+ }else if(scenario=="fragments_expired_during_collection"){
+  assert(f.feed(a)==0);f.partial();f.mesh.clock.now+=301;
+  assert(f.feed(b)==0&&f.reserved()==0&&f.mesh.queued==0&&f.calls==0);
+  assert(f.stats().find("parts=0")!=std::string::npos);
+  auto retry=envelope(f.publisher,f.mesh.clock.now,f.expected);
+  Packet first=part(retry,0),second=part(retry,1);
+  assert(f.feed(first)==0&&f.feed(second)==1&&f.reserved()==f.mesh.clock.now);
+ }else if(scenario=="fragments_conflicts"){
+  assert(f.feed(a)==0);f.partial();
+  Packet changed=changePart(a,[](uint8_t* frame){frame[11+90]^=1;});
+  assert(f.feed(changed)==0);f.partial();
+  Packet length=changePart(a,[](uint8_t* frame){storage::writeLE16(frame+8,307);});
+  assert(f.feed(length)==0);f.partial();
+  Packet index=changePart(a,[](uint8_t* frame){frame[10]=2;});
+  assert(f.feed(index)==0);f.partial();
+  assert(f.feed(b)==1&&f.calls==1&&f.reserved()==seq);
+ }else if(scenario=="fragments_fresh_retry"){
+  assert(f.feed(a)==0);f.partial();auto newer=envelope(f.publisher,seq+1,f.expected);
+  Packet fresh_first=part(newer,1),fresh_second=part(newer,0);
+  assert(f.feed(fresh_first)==0);f.partial();
+  assert(f.feed(b)==0);f.partial();assert(f.feed(fresh_second)==1&&f.reserved()==seq+1);
+  assert(f.feed(a)==0&&f.feed(b)==0&&f.calls==1&&f.mesh.queued==1);
+ }else if(scenario=="fragments_tamper"){
+  Packet mac=a;mac.payload[1]^=1;assert(f.feed(mac)==0&&f.stats().find("parts=0")!=std::string::npos);
+  assert(f.feed(a)==0);f.partial();const auto saved=f.disk.files.at(Path);
+  Packet signature=changePart(b,[](uint8_t* frame){frame[11+153]^=1;});
+  assert(f.feed(signature)==0&&f.calls==0&&f.disk.files.at(Path)==saved&&f.mesh.queued==0);
+  assert(f.stats().find("parts=0")!=std::string::npos);
+  assert(f.feed(b)==0&&f.feed(a)==1&&f.calls==1);
+ }else if(scenario=="fragments_wrong_signer"){
+  LocalIdentity stranger;auto forged=envelope(stranger,seq,f.expected);
+  Packet first=part(forged,0),second=part(forged,1);
+  assert(f.feed(first)==0);f.partial();assert(f.feed(second)==0&&f.reserved()==0&&f.mesh.queued==0);
+  assert(f.stats().find("parts=0")!=std::string::npos&&f.feed(a)==0&&f.feed(b)==1);
+ }else if(scenario=="fragments_fixed_timeout"){
+  assert(f.feed(a)==0);f.partial();assert(f.feed(a,nullptr,299999)==0);f.partial();
+  f.tick(1);assert(f.stats().find("parts=0")!=std::string::npos&&f.reserved()==0&&f.mesh.queued==0);
+  assert(f.feed(b)==0);f.partial();assert(f.feed(a)==1);
+ }else if(scenario=="fragments_timeout_wrap"){
+  fake_ms=UINT32_MAX-200000;assert(f.feed(b,nullptr,0)==0);f.partial();f.tick(299999);f.partial();
+  f.tick(1);assert(f.stats().find("parts=0")!=std::string::npos&&f.reserved()==0);
+  assert(f.feed(a)==0&&f.feed(b)==1);
+ }else if(scenario=="fragments_sequence_bounds"){
+  for(uint32_t invalid:{FleetCommand::MinEpoch-1,seq+61,seq-601}){
+   Packet changed=changePart(a,[&](uint8_t* frame){storage::writeLE32(frame+4,invalid);});
+   assert(f.feed(changed)==0&&f.stats().find("parts=0")!=std::string::npos&&f.reserved()==0);
+  }
+  Packet mismatch_a=changePart(a,[&](uint8_t* frame){storage::writeLE32(frame+4,seq+1);});
+  Packet mismatch_b=changePart(b,[&](uint8_t* frame){storage::writeLE32(frame+4,seq+1);});
+  const unsigned verifies=verify_calls;assert(f.feed(mismatch_a)==0&&f.feed(mismatch_b)==0);
+  assert(f.reserved()==0&&verify_calls==verifies&&f.stats().find("parts=0")!=std::string::npos);
+  assert(f.feed(a)==0&&f.feed(b)==1);
+ }else if(scenario=="fragments_target"){
+  LocalIdentity stranger;f.expected="set flood.rule.1 any "+std::string(150,' ')+"drop";
+  auto other=envelope(f.publisher,seq,f.expected,publicHex(stranger));
+  Packet other_a=part(other,0),other_b=part(other,1);assert(f.feed(other_a)==0);f.partial();
+  assert(f.feed(other_b)==0&&f.reserved()==0&&f.mesh.queued==0);
+  auto mine=envelope(f.publisher,seq,f.expected,publicHex(f.mesh.self_id));
+  Packet mine_a=part(mine,0),mine_b=part(mine,1);
+  assert(f.feed(mine_b)==0&&f.feed(mine_a)==1&&last_jitter_max==1500);
+ }else if(scenario=="fragments_config_reboot"){
+  assert(f.feed(a)==0);f.partial();assert(config(f.fleet,std::string("set fleet.channel ")+Key).rfind("OK",0)==0);
+  assert(f.stats().find("parts=0")!=std::string::npos&&f.feed(b)==0);f.partial();
+  assert(config(f.fleet,"set fleet.controller "+publicHex(f.publisher)).rfind("OK",0)==0);
+  assert(f.stats().find("parts=0")!=std::string::npos&&f.feed(a)==0);f.partial();
+  FleetChannel rebooted(&f.disk);unsigned calls=0;fake_ms+=1000;rebooted.receive(&b,f.mesh);
+  rebooted.service(f.mesh,f.profiles,"node",[&](uint32_t,const char*,char*){++calls;});
+  assert(calls==0&&f.reserved()==0&&f.mesh.queued==0);
+  fake_ms+=1000;rebooted.receive(&a,f.mesh);
+  rebooted.service(f.mesh,f.profiles,"node",[&](uint32_t sequence,const char* text,char* reply){
+   ++calls;assert(sequence==seq&&!strcmp(text,f.expected.c_str()));strcpy(reply,"OK");});
+  assert(calls==1&&f.reserved()==seq);
+ }else if(scenario=="fragments_reserve_failure"){
+  const auto saved=f.disk.files.at(Path);f.disk.fail_write=true;const unsigned writes=f.disk.writes;
+  assert(f.feed(b)==0);f.partial();assert(f.disk.writes==writes&&f.disk.files.at(Path)==saved);
+  assert(f.feed(a)==0&&f.disk.files.at(Path)==saved&&f.mesh.queued==0&&f.calls==0);
+  assert(config(f.fleet,"get fleet.channel").find("store=error")!=std::string::npos);
+  f.disk.fail_write=false;assert(f.feed(a)==0&&f.feed(b)==0);
+  FleetChannel rebooted(&f.disk);unsigned calls=0;
+  for(Packet* packet:{&a,&b}){fake_ms+=1000;rebooted.receive(packet,f.mesh);
+   rebooted.service(f.mesh,f.profiles,"node",[&](uint32_t,const char*,char* reply){++calls;strcpy(reply,"OK");});}
+  assert(calls==1&&f.reserved()==seq);
+ }else if(scenario=="fragments_scoped_radio_ack"){
+  f.mutation=true;f.mesh.fanout=true;TransportKey scope;memset(scope.key,0x12,sizeof(scope.key));
+  a.radio_profile=0;a.radio_generation=9;b.radio_profile=1;b.radio_generation=27;b.path_hash_size=3;
+  b.scoped=true;b.transport_codes[0]=scope.calcTransportCode(&b);
+  assert(f.feed(a)==0);f.partial();assert(f.feed(b,&scope)==1&&f.fleet.waiting()&&f.profiles.mutation);
+  assert(f.mesh.manager.outbound.size()==2&&f.mesh.ack_scoped&&f.mesh.ack_path_size==3);
+  assert(f.mesh.ack_codes[0]==scope.calcTransportCode(f.mesh.manager.outbound[0]));
+  assert(f.mesh.ack_codes[0]!=b.transport_codes[0]&&last_jitter_max==15500);
+  f.fleet.complete(f.mesh.manager.removeOutboundByIdx(0),f.profiles);
+  assert(f.fleet.waiting()&&f.profiles.mutation&&f.profiles.commits==0);
+  f.fleet.fail(f.mesh.manager.removeOutboundByIdx(0),f.profiles);
+  assert(!f.fleet.waiting()&&!f.profiles.mutation&&f.profiles.commits==1);
+  assert(f.feed(a)==0&&f.feed(b,&scope)==0&&f.calls==1);
+ }else if(scenario=="fragments_unknown_scope"){
+  f.mutation=true;a.scoped=true;a.transport_codes[0]=123;
+  assert(f.feed(b)==0&&f.feed(a)==1&&f.mesh.queued==0&&f.profiles.rollbacks==1);
+  assert(f.reserved()==seq&&!f.fleet.waiting()&&!f.profiles.mutation);
+ }else if(scenario=="fragments_single_interrupt"){
+  assert(f.feed(a)==0);f.partial();f.expected="get radio2";
+  Packet single=targetsCommand(f.publisher,seq+1,"all",f.expected.c_str());
+  assert(f.feed(single)==1&&f.reserved()==seq+1&&f.stats().find("parts=0")!=std::string::npos);
+  assert(f.feed(b)==0&&f.feed(a)==0&&f.calls==1&&f.mesh.queued==1);
+ }else if(scenario=="fragments_malformed_frame"){
+  for(unsigned damage=0;damage<4;++damage){
+   Packet malformed=changePart(a,[&](uint8_t* frame){
+    if(damage==0)storage::writeLE16(frame+8,309);
+    else if(damage==1)storage::writeLE16(frame+8,165);
+    else if(damage==2)frame[10]=2;
+    else frame[-1]=uint8_t(FleetCommand::MaxPayloadLength-1);
+   });assert(f.feed(malformed)==0&&f.stats().find("parts=0")!=std::string::npos);
+  }
+  assert(f.reserved()==0&&f.mesh.queued==0&&f.feed(a)==0&&f.feed(b)==1);
+ }else assert(false);
+ printf("fleet runtime %s passed\n",scenario.c_str());return 0;
+}
 int main(int argc,char** argv){
- assert(argc==2);const std::string scenario=argv[1];fs::FS fs;Mesh mesh;RadioProfileCLI profiles;
+ assert(argc==2);const std::string scenario=argv[1];
+ if(scenario.rfind("fragments_",0)==0)return fragmentScenario(scenario);
+ fs::FS fs;Mesh mesh;RadioProfileCLI profiles;
  FleetChannel fleet(&fs);LocalIdentity publisher;bind(fleet,mesh,profiles);
  if(scenario=="enrollment"){
   Packet packet=command(publisher,mesh.clock.now);assert(apply(fleet,fs,mesh,profiles,packet)==0);
@@ -588,6 +810,43 @@ int main(int argc,char** argv){
   assert(apply(fleet,fs,mesh,profiles,mutation)==0);
   FleetChannel rebooted(&fs);
   assert(apply(rebooted,fs,mesh,profiles,packet)==0&&apply(rebooted,fs,mesh,profiles,mutation)==0);
+ }else if(scenario=="fmc2_maximum_command"){
+  enroll(fleet,publisher);
+  const std::string secret(64,'1');
+  const std::string text="set flood.rule.1 5 c="+secret+" d";
+  const std::string oversized="set flood.rule.1 05 c="+secret+" d";
+  assert(text.size()==87&&text.size()==FleetCommand::SinglePacketMaxCommandLength);
+  assert(oversized.size()==88&&FleetCommand::commandAllowed(oversized.c_str()));
+  FleetCommand::Targets all;assert(FleetCommand::parseTargets("all",3,all));
+  uint8_t output[184];memset(output,0xA5,sizeof(output));
+  assert(FleetCommand::encode(publisher,channel_key,mesh.clock.now,mesh.clock.now+300,all,
+                             oversized.c_str(),output,sizeof(output))==166);
+  assert(output[166]==0xA5);
+  Packet packet=targetsCommand(publisher,mesh.clock.now,"all",text.c_str());
+  uint8_t plaintext[184]={};
+  const int size=Utils::MACThenDecrypt(channel_key,plaintext,packet.payload+1,packet.payload_len-1);
+  assert(size==176&&packet.payload_len==179&&packet.payload_len<=MAX_PACKET_PAYLOAD);
+  assert(plaintext[2]==FleetCommand::MaxPayloadLength&&plaintext[3+13]==87);
+  assert(!memcmp(plaintext+3,"FMC2",4)&&plaintext[3+12]==0);
+  assert(!memcmp(plaintext+3+14,text.data(),text.size()));
+  for(unsigned i=3+plaintext[2];i<unsigned(size);++i)assert(plaintext[i]==0);
+  const auto previous=fs.files.at(Path);unsigned calls=0;
+  struct Capture {char text[88]{};char guard='!';} capture;
+  const auto dispatch=[&](FleetChannel& receiver,Packet& request){
+   const unsigned before=calls;receiver.receive(&request,mesh);
+   receiver.service(mesh,profiles,"long-rule",[&](uint32_t seq,const char* command,char* reply){
+    ++calls;assert(storage::readLE32(fs.files.at(Path).data()+56)==seq);
+    assert(strlen(command)==text.size()&&!strcmp(command,text.c_str())&&command[86]=='d'&&command[87]==0);
+    memcpy(capture.text,command,text.size()+1);assert(capture.guard=='!');strcpy(reply,"OK");
+   });return calls-before;
+  };
+  plaintext[3+14+86]='s';Packet changed=raw(plaintext,size);
+  assert(dispatch(fleet,changed)==0&&fs.files.at(Path)==previous&&mesh.queued==0);
+  Packet too_long=compactUnchecked(publisher,mesh.clock.now,oversized.c_str());
+  assert(too_long.payload_len==179&&dispatch(fleet,too_long)==0&&fs.files.at(Path)==previous);
+  assert(dispatch(fleet,packet)==1&&last_jitter_max==60500&&mesh.queued==1);
+  assert(!strcmp(capture.text,text.c_str())&&capture.guard=='!');
+  assert(dispatch(fleet,packet)==0);FleetChannel rebooted(&fs);assert(dispatch(rebooted,packet)==0);
  }else if(scenario=="fmc2_short_prefix"){
   enroll(fleet,publisher);Packet packet=targetsCommand(publisher,mesh.clock.now,publicHex(mesh.self_id).substr(0,8));
   uint8_t plaintext[184]={};assert(Utils::MACThenDecrypt(channel_key,plaintext,packet.payload+1,packet.payload_len-1)>0);
@@ -770,6 +1029,9 @@ class SHA256 : public SHA256Base {{ public:
     def test_actual_zero_target_sender_reaches_receiver_and_keeps_ack_and_reboot_replay_safety(self):
         self.scenario("fmc2_encoded_all")
 
+    def test_single_packet_87_byte_body_is_intact_signed_and_88_bytes_require_fragmentation(self):
+        self.scenario("fmc2_maximum_command")
+
     def test_new_public_key_prefix_envelope_dispatches_and_staggers_possible_multiple_acks(self):
         self.scenario("fmc2_short_prefix")
 
@@ -785,6 +1047,60 @@ class SHA256 : public SHA256Base {{ public:
     def test_existing_remote_radio_mutation_blocks_fleet_without_reserving_sequence(self):
         self.scenario("existing_mutation")
 
+    def test_two_packets_deliver_exact_230_byte_command_in_either_order_once_across_reboot(self):
+        self.scenario("fragments_orders")
+
+    def test_identical_fragments_never_repeat_execution_or_reserve_partial_requests(self):
+        self.scenario("fragments_duplicates")
+
+    def test_smallest_fragmented_command_uses_a_short_second_packet_and_231_bytes_are_rejected(self):
+        self.scenario("fragments_minimum_size")
+
+    def test_command_expiring_between_parts_never_dispatches_and_fresh_retry_can_succeed(self):
+        self.scenario("fragments_expired_during_collection")
+
+    def test_conflicting_parts_cannot_destroy_a_valid_incomplete_transfer(self):
+        self.scenario("fragments_conflicts")
+
+    def test_fresh_retry_preempts_missing_part_and_delayed_old_parts_cannot_pollute_it(self):
+        self.scenario("fragments_fresh_retry")
+
+    def test_mac_and_last_signature_byte_tampering_never_authorize_partial_commands(self):
+        self.scenario("fragments_tamper")
+
+    def test_fragmented_request_from_another_real_signing_identity_is_rejected(self):
+        self.scenario("fragments_wrong_signer")
+
+    def test_fragment_deadline_is_fixed_despite_duplicates_and_missing_part(self):
+        self.scenario("fragments_fixed_timeout")
+
+    def test_fragment_deadline_handles_millisecond_timer_wrap(self):
+        self.scenario("fragments_timeout_wrap")
+
+    def test_fragment_sequence_plausibility_and_signed_sequence_match_precede_reservation(self):
+        self.scenario("fragments_sequence_bounds")
+
+    def test_fragmented_nonmatching_target_preserves_sequence_for_matching_retry(self):
+        self.scenario("fragments_target")
+
+    def test_channel_controller_changes_and_reboot_clear_partial_transfers(self):
+        self.scenario("fragments_config_reboot")
+
+    def test_fragment_completion_storage_failure_has_no_dispatch_or_reply(self):
+        self.scenario("fragments_reserve_failure")
+
+    def test_final_fragment_scope_and_radio_profile_feed_both_ack_copy_commit_barrier(self):
+        self.scenario("fragments_scoped_radio_ack")
+
+    def test_fragmented_radio_mutation_with_unknown_final_scope_rolls_back(self):
+        self.scenario("fragments_unknown_scope")
+
+    def test_new_valid_single_packet_can_interrupt_an_incomplete_two_packet_transfer(self):
+        self.scenario("fragments_single_interrupt")
+
+    def test_malformed_fragment_size_index_and_padding_never_allocate_partial_intent(self):
+        self.scenario("fragments_malformed_frame")
+
 
 if __name__ == "__main__":
     unittest.main()

test/test_fleet_command/test_fleet_command.cpp

@@ -2,8 +2,10 @@
 #include <Ed25519.h>
 #include <helpers/FleetCommand.h>
 #include <SHA256.h>
+#include <algorithm>
 #include <array>
 #include <string>
+#include <utility>
 #include <vector>
 
 namespace {
@@ -36,7 +38,7 @@ protected:
                               uint32_t sequence = Now,
                               uint32_t expires = Now + 120,
                               const uint8_t* target = nullptr) {
-    std::vector<uint8_t> bytes(Fleet::MaxPayloadLength);
+    std::vector<uint8_t> bytes(Fleet::MaxEnvelopeLength);
     const size_t size = Fleet::encode(publisher, ChannelKey, sequence, expires,
         target ? target : all.data(), command, bytes.data(), bytes.size());
     bytes.resize(size);
@@ -50,7 +52,7 @@ protected:
 
   std::vector<uint8_t> encodeTargets(const Fleet::Targets& targets,
                                     const char* command = "get radio2") {
-    std::vector<uint8_t> bytes(Fleet::MaxPayloadLength);
+    std::vector<uint8_t> bytes(Fleet::MaxEnvelopeLength);
     const size_t size = Fleet::encode(publisher, ChannelKey, Now, Now + 120,
                                      targets, command, bytes.data(), bytes.size());
     bytes.resize(size);
@@ -153,7 +155,7 @@ TEST_F(FleetCommandTest, ScheduledCommandsRejectMalformedTimesFieldsAndSelectors
       + std::string(Fleet::MaxCommandLength - prefix.size() - suffix.size(), '0') + suffix;
   ASSERT_EQ(Fleet::MaxCommandLength, maximum.size());
   EXPECT_TRUE(Fleet::commandAllowed(maximum.c_str()));
-  EXPECT_TRUE(decode(encode(maximum.c_str())));
+  EXPECT_TRUE(decode(encodeTargets(Fleet::Targets{}, maximum.c_str())));
   const std::string oversized = prefix + '0' + maximum.substr(prefix.size());
   EXPECT_FALSE(Fleet::commandAllowed(oversized.c_str()));
 }
@@ -289,21 +291,28 @@ TEST_F(FleetCommandTest, FailedSignatureVerificationNeverReturnsCommand) {
   EXPECT_STREQ("", decoded.command);
 }
 
-TEST_F(FleetCommandTest, FullPayloadLimitFitsWithoutOverflow) {
+TEST_F(FleetCommandTest, LegacyPayloadLimitFitsAndCannotBeRaisedByDestinationCapacity) {
   const std::string prefix = "set flood.rule.1 ";
-  const std::string maximum = prefix + std::string(Fleet::MaxCommandLength - prefix.size(), 'a');
+  const std::string maximum = prefix + std::string(Fleet::FixedTargetMaxCommandLength - prefix.size(), 'a');
   ASSERT_TRUE(Fleet::commandAllowed(maximum.c_str()));
   const auto bytes = encode(maximum.c_str());
-  ASSERT_EQ(Fleet::MaxPayloadLength, bytes.size());
+  ASSERT_EQ(Fleet::MaxEnvelopeLength, bytes.size());
   EXPECT_TRUE(decode(bytes));
   EXPECT_STREQ(maximum.c_str(), decoded.command);
-  EXPECT_FALSE(Fleet::commandAllowed((maximum + "a").c_str()));
+  EXPECT_TRUE(Fleet::commandAllowed((maximum + "a").c_str()));
   EXPECT_TRUE(encode((maximum + "a").c_str()).empty());
-  std::array<uint8_t, Fleet::MaxPayloadLength + 2> output;
+  std::array<uint8_t, Fleet::MaxEnvelopeLength + 64> output;
   output.fill(0xA5);
   EXPECT_EQ(0U, Fleet::encode(publisher, ChannelKey, Now, Now + 10, all.data(),
-                            maximum.c_str(), output.data() + 1, 164));
+                            maximum.c_str(), output.data() + 1, Fleet::MaxEnvelopeLength - 1));
   for (uint8_t value : output) EXPECT_EQ(0xA5, value);
+  for (size_t command_length : {Fleet::FixedTargetMaxCommandLength + 1, Fleet::MaxCommandLength}) {
+    const std::string oversized = prefix + std::string(command_length - prefix.size(), 'a');
+    ASSERT_TRUE(Fleet::commandAllowed(oversized.c_str()));
+    EXPECT_EQ(0U, Fleet::encode(publisher, ChannelKey, Now, Now + 10, all.data(),
+                              oversized.c_str(), output.data(), output.size()));
+    for (uint8_t value : output) EXPECT_EQ(0xA5, value);
+  }
 }
 
 TEST_F(FleetCommandTest, RejectsTruncationTrailingPaddingAndWrongMagic) {
@@ -439,7 +448,7 @@ TEST_F(FleetCommandTest, TargetRecordAndPayloadBudgetsAreSeparateAndBounded) {
     ASSERT_TRUE(Fleet::parseTargets(list.data(), list.size(), targets));
     EXPECT_EQ(count, targets.count);
     EXPECT_EQ(count * 5, targets.length);
-    EXPECT_EQ(count <= 15, !encodeTargets(targets).empty()) << count;
+    EXPECT_FALSE(encodeTargets(targets).empty()) << count;
   }
   Fleet::Targets targets{};
   list += ",01020304";
@@ -452,7 +461,7 @@ TEST_F(FleetCommandTest, TargetRecordAndPayloadBudgetsAreSeparateAndBounded) {
     ASSERT_TRUE(Fleet::parseTargets(list.data(), list.size(), targets));
     EXPECT_EQ(count * 17, targets.length);
   }
-  EXPECT_TRUE(encodeTargets(targets).empty());
+  EXPECT_FALSE(encodeTargets(targets).empty());
   list += ',' + full;
   EXPECT_FALSE(Fleet::parseTargets(list.data(), list.size(), targets));
 }
@@ -485,17 +494,25 @@ TEST_F(FleetCommandTest, CompactBroadcastCapacityAndCommandLimitRemainBounded) {
   const std::string prefix = "set flood.rule.1 ";
   const std::string maximum = prefix + std::string(Fleet::MaxCommandLength - prefix.size(), 'a');
   const auto bytes = encodeTargets(targets, maximum.c_str());
-  ASSERT_EQ(150U, bytes.size());
-  EXPECT_EQ(15U, encode(maximum.c_str()).size() - bytes.size());
+  ASSERT_EQ(230U, maximum.size());
+  ASSERT_EQ(Fleet::MaxEnvelopeLength, bytes.size());
+  ASSERT_TRUE(Fleet::commandAllowed(maximum.c_str()));
   EXPECT_TRUE(decode(bytes));
+  EXPECT_STREQ(maximum.c_str(), decoded.command);
   EXPECT_TRUE(decoded.broadcast);
-  std::array<uint8_t, Fleet::MaxPayloadLength> output;
+  std::array<uint8_t, Fleet::MaxEnvelopeLength> output;
   output.fill(0xA5);
   EXPECT_EQ(0U, Fleet::encode(publisher, ChannelKey, Now, Now + 120, targets,
                             maximum.c_str(), output.data(), bytes.size() - 1));
   for (uint8_t byte : output) EXPECT_EQ(0xA5, byte);
   const std::string oversized = maximum + 'a';
+  EXPECT_FALSE(Fleet::commandAllowed(oversized.c_str()));
   EXPECT_TRUE(encodeTargets(targets, oversized.c_str()).empty());
+  for (const char invalid : {'\n', ';', char(0x80)}) {
+    const std::string malformed = maximum.substr(0, maximum.size() - 1) + invalid;
+    EXPECT_FALSE(Fleet::commandAllowed(malformed.c_str()));
+    EXPECT_TRUE(encodeTargets(targets, malformed.c_str()).empty());
+  }
   for (size_t length = 0; length < bytes.size(); ++length) {
     const std::vector<uint8_t> truncated(bytes.begin(), bytes.begin() + length);
     EXPECT_FALSE(decode(truncated));
@@ -565,7 +582,7 @@ TEST_F(FleetCommandTest, AnyMatchingEntryIsAcceptedButMalformedLaterEntryIsRejec
   bytes[13 + targets.length] = 0;
   EXPECT_FALSE(decode(bytes));
   bytes = encodeTargets(targets);
-  bytes[13 + targets.length] = 73;
+  bytes[13 + targets.length] = Fleet::MaxCommandLength + 1;
   EXPECT_FALSE(decode(bytes));
   for (size_t length = 0; length < bytes.size(); ++length) {
     const std::vector<uint8_t> truncated(bytes.begin(), bytes.begin() + length);
@@ -615,21 +632,52 @@ TEST_F(FleetCommandTest, InvalidExternalTargetStructuresNeverWriteOutput) {
 }
 
 TEST_F(FleetCommandTest, LargestFmc2EnvelopeFitsAndOversizedCombinationWritesNothing) {
-  const std::string prefix = "set flood.rule.1 ";
-  const std::string maximum = prefix + std::string(Fleet::MaxCommandLength - prefix.size(), 'a');
   Fleet::Targets targets{};
   const char* three = "01020304,05060708,090A0B0C";
   ASSERT_TRUE(Fleet::parseTargets(three, strlen(three), targets));
-  EXPECT_EQ(Fleet::MaxPayloadLength, encodeTargets(targets, maximum.c_str()).size());
+  const std::string prefix = "set flood.rule.1 ";
+  const std::string maximum = prefix + std::string(
+      Fleet::MaxCommandLength - targets.length - prefix.size(), 'a');
+  EXPECT_EQ(Fleet::MaxEnvelopeLength, encodeTargets(targets, maximum.c_str()).size());
   const char* four = "01020304,05060708,090A0B0C,01020304";
   ASSERT_TRUE(Fleet::parseTargets(four, strlen(four), targets));
-  std::array<uint8_t, Fleet::MaxPayloadLength> output;
+  std::array<uint8_t, Fleet::MaxEnvelopeLength> output;
   output.fill(0xA5);
   EXPECT_EQ(0U, Fleet::encode(publisher, ChannelKey, Now, Now + 120,
                             targets, maximum.c_str(), output.data(), output.size()));
   for (uint8_t byte : output) EXPECT_EQ(0xA5, byte);
 }
 
+TEST_F(FleetCommandTest, EveryTargetFormUsesItsExactCommandBudgetWithoutPartialWrites) {
+  char full[PUB_KEY_SIZE * 2 + 1];
+  mesh::Utils::toHex(full, publisher.pub_key, PUB_KEY_SIZE);
+  const std::string short_key(full, 8), long_key(full, 12);
+  const std::string prefix = "set flood.rule.1 ";
+  const std::vector<std::pair<std::string, size_t>> cases = {
+    {"all", 230}, {short_key, 225}, {long_key, 223}, {full, 215},
+    {short_key + ',' + short_key, 220}, {short_key + ',' + long_key, 218},
+    {std::string(full) + ',' + short_key, 208}, {std::string(full) + ',' + full, 196}
+  };
+  for (const auto& item : cases) {
+    SCOPED_TRACE(item.first);
+    Fleet::Targets targets{};
+    ASSERT_TRUE(Fleet::parseTargets(item.first.data(), item.first.size(), targets));
+    const std::string maximum = prefix + std::string(item.second - prefix.size(), 'a');
+    ASSERT_TRUE(Fleet::commandAllowed(maximum.c_str()));
+    const auto bytes = encodeTargets(targets, maximum.c_str());
+    ASSERT_EQ(Fleet::MaxEnvelopeLength, bytes.size());
+    ASSERT_TRUE(decode(bytes));
+    EXPECT_STREQ(maximum.c_str(), decoded.command);
+    const std::string oversized = maximum + 'a';
+    EXPECT_TRUE(encodeTargets(targets, oversized.c_str()).empty());
+    std::array<uint8_t, Fleet::MaxEnvelopeLength + 64> output;
+    output.fill(0xA5);
+    EXPECT_EQ(0U, Fleet::encode(publisher, ChannelKey, Now, Now + 120, targets,
+                              oversized.c_str(), output.data(), output.size()));
+    for (uint8_t byte : output) EXPECT_EQ(0xA5, byte);
+  }
+}
+
 TEST_F(FleetCommandTest, UnrepresentableZeroDigestNeverBecomesAllDuringLegacyEncoding) {
   Fleet::Targets targets{};
   targets.count = 1;
@@ -641,6 +689,129 @@ TEST_F(FleetCommandTest, UnrepresentableZeroDigestNeverBecomesAllDuringLegacyEnc
   EXPECT_FALSE(decode(bytes));
 }
 
+TEST_F(FleetCommandTest, SinglePacketBoundariesRemainExactBeforeFragmentation) {
+  char full[PUB_KEY_SIZE * 2 + 1];
+  mesh::Utils::toHex(full, publisher.pub_key, PUB_KEY_SIZE);
+  const std::string prefix = "set flood.rule.1 ";
+  for (const auto& item : std::vector<std::pair<std::string, size_t>>{
+      {"all", 87}, {std::string(full, 8), 82}, {std::string(full, 12), 80}, {full, 72}}) {
+    Fleet::Targets targets{};
+    ASSERT_TRUE(Fleet::parseTargets(item.first.data(), item.first.size(), targets));
+    const std::string command = prefix + std::string(item.second - prefix.size(), 'a');
+    const auto single = encodeTargets(targets, command.c_str());
+    ASSERT_EQ(Fleet::MaxPayloadLength, single.size());
+    std::array<uint8_t, Fleet::MaxPayloadLength> output;
+    output.fill(0xA5);
+    EXPECT_EQ(0U, Fleet::fragment(single.data(), single.size(), 0, output.data(), output.size()));
+    for (uint8_t byte : output) EXPECT_EQ(0xA5, byte);
+    const auto longer = encodeTargets(targets, (command + 'a').c_str());
+    ASSERT_EQ(Fleet::MaxPayloadLength + 1, longer.size());
+    EXPECT_EQ(Fleet::MaxPayloadLength,
+              Fleet::fragment(longer.data(), longer.size(), 0, output.data(), output.size()));
+    EXPECT_EQ(Fleet::FragmentHeaderSize + longer.size() - Fleet::FragmentDataLength,
+              Fleet::fragment(longer.data(), longer.size(), 1, output.data(), output.size()));
+  }
+}
+
+TEST_F(FleetCommandTest, TwoExactFragmentsReassembleInEitherOrderAndVerifyOnlyWholeEnvelope) {
+  const std::string prefix = "set flood.rule.1 ";
+  for (size_t command_length : {88U, 89U, 229U, 230U}) {
+    const std::string command = prefix + std::string(command_length - prefix.size(), 'a');
+    const auto envelope = encodeTargets(Fleet::Targets{}, command.c_str());
+    ASSERT_EQ(Fleet::MinHeaderSize + command_length + Fleet::SignatureSize, envelope.size());
+    std::array<std::vector<uint8_t>, 2> fragments;
+    for (uint8_t index = 0; index < 2; ++index) {
+      fragments[index].resize(Fleet::MaxPayloadLength);
+      const size_t size = Fleet::fragment(envelope.data(), envelope.size(), index,
+                                         fragments[index].data(), fragments[index].size());
+      ASSERT_GT(size, 0U);
+      ASSERT_LE(size, Fleet::MaxPayloadLength);
+      fragments[index].resize(size);
+    }
+    for (bool reverse : {false, true}) {
+      std::vector<uint8_t> assembled(envelope.size());
+      const unsigned before = g_mock_ed25519_verify_calls;
+      for (unsigned step = 0; step < 2; ++step) {
+        const uint8_t index = reverse ? 1 - step : step;
+        Fleet::Fragment part;
+        ASSERT_TRUE(Fleet::parseFragment(fragments[index].data(), fragments[index].size(), part));
+        EXPECT_EQ(Now, part.sequence);
+        EXPECT_EQ(envelope.size(), part.total_length);
+        EXPECT_EQ(index, part.index);
+        EXPECT_EQ(index ? envelope.size() - Fleet::FragmentDataLength : Fleet::FragmentDataLength,
+                  part.length);
+        EXPECT_EQ(0, memcmp(part.data, envelope.data() + index * Fleet::FragmentDataLength, part.length));
+        std::copy_n(part.data, part.length, assembled.begin() + index * Fleet::FragmentDataLength);
+        EXPECT_FALSE(decode(fragments[index])); // A fragment cannot authorize a command.
+        EXPECT_EQ(before, g_mock_ed25519_verify_calls);
+        Fleet::Fragment duplicate;
+        ASSERT_TRUE(Fleet::parseFragment(fragments[index].data(), fragments[index].size(), duplicate));
+        EXPECT_EQ(part.data, duplicate.data);
+        EXPECT_EQ(part.length, duplicate.length);
+      }
+      EXPECT_EQ(envelope, assembled);
+      ASSERT_TRUE(decode(assembled));
+      EXPECT_EQ(before + 1, g_mock_ed25519_verify_calls);
+      EXPECT_STREQ(command.c_str(), decoded.command);
+      EXPECT_TRUE(decoded.broadcast);
+    }
+  }
+}
+
+TEST_F(FleetCommandTest, FragmentEncoderRejectsInvalidInputBeforeWriting) {
+  const std::string command = "set flood.rule.1 " + std::string(72, 'a');
+  auto envelope = encodeTargets(Fleet::Targets{}, command.c_str());
+  ASSERT_GT(envelope.size(), Fleet::MaxPayloadLength);
+  std::array<uint8_t, Fleet::MaxPayloadLength + 64> output;
+  output.fill(0xA5);
+  for (size_t length : {size_t(0), Fleet::MaxPayloadLength, Fleet::MaxEnvelopeLength + 1})
+    EXPECT_EQ(0U, Fleet::fragment(envelope.data(), length, 0, output.data(), output.size()));
+  EXPECT_EQ(0U, Fleet::fragment(nullptr, envelope.size(), 0, output.data(), output.size()));
+  EXPECT_EQ(0U, Fleet::fragment(envelope.data(), envelope.size(), 0, nullptr, output.size()));
+  for (uint8_t index : {uint8_t(2), uint8_t(255)})
+    EXPECT_EQ(0U, Fleet::fragment(envelope.data(), envelope.size(), index, output.data(), output.size()));
+  EXPECT_EQ(0U, Fleet::fragment(envelope.data(), envelope.size(), 0, output.data(), 164));
+  EXPECT_EQ(0U, Fleet::fragment(envelope.data(), envelope.size(), 1, output.data(),
+                               Fleet::FragmentHeaderSize + envelope.size() - Fleet::FragmentDataLength - 1));
+  envelope[0] = 'X';
+  EXPECT_EQ(0U, Fleet::fragment(envelope.data(), envelope.size(), 0, output.data(), output.size()));
+  for (uint8_t byte : output) EXPECT_EQ(0xA5, byte);
+}
+
+TEST_F(FleetCommandTest, FragmentParserRejectsMalformedLengthIndexAndMagicAndClearsOutput) {
+  const std::string command = "set flood.rule.1 " + std::string(72, 'a');
+  const auto envelope = encodeTargets(Fleet::Targets{}, command.c_str());
+  std::vector<uint8_t> valid(Fleet::MaxPayloadLength);
+  ASSERT_EQ(valid.size(), Fleet::fragment(envelope.data(), envelope.size(), 0, valid.data(), valid.size()));
+  const auto reject = [&](const uint8_t* bytes, size_t length) {
+    Fleet::Fragment part{123, 300, 1, valid.data(), 100};
+    EXPECT_FALSE(Fleet::parseFragment(bytes, length, part));
+    EXPECT_EQ(0U, part.sequence); EXPECT_EQ(0, part.total_length); EXPECT_EQ(0, part.index);
+    EXPECT_EQ(nullptr, part.data); EXPECT_EQ(0U, part.length);
+  };
+  reject(nullptr, valid.size());
+  for (size_t length = 0; length < valid.size(); ++length) reject(valid.data(), length);
+  auto malformed = valid;
+  malformed.push_back(0);
+  reject(malformed.data(), malformed.size());
+  malformed = valid; malformed[0] = 'X'; reject(malformed.data(), malformed.size());
+  for (uint16_t total : {uint16_t(0), uint16_t(165), uint16_t(309), uint16_t(65535)}) {
+    malformed = valid; malformed[8] = uint8_t(total); malformed[9] = uint8_t(total >> 8);
+    reject(malformed.data(), malformed.size());
+  }
+  for (uint8_t index : {uint8_t(1), uint8_t(2), uint8_t(255)}) {
+    malformed = valid; malformed[10] = index; reject(malformed.data(), malformed.size());
+  }
+  valid.resize(Fleet::MaxPayloadLength);
+  valid.resize(Fleet::fragment(envelope.data(), envelope.size(), 1, valid.data(), valid.size()));
+  ASSERT_FALSE(valid.empty());
+  malformed = valid; malformed.pop_back(); reject(malformed.data(), malformed.size());
+  malformed = valid; malformed.push_back(0); reject(malformed.data(), malformed.size());
+  Fleet::Fragment part;
+  ASSERT_TRUE(Fleet::parseFragment(valid.data(), valid.size(), part));
+  EXPECT_EQ(envelope.size() - Fleet::FragmentDataLength, part.length);
+}
+
 } // namespace
 
 int main(int argc, char** argv) {

test/test_fleet_command_crypto.py

@@ -34,6 +34,8 @@ FIXTURE = r'''
 #include <string>
 #include <vector>
 
+inline unsigned fleet_sign_calls = 0;
+
 // The production Identity interface is retained. OpenSSL implements its
 // signatures here so these tests cannot inherit native's always-true mock.
 namespace mesh {
@@ -46,6 +48,7 @@ LocalIdentity::LocalIdentity(RNG* rng) {
   EVP_PKEY_free(key);
 }
 void LocalIdentity::sign(uint8_t* signature, const uint8_t* message, int length) const {
+  ++fleet_sign_calls;
   EVP_PKEY* key = EVP_PKEY_new_raw_private_key(EVP_PKEY_ED25519, nullptr, prv_key, 32);
   EVP_MD_CTX* context = EVP_MD_CTX_new(); assert(key && context);
   assert(EVP_DigestSignInit(context, nullptr, nullptr, nullptr, key) == 1);
@@ -84,26 +87,83 @@ void hex(const uint8_t* bytes, size_t size) {
 
 int main(int argc, char** argv) {
   using Fleet = mesh::FleetCommand;
-  if (argc == 7 && std::string(argv[1]) == "encode") {
+  const std::string operation = argc > 1 ? argv[1] : "";
+  if (argc == 7 && (operation == "encode" || operation == "encode-wide")) {
     FixedRng rng; mesh::LocalIdentity publisher(&rng);
     const auto key = unhex(argv[2]); assert(key.size() == Fleet::KeySize);
     std::array<uint8_t, Fleet::TargetSize> target;
     if (!Fleet::parseTarget(argv[3], target.data())) { std::cout << "reject\n"; return 0; }
-    std::array<uint8_t, Fleet::MaxPayloadLength> output;
+    std::array<uint8_t, Fleet::MaxEnvelopeLength + 64> output;
+    output.fill(0xA5);
+    const unsigned signed_before = fleet_sign_calls;
+    const size_t capacity = operation == "encode-wide" ? output.size() : Fleet::MaxEnvelopeLength;
     const size_t size = Fleet::encode(publisher, key.data(), uint32_t(std::stoul(argv[4])),
-        uint32_t(std::stoul(argv[5])), target.data(), argv[6], output.data(), output.size());
-    if (!size) std::cout << "reject\n"; else hex(output.data(), size);
+        uint32_t(std::stoul(argv[5])), target.data(), argv[6], output.data(), capacity);
+    assert(size <= Fleet::MaxEnvelopeLength);
+    if (!size) {
+      assert(fleet_sign_calls == signed_before);
+      for (uint8_t byte : output) assert(byte == 0xA5);
+      std::cout << "reject\n";
+    } else {
+      assert(fleet_sign_calls == signed_before + 1);
+      for (size_t i = size; i < output.size(); ++i) assert(output[i] == 0xA5);
+      hex(output.data(), size);
+    }
     return 0;
   }
-  if (argc == 7 && std::string(argv[1]) == "encode2") {
+  if (argc == 7 && (operation == "encode2" || operation == "encode2-wide")) {
     FixedRng rng; mesh::LocalIdentity publisher(&rng);
     const auto key = unhex(argv[2]); assert(key.size() == Fleet::KeySize);
     Fleet::Targets targets;
     if (!Fleet::parseTargets(argv[3], strlen(argv[3]), targets)) { std::cout << "reject\n"; return 0; }
-    std::array<uint8_t, Fleet::MaxPayloadLength> output;
+    std::array<uint8_t, Fleet::MaxEnvelopeLength + 64> output;
+    output.fill(0xA5);
+    const unsigned signed_before = fleet_sign_calls;
+    const size_t capacity = operation == "encode2-wide" ? output.size() : Fleet::MaxEnvelopeLength;
     const size_t size = Fleet::encode(publisher, key.data(), uint32_t(std::stoul(argv[4])),
-        uint32_t(std::stoul(argv[5])), targets, argv[6], output.data(), output.size());
-    if (!size) std::cout << "reject\n"; else hex(output.data(), size);
+        uint32_t(std::stoul(argv[5])), targets, argv[6], output.data(), capacity);
+    assert(size <= Fleet::MaxEnvelopeLength);
+    if (!size) {
+      assert(fleet_sign_calls == signed_before);
+      for (uint8_t byte : output) assert(byte == 0xA5);
+      std::cout << "reject\n";
+    } else {
+      assert(fleet_sign_calls == signed_before + 1);
+      for (size_t i = size; i < output.size(); ++i) assert(output[i] == 0xA5);
+      hex(output.data(), size);
+    }
+    return 0;
+  }
+  if (argc == 5 && operation == "fragment") {
+    const auto envelope = unhex(argv[2]);
+    const unsigned index = unsigned(std::stoul(argv[3]));
+    assert(index <= 255);
+    const size_t capacity = std::stoul(argv[4]);
+    std::array<uint8_t, Fleet::MaxPayloadLength + 64> output;
+    assert(capacity <= output.size()); output.fill(0xA5);
+    const unsigned signed_before = fleet_sign_calls;
+    const size_t size = Fleet::fragment(envelope.data(), envelope.size(), uint8_t(index),
+                                        output.data(), capacity);
+    assert(fleet_sign_calls == signed_before && size <= Fleet::MaxPayloadLength);
+    if (!size) {
+      for (uint8_t byte : output) assert(byte == 0xA5);
+      std::cout << "reject\n";
+    } else {
+      for (size_t i = size; i < output.size(); ++i) assert(output[i] == 0xA5);
+      hex(output.data(), size);
+    }
+    return 0;
+  }
+  if (argc == 3 && operation == "parse-fragment") {
+    const auto payload = unhex(argv[2]);
+    Fleet::Fragment output{123, 300, 1, payload.data(), 100};
+    if (!Fleet::parseFragment(payload.data(), payload.size(), output)) {
+      assert(!output.sequence && !output.total_length && !output.index && !output.data && !output.length);
+      std::cout << "reject\n";
+    } else {
+      std::cout << output.sequence << '\n' << output.total_length << '\n' << unsigned(output.index) << '\n';
+      hex(output.data, output.length);
+    }
     return 0;
   }
   if (argc == 7 && (std::string(argv[1]) == "decode" || std::string(argv[1]) == "decode2")) {
@@ -142,6 +202,13 @@ def envelope2(entries=(), command="get radio2", sequence=NOW, expires=NOW + 120,
     return unsigned + publisher.sign(DOMAIN + channel + unsigned)
 
 
+def fragment_frame(payload, index):
+    """Independent wire framing; the signature stays in the whole envelope."""
+    sequence = struct.unpack("<I", payload[4:8])[0]
+    return (b"FMP1" + struct.pack("<IHB", sequence, len(payload), index)
+            + payload[index * 154:(index + 1) * 154])
+
+
 class FleetCommandCryptoTests(unittest.TestCase):
     @classmethod
     def setUpClass(cls):
@@ -299,8 +366,18 @@ class FleetCommandCryptoTests(unittest.TestCase):
         self.assertEqual(len(maximum), 72)
         self.assertNotEqual(self.decode(envelope(maximum)), "reject")
         self.assertNotEqual(self.decode(envelope2(entries, maximum)), "reject")
-        oversized = prefix + "0" + maximum[len(prefix):]
-        self.assertEqual(self.decode(envelope(oversized)), "reject")
+        larger = prefix + "0" + maximum[len(prefix):]
+        self.assertNotEqual(self.decode(envelope(larger)), "reject")
+        self.assertNotEqual(self.decode(envelope2(entries, larger)), "reject")
+        target_maximum = prefix + "0" * (218 - len(prefix) - len(suffix)) + suffix
+        self.assertNotEqual(self.decode(envelope2(entries, target_maximum)), "reject")
+        too_large = prefix + "0" + target_maximum[len(prefix):]
+        self.assertEqual(self.decode(envelope2(entries, too_large)), "reject")
+        broadcast_maximum = prefix + "0" * (230 - len(prefix) - len(suffix)) + suffix
+        expected = envelope2(command=broadcast_maximum)
+        self.assertEqual(self.run_tool("encode2", CHANNEL.hex(), "all", NOW, NOW + 120,
+                                       broadcast_maximum), expected.hex())
+        self.assertNotEqual(self.decode(expected), "reject")
 
     def test_clock_management_has_real_signatures_and_no_prefix_aliases(self):
         for command in ["clock", "clock sync", "time 1735689600", "time 1800000000", "time 4294967295"]:
@@ -333,12 +410,18 @@ class FleetCommandCryptoTests(unittest.TestCase):
                 self.assertEqual(self.decode(forged), "reject")
                 self.assertNotEqual(self.decode(original, now=NOW + 120), "reject")
 
-    def test_exact_and_oversized_command_boundaries(self):
-        maximum = "set flood.rule.1 " + "a" * (72 - len("set flood.rule.1 "))
+    def test_fixed_target_exact_limit_cannot_be_raised_by_destination_capacity(self):
+        maximum = "set flood.rule.1 " + "a" * (215 - len("set flood.rule.1 "))
         accepted = envelope(maximum)
-        self.assertEqual(len(accepted), 165)
+        self.assertEqual(len(accepted), 308)
         self.assertNotEqual(self.decode(accepted), "reject")
         self.assertEqual(self.decode(envelope(maximum + "a")), "reject")
+        for operation in ["encode", "encode-wide"]:
+            self.assertEqual(self.run_tool(operation, CHANNEL.hex(), "all", NOW, NOW + 120, maximum),
+                             accepted.hex())
+            for length in [216, 230, 231]:
+                command = "set flood.rule.1 " + "a" * (length - len("set flood.rule.1 "))
+                self.assertEqual(self.run_tool(operation, CHANNEL.hex(), "all", NOW, NOW + 120, command), "reject")
 
     def test_valid_signature_does_not_override_clock_policy(self):
         for sequence, expires, now in [
@@ -412,23 +495,193 @@ class FleetCommandCryptoTests(unittest.TestCase):
         self.assertEqual(self.decode(original, publisher=OTHER_PUBLIC), "reject")
         self.assertEqual(self.decode(original, channel=bytes(range(17, 33))), "reject")
 
-    def test_compact_broadcast_keeps_72_byte_command_and_packet_framing_limits(self):
+    def test_compact_broadcast_fits_230_byte_signed_command_and_framing_limits(self):
         prefix = "set flood.rule.1 "
-        maximum = prefix + "a" * (72 - len(prefix))
+        maximum = prefix + "a" * (230 - len(prefix))
         expected = envelope2(command=maximum)
         actual = self.run_tool("encode2", CHANNEL.hex(), "all", NOW, NOW + 120, maximum)
         self.assertEqual(actual, expected.hex())
-        self.assertEqual(len(expected), 150)
-        self.assertEqual(len(envelope(maximum)) - len(expected), 15)
-        self.assertNotEqual(self.decode(expected), "reject")
+        self.assertEqual(len(expected), 308)
+        PUBLISHER.public_key().verify(expected[-64:], DOMAIN + CHANNEL + expected[:-64])
+        for node in [PUBLIC, OTHER_PUBLIC]:
+            self.assertEqual(self.decode(expected, self_key=node, metadata=True),
+                             f"{NOW}\n{NOW + 120}\n{maximum}\n1")
         oversized = maximum + "a"
         self.assertEqual(self.decode(envelope2(command=oversized)), "reject")
         self.assertEqual(self.run_tool("encode2", CHANNEL.hex(), "all", NOW, NOW + 120, oversized), "reject")
+        self.assertEqual(self.run_tool("encode2-wide", CHANNEL.hex(), "all", NOW, NOW + 120, oversized), "reject")
+        for invalid in [b"\n", b";", b"\x80", b"\0"]:
+            malformed = maximum[:-1].encode("ascii") + invalid
+            self.assertEqual(self.decode(envelope2(command=malformed)), "reject")
         for length in [0, 12, 13, 14, len(expected) - 64, len(expected) - 1]:
             with self.subTest(length=length):
                 self.assertEqual(self.decode(expected[:length]), "reject")
         self.assertEqual(self.decode(expected + b"\0"), "reject")
 
+    def test_each_target_form_has_exact_packet_budget_and_atomic_oversize_rejection(self):
+        short, long = PUBLIC[:4].hex(), PUBLIC[:6].hex()
+        digest = hashlib.sha256(PUBLIC).digest()[:16]
+        for token, entries, maximum_length in [
+                ("all", [], 230), (short, [(4, PUBLIC[:4])], 225),
+                (long, [(6, PUBLIC[:6])], 223), (PUBLIC.hex(), None, 215),
+                (f"{short},{short}", [(4, PUBLIC[:4])] * 2, 220),
+                (f"{short},{long}", [(4, PUBLIC[:4]), (6, PUBLIC[:6])], 218),
+                (f"{PUBLIC.hex()},{short}", [(16, digest), (4, PUBLIC[:4])], 208),
+                (f"{PUBLIC.hex()},{PUBLIC.hex()}", [(16, digest)] * 2, 196)]:
+            with self.subTest(token=token, maximum=maximum_length):
+                prefix = "set flood.rule.1 "
+                maximum = prefix + "a" * (maximum_length - len(prefix))
+                expected = (envelope(maximum, target=digest) if entries is None
+                            else envelope2(entries, maximum))
+                self.assertEqual(len(expected), 308)
+                self.assertEqual(self.run_tool("encode2", CHANNEL.hex(), token, NOW, NOW + 120, maximum),
+                                 expected.hex())
+                self.assertNotEqual(self.decode(expected), "reject")
+                # The fixture verifies no writes and no signature calls on
+                # rejection, even when its output buffer exceeds packet size.
+                oversized = maximum + "a"
+                for operation in ["encode2", "encode2-wide"]:
+                    self.assertEqual(self.run_tool(operation, CHANNEL.hex(), token, NOW, NOW + 120,
+                                                   oversized), "reject")
+                signed_oversized = (envelope(oversized, target=digest) if entries is None
+                                    else envelope2(entries, oversized))
+                self.assertEqual(self.decode(signed_oversized), "reject")
+                altered = bytearray(expected)
+                altered[-65] = ord("b")
+                self.assertEqual(self.decode(altered), "reject")
+
+    def test_maximum_broadcast_signature_covers_final_byte_and_230_byte_length(self):
+        prefix = "set flood.rule.1 "
+        maximum = prefix + "a" * (230 - len(prefix))
+        original = envelope2(command=maximum)
+        self.assertNotEqual(self.decode(original), "reject")
+        for command in [maximum[:-1] + "b", maximum[:-1]]:
+            changed = envelope2(command=command)
+            self.assertNotEqual(self.decode(changed), "reject")
+            self.assertEqual(self.decode(changed[:-64] + original[-64:]), "reject")
+        for length_byte in [0, 87, 229, 231, 255]:
+            changed = bytearray(original)
+            changed[13] = length_byte
+            self.assertEqual(self.decode(changed), "reject")
+        for offset in range(len(original) - 64, len(original)):
+            with self.subTest(signature_byte=offset):
+                changed = bytearray(original)
+                changed[offset] ^= 1
+                self.assertEqual(self.decode(changed), "reject")
+
+    def test_single_packet_boundaries_and_first_two_fragment_command(self):
+        digest = hashlib.sha256(PUBLIC).digest()[:16]
+        for token, entries, maximum in [
+                ("all", [], 87), (PUBLIC[:4].hex(), [(4, PUBLIC[:4])], 82),
+                (PUBLIC[:6].hex(), [(6, PUBLIC[:6])], 80),
+                (PUBLIC.hex(), None, 72)]:
+            with self.subTest(token=token):
+                prefix = "set flood.rule.1 "
+                command = prefix + "a" * (maximum - len(prefix))
+                single = (envelope(command, target=digest) if entries is None
+                          else envelope2(entries, command))
+                self.assertEqual(len(single), 165)
+                self.assertEqual(self.run_tool("encode2", CHANNEL.hex(), token, NOW, NOW + 120,
+                                               command), single.hex())
+                self.assertNotEqual(self.decode(single), "reject")
+                for index in [0, 1]:
+                    self.assertEqual(self.run_tool("fragment", single.hex(), index, 165), "reject")
+                longer = (envelope(command + "a", target=digest) if entries is None
+                          else envelope2(entries, command + "a"))
+                self.assertEqual(len(longer), 166)
+                self.assertEqual(self.run_tool("encode2", CHANNEL.hex(), token, NOW, NOW + 120,
+                                               command + "a"), longer.hex())
+                self.assertNotEqual(self.decode(longer), "reject")
+                for index, size in [(0, 165), (1, 23)]:
+                    actual = bytes.fromhex(self.run_tool("fragment", longer.hex(), index, 165))
+                    self.assertEqual(actual, fragment_frame(longer, index))
+                    self.assertEqual(len(actual), size)
+
+    def test_two_fragments_reassemble_out_of_order_with_one_real_signature(self):
+        for size in [166, 167, 307, 308]:
+            with self.subTest(size=size):
+                prefix = "set flood.rule.1 "
+                command = prefix + "a" * (size - 78 - len(prefix))
+                original = envelope2(command=command)
+                self.assertEqual(len(original), size)
+                # The fixture requires exactly one signature call for encode
+                # and no signature calls while cutting either fragment.
+                self.assertEqual(self.run_tool("encode2", CHANNEL.hex(), "all", NOW, NOW + 120,
+                                               command), original.hex())
+                parts = [bytes.fromhex(self.run_tool("fragment", original.hex(), index, 165))
+                         for index in [0, 1]]
+                for index, part in enumerate(parts):
+                    self.assertEqual(part, fragment_frame(original, index))
+                    self.assertLessEqual(len(part), 165)
+                    self.assertEqual(self.run_tool("parse-fragment", part.hex()),
+                                     f"{NOW}\n{size}\n{index}\n{part[11:].hex()}")
+                    self.assertEqual(self.decode(part), "reject")
+                for order in [(0, 1), (1, 0), (0, 0, 1), (1, 1, 0)]:
+                    with self.subTest(order=order):
+                        assembly = bytearray(size)
+                        for index in order:
+                            part = parts[index]
+                            assembly[index * 154:index * 154 + len(part) - 11] = part[11:]
+                        assembly = bytes(assembly)
+                        self.assertEqual(assembly, original)
+                        PUBLISHER.public_key().verify(assembly[-64:], DOMAIN + CHANNEL + assembly[:-64])
+                        for node in [PUBLIC, OTHER_PUBLIC]:
+                            self.assertEqual(self.decode(assembly, self_key=node, metadata=True),
+                                             f"{NOW}\n{NOW + 120}\n{command}\n1")
+
+    def test_fragment_codec_bounds_and_exact_shapes_reject_without_output_writes(self):
+        prefix = "set flood.rule.1 "
+        original = envelope2(command=prefix + "a" * (88 - len(prefix)))
+        parts = [fragment_frame(original, index) for index in [0, 1]]
+        for source in [b"", original[:165], original + b"a" * (309 - len(original)),
+                       b"FMX1" + original[4:], b"FMP1" + original[4:]]:
+            with self.subTest(source_length=len(source), magic=source[:4]):
+                self.assertEqual(self.run_tool("fragment", source.hex(), 0, 229), "reject")
+        for index in [2, 255]:
+            self.assertEqual(self.run_tool("fragment", original.hex(), index, 229), "reject")
+        for index, required in [(0, 165), (1, 23)]:
+            for capacity in [0, 10, required - 1]:
+                self.assertEqual(self.run_tool("fragment", original.hex(), index, capacity), "reject")
+            self.assertEqual(self.run_tool("fragment", original.hex(), index, 229), parts[index].hex())
+        malformed = [b"", b"FMP1", parts[0][:10], parts[0][:11],
+                     parts[0][:-1], parts[0] + b"\0", parts[1][:-1], parts[1] + b"\0",
+                     b"FMX1" + parts[0][4:]]
+        for total in [0, 154, 165, 309, 65535]:
+            malformed.append(parts[0][:8] + struct.pack("<H", total) + parts[0][10:])
+        for index in [1, 2, 255]:
+            malformed.append(parts[0][:10] + bytes([index]) + parts[0][11:])
+        malformed.append(parts[1][:10] + b"\0" + parts[1][11:])
+        for part in malformed:
+            with self.subTest(part=part.hex()):
+                self.assertEqual(self.run_tool("parse-fragment", part.hex()), "reject")
+
+    def test_fragment_tampering_and_mixed_commands_cannot_forge_complete_signature(self):
+        prefix = "set flood.rule.1 "
+        command = prefix + "a" * (230 - len(prefix))
+        original = envelope2(command=command)
+        parts = [fragment_frame(original, index) for index in [0, 1]]
+        self.assertNotEqual(self.decode(original), "reject")
+        # Every chunk contains signed data. Matching frame metadata and an
+        # allowed changed command cannot substitute for the full signature.
+        for index, data_offset in [(0, 14 + len(prefix)), (1, 0), (1, 153)]:
+            changed = bytearray(parts[index])
+            changed[11 + data_offset] ^= 1
+            self.assertNotEqual(self.run_tool("parse-fragment", changed.hex()), "reject")
+            chunks = [parts[0][11:], parts[1][11:]]
+            chunks[index] = changed[11:]
+            self.assertEqual(self.decode(b"".join(chunks)), "reject")
+        different = envelope2(command=prefix + "b" * (230 - len(prefix)))
+        self.assertNotEqual(self.decode(different), "reject")
+        different_parts = [fragment_frame(different, index) for index in [0, 1]]
+        for combined in [parts[0][11:] + different_parts[1][11:],
+                         different_parts[0][11:] + parts[1][11:]]:
+            self.assertEqual(self.decode(combined), "reject")
+        # Frame metadata is deliberately unauthenticated and only admits a
+        # bounded assembly. Runtime must compare it to the signed sequence.
+        changed_sequence = parts[0][:4] + struct.pack("<I", NOW + 1) + parts[0][8:]
+        self.assertTrue(self.run_tool("parse-fragment", changed_sequence.hex()).startswith(f"{NOW + 1}\n"))
+        self.assertEqual(self.decode(changed_sequence), "reject")
+
     def test_real_signature_round_trip_for_mixed_prefixes_and_full_keys(self):
         entries = [(4, PUBLIC[:4]), (6, OTHER_PUBLIC[:6]), (16, hashlib.sha256(PUBLIC).digest()[:16])]
         token = f"{PUBLIC[:4].hex()},{OTHER_PUBLIC[:6].hex()},{PUBLIC.hex()}"
@@ -502,10 +755,10 @@ class FleetCommandCryptoTests(unittest.TestCase):
                 self.assertEqual(self.run_tool("encode2", CHANNEL.hex(), token, NOW, NOW + 120, "get radio2"), "reject")
 
     def test_fmc2_packet_capacity_is_checked_before_signing_or_sending(self):
-        maximum = "set flood.rule.1 " + "a" * (72 - len("set flood.rule.1 "))
+        maximum = "set flood.rule.1 " + "a" * (215 - len("set flood.rule.1 "))
         entries = [(4, PUBLIC[:4])] * 3
         expected = envelope2(entries, maximum)
-        self.assertEqual(len(expected), 165)
+        self.assertEqual(len(expected), 308)
         token = ",".join([PUBLIC[:4].hex()] * 3)
         actual = self.run_tool("encode2", CHANNEL.hex(), token, NOW, NOW + 120, maximum)
         self.assertEqual(actual, expected.hex())
@@ -516,7 +769,7 @@ class FleetCommandCryptoTests(unittest.TestCase):
         for count in [15, 16, 17]:
             token = ",".join([PUBLIC[:4].hex()] * count)
             result = self.run_tool("encode2", CHANNEL.hex(), token, NOW, NOW + 120, "get radio2")
-            self.assertEqual(result == "reject", count > 15)
+            self.assertNotEqual(result, "reject")
 
     def test_fmc2_rejects_short_trailing_and_signed_forbidden_commands(self):
         original = envelope2([(4, PUBLIC[:4]), (6, OTHER_PUBLIC[:6])])

test/test_nrf52_extrafs_contract.py

@@ -220,8 +220,8 @@ class Nrf52ExtraFsContractTest(unittest.TestCase):
         next_at = handler.index("cmd_frame[0] == CMD_SET_ADVERT_NAME", get_at)
         get_contacts = handler[get_at:next_at]
         incomplete_at = get_contacts.index("hasIncompleteContactLoad()")
-        lock_at = get_contacts.index("_serial->lockReplyRoute()")
-        self.assertLess(incomplete_at, lock_at)
+        route_at = get_contacts.index("_iter_reply_route = _serial->captureReplyRoute()")
+        self.assertLess(incomplete_at, route_at)
         self.assertIn("ERR_CODE_FILE_IO_ERROR", get_contacts)
 
         base = (ROOT / "src" / "helpers" / "BaseChatMesh.cpp").read_text(

test/test_transport_filter/test_transport_filter.cpp

@@ -204,6 +204,48 @@ TEST_F(TransportFilter, InvalidModesAndActionsDoNotReplaceSavedRule) {
   EXPECT_FALSE(blocked(p, RULE_MODE_RADIO));
 }
 
+TEST_F(TransportFilter, TwoPacketFleetRulePreservesActionAfterLongParameters) {
+  const std::string prefix = "set flood.rule.1 type=grp_txt channel=" + std::string(64, '1')
+      + " prefix=AABBCC,DDEEFF,112233 hops=6+ priority=255 suspend=tempradio"
+        " mode=radio hashbytes=3 in=any";
+  const std::string action = " action=drop";
+  ASSERT_LT(prefix.size() + action.size(), 230U);
+  const std::string setter = prefix + std::string(230 - prefix.size() - action.size(), ' ') + action;
+  ASSERT_EQ(230U, setter.size());
+  ASSERT_GT(setter.size() - strlen("set flood.rule.1 "), 192U);
+  command(setter.c_str()); ASSERT_EQ(0, strncmp(reply, "OK", 2)) << reply;
+  const auto saved = fs.files["/flood_filter"];
+
+  mesh::Packet p;
+  p.header = (PAYLOAD_TYPE_GRP_TXT << PH_TYPE_SHIFT) | ROUTE_TYPE_FLOOD;
+  p.setPathHashSizeAndCount(3, 6);
+  memset(p.path, 0, p.getPathByteLen());
+  const uint8_t path[] = {0xAA,0xBB,0xCC,0xDD,0xEE,0xFF,0x11,0x22,0x33};
+  memcpy(p.path, path, sizeof(path));
+  uint8_t key[PUB_KEY_SIZE];memset(key, 0x11, sizeof(key));
+  mesh::Utils::sha256(p.payload, 1, key, sizeof(key));
+  const uint8_t text[] = "fleet: sample";
+  p.payload_len = 1 + mesh::Utils::encryptThenMAC(key, p.payload + 1, text, sizeof(text));
+  EXPECT_TRUE(blocked(p, RULE_MODE_RADIO));
+
+  std::string stop = setter;stop.replace(stop.size() - 4, 4, "stop");
+  command(stop.c_str()); ASSERT_EQ(0, strncmp(reply, "OK", 2)) << reply;
+  EXPECT_FALSE(blocked(p, RULE_MODE_RADIO));
+  EXPECT_NE(saved, fs.files["/flood_filter"]);
+
+  const auto unchanged = fs.files["/flood_filter"];
+  command(("set flood.rule.1 any " + std::string(256, ' ') + "drop").c_str());
+  EXPECT_EQ(0, strncmp(reply, "Err - rule parameters too long", 29));
+  EXPECT_EQ(unchanged, fs.files["/flood_filter"]);
+}
+
+TEST_F(TransportFilter, OverlongRateTokenCannotDiscardItsInvalidSuffix) {
+  install("radio");const auto saved = fs.files["/flood_filter"];
+  command(("set flood.rule.3 any q=" + std::string(23, '0') + "x stop").c_str());
+  EXPECT_EQ(0, strncmp(reply, "Err - rate value too long", 24)) << reply;
+  EXPECT_EQ(saved, fs.files["/flood_filter"]);
+}
+
 TEST_F(TransportFilter, CombinedModeSharesRateAndResetsAfterMinute) {
   command("set flood.rule.3 type=any mode=bridge,cross rate=1/min");
   ASSERT_EQ(0, strncmp(reply, "OK", 2)) << reply;
@@ -490,6 +532,35 @@ TEST_F(TransportFilter, PrivateKeyReferencesCopyLocallyWithoutDisclosingKey) {
   EXPECT_EQ(empty, fs.files["/flood_filter"]);
 }
 
+TEST_F(TransportFilter, WholeFleetMaximumRulePreservesItsFinalDropAction) {
+  const std::string setter = "set flood.rule.1 5 c=" + std::string(64, '1') + " d";
+  ASSERT_EQ(87U, setter.size());
+  command(setter.c_str());
+  ASSERT_EQ(0, strncmp(reply, "OK", 2)) << reply;
+
+  mesh::Packet p;
+  p.header = (PAYLOAD_TYPE_GRP_TXT << PH_TYPE_SHIFT) | ROUTE_TYPE_FLOOD;
+  p.setPathHashSizeAndCount(1, 0);
+  uint8_t key[PUB_KEY_SIZE];
+  memset(key, 0x11, sizeof(key));
+  mesh::Utils::sha256(p.payload, 1, key, sizeof(key));
+  const uint8_t text[] = "fleet: sample";
+  p.payload_len = 1 + mesh::Utils::encryptThenMAC(key, p.payload + 1, text, sizeof(text));
+  EXPECT_TRUE(blocked(p, RULE_MODE_RADIO));
+
+  command("get fr.1");
+  const std::string formatted = reply;
+  ASSERT_GE(formatted.size(), 2U);
+  EXPECT_EQ(" d", formatted.substr(formatted.size() - 2));
+
+  // A stop action at the same final byte has different forwarding behavior.
+  std::string stop = setter;
+  stop.back() = 's';
+  command(stop.c_str());
+  ASSERT_EQ(0, strncmp(reply, "OK", 2)) << reply;
+  EXPECT_FALSE(blocked(p, RULE_MODE_RADIO));
+}
+
 TEST(TransportModes, KeyFingerprintCollisionsCannotSelectAnArbitrarySecret) {
   struct Row {
     bool active = true;