feat(mesh): meshtastic PKI E2E pill — surface pki_encrypted on received DMs

The synthetic meshcore-style frame the meshtastic driver builds can't carry the
radio's PKI-encryption status, so received meshtastic DMs never lit the E2E pill.
Thread it out-of-band: the device records `last_rx_encrypted` (= packet
pki_encrypted) when it yields a text frame; the session loop reads it via
`take_rx_encrypted()` right after dispatch and stamps the just-stored received
message E2E (dispatch::stamp_received_encrypted, monotonic-id keyed). Meshcore
returns false here (its E2E is derived in the frames decrypt path). Pure
out-of-band signal — no change to the shared meshcore wire format.

Built + deployed live in binary d937814e on .116/.198. cargo check green.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

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

@@ -122,6 +122,20 @@ pub(crate) async fn stamp_received_transport(
     }
 }
 
+/// Mark every RECEIVED message stored since `after_id` as end-to-end encrypted.
+/// Used by the session loop to stamp the E2E pill on a meshtastic frame the radio
+/// reported PKI-encrypted (the synthetic frame can't carry that flag, and the
+/// typed-dispatch store path defaults `encrypted` to false). One inbound frame
+/// yields at most one received message, so no sender filter is needed.
+pub(crate) async fn stamp_received_encrypted(state: &Arc<MeshState>, after_id: u64) {
+    let mut messages = state.messages.write().await;
+    for m in messages.iter_mut() {
+        if m.id > after_id && matches!(m.direction, MessageDirection::Received) {
+            m.encrypted = true;
+        }
+    }
+}
+
 /// Dispatch a pre-decoded TypedEnvelope. Shared between the radio receive
 /// path (handle_typed_message above) and the federation receive path
 /// (MeshService::inject_typed_from_federation) so both transports land the

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

@@ -4,7 +4,8 @@ use super::super::meshtastic::MeshtasticDevice;
 use super::super::serial::MeshcoreDevice;
 use super::super::types::*;
 use super::{
-    frames, MeshCommand, MeshState, ADVERT_INTERVAL, MAX_CONSECUTIVE_WRITE_FAILURES, SYNC_INTERVAL,
+    dispatch, frames, MeshCommand, MeshState, ADVERT_INTERVAL, MAX_CONSECUTIVE_WRITE_FAILURES,
+    SYNC_INTERVAL,
 };
 use anyhow::{Context, Result};
 use std::sync::Arc;
@@ -152,6 +153,15 @@ impl MeshRadioDevice {
             Self::Meshtastic(device) => device.try_recv_frame().await,
         }
     }
+
+    /// PKI-E2E status of the last inbound frame (meshtastic only; meshcore's
+    /// per-message E2E is derived in the frames decrypt path). Take-and-clear.
+    fn take_rx_encrypted(&mut self) -> bool {
+        match self {
+            Self::Meshcore(_) => false,
+            Self::Meshtastic(device) => device.take_rx_encrypted(),
+        }
+    }
 }
 
 /// Scan all candidate serial ports and open the first supported mesh device found.
@@ -710,11 +720,19 @@ pub(super) async fn run_mesh_session(
                     Ok(Some(frame)) => {
                         // Successful read resets the failure counter
                         consecutive_write_failures = 0;
+                        // For meshtastic, the PKI-E2E status of this frame can't
+                        // ride the synthetic meshcore frame — snapshot the message
+                        // id high-water mark, dispatch, then stamp the E2E pill on
+                        // whatever received message this frame produced.
+                        let before_id = dispatch::max_message_id(state).await;
                         let should_action = frames::handle_frame(
                             &frame,
                             state,
                             our_x25519_secret,
                         ).await;
+                        if device.take_rx_encrypted() {
+                            dispatch::stamp_received_encrypted(state, before_id).await;
+                        }
                         if should_action {
                             // Contact discovery or messages waiting — sync both
                             refresh_contacts(&mut device, state).await;

core/archipelago/src/mesh/meshtastic.rs

@@ -97,6 +97,11 @@ pub struct MeshtasticDevice {
     /// `None` until a primary `Channel` frame is seen.
     current_primary_channel: Option<(String, Vec<u8>)>,
     device_path: String,
+    /// PKI-encryption status of the most recent inbound text frame yielded by
+    /// `try_recv_frame`. The synthetic meshcore-style frame can't carry it, so
+    /// the session loop reads it via `take_rx_encrypted()` right after dispatch
+    /// to stamp the message's E2E pill. Set true only for `pki_encrypted` DMs.
+    last_rx_encrypted: bool,
 }
 
 impl MeshtasticDevice {
@@ -126,6 +131,7 @@ impl MeshtasticDevice {
             current_region: None,
             current_primary_channel: None,
             device_path: path.to_string(),
+            last_rx_encrypted: false,
         })
     }
 
@@ -777,12 +783,22 @@ impl MeshtasticDevice {
         payload.push(0); // text type
         payload.extend_from_slice(&0u32.to_le_bytes());
         payload.extend_from_slice(&packet.payload);
+        // Carry the PKI-E2E status out-of-band (the synthetic frame can't hold
+        // it); the session loop reads it via take_rx_encrypted() to set the pill.
+        self.last_rx_encrypted = packet.pki_encrypted;
         Some(InboundFrame {
             code: super::protocol::RESP_CONTACT_MSG_V3,
             data: payload,
             bytes_consumed: 0,
         })
     }
+
+    /// Take + clear the PKI-E2E status of the last inbound text frame. The
+    /// session loop calls this right after dispatching a received frame to stamp
+    /// the message's E2E pill (meshtastic DMs are E2E only when PKI-encrypted).
+    pub fn take_rx_encrypted(&mut self) -> bool {
+        std::mem::take(&mut self.last_rx_encrypted)
+    }
 }
 
 fn decode_serial_frame(buf: &mut Vec<u8>) -> Option<Vec<u8>> {