integration: preserve deployed 1.8.0 OTA work

core/archipelago/src/mesh/listener/assist.rs

@@ -181,7 +181,10 @@ async fn is_sender_allowed(
         match peers.get(&sender_contact_id) {
             // Match identity on the bound archipelago key (stable, advert/
             // federation-verified), not the firmware routing key.
-            Some(p) => (p.identity_pubkey_hex().map(|s| s.to_string()), p.did.clone()),
+            Some(p) => (
+                p.identity_pubkey_hex().map(|s| s.to_string()),
+                p.did.clone(),
+            ),
             None => (None, None),
         }
     };

core/archipelago/src/mesh/listener/decode.rs

@@ -314,17 +314,66 @@ pub(super) async fn try_chunk_reassemble(
 
 /// Look up a peer by pubkey hex prefix. Returns (contact_id, display_name).
 pub(super) async fn resolve_peer(state: &Arc<MeshState>, sender_prefix: &str) -> (u32, String) {
-    let peers = state.peers.read().await;
-    peers
-        .values()
-        .find(|p| {
+    {
+        let peers = state.peers.read().await;
+        if let Some(peer) = peers.values().find(|p| {
             p.pubkey_hex
                 .as_ref()
                 .map(|k| k.starts_with(sender_prefix))
                 .unwrap_or(false)
-        })
-        .map(|p| (p.contact_id, p.advert_name.clone()))
-        .unwrap_or((0, sender_prefix.to_string()))
+        }) {
+            return (peer.contact_id, peer.advert_name.clone());
+        }
+    }
+
+    if let Some((node_num, pubkey_hex, name)) = meshtastic_peer_from_prefix(sender_prefix) {
+        let peer = MeshPeer {
+            contact_id: node_num,
+            advert_name: name.clone(),
+            did: None,
+            pubkey_hex: Some(pubkey_hex),
+            arch_pubkey_hex: None,
+            x25519_pubkey: None,
+            rssi: None,
+            snr: None,
+            last_heard: chrono::Utc::now().to_rfc3339(),
+            hops: 0xff,
+            last_advert: 0,
+            reachable: true,
+        };
+        let is_new = {
+            let mut peers = state.peers.write().await;
+            peers.insert(node_num, peer.clone()).is_none()
+        };
+        state.update_peer_count().await;
+        let _ = state.event_tx.send(if is_new {
+            MeshEvent::PeerDiscovered(peer)
+        } else {
+            MeshEvent::PeerUpdated(peer)
+        });
+        return (node_num, name);
+    }
+
+    (0, sender_prefix.to_string())
+}
+
+fn meshtastic_peer_from_prefix(sender_prefix: &str) -> Option<(u32, String, String)> {
+    if sender_prefix.len() < 12 {
+        return None;
+    }
+    let bytes = hex::decode(&sender_prefix[..12]).ok()?;
+    if bytes.len() != 6 || bytes[4] != b'm' || bytes[5] != b'e' {
+        return None;
+    }
+    let node_num = u32::from_le_bytes([bytes[0], bytes[1], bytes[2], bytes[3]]);
+    if node_num == 0 || node_num == u32::MAX {
+        return None;
+    }
+    let mut full_key = [0u8; 32];
+    full_key[..4].copy_from_slice(&node_num.to_le_bytes());
+    full_key[4..15].copy_from_slice(b"meshtastic:");
+    let name = format!("Meshtastic !{:08x}", node_num);
+    Some((node_num, hex::encode(full_key), name))
 }
 
 /// Store a plain-text (non-typed) message and emit an event.
@@ -333,6 +382,16 @@ pub(super) async fn store_plain_message(
     contact_id: u32,
     peer_name: &str,
     text: &str,
+) {
+    store_plain_message_with_encryption(state, contact_id, peer_name, text, false).await;
+}
+
+pub(super) async fn store_plain_message_with_encryption(
+    state: &Arc<MeshState>,
+    contact_id: u32,
+    peer_name: &str,
+    text: &str,
+    encrypted: bool,
 ) {
     let msg_id = state.next_id().await;
     let msg = MeshMessage {
@@ -343,7 +402,7 @@ pub(super) async fn store_plain_message(
         plaintext: text.to_string(),
         timestamp: chrono::Utc::now().to_rfc3339(),
         delivered: true,
-        encrypted: false,
+        encrypted,
         message_type: "text".to_string(),
         typed_payload: None,
         sender_pubkey: None,

core/archipelago/src/mesh/listener/frames.rs

@@ -4,7 +4,8 @@ use super::super::message_types::TypedEnvelope;
 use super::super::protocol;
 use super::decode::{
     handle_identity_received, is_mc_chunk_frame, resolve_peer, store_plain_message,
-    try_base64_typed, try_chunk_reassemble, try_decrypt_base64, try_decrypt_ratchet_base64,
+    store_plain_message_with_encryption, try_base64_typed, try_chunk_reassemble,
+    try_decrypt_base64, try_decrypt_ratchet_base64,
 };
 use super::dispatch::handle_typed_message;
 use super::MeshState;
@@ -62,11 +63,12 @@ pub(super) async fn handle_frame(
             return true; // Signal caller to sync immediately
         }
 
-        protocol::RESP_CONTACT_MSG_V3 => {
+        protocol::RESP_CONTACT_MSG_V3 | protocol::RESP_CONTACT_MSG_V3_E2E => {
             // Direct message received (v3 format) — check for typed envelope first
             match protocol::parse_contact_msg_v3_raw(&frame.data) {
                 Ok((sender_prefix, payload, _snr)) => {
                     if !payload.is_empty() {
+                        let encrypted = frame.code == protocol::RESP_CONTACT_MSG_V3_E2E;
                         let (contact_id, name) = resolve_peer(state, &sender_prefix).await;
                         if TypedEnvelope::is_typed(&payload) {
                             handle_typed_message(&payload, contact_id, &name, state).await;
@@ -86,7 +88,10 @@ pub(super) async fn handle_frame(
                             handle_typed_message(&decoded, contact_id, &name, state).await;
                         } else if !payload.starts_with(b"MC") {
                             let text = String::from_utf8_lossy(&payload).to_string();
-                            store_plain_message(state, contact_id, &name, &text).await;
+                            store_plain_message_with_encryption(
+                                state, contact_id, &name, &text, encrypted,
+                            )
+                            .await;
                             info!(from = %sender_prefix, "Received mesh DM (v3)");
                         }
                     }

core/archipelago/src/mesh/listener/session.rs

@@ -407,8 +407,13 @@ async fn refresh_contacts(device: &mut MeshRadioDevice, state: &Arc<MeshState>)
             // user-controlled feature; until then every firmware contact is
             // surfaced. `radio_contact_blocklist` is retained but unused.
             let mut peers = state.peers.write().await;
+            let is_meshtastic = matches!(device.device_type(), DeviceType::Meshtastic);
             for (idx, contact) in contacts.iter().enumerate() {
-                let contact_id = idx as u32;
+                let contact_id = if is_meshtastic {
+                    meshtastic_contact_id(&contact.public_key_hex).unwrap_or(idx as u32)
+                } else {
+                    idx as u32
+                };
                 let existing = peers.get(&contact_id);
                 let peer = super::super::types::MeshPeer {
                     contact_id,
@@ -482,6 +487,19 @@ async fn refresh_contacts(device: &mut MeshRadioDevice, state: &Arc<MeshState>)
     }
 }
 
+fn meshtastic_contact_id(public_key_hex: &str) -> Option<u32> {
+    let bytes = hex::decode(public_key_hex).ok()?;
+    if bytes.len() < 15 || &bytes[4..15] != b"meshtastic:" {
+        return None;
+    }
+    let node_num = u32::from_le_bytes([bytes[0], bytes[1], bytes[2], bytes[3]]);
+    if node_num == 0 || node_num == u32::MAX {
+        None
+    } else {
+        Some(node_num)
+    }
+}
+
 /// Drain any queued messages from the device.
 /// Returns `true` if a write/communication error occurred (for failure tracking).
 async fn sync_queued_messages(

core/archipelago/src/mesh/meshtastic.rs

@@ -22,6 +22,7 @@ const START2: u8 = 0xc3;
 const TO_RADIO_MAX: usize = 512;
 const BROADCAST_NUM: u32 = 0xffff_ffff;
 const TEXT_MESSAGE_APP: u32 = 1;
+const POSITION_APP: u32 = 3;
 /// Meshtastic PortNum for NodeInfo (identity) packets — used to actively
 /// advertise ourselves over the air so neighbours discover us, the parity
 /// equivalent of meshcore's self-advert.
@@ -33,6 +34,8 @@ const ADMIN_SET_OWNER_FIELD: u64 = 32;
 /// Meshtastic firmware caps long_name at ~40 bytes and short_name at 4 bytes.
 const MESHTASTIC_LONG_NAME_MAX: usize = 39;
 const MESHTASTIC_SHORT_NAME_MAX: usize = 4;
+const STALE_RX_SECS: u32 = 24 * 60 * 60;
+const PLAUSIBLE_RX_EPOCH_SECS: u32 = 1_700_000_000; // 2023-11-14
 
 const TO_RADIO_PACKET: u64 = 1;
 const TO_RADIO_WANT_CONFIG_ID: u64 = 3;
@@ -45,6 +48,11 @@ const FROM_RADIO_NODE_INFO: u64 = 4;
 const FROM_RADIO_CONFIG: u64 = 5;
 const FROM_RADIO_CONFIG_COMPLETE_ID: u64 = 7;
 const FROM_RADIO_REBOOTED: u64 = 8;
+/// Upper bound for a single Meshtastic serial API protobuf frame. The serial
+/// stream can contain firmware log text, so this is also used to reject false
+/// 0x94c3 markers found inside logs instead of waiting forever for a bogus
+/// length.
+const FROM_RADIO_MAX: usize = 4096;
 
 /// AdminMessage.set_config oneof field number (carries a `Config`). NB: 33 is
 /// `set_channel` — `set_config` is 34 (verified against meshtastic/protobufs).
@@ -53,6 +61,15 @@ const ADMIN_SET_CONFIG_FIELD: u64 = 34;
 const ADMIN_SET_CHANNEL_FIELD: u64 = 33;
 /// FromRadio.channel (field 10): a `Channel` streamed during want_config.
 const FROM_RADIO_CHANNEL: u64 = 10;
+const FROM_RADIO_QUEUE_STATUS: u64 = 11;
+const FROM_RADIO_XMODEM_PACKET: u64 = 12;
+const FROM_RADIO_METADATA: u64 = 13;
+const FROM_RADIO_MQTT_CLIENT_PROXY_MESSAGE: u64 = 14;
+const FROM_RADIO_FILE_INFO: u64 = 15;
+const FROM_RADIO_CLIENT_NOTIFICATION: u64 = 16;
+const FROM_RADIO_DEVICE_UI_CONFIG: u64 = 17;
+const FROM_RADIO_LOCKDOWN_STATUS: u64 = 18;
+const FROM_RADIO_REGION_PRESETS: u64 = 19;
 /// Channel.role value for the PRIMARY channel (broadcasts ride here).
 const CHANNEL_ROLE_PRIMARY: u64 = 1;
 /// Config.lora oneof field number (carries a `LoRaConfig`).
@@ -386,7 +403,23 @@ impl MeshtasticDevice {
     }
 
     pub async fn send_self_advert(&mut self) -> Result<()> {
-        self.send_to_radio(&encode_heartbeat()).await
+        self.send_to_radio(&encode_heartbeat()).await?;
+        self.send_time_broadcast().await
+    }
+
+    /// Broadcast a minimal Position payload carrying current epoch time. The
+    /// Meshtastic protobuf explicitly documents `Position.time` as the path for
+    /// phone/API clients to set time on mesh devices without GPS/RTC. This keeps
+    /// stock Meshtastic clients from rendering incoming Archipelago-originated
+    /// packets as Jan 1 1970 when their radio clock is unset.
+    pub async fn send_time_broadcast(&mut self) -> Result<()> {
+        let now = now_unix_secs();
+        let mut position = Vec::new();
+        encode_fixed32_field(4, now, &mut position);
+        encode_fixed32_field(7, now, &mut position);
+        let packet = encode_mesh_packet(BROADCAST_NUM, POSITION_APP, &position);
+        self.send_to_radio(&encode_to_radio_variant(TO_RADIO_PACKET, &packet))
+            .await
     }
 
     /// Build our own `User` protobuf (id/long_name/short_name) for a NodeInfo
@@ -595,8 +628,10 @@ impl MeshtasticDevice {
                     // Bound memory if it's a pure-debug flood with no frames:
                     // keep only from the last possible frame-start marker.
                     if self.read_buf.len() > 64 * 1024 {
-                        if let Some(pos) =
-                            self.read_buf.windows(2).rposition(|w| w == [START1, START2])
+                        if let Some(pos) = self
+                            .read_buf
+                            .windows(2)
+                            .rposition(|w| w == [START1, START2])
                         {
                             self.read_buf.drain(..pos);
                         } else {
@@ -653,7 +688,17 @@ impl MeshtasticDevice {
                 }
                 None
             }
-            FROM_RADIO_CONFIG_COMPLETE_ID | FROM_RADIO_REBOOTED => None,
+            FROM_RADIO_CONFIG_COMPLETE_ID
+            | FROM_RADIO_REBOOTED
+            | FROM_RADIO_QUEUE_STATUS
+            | FROM_RADIO_XMODEM_PACKET
+            | FROM_RADIO_METADATA
+            | FROM_RADIO_MQTT_CLIENT_PROXY_MESSAGE
+            | FROM_RADIO_FILE_INFO
+            | FROM_RADIO_CLIENT_NOTIFICATION
+            | FROM_RADIO_DEVICE_UI_CONFIG
+            | FROM_RADIO_LOCKDOWN_STATUS
+            | FROM_RADIO_REGION_PRESETS => None,
             other => {
                 debug!(
                     field = other,
@@ -700,73 +745,135 @@ impl MeshtasticDevice {
     }
 
     fn packet_to_inbound_frame(&mut self, data: &[u8]) -> Option<InboundFrame> {
-        let packet = parse_mesh_packet(data)?;
-        if packet.portnum != TEXT_MESSAGE_APP || packet.payload.is_empty() {
-            return None;
-        }
-        let from = packet.from.unwrap_or(0);
-        if Some(from) == self.node_num {
-            return None;
-        }
-        info!(
-            from = format!("!{:08x}", from),
-            len = packet.payload.len(),
-            pki = packet.pki_encrypted,
-            "Meshtastic received text packet over the air"
-        );
-        // Record E2E status: a `pki_encrypted` packet (or one carrying the
-        // sender's `public_key`) proves this DM arrived end-to-end encrypted via
-        // the PKI, not the shared channel PSK. We learn the sender's key here too
-        // — but keep it OUT of the routing `public_key_hex` (synthetic) so the
-        // device interface stays identical to meshcore's and the two remain
-        // hot-swappable behind the mesh listener.
-        if let Some(pk) = packet.public_key.as_ref() {
-            self.peer_pubkeys.entry(from).or_insert_with(|| pk.clone());
-        }
-        if packet.pki_encrypted {
-            debug!(node = from, "Meshtastic DM received end-to-end encrypted (PKI)");
-        }
-        let from_key = synthetic_pubkey(from);
-        self.contacts.entry(from).or_insert_with(|| ParsedContact {
-            public_key_hex: hex::encode(synthetic_pubkey(from)),
-            advert_name: format!("Meshtastic !{:08x}", from),
-            last_advert: 0,
-            contact_type: 1,
-            path_len: 0xff,
-            flags: 0,
-        });
-
-        let mut payload = Vec::with_capacity(15 + packet.payload.len());
-        payload.push(0); // SNR unknown
-        payload.extend_from_slice(&[0, 0]); // reserved
-        payload.extend_from_slice(&from_key[..6]);
-        payload.push(0xff); // unknown/flood path
-        payload.push(0); // text type
-        payload.extend_from_slice(&0u32.to_le_bytes());
-        payload.extend_from_slice(&packet.payload);
-        Some(InboundFrame {
-            code: super::protocol::RESP_CONTACT_MSG_V3,
-            data: payload,
-            bytes_consumed: 0,
-        })
+        packet_to_inbound_frame(
+            data,
+            self.node_num,
+            &mut self.contacts,
+            &mut self.peer_pubkeys,
+        )
     }
 }
 
+fn packet_to_inbound_frame(
+    data: &[u8],
+    local_node_num: Option<u32>,
+    contacts: &mut HashMap<u32, ParsedContact>,
+    peer_pubkeys: &mut HashMap<u32, Vec<u8>>,
+) -> Option<InboundFrame> {
+    let packet = parse_mesh_packet(data)?;
+    if packet.portnum != TEXT_MESSAGE_APP || packet.payload.is_empty() {
+        return None;
+    }
+    if packet_is_stale(packet.rx_time) {
+        debug!(
+            from = ?packet.from.map(|n| format!("!{:08x}", n)),
+            rx_time = ?packet.rx_time,
+            "Dropping stale Meshtastic text packet from radio backlog"
+        );
+        return None;
+    }
+    let from = packet.from.unwrap_or(0);
+    if Some(from) == local_node_num {
+        return None;
+    }
+    info!(
+        from = format!("!{:08x}", from),
+        len = packet.payload.len(),
+        pki = packet.pki_encrypted,
+        "Meshtastic received text packet over the air"
+    );
+    // Record E2E status without overwriting the synthetic routing key used by
+    // the shared mesh listener.
+    if let Some(pk) = packet.public_key.as_ref() {
+        peer_pubkeys.entry(from).or_insert_with(|| pk.clone());
+    }
+    if packet.pki_encrypted {
+        debug!(
+            node = from,
+            "Meshtastic DM received end-to-end encrypted (PKI)"
+        );
+    }
+    let from_key = synthetic_pubkey(from);
+    contacts.entry(from).or_insert_with(|| ParsedContact {
+        public_key_hex: hex::encode(synthetic_pubkey(from)),
+        advert_name: format!("Meshtastic !{:08x}", from),
+        last_advert: 0,
+        contact_type: 1,
+        path_len: 0xff,
+        flags: 0,
+    });
+
+    let mut payload = Vec::with_capacity(15 + packet.payload.len());
+    payload.push(0); // SNR unknown
+    payload.extend_from_slice(&[0, 0]); // reserved
+    payload.extend_from_slice(&from_key[..6]);
+    payload.push(0xff); // unknown/flood path
+    payload.push(0); // text type
+    payload.extend_from_slice(&packet.rx_time.unwrap_or_else(now_unix_secs).to_le_bytes());
+    payload.extend_from_slice(&packet.payload);
+    Some(InboundFrame {
+        code: if packet.pki_encrypted {
+            super::protocol::RESP_CONTACT_MSG_V3_E2E
+        } else {
+            super::protocol::RESP_CONTACT_MSG_V3
+        },
+        data: payload,
+        bytes_consumed: 0,
+    })
+}
+
+fn packet_is_stale(rx_time: Option<u32>) -> bool {
+    let Some(rx_time) = rx_time else {
+        return false;
+    };
+    // Radios without GPS/RTC can report tiny nonzero epoch values until their
+    // clock is set. Treat those as unknown, not stale, or live LoRa packets from
+    // stock Meshtastic peers disappear before reaching mesh.messages.
+    if rx_time < PLAUSIBLE_RX_EPOCH_SECS {
+        return false;
+    }
+    let now = now_unix_secs();
+    if now < PLAUSIBLE_RX_EPOCH_SECS || rx_time > now.saturating_add(60) {
+        return false;
+    }
+    rx_time.saturating_add(STALE_RX_SECS) < now
+}
+
 fn decode_serial_frame(buf: &mut Vec<u8>) -> Option<Vec<u8>> {
-    let start = buf.windows(2).position(|w| w == [START1, START2])?;
-    if start > 0 {
-        buf.drain(..start);
+    loop {
+        let start = buf.windows(2).position(|w| w == [START1, START2])?;
+        if start > 0 {
+            buf.drain(..start);
+        }
+        if buf.len() < 4 {
+            return None;
+        }
+        let len = u16::from_be_bytes([buf[2], buf[3]]) as usize;
+        if len == 0 || len > FROM_RADIO_MAX {
+            debug!(
+                len,
+                head = %hex::encode(&buf[..buf.len().min(16)]),
+                "Discarding invalid Meshtastic serial frame marker"
+            );
+            buf.drain(..1);
+            continue;
+        }
+        if buf.len() < 4 + len {
+            return None;
+        }
+        let payload = buf[4..4 + len].to_vec();
+        if decode_top_level_variant(&payload).is_none() {
+            debug!(
+                len,
+                head = %hex::encode(&buf[..buf.len().min(16)]),
+                "Discarding invalid Meshtastic serial frame payload"
+            );
+            buf.drain(..1);
+            continue;
+        }
+        buf.drain(..4 + len);
+        return Some(payload);
     }
-    if buf.len() < 4 {
-        return None;
-    }
-    let len = u16::from_be_bytes([buf[2], buf[3]]) as usize;
-    if buf.len() < 4 + len {
-        return None;
-    }
-    let payload = buf[4..4 + len].to_vec();
-    buf.drain(..4 + len);
-    Some(payload)
 }
 
 fn encode_want_config() -> Vec<u8> {
@@ -844,9 +951,7 @@ fn parse_primary_channel(data: &[u8]) -> Option<(String, Vec<u8>)> {
                     j = sn;
                     match (sf, sv) {
                         (2, FieldValue::Bytes(p)) => psk = p.to_vec(),
-                        (3, FieldValue::Bytes(n)) => {
-                            name = String::from_utf8_lossy(n).to_string()
-                        }
+                        (3, FieldValue::Bytes(n)) => name = String::from_utf8_lossy(n).to_string(),
                         _ => {}
                     }
                 }
@@ -918,6 +1023,8 @@ fn encode_mesh_packet(to: u32, portnum: u32, payload: &[u8]) -> Vec<u8> {
     let mut packet = Vec::new();
     encode_fixed32_field(2, to, &mut packet);
     encode_len_field(4, &decoded, &mut packet);
+    encode_fixed32_field(6, next_packet_id(), &mut packet);
+    encode_fixed32_field(7, now_unix_secs(), &mut packet);
     packet
 }
 
@@ -949,6 +1056,15 @@ fn decode_top_level_variant(buf: &[u8]) -> Option<(u64, &[u8])> {
                         | FROM_RADIO_NODE_INFO
                         | FROM_RADIO_CONFIG
                         | FROM_RADIO_CHANNEL
+                        | FROM_RADIO_QUEUE_STATUS
+                        | FROM_RADIO_XMODEM_PACKET
+                        | FROM_RADIO_METADATA
+                        | FROM_RADIO_MQTT_CLIENT_PROXY_MESSAGE
+                        | FROM_RADIO_FILE_INFO
+                        | FROM_RADIO_CLIENT_NOTIFICATION
+                        | FROM_RADIO_DEVICE_UI_CONFIG
+                        | FROM_RADIO_LOCKDOWN_STATUS
+                        | FROM_RADIO_REGION_PRESETS
                 ) {
                     return Some((field, &buf[idx..end]));
                 }
@@ -1056,6 +1172,9 @@ struct ParsedPacket {
     from: Option<u32>,
     portnum: u32,
     payload: Vec<u8>,
+    #[allow(dead_code)]
+    id: Option<u32>,
+    rx_time: Option<u32>,
     /// MeshPacket.pki_encrypted (field 17): the firmware decrypted this packet
     /// with the PKI (Curve25519) key, i.e. it arrived end-to-end encrypted
     /// rather than via the shared channel PSK.
@@ -1068,6 +1187,8 @@ fn parse_mesh_packet(data: &[u8]) -> Option<ParsedPacket> {
     let mut idx = 0;
     let mut from = None;
     let mut decoded = None;
+    let mut id = None;
+    let mut rx_time = None;
     let mut pki_encrypted = false;
     let mut public_key = None;
     while idx < data.len() {
@@ -1076,6 +1197,8 @@ fn parse_mesh_packet(data: &[u8]) -> Option<ParsedPacket> {
         match (field, value) {
             (1, FieldValue::Fixed32(v)) => from = Some(v),
             (4, FieldValue::Bytes(b)) => decoded = Some(b),
+            (6, FieldValue::Fixed32(v)) => id = Some(v),
+            (7, FieldValue::Fixed32(v)) if v != 0 => rx_time = Some(v),
             (16, FieldValue::Bytes(b)) if !b.is_empty() => public_key = Some(b.to_vec()),
             (17, FieldValue::Varint(v)) => pki_encrypted = v != 0,
             _ => {}
@@ -1098,11 +1221,30 @@ fn parse_mesh_packet(data: &[u8]) -> Option<ParsedPacket> {
         from,
         portnum,
         payload,
+        id,
+        rx_time,
         pki_encrypted,
         public_key,
     })
 }
 
+fn now_unix_secs() -> u32 {
+    std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .map(|d| d.as_secs() as u32)
+        .unwrap_or(0)
+}
+
+fn next_packet_id() -> u32 {
+    static COUNTER: std::sync::atomic::AtomicU32 = std::sync::atomic::AtomicU32::new(1);
+    let ctr = COUNTER.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
+    let nanos = std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .map(|d| d.subsec_nanos())
+        .unwrap_or(0);
+    nanos ^ ctr.rotate_left(16)
+}
+
 enum FieldValue<'a> {
     Varint(u64),
     Fixed32(u32),
@@ -1204,3 +1346,134 @@ fn synthetic_pubkey(node_num: u32) -> [u8; 32] {
     out[4..15].copy_from_slice(b"meshtastic:");
     out
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::mesh::protocol;
+
+    fn serial_frame(payload: &[u8]) -> Vec<u8> {
+        let mut frame = Vec::new();
+        frame.push(START1);
+        frame.push(START2);
+        frame.extend_from_slice(&(payload.len() as u16).to_be_bytes());
+        frame.extend_from_slice(payload);
+        frame
+    }
+
+    #[test]
+    fn decode_serial_frame_skips_false_marker_with_impossible_length() {
+        let valid_payload = encode_varint_field(FROM_RADIO_CONFIG_COMPLETE_ID, 1);
+        let mut buf = vec![b'l', b'o', b'g', START1, START2, 0xff, 0xff, b'x'];
+        buf.extend_from_slice(&serial_frame(&valid_payload));
+
+        let decoded = decode_serial_frame(&mut buf).expect("valid frame after false marker");
+        assert_eq!(decoded, valid_payload);
+    }
+
+    #[test]
+    fn decode_serial_frame_skips_false_marker_with_invalid_payload() {
+        let valid_payload = encode_varint_field(FROM_RADIO_CONFIG_COMPLETE_ID, 1);
+        let mut buf = vec![START1, START2, 0x00, 0x03, b'b', b'a', b'd'];
+        buf.extend_from_slice(&serial_frame(&valid_payload));
+
+        let decoded = decode_serial_frame(&mut buf).expect("valid frame after invalid payload");
+        assert_eq!(decoded, valid_payload);
+    }
+
+    #[test]
+    fn decode_serial_frame_accepts_queue_status_variant() {
+        let mut queue_status = Vec::new();
+        encode_len_field(
+            FROM_RADIO_QUEUE_STATUS,
+            &[0x10, 0x0e, 0x18, 0x10],
+            &mut queue_status,
+        );
+        let mut buf = serial_frame(&queue_status);
+
+        let decoded =
+            decode_serial_frame(&mut buf).expect("queue status is a valid FromRadio frame");
+        assert_eq!(decoded, queue_status);
+    }
+
+    #[test]
+    fn encode_mesh_packet_sets_nonzero_id_and_time() {
+        let before = now_unix_secs();
+        let packet = encode_mesh_packet(0x1122_3344, TEXT_MESSAGE_APP, b"hello");
+        let after = now_unix_secs();
+        let parsed = parse_mesh_packet(&packet).expect("packet should parse");
+
+        assert_eq!(parsed.portnum, TEXT_MESSAGE_APP);
+        assert_eq!(parsed.payload, b"hello");
+        assert!(parsed.id.unwrap_or(0) != 0);
+        let rx_time = parsed.rx_time.expect("rx_time should be set");
+        assert!(rx_time >= before.saturating_sub(1));
+        assert!(rx_time <= after.saturating_add(1));
+    }
+
+    #[test]
+    fn packet_to_inbound_frame_accepts_stock_peer_with_unset_clock() {
+        let from = 0x0000_3ccc;
+        let mut contacts = HashMap::new();
+        let mut peer_pubkeys = HashMap::new();
+        let mut decoded = Vec::new();
+        encode_varint_field_into(1, TEXT_MESSAGE_APP as u64, &mut decoded);
+        encode_len_field(2, b"hello from 3ccc", &mut decoded);
+
+        let mut packet = Vec::new();
+        encode_fixed32_field(1, from, &mut packet);
+        encode_fixed32_field(2, BROADCAST_NUM, &mut packet);
+        encode_len_field(4, &decoded, &mut packet);
+        encode_fixed32_field(7, 12_345, &mut packet);
+
+        let frame =
+            packet_to_inbound_frame(&packet, Some(0x1111_1111), &mut contacts, &mut peer_pubkeys)
+                .expect("live packet with unset radio clock must not be dropped");
+        assert_eq!(frame.code, protocol::RESP_CONTACT_MSG_V3);
+
+        let (sender_prefix, payload, _snr) =
+            protocol::parse_contact_msg_v3_raw(&frame.data).unwrap();
+        assert_eq!(sender_prefix, "cc3c00006d65");
+        assert_eq!(payload, b"hello from 3ccc");
+        assert!(contacts.contains_key(&from));
+    }
+
+    #[test]
+    fn packet_to_inbound_frame_accepts_recent_meshtastic_backlog() {
+        let from = 0x433e_3ccc;
+        let mut contacts = HashMap::new();
+        let mut peer_pubkeys = HashMap::new();
+        let mut decoded = Vec::new();
+        encode_varint_field_into(1, TEXT_MESSAGE_APP as u64, &mut decoded);
+        encode_len_field(2, b"recent backlog", &mut decoded);
+
+        let mut packet = Vec::new();
+        encode_fixed32_field(1, from, &mut packet);
+        encode_fixed32_field(2, BROADCAST_NUM, &mut packet);
+        encode_len_field(4, &decoded, &mut packet);
+        encode_fixed32_field(7, now_unix_secs().saturating_sub(60 * 60), &mut packet);
+
+        let frame =
+            packet_to_inbound_frame(&packet, Some(0x1111_1111), &mut contacts, &mut peer_pubkeys)
+                .expect("recent radio backlog must surface in mesh.messages");
+        let (sender_prefix, payload, _snr) =
+            protocol::parse_contact_msg_v3_raw(&frame.data).unwrap();
+        assert_eq!(sender_prefix, "cc3c3e436d65");
+        assert_eq!(payload, b"recent backlog");
+    }
+
+    #[test]
+    fn stale_filter_keeps_packets_from_radios_with_unset_clock() {
+        assert!(!packet_is_stale(None));
+        assert!(!packet_is_stale(Some(0)));
+        assert!(!packet_is_stale(Some(12_345)));
+    }
+
+    #[test]
+    fn stale_filter_drops_only_plausibly_old_packets() {
+        let old = now_unix_secs().saturating_sub(STALE_RX_SECS + 60);
+        if old >= PLAUSIBLE_RX_EPOCH_SECS {
+            assert!(packet_is_stale(Some(old)));
+        }
+    }
+}

core/archipelago/src/mesh/mod.rs

@@ -1109,15 +1109,13 @@ impl MeshService {
         // (FIPS→Tor) instead of handing it to a radio that physically cannot
         // deliver it. Reachable radio peers stay on the mesh; oversized
         // envelopes (file shares etc.) always take the federation path.
-        let radio_federated_unreachable = !is_federation_synthetic
-            && !exceeds_lora
-            && {
-                let peers = self.state.peers.read().await;
-                peers
-                    .get(&contact_id)
-                    .map(|p| !p.reachable && p.arch_pubkey_hex.is_some())
-                    .unwrap_or(false)
-            };
+        let radio_federated_unreachable = !is_federation_synthetic && !exceeds_lora && {
+            let peers = self.state.peers.read().await;
+            peers
+                .get(&contact_id)
+                .map(|p| !p.reachable && p.arch_pubkey_hex.is_some())
+                .unwrap_or(false)
+        };
         if is_federation_synthetic || exceeds_lora || radio_federated_unreachable {
             // Resolve the peer's pubkey/did. Prefer the live mesh peer table,
             // but fall back to federation storage for federation-synthetic ids
@@ -1508,9 +1506,12 @@ impl MeshService {
         // it stays backward compatible. (Federation/Tor sends already sign in
         // `send_typed_wire_via_federation`.) `with_seq` is applied after signing
         // — seq is not covered by the signature.
-        let envelope =
-            TypedEnvelope::new_signed(MeshMessageType::Text, text.as_bytes().to_vec(), &self.signing_key)
-                .with_seq(seq);
+        let envelope = TypedEnvelope::new_signed(
+            MeshMessageType::Text,
+            text.as_bytes().to_vec(),
+            &self.signing_key,
+        )
+        .with_seq(seq);
         let wire = envelope.to_wire()?;
         self.send_typed_wire(contact_id, wire, "text", text, None, seq)
             .await
@@ -1653,7 +1654,13 @@ impl MeshService {
     /// Recently-denied `!ai` askers (newest first) so the UI can offer to allow
     /// them. Cleared implicitly as new denials rotate older ones out.
     pub async fn assistant_denied_askers(&self) -> Vec<listener::DeniedAsker> {
-        self.state.assist_denied.read().await.iter().cloned().collect()
+        self.state
+            .assist_denied
+            .read()
+            .await
+            .iter()
+            .cloned()
+            .collect()
     }
 
     /// Update the mesh-AI assistant settings live (no listener restart) and

core/archipelago/src/mesh/protocol.rs

@@ -64,6 +64,11 @@ pub const RESP_CONTACT_MSG_V3: u8 = 0x10;
 pub const RESP_CHANNEL_MSG_V3: u8 = 0x11;
 pub const RESP_CHANNEL_INFO: u8 = 0x12;
 pub const RESP_STATS: u8 = 0x18;
+/// Archipelago-internal synthetic response code used by the Meshtastic adapter
+/// for text DMs that the firmware reports as PKI-encrypted. Meshcore firmware
+/// never emits this code; it lets the shared listener persist the E2E badge
+/// without changing the on-wire Meshcore frame format.
+pub const RESP_CONTACT_MSG_V3_E2E: u8 = 0x13;
 
 // --- Push notification codes (device -> host, async, >= 0x80) ---
 pub const PUSH_CONTACT_ADVERT: u8 = 0x80;