diff --git a/.github/workflows/wire.yml b/.github/workflows/wire.yml
index 32d2938..60786fa 100644
--- a/.github/workflows/wire.yml
+++ b/.github/workflows/wire.yml
@@ -26,6 +26,24 @@ jobs:
       - name: Golden fixture round-trips through views
         run: cargo test -p thrum-core
 
+  mesh:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: dtolnay/rust-toolchain@stable
+        with:
+          components: clippy
+      - uses: Swatinem/rust-cache@v2
+      - name: Clippy on the crates this layer owns
+        # type_complexity — a tuple-of-tuples in thrum-core's build script.
+        # double_must_use — async_trait stamps #[must_use] onto a
+        #   Pin<Box<dyn Future>>; the lint fires in macro expansion, not our source.
+        # Both lints landed after this branch's local toolchain (1.98.1), so they
+        # are allowed rather than fixed blind.
+        run: cargo clippy -p ensemble -p hum-paths --all-targets -- -D warnings -A clippy::type_complexity -A clippy::double_must_use
+      - name: Ensemble, daemon and mesh simulation
+        run: cargo test --locked -p ensemble -p sim -p humd -p hum -p humctl
+
   go:
     runs-on: ubuntu-latest
     steps:
diff --git a/codegen/src/protocol.rs b/codegen/src/protocol.rs
index a9137a6..f63b51a 100644
--- a/codegen/src/protocol.rs
+++ b/codegen/src/protocol.rs
@@ -120,15 +120,12 @@ fn simplify(ty: &Type) -> (bool, Repr) {
     if let Type::Path(p) = ty {
         // Check the leading segment name directly: syn's `is_ident` rejects
         // paths with generic arguments, so `Option<String>` would slip by.
-        if let Some(seg) = p.path.segments.first() {
-            if seg.ident == "Option" {
-                if let syn::PathArguments::AngleBracketed(a) = &seg.arguments {
-                    if let Some(syn::GenericArgument::Type(inner)) = a.args.first() {
+        if let Some(seg) = p.path.segments.first()
+            && seg.ident == "Option"
+                && let syn::PathArguments::AngleBracketed(a) = &seg.arguments
+                    && let Some(syn::GenericArgument::Type(inner)) = a.args.first() {
                         return (true, simplify(inner).1);
                     }
-                }
-            }
-        }
     }
     (false, repr_of(ty))
 }
@@ -145,12 +142,12 @@ fn repr_of(ty: &Type) -> Repr {
         "bool" => Repr::Bool,
         "Value" => Repr::Json,
         "Vec" => {
-            let inner = angle_ty(&seg.arguments, 0).unwrap_or_else(|| Repr::Json);
+            let inner = angle_ty(&seg.arguments, 0).unwrap_or(Repr::Json);
             Repr::Arr(Box::new(inner))
         }
         "BTreeMap" => {
             // The value type is the SECOND generic arg (first is the key).
-            let inner = angle_ty(&seg.arguments, 1).unwrap_or_else(|| Repr::Json);
+            let inner = angle_ty(&seg.arguments, 1).unwrap_or(Repr::Json);
             Repr::Map(Box::new(inner))
         }
         other => {
@@ -169,11 +166,10 @@ fn repr_of(ty: &Type) -> Repr {
 }
 
 fn angle_ty(args: &syn::PathArguments, nth: usize) -> Option<Repr> {
-    if let syn::PathArguments::AngleBracketed(a) = args {
-        if let Some(syn::GenericArgument::Type(t)) = a.args.iter().nth(nth) {
+    if let syn::PathArguments::AngleBracketed(a) = args
+        && let Some(syn::GenericArgument::Type(t)) = a.args.iter().nth(nth) {
             return Some(simplify(t).1);
         }
-    }
     None
 }
 
@@ -576,8 +572,8 @@ mod tests {
         let f = &ts[0].fields;
         assert_eq!(f.len(), 4);
         assert_eq!(f[0].json, "chunkType");
-        assert_eq!(f[0].repr.matches(&Repr::Str), true);
-        assert_eq!(f[1].optional, true);
+        assert!(f[0].repr.matches(&Repr::Str));
+        assert!(f[1].optional);
         assert!(matches!(f[1].repr, Repr::Json));
         assert!(matches!(f[2].repr, Repr::Arr(_)));
         assert!(matches!(f[3].repr, Repr::Map(_)));
diff --git a/ensemble/src/framing.rs b/ensemble/src/framing.rs
new file mode 100644
index 0000000..c738eeb
--- /dev/null
+++ b/ensemble/src/framing.rs
@@ -0,0 +1,203 @@
+use anyhow::{Result, anyhow};
+use tokio::io::{AsyncBufReadExt, AsyncRead, AsyncWrite, AsyncWriteExt, BufReader};
+use tokio::sync::mpsc;
+
+use crate::Tone;
+
+pub const MAX_FRAME_BYTES: usize = 256 * 1024;
+
+pub async fn write_frame<W: AsyncWrite + Unpin>(writer: &mut W, tone: &Tone) -> Result<()> {
+    let mut line = serde_json::to_vec(tone).map_err(|e| anyhow!("encode: {e}"))?;
+    line.push(b'\n');
+    writer
+        .write_all(&line)
+        .await
+        .map_err(|e| anyhow!("write: {e}"))
+}
+
+pub async fn pump<R>(reader: R, tx: mpsc::Sender<Tone>, transport: &'static str)
+where
+    R: AsyncRead + Unpin,
+{
+    let mut reader = BufReader::new(reader);
+    let mut line: Vec<u8> = Vec::new();
+    let mut oversize = false;
+    let mut discarded = 0u64;
+
+    loop {
+        let chunk = match reader.fill_buf().await {
+            Ok(c) => c,
+            Err(e) => {
+                tracing::trace!(target: "ensemble.frame", transport, error = %e, "read.failed");
+                break;
+            }
+        };
+        if chunk.is_empty() {
+            if !line.is_empty() && !oversize {
+                forward(&line, &tx, transport).await;
+            }
+            break;
+        }
+        let newline = chunk.iter().position(|b| *b == b'\n');
+        let take = newline.unwrap_or(chunk.len());
+        let consumed = take + usize::from(newline.is_some());
+
+        if line.len() + take > MAX_FRAME_BYTES {
+            if !oversize {
+                oversize = true;
+                tracing::warn!(
+                    target: "ensemble.frame",
+                    transport,
+                    cap = MAX_FRAME_BYTES,
+                    "frame.oversize",
+                );
+            }
+        } else {
+            line.extend_from_slice(&chunk[..take]);
+        }
+        reader.consume(consumed);
+
+        if newline.is_none() {
+            continue;
+        }
+        if oversize {
+            oversize = false;
+            discarded += 1;
+            line.clear();
+            continue;
+        }
+        if !forward(&line, &tx, transport).await {
+            break;
+        }
+        line.clear();
+    }
+
+    if discarded > 0 {
+        tracing::warn!(target: "ensemble.frame", transport, discarded, "frame.oversize.total");
+    }
+}
+
+async fn forward(line: &[u8], tx: &mpsc::Sender<Tone>, transport: &'static str) -> bool {
+    if line.iter().all(|b| b.is_ascii_whitespace()) {
+        return true;
+    }
+    match serde_json::from_slice::<Tone>(line) {
+        Ok(tone) => tx.send(tone).await.is_ok(),
+        Err(e) => {
+            tracing::trace!(target: "ensemble.frame", transport, error = %e, "frame.rejected");
+            true
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use serde_json::json;
+
+    async fn frames(input: &[u8]) -> Vec<Tone> {
+        let (tx, mut rx) = mpsc::channel(64);
+        let (mut w, r) = tokio::io::duplex(input.len() + 64);
+        w.write_all(input).await.unwrap();
+        w.shutdown().await.unwrap();
+        pump(r, tx, "test").await;
+        let mut out = Vec::new();
+        while let Ok(t) = rx.try_recv() {
+            out.push(t);
+        }
+        out
+    }
+
+    #[tokio::test]
+    async fn a_frame_survives_arriving_in_pieces() {
+        let got = frames(b"{\"chi\":\"a\"}\n{\"chi\":\"b\"}\n").await;
+        assert_eq!(got.len(), 2);
+        assert_eq!(got[0]["chi"], "a");
+        assert_eq!(got[1]["chi"], "b");
+    }
+
+    #[tokio::test]
+    async fn one_byte_at_a_time_reassembles() {
+        let raw = b"{\"chi\":\"drip\"}\n";
+        let (tx, mut rx) = mpsc::channel(4);
+        let (mut w, r) = tokio::io::duplex(64);
+        let feeder = tokio::spawn(async move {
+            for b in raw {
+                w.write_all(&[*b]).await.unwrap();
+                tokio::task::yield_now().await;
+            }
+            w.shutdown().await.unwrap();
+        });
+        pump(r, tx, "test").await;
+        feeder.await.unwrap();
+        let tone = rx.try_recv().unwrap();
+        assert_eq!(tone["chi"], "drip");
+    }
+
+    #[tokio::test]
+    async fn a_trailing_frame_without_a_newline_still_arrives() {
+        let got = frames(b"{\"chi\":\"a\"}\n{\"chi\":\"tail\"}").await;
+        assert_eq!(got.len(), 2);
+        assert_eq!(got[1]["chi"], "tail");
+    }
+
+    #[tokio::test]
+    async fn garbage_is_skipped_and_the_next_frame_still_arrives() {
+        let got = frames(b"not json\n{\"chi\":\"ok\"}\n").await;
+        assert_eq!(got.len(), 1);
+        assert_eq!(got[0]["chi"], "ok");
+    }
+
+    #[tokio::test]
+    async fn a_blank_line_is_a_keepalive_not_a_tone() {
+        let got = frames(b"\n\n{\"chi\":\"ok\"}\n").await;
+        assert_eq!(got.len(), 1);
+    }
+
+    #[tokio::test]
+    async fn an_oversize_frame_is_discarded_and_the_link_resyncs() {
+        let mut input = vec![b'x'; MAX_FRAME_BYTES + 4096];
+        input.push(b'\n');
+        input.extend_from_slice(b"{\"chi\":\"after\"}\n");
+        let got = frames(&input).await;
+        assert_eq!(got.len(), 1, "the oversize frame must not be parsed");
+        assert_eq!(got[0]["chi"], "after");
+    }
+
+    #[tokio::test]
+    async fn a_frame_at_exactly_the_cap_is_accepted() {
+        let mut input = br#"{"chi":""#.to_vec();
+        input.extend(std::iter::repeat_n(b'p', MAX_FRAME_BYTES - input.len() - 2));
+        input.extend_from_slice(br#""}"#);
+        assert_eq!(input.len(), MAX_FRAME_BYTES);
+        input.push(b'\n');
+        let got = frames(&input).await;
+        assert_eq!(got.len(), 1);
+    }
+
+    #[tokio::test]
+    async fn pump_stops_when_the_receiver_is_gone() {
+        let (tx, rx) = mpsc::channel(1);
+        drop(rx);
+        let (mut w, r) = tokio::io::duplex(64);
+        w.write_all(b"{\"chi\":\"a\"}\n{\"chi\":\"b\"}\n")
+            .await
+            .unwrap();
+        pump(r, tx, "test").await;
+    }
+
+    #[tokio::test]
+    async fn write_frame_emits_exactly_one_newline_terminated_line() {
+        let (mut w, r) = tokio::io::duplex(256);
+        let mut r = BufReader::new(r);
+        write_frame(&mut w, &json!({"chi": "x", "n": 1}))
+            .await
+            .unwrap();
+        drop(w);
+        let mut first = String::new();
+        r.read_line(&mut first).await.unwrap();
+        assert_eq!(first, "{\"chi\":\"x\",\"n\":1}\n");
+        let mut rest = String::new();
+        assert_eq!(r.read_line(&mut rest).await.unwrap(), 0);
+    }
+}
diff --git a/ensemble/src/handshake.rs b/ensemble/src/handshake.rs
index def1d88..f6fe3c8 100644
--- a/ensemble/src/handshake.rs
+++ b/ensemble/src/handshake.rs
@@ -7,7 +7,7 @@ use ed25519_dalek::{Signature, Signer, SigningKey, Verifier, VerifyingKey};
 use serde::{Deserialize, Serialize};
 use tokio::sync::mpsc;
 
-use super::{headroom, now_ms, Tone, HANDSHAKE_DOMAIN, HANDSHAKE_SKEW_MS};
+use super::{HANDSHAKE_DOMAIN, HANDSHAKE_SKEW_MS, Tone, headroom, now_ms};
 
 pub use hum_identity::{Hid, HidParseError, HidPrefix};
 
@@ -47,7 +47,12 @@ pub struct HumdAddr {
 }
 
 impl HumdAddr {
-    pub fn new(id: Hid) -> Self { Self { id, hints: Vec::new() } }
+    pub fn new(id: Hid) -> Self {
+        Self {
+            id,
+            hints: Vec::new(),
+        }
+    }
     pub fn with_hint(mut self, h: impl Into<String>) -> Self {
         self.hints.push(h.into());
         self
@@ -123,6 +128,11 @@ pub trait PeerConnection: Send + Sync {
     async fn send(&self, tone: Tone) -> Result<()>;
     fn take_receiver(&self) -> Option<mpsc::Receiver<Tone>>;
     fn close(&self);
+
+    fn arm_opening(&self) {}
+    async fn send_opening(&self, tone: Tone) -> Result<()> {
+        self.send(tone).await
+    }
 }
 
 #[async_trait]
@@ -196,30 +206,46 @@ pub fn parse_hello_caps(tone: &Tone) -> Option<(Hid, PeerCapabilities)> {
     let nests = tone
         .get("nests")
         .and_then(|v| v.as_array())
-        .map(|a| a.iter().filter_map(|v| v.as_str().map(String::from)).collect())
+        .map(|a| {
+            a.iter()
+                .filter_map(|v| v.as_str().map(String::from))
+                .collect()
+        })
         .unwrap_or_default();
     let hosts = tone
         .get("hosts")
         .and_then(|v| v.as_array())
-        .map(|a| a.iter().filter_map(|v| v.as_str().map(String::from)).collect())
+        .map(|a| {
+            a.iter()
+                .filter_map(|v| v.as_str().map(String::from))
+                .collect()
+        })
         .unwrap_or_default();
-    let can_relay = tone.get("can_relay").and_then(|v| v.as_bool()).unwrap_or(false);
-    let free_slots = tone
-        .get("free_slots")
-        .and_then(|v| {
-            if v.is_null() {
-                None
-            } else {
-                v.as_u64().map(|n| n as usize)
-            }
-        });
+    let can_relay = tone
+        .get("can_relay")
+        .and_then(|v| v.as_bool())
+        .unwrap_or(false);
+    let free_slots = tone.get("free_slots").and_then(|v| {
+        if v.is_null() {
+            None
+        } else {
+            v.as_u64().map(|n| n as usize)
+        }
+    });
     let headroom = tone
         .get("headroom")
         .and_then(|v| serde_json::from_value::<headroom::CellHeadroom>(v.clone()).ok())
         .unwrap_or_default();
     Some((
         claimed_id,
-        PeerCapabilities { proto_version, nests, hosts, can_relay, free_slots, headroom },
+        PeerCapabilities {
+            proto_version,
+            nests,
+            hosts,
+            can_relay,
+            free_slots,
+            headroom,
+        },
     ))
 }
 
@@ -247,24 +273,46 @@ pub fn parse_hello(tone: &Tone) -> HelloParse {
         let nests = tone
             .get("nests")
             .and_then(|v| v.as_array())
-            .map(|a| a.iter().filter_map(|v| v.as_str().map(String::from)).collect())
+            .map(|a| {
+                a.iter()
+                    .filter_map(|v| v.as_str().map(String::from))
+                    .collect()
+            })
             .unwrap_or_default();
         let hosts = tone
             .get("hosts")
             .and_then(|v| v.as_array())
-            .map(|a| a.iter().filter_map(|v| v.as_str().map(String::from)).collect())
+            .map(|a| {
+                a.iter()
+                    .filter_map(|v| v.as_str().map(String::from))
+                    .collect()
+            })
             .unwrap_or_default();
-        let can_relay = tone.get("can_relay").and_then(|v| v.as_bool()).unwrap_or(false);
-        let free_slots = tone
-            .get("free_slots")
-            .and_then(|v| if v.is_null() { None } else { v.as_u64().map(|n| n as usize) });
+        let can_relay = tone
+            .get("can_relay")
+            .and_then(|v| v.as_bool())
+            .unwrap_or(false);
+        let free_slots = tone.get("free_slots").and_then(|v| {
+            if v.is_null() {
+                None
+            } else {
+                v.as_u64().map(|n| n as usize)
+            }
+        });
         let headroom = tone
             .get("headroom")
             .and_then(|v| serde_json::from_value::<headroom::CellHeadroom>(v.clone()).ok())
             .unwrap_or_default();
         HelloParse::Unsigned(
             claimed_id,
-            PeerCapabilities { proto_version, nests, hosts, can_relay, free_slots, headroom },
+            PeerCapabilities {
+                proto_version,
+                nests,
+                hosts,
+                can_relay,
+                free_slots,
+                headroom,
+            },
         )
     }
 }
diff --git a/ensemble/src/headroom.rs b/ensemble/src/headroom.rs
index 758de25..aa040f6 100644
--- a/ensemble/src/headroom.rs
+++ b/ensemble/src/headroom.rs
@@ -15,20 +15,15 @@ use serde::{Deserialize, Serialize};
 ///   Refuse — > 95%, do not route new prompts here (humd will reject)
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
 #[serde(rename_all = "kebab-case")]
+#[derive(Default)]
 pub enum Pressure {
+    #[default]
     Cool,
     Warm,
     Hot,
     Refuse,
 }
 
-impl Default for Pressure {
-    fn default() -> Self {
-        // No-op advertise — an empty humd is at rest, not refusing.
-        Pressure::Cool
-    }
-}
-
 impl Pressure {
     /// Compute from a slot-occupancy ratio (0.0 .. 1.0). Out-of-range
     /// values clamp.
@@ -78,11 +73,7 @@ impl CellHeadroom {
     /// When `total_slots == 0` (no nest), pressure is `Cool` and
     /// `free_slots` is forced to 0 — the humd is honest about having
     /// nothing to offer rather than misrepresenting itself as full.
-    pub fn from_counts(
-        free_slots: u32,
-        total_slots: u32,
-        p95_latency_ms: Option<u32>,
-    ) -> Self {
+    pub fn from_counts(free_slots: u32, total_slots: u32, p95_latency_ms: Option<u32>) -> Self {
         if total_slots == 0 {
             return Self {
                 free_slots: 0,
@@ -216,6 +207,10 @@ mod tests {
         // skip_serializing_if for None p95.
         let no_latency = CellHeadroom::from_counts(5, 10, None);
         let s = serde_json::to_string(&no_latency).unwrap();
-        assert!(!s.contains("p95_latency_ms"), "None should be skipped: {}", s);
+        assert!(
+            !s.contains("p95_latency_ms"),
+            "None should be skipped: {}",
+            s
+        );
     }
 }
diff --git a/ensemble/src/hives.rs b/ensemble/src/hives.rs
index 8a36a2e..e3d8ece 100644
--- a/ensemble/src/hives.rs
+++ b/ensemble/src/hives.rs
@@ -213,7 +213,7 @@ pub enum HiveAnnounce {
     /// Bee is live on `humd_id`.
     Advertise {
         humd_id: String,
-        manifest: HiveManifest,
+        manifest: Box<HiveManifest>,
     },
     /// Bee has shut down on `humd_id`.
     Retract { humd_id: String, name: String },
@@ -248,14 +248,17 @@ mod tests {
         let back: HiveManifest = serde_json::from_value(j).unwrap();
         assert_eq!(back.chis.len(), 4);
         assert_eq!(back.propensity.statefulness.as_deref(), Some("stateless"));
-        assert_eq!(back.source.as_deref(), Some("https://github.com/example/mm-bee"));
+        assert_eq!(
+            back.source.as_deref(),
+            Some("https://github.com/example/mm-bee")
+        );
     }
 
     #[test]
     fn announce_envelope_tags_kind() {
         let env = HiveAnnounce::Advertise {
             humd_id: "deadbeef".into(),
-            manifest: HiveManifest::new("mm", "0.1.0", "0.7.0"),
+            manifest: Box::new(HiveManifest::new("mm", "0.1.0", "0.7.0")),
         };
         let s = serde_json::to_string(&env).unwrap();
         assert!(s.contains("\"kind\":\"advertise\""));
diff --git a/ensemble/src/iroh.rs b/ensemble/src/iroh.rs
index d79c855..62d7988 100644
--- a/ensemble/src/iroh.rs
+++ b/ensemble/src/iroh.rs
@@ -1,16 +1,17 @@
-
 use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, Ordering};
 
-use anyhow::{anyhow, Context, Result};
+use anyhow::{Context, Result, anyhow};
 use async_trait::async_trait;
 use sha2::{Digest, Sha256};
-use tokio::io::{AsyncBufReadExt, BufReader};
-use tokio::sync::{mpsc, Mutex};
+use tokio::sync::{Mutex, mpsc};
 
 use iroh::endpoint::{Connection, RecvStream, SendStream};
 use iroh::{Endpoint, EndpointAddr, EndpointId, PublicKey, TransportAddr};
 
-use crate::{HumdAddr, Hid, PeerCapabilities, PeerConnection, Tone, Transport};
+use crate::framing;
+use crate::opening::Gate;
+use crate::{Hid, HumdAddr, PeerCapabilities, PeerConnection, Tone, Transport};
 
 pub const IROH_ALPN: &[u8] = b"hum/0.5";
 
@@ -46,8 +47,10 @@ pub struct IrohEndpoint {
     node_id: EndpointId,
     caps: PeerCapabilities,
     _connection: Connection,
-    sender: Mutex<Option<SendStream>>,
+    sender: Arc<Mutex<Option<SendStream>>>,
     rx: parking_lot::Mutex<Option<mpsc::Receiver<Tone>>>,
+    closed: AtomicBool,
+    gate: Gate,
 }
 
 impl IrohEndpoint {
@@ -59,17 +62,20 @@ impl IrohEndpoint {
         caps: PeerCapabilities,
     ) -> Arc<Self> {
         let humd_id = humd_id_from_node_id(&node_id);
-        let peer = HumdAddr::new(humd_id).with_hint(format!("{IROH_HINT}{}", hex::encode(node_id.as_bytes())));
+        let peer = HumdAddr::new(humd_id)
+            .with_hint(format!("{IROH_HINT}{}", hex::encode(node_id.as_bytes())));
         let (tx, rx) = mpsc::channel::<Tone>(RECV_CAP);
         let me = Arc::new(Self {
             peer,
             node_id,
             caps,
             _connection: conn,
-            sender: Mutex::new(Some(send)),
+            sender: Arc::new(Mutex::new(Some(send))),
             rx: parking_lot::Mutex::new(Some(rx)),
+            closed: AtomicBool::new(false),
+            gate: Gate::new(),
         });
-        tokio::spawn(read_loop(recv, tx));
+        tokio::spawn(framing::pump(recv, tx, "iroh"));
         me
     }
 
@@ -98,6 +104,19 @@ impl IrohEndpoint {
     }
 }
 
+impl IrohEndpoint {
+    async fn write(&self, tone: Tone) -> Result<()> {
+        let mut guard = self.sender.lock().await;
+        if self.closed.load(Ordering::SeqCst) {
+            return Err(anyhow!("iroh send: link closed"));
+        }
+        let s = guard
+            .as_mut()
+            .ok_or_else(|| anyhow!("iroh send: stream closed"))?;
+        framing::write_frame(s, &tone).await
+    }
+}
+
 #[async_trait]
 impl PeerConnection for IrohEndpoint {
     fn peer(&self) -> &HumdAddr {
@@ -108,61 +127,34 @@ impl PeerConnection for IrohEndpoint {
     }
 
     async fn send(&self, tone: Tone) -> Result<()> {
-        let mut line = serde_json::to_vec(&tone)
-            .map_err(|e| anyhow!("iroh send: serialize: {e}"))?;
-        line.push(b'\n');
-        let mut guard = self.sender.lock().await;
-        let s = guard
-            .as_mut()
-            .ok_or_else(|| anyhow!("iroh send: stream closed"))?;
-        s.write_all(&line)
-            .await
-            .map_err(|e| anyhow!("iroh send: write: {e}"))?;
-        Ok(())
+        self.gate.wait().await;
+        self.write(tone).await
+    }
+
+    async fn send_opening(&self, tone: Tone) -> Result<()> {
+        let result = self.write(tone).await;
+        self.gate.opened();
+        result
     }
 
     fn take_receiver(&self) -> Option<mpsc::Receiver<Tone>> {
         self.rx.lock().take()
     }
 
-    fn close(&self) {
-        let slot = match self.sender.try_lock() {
-            Ok(mut g) => g.take(),
-            Err(_) => None,
-        };
-        if let Some(mut s) = slot {
-            tokio::spawn(async move {
-                let _ = s.finish();
-            });
-        }
-        let _ = self.rx.lock().take();
+    fn arm_opening(&self) {
+        self.gate.arm();
     }
-}
 
-async fn read_loop(recv: RecvStream, tx: mpsc::Sender<Tone>) {
-    let mut lines = BufReader::new(recv).lines();
-    loop {
-        let line = match lines.next_line().await {
-            Ok(Some(l)) => l,
-            Ok(None) => break,
-            Err(e) => {
-                tracing::trace!(target: "ensemble.iroh", err = %e, "iroh.read.failed");
-                break;
-            }
-        };
-        if line.is_empty() {
-            continue;
-        }
-        let tone: Tone = match serde_json::from_str(&line) {
-            Ok(v) => v,
-            Err(e) => {
-                tracing::trace!(target: "ensemble.iroh", err = %e, "iroh.parse.failed");
-                continue;
+    fn close(&self) {
+        self.closed.store(true, Ordering::SeqCst);
+        self.gate.shut();
+        let sender = self.sender.clone();
+        tokio::spawn(async move {
+            if let Some(mut s) = sender.lock().await.take() {
+                let _ = s.finish();
             }
-        };
-        if tx.send(tone).await.is_err() {
-            break;
-        }
+        });
+        let _ = self.rx.lock().take();
     }
 }
 
@@ -284,12 +276,9 @@ impl Transport for IrohTransport {
             .map(TransportAddr::Ip)
             .collect();
         let endpoint_addr = EndpointAddr::from_parts(node_id, direct_addrs);
-        let endpoint = IrohEndpoint::connect(
-            &self.endpoint,
-            endpoint_addr,
-            PeerCapabilities::default(),
-        )
-        .await?;
+        let endpoint =
+            IrohEndpoint::connect(&self.endpoint, endpoint_addr, PeerCapabilities::default())
+                .await?;
         Ok(endpoint as Arc<dyn PeerConnection>)
     }
 }
@@ -317,7 +306,12 @@ mod dial_tests {
 
     #[test]
     fn a_concrete_address_is_left_alone() {
-        for s in ["192.168.1.5:9000", "10.0.0.7:443", "[fe80::1]:9000", "127.0.0.1:5000"] {
+        for s in [
+            "192.168.1.5:9000",
+            "10.0.0.7:443",
+            "[fe80::1]:9000",
+            "127.0.0.1:5000",
+        ] {
             let before: SocketAddr = s.parse().expect("parse");
             assert_eq!(dialable_addr(before), before, "{s} must be untouched");
         }
@@ -330,6 +324,9 @@ mod dial_tests {
         assert_eq!(dialable_addr(once), once);
         assert!(!once.ip().is_unspecified());
         assert_ne!(once.ip(), IpAddr::V4(Ipv4Addr::UNSPECIFIED));
-        assert_ne!(dialable_addr(SocketAddr::new(IpAddr::V6(Ipv6Addr::UNSPECIFIED), 1)).ip(), IpAddr::V6(Ipv6Addr::UNSPECIFIED));
+        assert_ne!(
+            dialable_addr(SocketAddr::new(IpAddr::V6(Ipv6Addr::UNSPECIFIED), 1)).ip(),
+            IpAddr::V6(Ipv6Addr::UNSPECIFIED)
+        );
     }
 }
diff --git a/ensemble/src/kad.rs b/ensemble/src/kad.rs
index 30d43ca..38f12c6 100644
--- a/ensemble/src/kad.rs
+++ b/ensemble/src/kad.rs
@@ -50,7 +50,7 @@ use rand::RngCore;
 use serde_json::Value;
 use tokio::sync::oneshot;
 
-use crate::{HumdAddr, Hid, Tone};
+use crate::{Hid, HumdAddr, Tone};
 
 /// Kademlia replication / bucket-size parameter (paper default).
 pub const KAD_K: usize = 20;
@@ -82,8 +82,8 @@ impl XorDistance {
     /// 256-bit big-endian number), the closer the two ids.
     pub fn distance(a: &Hid, b: &Hid) -> [u8; 32] {
         let mut out = [0u8; 32];
-        for i in 0..32 {
-            out[i] = a.as_bytes()[i] ^ b.as_bytes()[i];
+        for (slot, (x, y)) in out.iter_mut().zip(a.as_bytes().iter().zip(b.as_bytes())) {
+            *slot = x ^ y;
         }
         out
     }
@@ -116,12 +116,20 @@ pub struct KBucket {
 
 impl KBucket {
     pub fn new() -> Self {
-        Self { entries: VecDeque::with_capacity(KAD_K) }
+        Self {
+            entries: VecDeque::with_capacity(KAD_K),
+        }
     }
 
-    pub fn len(&self) -> usize { self.entries.len() }
-    pub fn is_empty(&self) -> bool { self.entries.is_empty() }
-    pub fn iter(&self) -> impl Iterator<Item = &HumdAddr> { self.entries.iter() }
+    pub fn len(&self) -> usize {
+        self.entries.len()
+    }
+    pub fn is_empty(&self) -> bool {
+        self.entries.is_empty()
+    }
+    pub fn iter(&self) -> impl Iterator<Item = &HumdAddr> {
+        self.entries.iter()
+    }
 
     /// Insert (or refresh) `addr`. If the bucket already holds an entry
     /// with the same id, that entry is moved to the back (most-recent
@@ -166,7 +174,9 @@ impl RoutingTable {
         }
     }
 
-    pub fn me(&self) -> Hid { self.me }
+    pub fn me(&self) -> Hid {
+        self.me
+    }
 
     /// Insert `addr` into the appropriate bucket. Inserting the table's
     /// own id is a no-op (returns false). Returns true on insert/refresh.
@@ -198,12 +208,31 @@ impl RoutingTable {
         all
     }
 
+    pub fn forget(&mut self, id: &Hid) -> bool {
+        let mut dropped = false;
+        for bucket in &mut self.buckets {
+            let before = bucket.len();
+            bucket.entries.retain(|a| &a.id != id);
+            dropped |= bucket.len() != before;
+        }
+        dropped
+    }
+
+    pub fn entries(&self) -> Vec<HumdAddr> {
+        self.buckets
+            .iter()
+            .flat_map(|b| b.iter().cloned())
+            .collect()
+    }
+
     /// Total number of peers currently in the table.
     pub fn len(&self) -> usize {
         self.buckets.iter().map(|b| b.len()).sum()
     }
 
-    pub fn is_empty(&self) -> bool { self.len() == 0 }
+    pub fn is_empty(&self) -> bool {
+        self.len() == 0
+    }
 
     /// Direct lookup: does the table hold a HumdAddr for `id`?
     pub fn get(&self, id: &Hid) -> Option<HumdAddr> {
@@ -211,10 +240,7 @@ impl RoutingTable {
         if idx >= 256 {
             return None;
         }
-        self.buckets[idx]
-            .iter()
-            .find(|a| &a.id == id)
-            .cloned()
+        self.buckets[idx].iter().find(|a| &a.id == id).cloned()
     }
 }
 
@@ -241,6 +267,22 @@ impl KadState {
     /// Stash a HumdAddr into the routing table. Used at install-time
     /// (bootstrap a peer we just connected to) and when FIND_NODE resps
     /// advertise new peers.
+    pub fn forget(&self, id: &Hid) -> bool {
+        self.table.lock().forget(id)
+    }
+
+    pub fn snapshot(&self) -> Vec<HumdAddr> {
+        self.table.lock().entries()
+    }
+
+    pub fn restore(self: &Arc<Self>, addrs: Vec<HumdAddr>) -> usize {
+        let mut table = self.table.lock();
+        addrs
+            .into_iter()
+            .filter(|a| table.insert(a.clone()))
+            .count()
+    }
+
     pub fn note_peer(&self, addr: HumdAddr) {
         self.table.lock().insert(addr);
     }
@@ -304,12 +346,7 @@ pub fn find_node_tone(rid: &str, query_id: &str, target: &Hid, from: &Hid) -> To
 }
 
 /// Build a `chi:"kad-find-node-resp"` tone carrying up to K HumdAddrs.
-pub fn find_node_resp_tone(
-    rid: &str,
-    query_id: &str,
-    from: &Hid,
-    closest: &[HumdAddr],
-) -> Tone {
+pub fn find_node_resp_tone(rid: &str, query_id: &str, from: &Hid, closest: &[HumdAddr]) -> Tone {
     serde_json::json!({
         "chi": KAD_FIND_NODE_RESP_CHI,
         "rid": rid,
@@ -348,7 +385,12 @@ pub fn parse_find_node(tone: &Tone) -> Option<ParsedFindNode> {
         .and_then(|v| v.as_str())
         .unwrap_or("")
         .to_string();
-    Some(ParsedFindNode { query_id, target, from, rid })
+    Some(ParsedFindNode {
+        query_id,
+        target,
+        from,
+        rid,
+    })
 }
 
 /// Pull fields out of a `chi:"kad-find-node-resp"` tone.
@@ -360,7 +402,11 @@ pub fn parse_find_node_resp(tone: &Tone) -> Option<ParsedFindNodeResp> {
         .iter()
         .filter_map(|v| serde_json::from_value::<HumdAddr>(v.clone()).ok())
         .collect();
-    Some(ParsedFindNodeResp { query_id, from, closest })
+    Some(ParsedFindNodeResp {
+        query_id,
+        from,
+        closest,
+    })
 }
 
 // ── Lookup driver ──────────────────────────────────────────────────────────
@@ -455,6 +501,7 @@ impl LookupShortlist {
 pub(crate) async fn query_peer(
     kad: &Arc<KadState>,
     conn: &Arc<dyn crate::PeerConnection>,
+    stats: &Arc<crate::SendStats>,
     me: &Hid,
     target: &Hid,
     per_query_timeout: Duration,
@@ -463,7 +510,7 @@ pub(crate) async fn query_peer(
     let rid = format!("kad-find-{}", &query_id[..8]);
     let rx = kad.register_query(query_id.clone());
     let tone = find_node_tone(&rid, &query_id, target, me);
-    if conn.send(tone).await.is_err() {
+    if crate::send_bounded(conn, tone, stats).await.is_err() {
         kad.cancel_query(&query_id);
         return Vec::new();
     }
@@ -587,7 +634,7 @@ mod tests {
     fn parse_find_node_resp_round_trip() {
         let me = Hid::random_humd();
         let peer = HumdAddr::new(Hid::random_humd()).with_hint("tcp:1.2.3.4:9000");
-        let tone = find_node_resp_tone("rid-1", "qid-1", &me, &[peer.clone()]);
+        let tone = find_node_resp_tone("rid-1", "qid-1", &me, std::slice::from_ref(&peer));
         let parsed = parse_find_node_resp(&tone).expect("parse");
         assert_eq!(parsed.query_id, "qid-1");
         assert_eq!(parsed.from, me);
diff --git a/ensemble/src/lib.rs b/ensemble/src/lib.rs
index 2818274..bf00418 100644
--- a/ensemble/src/lib.rs
+++ b/ensemble/src/lib.rs
@@ -1,7 +1,7 @@
 use std::collections::HashMap;
 use std::ops::Deref;
-use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
 use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
 use std::time::Duration;
 
 use anyhow::Result;
@@ -11,14 +11,13 @@ use tokio::task::JoinSet;
 
 pub mod handshake;
 pub use handshake::{
-    hello_tone, hello_tone_unsigned, parse_hello, parse_hello_caps, EnsembleHello, Hid,
-    HidParseError, HidPrefix, HelloParse, HumdAddr, HumdKey, PeerCapabilities, PeerConnection,
-    Transport,
+    EnsembleHello, HelloParse, Hid, HidParseError, HidPrefix, HumdAddr, HumdKey, PeerCapabilities,
+    PeerConnection, Transport, hello_tone, hello_tone_unsigned, parse_hello, parse_hello_caps,
 };
 
 pub mod link;
 pub use link::{
-    InMemoryEndpoint, LinkCounters, LinkFaults, Noise, Script, Verdict, PARTITION_BUFFER_CAP,
+    InMemoryEndpoint, LinkCounters, LinkFaults, Noise, PARTITION_BUFFER_CAP, Script, Verdict,
 };
 
 pub mod tcp;
@@ -26,39 +25,42 @@ pub use tcp::{TcpEndpoint, TcpListener, TcpTransport};
 
 pub mod tls;
 pub use tls::{
-    cert_fingerprint, client_config_pinned, PinnedFingerprintVerifier, TlsTcpEndpoint,
-    TlsTcpListener, TlsTcpTransport, TLS_FP_HINT, TLS_HINT,
+    PinnedFingerprintVerifier, TLS_FP_HINT, TLS_HINT, TlsTcpEndpoint, TlsTcpListener,
+    TlsTcpTransport, cert_fingerprint, client_config_pinned,
 };
 
 pub mod iroh;
-pub use iroh::{dialable_addr, IrohEndpoint, IrohTransport, IROH_ALPN, IROH_IP_HINT};
+pub use iroh::{IROH_ALPN, IROH_IP_HINT, IrohEndpoint, IrohTransport, dialable_addr};
 
 pub mod delivery;
+pub mod framing;
+pub mod opening;
 pub mod send;
-pub use send::{send_bounded, SendError, SendStats, SEND_TIMEOUT};
-pub use delivery::{DeliveryState, DELIVERY_SEEN_CAP};
+pub use delivery::{DELIVERY_SEEN_CAP, DeliveryState};
+pub use framing::MAX_FRAME_BYTES;
+pub use opening::Gate;
+pub use send::{SEND_TIMEOUT, SendError, SendStats, fanout, send_bounded};
 
 pub mod gossip;
 pub use gossip::{
-    gossip_tone, gossip_tone_with_dusk, mint_msg_id, GossipState, GOSSIP_CHI,
-    GOSSIP_SEEN_CAP,
+    GOSSIP_CHI, GOSSIP_SEEN_CAP, GossipState, gossip_tone, gossip_tone_with_dusk, mint_msg_id,
 };
 
 pub mod liveness;
 pub use liveness::{
-    ping_tone, pong_tone, probe_seq, Lease, Liveness, LivenessSignal, PING_CHI, PONG_CHI,
+    Lease, Liveness, LivenessSignal, PING_CHI, PONG_CHI, ping_tone, pong_tone, probe_seq,
 };
 
 pub mod kad;
 pub use kad::{
-    find_node_resp_tone, find_node_tone, mint_query_id, parse_find_node, parse_find_node_resp,
-    KBucket, KadFindOutcome, KadState, RoutingTable, XorDistance, KAD_ALPHA,
-    KAD_FIND_NODE_CHI, KAD_FIND_NODE_RESP_CHI, KAD_K, KAD_MAX_ROUNDS,
+    KAD_ALPHA, KAD_FIND_NODE_CHI, KAD_FIND_NODE_RESP_CHI, KAD_K, KAD_MAX_ROUNDS, KBucket,
+    KadFindOutcome, KadState, RoutingTable, XorDistance, find_node_resp_tone, find_node_tone,
+    mint_query_id, parse_find_node, parse_find_node_resp,
 };
 
 pub mod hives;
 pub mod uri;
-pub use hives::{BindAddr, HiveAnnounce, HiveManifest, Propensity, ToolEntry, ANNOUNCE_TOPIC};
+pub use hives::{ANNOUNCE_TOPIC, BindAddr, HiveAnnounce, HiveManifest, Propensity, ToolEntry};
 pub use uri::{AliasResolver, HostRef, HumUri, UriParseError};
 
 pub mod headroom;
@@ -96,6 +98,7 @@ pub struct Ensemble {
     expired_dusk: Arc<AtomicU64>,
     delivery: Arc<DeliveryState>,
     send_stats: Arc<SendStats>,
+    unverified: Arc<AtomicU64>,
 }
 
 #[derive(Clone)]
@@ -108,7 +111,11 @@ pub struct Inbox {
 impl Inbox {
     fn new() -> Self {
         let (tx, _) = broadcast::channel(256);
-        Self { tx, subscribers: Arc::new(AtomicUsize::new(0)), dropped: Arc::new(AtomicU64::new(0)) }
+        Self {
+            tx,
+            subscribers: Arc::new(AtomicUsize::new(0)),
+            dropped: Arc::new(AtomicU64::new(0)),
+        }
     }
 
     pub fn publish(&self, tone: Tone) -> bool {
@@ -126,12 +133,19 @@ impl Inbox {
 
     pub fn subscribe(&self) -> InboxSub {
         self.subscribers.fetch_add(1, Ordering::SeqCst);
-        InboxSub { rx: self.tx.subscribe(), subscribers: self.subscribers.clone() }
+        InboxSub {
+            rx: self.tx.subscribe(),
+            subscribers: self.subscribers.clone(),
+        }
     }
 
-    pub fn has_subscribers(&self) -> bool { self.subscribers.load(Ordering::SeqCst) > 0 }
+    pub fn has_subscribers(&self) -> bool {
+        self.subscribers.load(Ordering::SeqCst) > 0
+    }
 
-    pub fn dropped(&self) -> u64 { self.dropped.load(Ordering::Relaxed) }
+    pub fn dropped(&self) -> u64 {
+        self.dropped.load(Ordering::Relaxed)
+    }
 }
 
 pub struct InboxSub {
@@ -141,11 +155,15 @@ pub struct InboxSub {
 
 impl Deref for InboxSub {
     type Target = broadcast::Receiver<Tone>;
-    fn deref(&self) -> &Self::Target { &self.rx }
+    fn deref(&self) -> &Self::Target {
+        &self.rx
+    }
 }
 
 impl std::ops::DerefMut for InboxSub {
-    fn deref_mut(&mut self) -> &mut Self::Target { &mut self.rx }
+    fn deref_mut(&mut self) -> &mut Self::Target {
+        &mut self.rx
+    }
 }
 
 impl Drop for InboxSub {
@@ -180,10 +198,11 @@ impl Ensemble {
             inbox: Inbox::new(),
             gossip: GossipState::new(),
             kad: KadState::new(me),
-            strict_auth: false,
+            strict_auth: true,
             expired_dusk: Arc::new(AtomicU64::new(0)),
             delivery: DeliveryState::new(),
             send_stats: SendStats::new(),
+            unverified: Arc::new(AtomicU64::new(0)),
         }
     }
 
@@ -193,9 +212,13 @@ impl Ensemble {
         e
     }
 
-    pub fn me(&self) -> Hid { self.me }
+    pub fn me(&self) -> Hid {
+        self.me
+    }
 
-    pub fn strict_auth(&self) -> bool { self.strict_auth }
+    pub fn strict_auth(&self) -> bool {
+        self.strict_auth
+    }
 
     pub fn install(
         &self,
@@ -203,100 +226,7 @@ impl Ensemble {
         my_caps: PeerCapabilities,
         my_key: &HumdKey,
     ) {
-        let id = conn.peer().id;
-        let hello = hello_tone(&self.me, my_key, &my_caps);
-        let hello_conn = conn.clone();
-        tokio::spawn(async move {
-            let _ = hello_conn.send(hello).await;
-        });
-
-        let rx = conn.take_receiver();
-        self.peers.write().insert(
-            id,
-            Peer { conn: conn.clone(), learned_caps: None, lease: Lease::new() },
-        );
-        self.kad.note_peer(conn.peer().clone());
-
-        if let Some(mut rx) = rx {
-            let peers = self.peers.clone();
-            let inbox = self.inbox.clone();
-            let conn_for_drain = conn.clone();
-            let strict = self.strict_auth;
-            let gossip = self.gossip.clone();
-            let expired_dusk = self.expired_dusk.clone();
-            let delivery = self.delivery.clone();
-            let send_stats = self.send_stats.clone();
-            let kad = self.kad.clone();
-            let my_id = self.me;
-            tokio::spawn(async move {
-                let mut id = id;
-                let mut handshake_seen = false;
-                while let Some(tone) = rx.recv().await {
-                    let is_hello = tone.get("chi").and_then(|v| v.as_str()) == Some("hello");
-                    if is_hello && !handshake_seen {
-                        handshake_seen = true;
-                        match parse_hello(&tone) {
-                            HelloParse::Verified(claimed_id, caps) => {
-                                if claimed_id != id {
-                                    rekey_peer(&peers, &kad, &conn_for_drain, id, claimed_id);
-                                    id = claimed_id;
-                                }
-                                if let Some(p) = peers.write().get_mut(&id) {
-                                    p.learned_caps = Some(caps);
-                                }
-                            }
-                            HelloParse::Unsigned(claimed_id, caps) => {
-                                if strict {
-                                    tracing::warn!(
-                                        target: "ensemble",
-                                        transport_id = %id.short(),
-                                        claimed_id = %claimed_id.short(),
-                                        "hello.rejected: strict_auth requires signed hello"
-                                    );
-                                    peers.write().remove(&id);
-                                    conn_for_drain.close();
-                                    return;
-                                }
-                                if claimed_id == id {
-                                    if let Some(p) = peers.write().get_mut(&id) {
-                                        p.learned_caps = Some(caps);
-                                    }
-                                }
-                            }
-                            HelloParse::Invalid => {
-                                peers.write().remove(&id);
-                                conn_for_drain.close();
-                                return;
-                            }
-                        }
-                        continue;
-                    }
-                    if handle_liveness(&peers, &my_id, &id, &conn_for_drain, &tone).await {
-                        continue;
-                    }
-                    if !delivery::dispatchable(&tone, &delivery, &expired_dusk) {
-                        continue;
-                    }
-                    if tone.get("chi").and_then(|v| v.as_str()) == Some(GOSSIP_CHI) {
-                        if handle_gossip(&send_stats, &gossip, &peers, &id, &tone).await {
-                            continue;
-                        }
-                    }
-                    let chi_val = tone.get("chi").and_then(|v| v.as_str());
-                    if chi_val == Some(KAD_FIND_NODE_CHI)
-                        || chi_val == Some(KAD_FIND_NODE_RESP_CHI)
-                    {
-                        if handle_kad(&send_stats, &kad, &peers, &id, &my_id, &tone).await {
-                            continue;
-                        }
-                    }
-                    inbox.publish(tone);
-                }
-                if let Some(p) = peers.write().get_mut(&id) {
-                    p.lease.observe(LivenessSignal::TransportClosed);
-                }
-            });
-        }
+        self.admit(conn, hello_tone(&self.me, my_key, &my_caps), true);
     }
 
     pub fn add_peer(&self, conn: Arc<dyn PeerConnection>) {
@@ -307,105 +237,51 @@ impl Ensemble {
         self.install_unsigned(conn, caps);
     }
 
-    pub fn install_unsigned(
-        &self,
-        conn: Arc<dyn PeerConnection>,
-        my_caps: PeerCapabilities,
-    ) {
+    pub fn install_unsigned(&self, conn: Arc<dyn PeerConnection>, my_caps: PeerCapabilities) {
+        self.admit(conn, hello_tone_unsigned(&self.me, &my_caps), false);
+    }
+
+    fn admit(&self, conn: Arc<dyn PeerConnection>, hello: Tone, rekey: bool) {
         let id = conn.peer().id;
-        let hello = hello_tone_unsigned(&self.me, &my_caps);
+        conn.arm_opening();
+
+        let stats = self.send_stats.clone();
         let hello_conn = conn.clone();
         tokio::spawn(async move {
-            let _ = hello_conn.send(hello).await;
+            if send::send_opening_bounded(&hello_conn, hello, &stats, send::SEND_TIMEOUT)
+                .await
+                .is_err()
+            {
+                hello_conn.close();
+            }
         });
 
         let rx = conn.take_receiver();
         self.peers.write().insert(
             id,
-            Peer { conn: conn.clone(), learned_caps: None, lease: Lease::new() },
+            Peer {
+                conn: conn.clone(),
+                learned_caps: None,
+                lease: Lease::new(),
+            },
         );
         self.kad.note_peer(conn.peer().clone());
 
-        if let Some(mut rx) = rx {
-            let peers = self.peers.clone();
-            let inbox = self.inbox.clone();
-            let conn_for_drain = conn.clone();
-            let strict = self.strict_auth;
-            let gossip = self.gossip.clone();
-            let expired_dusk = self.expired_dusk.clone();
-            let delivery = self.delivery.clone();
-            let send_stats = self.send_stats.clone();
-            let kad = self.kad.clone();
-            let my_id = self.me;
-            tokio::spawn(async move {
-                let mut handshake_seen = false;
-                while let Some(tone) = rx.recv().await {
-                    let is_hello = tone.get("chi").and_then(|v| v.as_str()) == Some("hello");
-                    if is_hello && !handshake_seen {
-                        handshake_seen = true;
-                        match parse_hello(&tone) {
-                            HelloParse::Verified(claimed_id, caps) if claimed_id == id => {
-                                if let Some(p) = peers.write().get_mut(&id) {
-                                    p.learned_caps = Some(caps);
-                                }
-                            }
-                            HelloParse::Verified(claimed_id, _) => {
-                                tracing::warn!(
-                                    target: "ensemble",
-                                    transport_id = %id.short(),
-                                    claimed_id = %claimed_id.short(),
-                                    "hello.rejected: claimed humd_id does not match transport-peer id"
-                                );
-                                peers.write().remove(&id);
-                                conn_for_drain.close();
-                                return;
-                            }
-                            HelloParse::Unsigned(claimed_id, caps) => {
-                                if strict {
-                                    peers.write().remove(&id);
-                                    conn_for_drain.close();
-                                    return;
-                                }
-                                if claimed_id == id {
-                                    if let Some(p) = peers.write().get_mut(&id) {
-                                        p.learned_caps = Some(caps);
-                                    }
-                                }
-                            }
-                            HelloParse::Invalid => {
-                                peers.write().remove(&id);
-                                conn_for_drain.close();
-                                return;
-                            }
-                        }
-                        continue;
-                    }
-                    if handle_liveness(&peers, &my_id, &id, &conn_for_drain, &tone).await {
-                        continue;
-                    }
-                    if !delivery::dispatchable(&tone, &delivery, &expired_dusk) {
-                        continue;
-                    }
-                    if tone.get("chi").and_then(|v| v.as_str()) == Some(GOSSIP_CHI) {
-                        if handle_gossip(&send_stats, &gossip, &peers, &id, &tone).await {
-                            continue;
-                        }
-                    }
-                    let chi_val = tone.get("chi").and_then(|v| v.as_str());
-                    if chi_val == Some(KAD_FIND_NODE_CHI)
-                        || chi_val == Some(KAD_FIND_NODE_RESP_CHI)
-                    {
-                        if handle_kad(&send_stats, &kad, &peers, &id, &my_id, &tone).await {
-                            continue;
-                        }
-                    }
-                    inbox.publish(tone);
-                }
-                if let Some(p) = peers.write().get_mut(&id) {
-                    p.lease.observe(LivenessSignal::TransportClosed);
-                }
-            });
-        }
+        let Some(rx) = rx else { return };
+        let drain = Drain {
+            me: self.me,
+            peers: self.peers.clone(),
+            inbox: self.inbox.clone(),
+            gossip: self.gossip.clone(),
+            kad: self.kad.clone(),
+            delivery: self.delivery.clone(),
+            expired_dusk: self.expired_dusk.clone(),
+            send_stats: self.send_stats.clone(),
+            unverified: self.unverified.clone(),
+            strict: self.strict_auth,
+            rekey,
+        };
+        tokio::spawn(drain.run(rx, conn));
     }
 
     pub async fn probe_all(&self, seq: u64) {
@@ -415,8 +291,12 @@ impl Ensemble {
             .iter()
             .map(|(id, p)| (*id, p.conn.clone()))
             .collect();
-        for (id, conn) in peers {
-            let _ = send_bounded(&conn, ping_tone(&self.me, &id, seq), &self.send_stats).await;
+        let sends = peers
+            .into_iter()
+            .map(|(id, conn)| (conn, ping_tone(&self.me, &id, seq)))
+            .collect();
+        for e in fanout(sends, &self.send_stats).await {
+            tracing::debug!(target: "ensemble.liveness", error = %e, "probe.failed");
         }
     }
 
@@ -455,11 +335,20 @@ impl Ensemble {
     }
 
     pub fn remove_peer(&self, id: &Hid) {
+        self.kad.forget(id);
         if let Some(p) = self.peers.write().remove(id) {
             p.conn.close();
         }
     }
 
+    pub fn routing_snapshot(&self) -> Vec<HumdAddr> {
+        self.kad.snapshot()
+    }
+
+    pub fn routing_restore(&self, addrs: Vec<HumdAddr>) -> usize {
+        self.kad.restore(addrs)
+    }
+
     pub fn peers(&self) -> Vec<Hid> {
         self.peers.read().keys().copied().collect()
     }
@@ -473,14 +362,23 @@ impl Ensemble {
     }
 
     pub fn handshake_done(&self, id: &Hid) -> bool {
-        self.peers.read().get(id).is_some_and(|p| p.learned_caps.is_some())
+        self.peers
+            .read()
+            .get(id)
+            .is_some_and(|p| p.learned_caps.is_some())
     }
 
-    pub fn subscribe(&self) -> InboxSub { self.inbox.subscribe() }
+    pub fn subscribe(&self) -> InboxSub {
+        self.inbox.subscribe()
+    }
 
-    pub fn has_subscribers(&self) -> bool { self.inbox.has_subscribers() }
+    pub fn has_subscribers(&self) -> bool {
+        self.inbox.has_subscribers()
+    }
 
-    pub fn inbox_dropped(&self) -> u64 { self.inbox.dropped() }
+    pub fn inbox_dropped(&self) -> u64 {
+        self.inbox.dropped()
+    }
 
     pub fn expired_dusk(&self) -> u64 {
         self.expired_dusk.load(Ordering::SeqCst)
@@ -494,6 +392,10 @@ impl Ensemble {
         self.send_stats.timed_out()
     }
 
+    pub fn tones_before_handshake(&self) -> u64 {
+        self.unverified.load(Ordering::SeqCst)
+    }
+
     pub fn send_failures(&self) -> u64 {
         self.send_stats.failed()
     }
@@ -512,20 +414,14 @@ impl Ensemble {
         let rid = format!("gossip-{msg_id}");
         self.gossip.note_seen(&msg_id);
         let tone = gossip_tone_with_dusk(topic, &rid, &self.me, payload, &msg_id, dusk_ms);
-        let conns: Vec<Arc<dyn PeerConnection>> = {
-            let peers = self.peers.read();
-            peers.values().map(|p| p.conn.clone()).collect()
-        };
-        for conn in conns {
-            if let Err(e) = send_bounded(&conn, tone.clone(), &self.send_stats).await {
-                tracing::debug!(
-                    target: "ensemble.gossip",
-                    peer = %conn.peer().id.short(),
-                    topic = topic,
-                    error = %e,
-                    "publish send failed"
-                );
-            }
+        let sends: Vec<(Arc<dyn PeerConnection>, Tone)> = self
+            .peers
+            .read()
+            .values()
+            .map(|p| (p.conn.clone(), tone.clone()))
+            .collect();
+        for e in fanout(sends, &self.send_stats).await {
+            tracing::debug!(target: "ensemble.gossip", topic, error = %e, "publish.failed");
         }
     }
 
@@ -536,7 +432,7 @@ impl Ensemble {
     pub async fn hive_advertise(&self, manifest: hives::HiveManifest) {
         let env = hives::HiveAnnounce::Advertise {
             humd_id: self.me.to_hex(),
-            manifest,
+            manifest: Box::new(manifest),
         };
         match serde_json::to_value(&env) {
             Ok(payload) => self.publish(hives::ANNOUNCE_TOPIC, payload).await,
@@ -582,7 +478,10 @@ impl Ensemble {
         rx
     }
 
-    pub fn hive_discover(&self, name: impl Into<String>) -> mpsc::Receiver<(Hid, hives::HiveManifest)> {
+    pub fn hive_discover(
+        &self,
+        name: impl Into<String>,
+    ) -> mpsc::Receiver<(Hid, hives::HiveManifest)> {
         let needle = name.into();
         let mut raw = self.subscribe_topic(hives::ANNOUNCE_TOPIC);
         let (tx, rx) = mpsc::channel(64);
@@ -590,16 +489,15 @@ impl Ensemble {
             loop {
                 match raw.recv().await {
                     Ok(v) => {
-                        let parsed: Result<hives::HiveAnnounce, _> =
-                            serde_json::from_value(v);
+                        let parsed: Result<hives::HiveAnnounce, _> = serde_json::from_value(v);
                         if let Ok(hives::HiveAnnounce::Advertise { humd_id, manifest }) = parsed {
                             if manifest.name != needle {
                                 continue;
                             }
-                            if let Ok(id) = Hid::from_hex(&humd_id) {
-                                if tx.send((id, manifest)).await.is_err() {
-                                    break;
-                                }
+                            if let Ok(id) = Hid::from_hex(&humd_id)
+                                && tx.send((id, *manifest)).await.is_err()
+                            {
+                                break;
                             }
                         }
                     }
@@ -654,28 +552,35 @@ impl Ensemble {
                 };
                 if let Some(conn) = conn {
                     let kad = self.kad.clone();
+                    let stats = self.send_stats.clone();
                     let me = self.me;
                     let tgt = target;
                     joinset.spawn(async move {
-                        kad::query_peer(&kad, &conn, &me, &tgt, per_query_timeout).await
+                        kad::query_peer(&kad, &conn, &stats, &me, &tgt, per_query_timeout).await
                     });
                 }
             }
             if joinset.is_empty() {
                 continue;
             }
-            while let Some(res) = joinset.join_next().await {
-                let advertised_list = match res {
-                    Ok(list) => list,
-                    Err(_) => continue,
-                };
-                for advertised in advertised_list {
-                    self.kad.note_peer(advertised.clone());
-                    if advertised.id == self.me {
-                        continue;
+            let drain = async {
+                while let Some(res) = joinset.join_next().await {
+                    let advertised_list = match res {
+                        Ok(list) => list,
+                        Err(_) => continue,
+                    };
+                    for advertised in advertised_list {
+                        self.kad.note_peer(advertised.clone());
+                        if advertised.id == self.me {
+                            continue;
+                        }
+                        shortlist.insert(advertised);
                     }
-                    shortlist.insert(advertised);
                 }
+            };
+            if tokio::time::timeout_at(deadline, drain).await.is_err() {
+                joinset.abort_all();
+                return self.kad.get(&target);
             }
 
             if let Some(addr) = self.kad.get(&target) {
@@ -721,6 +626,122 @@ impl Ensemble {
     }
 }
 
+struct Drain {
+    me: Hid,
+    peers: Arc<RwLock<HashMap<Hid, Peer>>>,
+    inbox: Inbox,
+    gossip: Arc<GossipState>,
+    kad: Arc<KadState>,
+    delivery: Arc<DeliveryState>,
+    expired_dusk: Arc<AtomicU64>,
+    send_stats: Arc<SendStats>,
+    unverified: Arc<AtomicU64>,
+    strict: bool,
+    rekey: bool,
+}
+
+impl Drain {
+    async fn run(self, mut rx: mpsc::Receiver<Tone>, conn: Arc<dyn PeerConnection>) {
+        let mut id = conn.peer().id;
+        let mut handshook = false;
+
+        while let Some(tone) = rx.recv().await {
+            let chi = tone.get("chi").and_then(|v| v.as_str());
+
+            if !handshook {
+                if chi == Some("hello") {
+                    handshook = true;
+                    if self.hello(&mut id, &conn, parse_hello(&tone)).is_err() {
+                        self.peers.write().remove(&id);
+                        conn.close();
+                        return;
+                    }
+                } else {
+                    self.unverified.fetch_add(1, Ordering::SeqCst);
+                    continue;
+                }
+                continue;
+            }
+            if handle_liveness(&self.peers, &self.send_stats, &self.me, &id, &conn, &tone).await {
+                continue;
+            }
+            if !delivery::dispatchable(&tone, &self.delivery, &self.expired_dusk) {
+                continue;
+            }
+            if chi == Some(GOSSIP_CHI)
+                && handle_gossip(&self.send_stats, &self.gossip, &self.peers, &id, &tone).await
+            {
+                continue;
+            }
+            if matches!(chi, Some(KAD_FIND_NODE_CHI) | Some(KAD_FIND_NODE_RESP_CHI))
+                && handle_kad(
+                    &self.send_stats,
+                    &self.kad,
+                    &self.peers,
+                    &id,
+                    &self.me,
+                    &tone,
+                )
+                .await
+            {
+                continue;
+            }
+            self.inbox.publish(tone);
+        }
+
+        if let Some(p) = self.peers.write().get_mut(&id) {
+            p.lease.observe(LivenessSignal::TransportClosed);
+        }
+    }
+
+    fn hello(
+        &self,
+        id: &mut Hid,
+        conn: &Arc<dyn PeerConnection>,
+        parsed: HelloParse,
+    ) -> Result<(), ()> {
+        match parsed {
+            HelloParse::Verified(claimed, caps) => {
+                if claimed != *id && !self.rekey {
+                    tracing::warn!(
+                        target: "ensemble",
+                        transport_id = %id.short(),
+                        claimed_id = %claimed.short(),
+                        "hello.rejected: claimed id differs from the id this link was admitted under"
+                    );
+                    return Err(());
+                }
+                if claimed != *id {
+                    rekey_peer(&self.peers, &self.kad, conn, *id, claimed);
+                    *id = claimed;
+                }
+                if let Some(p) = self.peers.write().get_mut(id) {
+                    p.learned_caps = Some(caps);
+                }
+                Ok(())
+            }
+            HelloParse::Unsigned(claimed, caps) => {
+                if self.strict {
+                    tracing::warn!(
+                        target: "ensemble",
+                        transport_id = %id.short(),
+                        claimed_id = %claimed.short(),
+                        "hello.rejected: strict_auth requires a signed hello"
+                    );
+                    return Err(());
+                }
+                if claimed == *id
+                    && let Some(p) = self.peers.write().get_mut(id)
+                {
+                    p.learned_caps = Some(caps);
+                }
+                Ok(())
+            }
+            HelloParse::Invalid => Err(()),
+        }
+    }
+}
+
 fn rekey_peer(
     peers: &Arc<RwLock<HashMap<Hid, Peer>>>,
     kad: &Arc<KadState>,
@@ -753,21 +774,24 @@ async fn handle_kad(
             None => return false,
         };
         let closest = kad.closest_to(&parsed.target, KAD_K);
-        let resp_rid = format!("kad-resp-{}", &parsed.query_id[..8.min(parsed.query_id.len())]);
+        let resp_rid = format!(
+            "kad-resp-{}",
+            &parsed.query_id[..8.min(parsed.query_id.len())]
+        );
         let resp = kad::find_node_resp_tone(&resp_rid, &parsed.query_id, me, &closest);
         let conn = {
             let peers = peers.read();
             peers.get(arrived_from).map(|p| p.conn.clone())
         };
-        if let Some(conn) = conn {
-            if let Err(e) = send_bounded(&conn, resp, send_stats).await {
-                tracing::debug!(
-                    target: "ensemble.kad",
-                    peer = %arrived_from.short(),
-                    error = %e,
-                    "find-node response send failed"
-                );
-            }
+        if let Some(conn) = conn
+            && let Err(e) = send_bounded(&conn, resp, send_stats).await
+        {
+            tracing::debug!(
+                target: "ensemble.kad",
+                peer = %arrived_from.short(),
+                error = %e,
+                "find-node response send failed"
+            );
         }
         true
     } else if chi_val == Some(kad::KAD_FIND_NODE_RESP_CHI) {
@@ -787,6 +811,7 @@ async fn handle_kad(
 
 async fn handle_liveness(
     peers: &Arc<RwLock<HashMap<Hid, Peer>>>,
+    send_stats: &Arc<SendStats>,
     me: &Hid,
     arrived_from: &Hid,
     conn: &Arc<dyn PeerConnection>,
@@ -803,13 +828,13 @@ async fn handle_liveness(
         return false;
     }
     if let Some(seq) = probe_seq(tone) {
-        let _ = conn.send(pong_tone(me, arrived_from, seq)).await;
+        let _ = send_bounded(conn, pong_tone(me, arrived_from, seq), send_stats).await;
     }
     true
 }
 
 async fn handle_gossip(
-    send_stats: &SendStats,
+    send_stats: &Arc<SendStats>,
     gossip: &Arc<gossip::GossipState>,
     peers: &Arc<RwLock<HashMap<Hid, Peer>>>,
     arrived_from: &Hid,
@@ -825,23 +850,14 @@ async fn handle_gossip(
     if let Some(tx) = gossip.sender(parsed.topic) {
         let _ = tx.send(parsed.payload.clone());
     }
-    let others: Vec<Arc<dyn PeerConnection>> = {
-        let peers = peers.read();
-        peers
-            .iter()
-            .filter(|(id, _)| *id != arrived_from)
-            .map(|(_, p)| p.conn.clone())
-            .collect()
-    };
-    for conn in others {
-        if let Err(e) = send_bounded(&conn, tone.clone(), send_stats).await {
-            tracing::debug!(
-                target: "ensemble.gossip",
-                peer = %conn.peer().id.short(),
-                error = %e,
-                "re-fan send failed"
-            );
-        }
+    let sends: Vec<(Arc<dyn PeerConnection>, Tone)> = peers
+        .read()
+        .iter()
+        .filter(|(id, _)| *id != arrived_from)
+        .map(|(_, p)| (p.conn.clone(), tone.clone()))
+        .collect();
+    for e in fanout(sends, send_stats).await {
+        tracing::debug!(target: "ensemble.gossip", error = %e, "refan.failed");
     }
     true
 }
@@ -877,7 +893,10 @@ mod tests {
     }
 
     fn rids(tones: &[Tone]) -> Vec<String> {
-        tones.iter().map(|t| t["rid"].as_str().unwrap().to_string()).collect()
+        tones
+            .iter()
+            .map(|t| t["rid"].as_str().unwrap().to_string())
+            .collect()
     }
 
     #[tokio::test]
@@ -890,7 +909,14 @@ mod tests {
         let got = rids(&drain(&mut rx).await);
         assert_eq!(got, ["t0", "t1", "t2", "t3", "t4"]);
         let c = a.counters();
-        assert_eq!(c, LinkCounters { offered: 5, delivered: 5, ..Default::default() });
+        assert_eq!(
+            c,
+            LinkCounters {
+                offered: 5,
+                delivered: 5,
+                ..Default::default()
+            }
+        );
     }
 
     #[tokio::test]
@@ -975,9 +1001,16 @@ mod tests {
             drain(&mut rx).await.len()
         }
         assert_eq!(run(42).await, run(42).await, "same seed, same pattern");
-        assert_ne!(run(1).await, run(2).await, "different seeds, different patterns");
+        assert_ne!(
+            run(1).await,
+            run(2).await,
+            "different seeds, different patterns"
+        );
         let survived = run(42).await;
-        assert!((120..170).contains(&survived), "survived {survived} of 200 at 30% loss");
+        assert!(
+            (120..170).contains(&survived),
+            "survived {survived} of 200 at 30% loss"
+        );
     }
 
     #[tokio::test]
@@ -1079,10 +1112,16 @@ mod tests {
 
         let c = a.counters();
         let arrived = drain(&mut rx).await.len() as u64;
-        assert_eq!(c.delivered, arrived, "delivered must match what the peer saw");
+        assert_eq!(
+            c.delivered, arrived,
+            "delivered must match what the peer saw"
+        );
         assert_eq!(c.offered, 43);
         assert_eq!(c.buffered, 3);
-        assert_eq!(c.offered, c.delivered - c.duplicated + c.dropped + c.lost_on_heal);
+        assert_eq!(
+            c.offered,
+            c.delivered - c.duplicated + c.dropped + c.lost_on_heal
+        );
     }
 
     #[tokio::test]
@@ -1093,7 +1132,11 @@ mod tests {
         a.set_faults(LinkFaults::default().drop_next(10));
         b.send(tone("survivor")).await.unwrap();
         a.send(tone("doomed")).await.unwrap();
-        assert_eq!(rids(&drain(&mut a_rx).await), ["survivor"], "b→a is unaffected");
+        assert_eq!(
+            rids(&drain(&mut a_rx).await),
+            ["survivor"],
+            "b→a is unaffected"
+        );
         assert!(drain(&mut b_rx).await.is_empty(), "a→b drops");
     }
 
@@ -1150,11 +1193,21 @@ mod tests {
         let a_id = Hid::random_humd();
         let b_id = Hid::random_humd();
         let (a, b) = InMemoryEndpoint::pair(
-            a_id, PeerCapabilities { proto_version: "0.2.0".into(), ..Default::default() },
-            b_id, PeerCapabilities { proto_version: "0.2.0".into(), ..Default::default() },
+            a_id,
+            PeerCapabilities {
+                proto_version: "0.2.0".into(),
+                ..Default::default()
+            },
+            b_id,
+            PeerCapabilities {
+                proto_version: "0.2.0".into(),
+                ..Default::default()
+            },
         );
         let mut rx_b = b.take_receiver().unwrap();
-        a.send(json!({"chi": "hello", "rid": "1", "from": a_id.to_hex()})).await.unwrap();
+        a.send(json!({"chi": "hello", "rid": "1", "from": a_id.to_hex()}))
+            .await
+            .unwrap();
         let received = rx_b.recv().await.unwrap();
         assert_eq!(received.get("chi").unwrap(), "hello");
     }
@@ -1167,8 +1220,10 @@ mod tests {
 
         let ensemble = Ensemble::new(me);
         let (mine, theirs) = InMemoryEndpoint::pair(
-            me, PeerCapabilities::default(),
-            peer_id, PeerCapabilities::default(),
+            me,
+            PeerCapabilities::default(),
+            peer_id,
+            PeerCapabilities::default(),
         );
         ensemble.add_peer(mine);
         let mut rx = theirs.take_receiver().unwrap();
@@ -1212,10 +1267,7 @@ mod tests {
             free_slots: None,
             headroom: headroom::CellHeadroom::default(),
         };
-        let (a_side, b_side) = InMemoryEndpoint::pair(
-            a_id, b_caps.clone(),
-            b_id, a_caps.clone(),
-        );
+        let (a_side, b_side) = InMemoryEndpoint::pair(a_id, b_caps.clone(), b_id, a_caps.clone());
 
         let ensemble_a = Ensemble::new(a_id);
         let ensemble_b = Ensemble::new(b_id);
@@ -1261,16 +1313,32 @@ mod tests {
         let me = me_key.hid();
         let peer_id = peer_key.hid();
         let (mine, theirs) = InMemoryEndpoint::pair(
-            me, PeerCapabilities::default(),
-            peer_id, PeerCapabilities::default(),
+            me,
+            PeerCapabilities::default(),
+            peer_id,
+            PeerCapabilities::default(),
         );
 
         let ensemble = Ensemble::new(me);
         let mut sub = ensemble.subscribe();
-        ensemble.install(mine, PeerCapabilities { proto_version: "0.3.0".into(), ..Default::default() }, &me_key);
+        ensemble.install(
+            mine,
+            PeerCapabilities {
+                proto_version: "0.3.0".into(),
+                ..Default::default()
+            },
+            &me_key,
+        );
 
         theirs
-            .send(hello_tone(&peer_id, &peer_key, &PeerCapabilities { proto_version: "0.3.0".into(), ..Default::default() }))
+            .send(hello_tone(
+                &peer_id,
+                &peer_key,
+                &PeerCapabilities {
+                    proto_version: "0.3.0".into(),
+                    ..Default::default()
+                },
+            ))
             .await
             .unwrap();
         theirs
@@ -1299,16 +1367,32 @@ mod tests {
         let me = me_key.hid();
         let peer_id = peer_key.hid();
         let (mine, theirs) = InMemoryEndpoint::pair(
-            me, PeerCapabilities::default(),
-            peer_id, PeerCapabilities::default(),
+            me,
+            PeerCapabilities::default(),
+            peer_id,
+            PeerCapabilities::default(),
         );
 
         let ensemble = Ensemble::new(me);
         let mut sub = ensemble.subscribe();
-        ensemble.install(mine, PeerCapabilities { proto_version: "0.2.0".into(), ..Default::default() }, &me_key);
+        ensemble.install(
+            mine,
+            PeerCapabilities {
+                proto_version: "0.2.0".into(),
+                ..Default::default()
+            },
+            &me_key,
+        );
 
         theirs
-            .send(hello_tone(&peer_id, &peer_key, &PeerCapabilities { proto_version: "0.2.0".into(), ..Default::default() }))
+            .send(hello_tone(
+                &peer_id,
+                &peer_key,
+                &PeerCapabilities {
+                    proto_version: "0.2.0".into(),
+                    ..Default::default()
+                },
+            ))
             .await
             .unwrap();
         theirs
@@ -1324,3 +1408,126 @@ mod tests {
         assert_eq!(got.get("rid").unwrap(), "r1");
     }
 }
+
+#[cfg(test)]
+mod invariants {
+    use super::*;
+    use serde_json::json;
+
+    fn caps() -> PeerCapabilities {
+        PeerCapabilities::default()
+    }
+
+    #[tokio::test]
+    async fn the_opening_frame_precedes_application_traffic() {
+        let me = Hid::random_humd();
+        let peer = Hid::random_humd();
+        let key = HumdKey::generate();
+        let ensemble = Ensemble::new(me);
+        let (mine, theirs) = InMemoryEndpoint::pair_concrete(me, caps(), peer, caps());
+        let mut rx = theirs
+            .take_receiver()
+            .expect("far end must expose a receiver");
+
+        ensemble.install(mine, caps(), &key);
+        ensemble
+            .route(json!({"chi": "prompt", "rid": "r1", "to": peer.to_hex()}))
+            .await
+            .expect("routing to the only peer cannot fail");
+
+        let first = rx.recv().await.expect("receiver closed");
+        assert_eq!(first["chi"], "hello", "the hello must be the first frame");
+        let second = rx.recv().await.expect("receiver closed");
+        assert_eq!(second["rid"], "r1", "application traffic follows the hello");
+    }
+
+    #[tokio::test]
+    async fn a_tone_that_outruns_the_handshake_is_never_published() {
+        let me = Hid::random_humd();
+        let peer = Hid::random_humd();
+        let key = HumdKey::generate();
+        let ensemble = Ensemble::new(me);
+        let (mine, theirs) = InMemoryEndpoint::pair_concrete(me, caps(), peer, caps());
+
+        ensemble.install(mine, caps(), &key);
+
+        let mut inbox = ensemble.subscribe();
+        theirs
+            .send(json!({"chi": "prompt", "rid": "early", "to": me.to_hex()}))
+            .await
+            .expect("in-memory send cannot fail");
+        tokio::time::sleep(Duration::from_millis(50)).await;
+
+        assert_eq!(
+            ensemble.tones_before_handshake(),
+            1,
+            "the drop must be counted"
+        );
+        assert!(
+            inbox.try_recv().is_err(),
+            "an unauthenticated tone must not publish"
+        );
+    }
+
+    #[tokio::test]
+    async fn an_unsigned_hello_is_refused_under_the_default_posture() {
+        let me = Hid::random_humd();
+        let peer = Hid::random_humd();
+        let ensemble = Ensemble::new(me);
+        assert!(
+            ensemble.strict_auth(),
+            "strict auth must be the default posture"
+        );
+        let (mine, theirs) = InMemoryEndpoint::pair_concrete(me, caps(), peer, caps());
+
+        ensemble.install_unsigned(mine, caps());
+        theirs
+            .send(hello_tone_unsigned(&peer, &caps()))
+            .await
+            .expect("in-memory send cannot fail");
+        tokio::time::sleep(Duration::from_millis(50)).await;
+
+        assert!(
+            !ensemble.peers().contains(&peer),
+            "an unsigned peer must not be admitted under strict auth"
+        );
+    }
+
+    #[tokio::test]
+    async fn the_same_unsigned_hello_is_accepted_when_strictness_is_waived() {
+        let me = Hid::random_humd();
+        let peer = Hid::random_humd();
+        let ensemble = Ensemble::with_strict_auth(me, false);
+        let (mine, theirs) = InMemoryEndpoint::pair_concrete(me, caps(), peer, caps());
+
+        ensemble.install_unsigned(mine, caps());
+        theirs
+            .send(hello_tone_unsigned(&peer, &caps()))
+            .await
+            .expect("in-memory send cannot fail");
+        tokio::time::sleep(Duration::from_millis(50)).await;
+
+        assert!(
+            ensemble.handshake_done(&peer),
+            "waived strictness must admit the peer"
+        );
+    }
+
+    #[tokio::test]
+    async fn a_removed_peer_leaves_the_routing_table() {
+        let me = Hid::random_humd();
+        let peer = Hid::random_humd();
+        let ensemble = Ensemble::new(me);
+        let (mine, _theirs) = InMemoryEndpoint::pair_concrete(me, caps(), peer, caps());
+
+        ensemble.add_peer(mine);
+        assert!(ensemble.kad_closest(&peer, 4).iter().any(|a| a.id == peer));
+
+        ensemble.remove_peer(&peer);
+        assert_eq!(
+            ensemble.kad_routing_table_len(),
+            0,
+            "a dead peer must not route"
+        );
+    }
+}
diff --git a/ensemble/src/link.rs b/ensemble/src/link.rs
index c946a91..596fec4 100644
--- a/ensemble/src/link.rs
+++ b/ensemble/src/link.rs
@@ -4,11 +4,12 @@ use parking_lot::Mutex;
 use rand::rngs::StdRng;
 use rand::{Rng, SeedableRng};
 use std::collections::VecDeque;
-use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, Ordering};
 use tokio::sync::mpsc;
 
 use super::{Hid, HumdAddr, PeerCapabilities, PeerConnection, Tone};
+use crate::opening::Gate;
 
 #[derive(Debug, Clone, Default, PartialEq, Eq)]
 pub struct Script {
@@ -22,7 +23,7 @@ pub struct Script {
 impl Script {
     fn take(counted: &mut usize, every: usize, offered: u64) -> bool {
         match *counted {
-            0 => every > 0 && offered % every as u64 == 0,
+            0 => every > 0 && offered.is_multiple_of(every as u64),
             n => {
                 *counted = n - 1;
                 true
@@ -68,7 +69,12 @@ pub struct LinkFaults {
 
 impl Default for LinkFaults {
     fn default() -> Self {
-        Self { script: Script::default(), noise: Noise::default(), seed: 0x5EED_C0DE, offered: 0 }
+        Self {
+            script: Script::default(),
+            noise: Noise::default(),
+            seed: 0x5EED_C0DE,
+            offered: 0,
+        }
     }
 }
 
@@ -111,7 +117,9 @@ impl LinkFaults {
 
     fn verdict(&mut self, rng: &mut impl Rng) -> Verdict {
         self.offered += 1;
-        self.script.verdict(self.offered).unwrap_or_else(|| self.noise.verdict(rng))
+        self.script
+            .verdict(self.offered)
+            .unwrap_or_else(|| self.noise.verdict(rng))
     }
 }
 
@@ -158,6 +166,7 @@ pub struct InMemoryEndpoint {
     reorder_hold: Mutex<Option<Tone>>,
     killed: AtomicBool,
     stalled: AtomicBool,
+    gate: Gate,
 }
 
 struct PartitionState {
@@ -195,10 +204,14 @@ impl InMemoryEndpoint {
                 partitioned: false,
                 buffer: VecDeque::new(),
             }),
-            faults: Mutex::new(LinkFaults { seed: seed_a, ..Default::default() }),
+            faults: Mutex::new(LinkFaults {
+                seed: seed_a,
+                ..Default::default()
+            }),
             rng: Mutex::new(StdRng::seed_from_u64(seed_a)),
             counters: Mutex::new(LinkCounters::default()),
             reorder_hold: Mutex::new(None),
+            gate: Gate::new(),
             killed: AtomicBool::new(false),
             stalled: AtomicBool::new(false),
         });
@@ -211,10 +224,14 @@ impl InMemoryEndpoint {
                 partitioned: false,
                 buffer: VecDeque::new(),
             }),
-            faults: Mutex::new(LinkFaults { seed: seed_b, ..Default::default() }),
+            faults: Mutex::new(LinkFaults {
+                seed: seed_b,
+                ..Default::default()
+            }),
             rng: Mutex::new(StdRng::seed_from_u64(seed_b)),
             counters: Mutex::new(LinkCounters::default()),
             reorder_hold: Mutex::new(None),
+            gate: Gate::new(),
             killed: AtomicBool::new(false),
             stalled: AtomicBool::new(false),
         });
@@ -250,7 +267,10 @@ impl InMemoryEndpoint {
     async fn emit(&self, tone: Tone) -> Result<()> {
         if self.stalled.load(Ordering::SeqCst) {
             self.counters.lock().stalled_sends += 1;
-            return Ok(std::future::pending::<()>().await);
+            return {
+                let _: () = std::future::pending::<()>().await;
+                Ok(())
+            };
         }
         let tx = {
             let mut guard = self.tx.lock();
@@ -298,13 +318,21 @@ impl InMemoryEndpoint {
         *self.rng.lock() = StdRng::seed_from_u64(seed);
     }
 
-    pub fn faults(&self) -> LinkFaults { self.faults.lock().clone() }
+    pub fn faults(&self) -> LinkFaults {
+        self.faults.lock().clone()
+    }
 
-    pub fn counters(&self) -> LinkCounters { *self.counters.lock() }
+    pub fn counters(&self) -> LinkCounters {
+        *self.counters.lock()
+    }
 
-    pub fn buffered(&self) -> usize { self.partition.lock().buffer.len() }
+    pub fn buffered(&self) -> usize {
+        self.partition.lock().buffer.len()
+    }
 
-    pub fn is_partitioned(&self) -> bool { self.partition.lock().partitioned }
+    pub fn is_partitioned(&self) -> bool {
+        self.partition.lock().partitioned
+    }
 
     pub fn flush_reorder(&self) -> bool {
         match self.reorder_hold.lock().take() {
@@ -337,10 +365,41 @@ impl InMemoryEndpoint {
 
 #[async_trait]
 impl PeerConnection for InMemoryEndpoint {
-    fn peer(&self) -> &HumdAddr { &self.peer }
-    fn capabilities(&self) -> &PeerCapabilities { &self.caps }
+    fn peer(&self) -> &HumdAddr {
+        &self.peer
+    }
+    fn capabilities(&self) -> &PeerCapabilities {
+        &self.caps
+    }
 
     async fn send(&self, tone: Tone) -> Result<()> {
+        self.gate.wait().await;
+        self.deliver(tone).await
+    }
+
+    async fn send_opening(&self, tone: Tone) -> Result<()> {
+        let result = self.deliver(tone).await;
+        self.gate.opened();
+        result
+    }
+
+    fn take_receiver(&self) -> Option<mpsc::Receiver<Tone>> {
+        self.rx.lock().take()
+    }
+
+    fn arm_opening(&self) {
+        self.gate.arm();
+    }
+
+    fn close(&self) {
+        self.gate.shut();
+        self.tx.lock().take();
+        let _ = self.rx.lock().take();
+    }
+}
+
+impl InMemoryEndpoint {
+    async fn deliver(&self, tone: Tone) -> Result<()> {
         if self.killed.load(Ordering::SeqCst) {
             return Err(anyhow::anyhow!("link to {} is dead", self.peer.id.short()));
         }
@@ -389,13 +448,4 @@ impl PeerConnection for InMemoryEndpoint {
         }
         Ok(())
     }
-
-    fn take_receiver(&self) -> Option<mpsc::Receiver<Tone>> {
-        self.rx.lock().take()
-    }
-
-    fn close(&self) {
-        self.tx.lock().take();
-        let _ = self.rx.lock().take();
-    }
 }
diff --git a/ensemble/src/liveness.rs b/ensemble/src/liveness.rs
index 337609a..614531d 100644
--- a/ensemble/src/liveness.rs
+++ b/ensemble/src/liveness.rs
@@ -1,4 +1,3 @@
-
 use std::time::Duration;
 
 use crate::{Hid, Tone};
diff --git a/ensemble/src/opening.rs b/ensemble/src/opening.rs
new file mode 100644
index 0000000..dc5776b
--- /dev/null
+++ b/ensemble/src/opening.rs
@@ -0,0 +1,145 @@
+use std::sync::atomic::{AtomicBool, Ordering};
+
+use parking_lot::Mutex;
+use tokio::sync::Notify;
+
+/// `arm` is synchronous so it can be called from a non-async `install`.
+/// Every later `send` parks until the opening frame has landed, and a
+/// shut gate never reopens.
+#[derive(Debug)]
+pub struct Gate {
+    open: Mutex<bool>,
+    arrived: Notify,
+    closed: AtomicBool,
+}
+
+impl Gate {
+    pub fn new() -> Self {
+        Self {
+            open: Mutex::new(true),
+            arrived: Notify::new(),
+            closed: AtomicBool::new(false),
+        }
+    }
+
+    pub fn arm(&self) {
+        *self.open.lock() = false;
+    }
+
+    pub fn is_open(&self) -> bool {
+        *self.open.lock()
+    }
+
+    pub async fn wait(&self) {
+        loop {
+            // Register before reading the flag: `notify_waiters` reaches
+            // only waiters already enrolled, so checking first can drop a
+            // wakeup and park forever.
+            if self.is_open() || self.closed.load(Ordering::SeqCst) {
+                return;
+            }
+            self.arrived.notified().await;
+        }
+    }
+
+    pub fn opened(&self) {
+        if self.closed.load(Ordering::SeqCst) {
+            return;
+        }
+        *self.open.lock() = true;
+        self.arrived.notify_waiters();
+    }
+
+    pub fn shut(&self) {
+        self.closed.store(true, Ordering::SeqCst);
+        self.arrived.notify_waiters();
+    }
+}
+
+impl Default for Gate {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::sync::Arc;
+    use std::time::Duration;
+
+    #[tokio::test]
+    async fn an_armed_gate_parks_until_it_opens() {
+        let gate = Arc::new(Gate::new());
+        gate.arm();
+        assert!(!gate.is_open());
+
+        let waiter = tokio::spawn({
+            let gate = gate.clone();
+            async move {
+                gate.wait().await;
+                "through"
+            }
+        });
+
+        tokio::time::sleep(Duration::from_millis(20)).await;
+        assert!(!waiter.is_finished(), "an armed gate must hold traffic");
+
+        gate.opened();
+        assert_eq!(
+            tokio::time::timeout(Duration::from_secs(1), waiter)
+                .await
+                .unwrap()
+                .unwrap(),
+            "through"
+        );
+    }
+
+    #[tokio::test]
+    async fn a_gate_that_is_never_armed_passes_immediately() {
+        let gate = Gate::new();
+        tokio::time::timeout(Duration::from_millis(50), gate.wait())
+            .await
+            .expect("an open gate must not park");
+    }
+
+    #[tokio::test]
+    async fn shutting_releases_waiters_so_a_stalled_opening_frame_cannot_wedge_a_link() {
+        let gate = Arc::new(Gate::new());
+        gate.arm();
+        let waiter = tokio::spawn({
+            let gate = gate.clone();
+            async move {
+                gate.wait().await;
+                "released"
+            }
+        });
+        tokio::time::sleep(Duration::from_millis(20)).await;
+        gate.shut();
+        assert_eq!(
+            tokio::time::timeout(Duration::from_secs(1), waiter)
+                .await
+                .unwrap()
+                .unwrap(),
+            "released"
+        );
+    }
+
+    #[tokio::test]
+    async fn a_late_opening_frame_cannot_resurrect_a_shut_gate() {
+        let gate = Gate::new();
+        gate.arm();
+        gate.shut();
+        gate.opened();
+        assert!(!gate.is_open(), "a closed link must stay closed");
+        tokio::time::timeout(Duration::from_millis(50), gate.wait())
+            .await
+            .expect("a shut gate must not park");
+    }
+
+    #[tokio::test]
+    async fn a_default_gate_starts_open() {
+        let gate = Gate::default();
+        assert!(gate.is_open(), "default must match new");
+    }
+}
diff --git a/ensemble/src/send.rs b/ensemble/src/send.rs
index 2bbfc62..70c6f8a 100644
--- a/ensemble/src/send.rs
+++ b/ensemble/src/send.rs
@@ -1,8 +1,9 @@
-
-use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, Ordering};
 use std::time::Duration;
 
+use tokio::task::JoinSet;
+
 use crate::{PeerConnection, Tone};
 
 pub const SEND_TIMEOUT: Duration = Duration::from_secs(5);
@@ -35,6 +36,27 @@ impl SendStats {
     }
 }
 
+/// Resolves when every send has landed or hit its deadline.
+pub async fn fanout(
+    sends: Vec<(Arc<dyn PeerConnection>, Tone)>,
+    stats: &Arc<SendStats>,
+) -> Vec<SendError> {
+    let mut set = JoinSet::new();
+    for (conn, tone) in sends {
+        let stats = stats.clone();
+        set.spawn(async move { send_bounded(&conn, tone, &stats).await });
+    }
+    let mut out = Vec::new();
+    while let Some(joined) = set.join_next().await {
+        match joined {
+            Ok(Err(e)) => out.push(e),
+            Ok(Ok(())) => {}
+            Err(e) => tracing::error!(target: "ensemble.send", error = %e, "fanout.task.panicked"),
+        }
+    }
+    out
+}
+
 pub async fn send_bounded(
     conn: &Arc<dyn PeerConnection>,
     tone: Tone,
@@ -69,6 +91,33 @@ pub async fn send_bounded_with(
     }
 }
 
+pub async fn send_opening_bounded(
+    conn: &Arc<dyn PeerConnection>,
+    tone: Tone,
+    stats: &SendStats,
+    deadline: Duration,
+) -> Result<(), SendError> {
+    // `send_opening`, not `send`: `send` parks on the gate this frame opens.
+    match tokio::time::timeout(deadline, conn.send_opening(tone)).await {
+        Ok(Ok(())) => Ok(()),
+        Ok(Err(e)) => {
+            stats.failed.fetch_add(1, Ordering::SeqCst);
+            Err(SendError::Failed(e.to_string()))
+        }
+        Err(_) => {
+            stats.timed_out.fetch_add(1, Ordering::SeqCst);
+            tracing::warn!(
+                target: "ensemble.send",
+                peer = %conn.peer().id.short(),
+                timeout_ms = deadline.as_millis() as u64,
+                "opening.timed_out: peer never took the hello, closing"
+            );
+            conn.close();
+            Err(SendError::TimedOut)
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -109,7 +158,10 @@ mod tests {
 
     fn conn(stalls: bool) -> Arc<Stalled> {
         Arc::new(Stalled {
-            addr: HumdAddr { id: Hid::random_humd(), hints: vec![] },
+            addr: HumdAddr {
+                id: Hid::random_humd(),
+                hints: vec![],
+            },
             caps: PeerCapabilities::default(),
             stalled: AtomicBool::new(stalls),
             closed: AtomicBool::new(false),
@@ -121,6 +173,98 @@ mod tests {
         c.clone()
     }
 
+    struct Slow {
+        addr: HumdAddr,
+        caps: PeerCapabilities,
+        delay: Duration,
+        sent: AtomicU64,
+    }
+
+    #[async_trait::async_trait]
+    impl PeerConnection for Slow {
+        fn peer(&self) -> &HumdAddr {
+            &self.addr
+        }
+        fn capabilities(&self) -> &PeerCapabilities {
+            &self.caps
+        }
+        async fn send(&self, _tone: Tone) -> anyhow::Result<()> {
+            tokio::time::sleep(self.delay).await;
+            self.sent.fetch_add(1, Ordering::SeqCst);
+            Ok(())
+        }
+        fn take_receiver(&self) -> Option<tokio::sync::mpsc::Receiver<Tone>> {
+            None
+        }
+        fn close(&self) {}
+    }
+
+    #[tokio::test]
+    async fn a_stalled_opening_hits_its_deadline_and_closes_the_link() {
+        let c = conn(false);
+        c.stalled.store(true, Ordering::SeqCst);
+        let stats = SendStats::new();
+
+        let got = tokio::time::timeout(
+            Duration::from_secs(10),
+            send_opening_bounded(
+                &as_conn(&c),
+                serde_json::json!({"chi": "hello"}),
+                &stats,
+                Duration::from_millis(50),
+            ),
+        )
+        .await
+        .expect("a stalled opening must return, not hang");
+
+        assert_eq!(got, Err(SendError::TimedOut));
+        assert!(c.closed.load(Ordering::SeqCst), "a stalled opening must close the link");
+        assert_eq!(stats.timed_out(), 1, "the timeout must be accounted for");
+    }
+
+    #[tokio::test]
+    async fn a_fanout_costs_the_slowest_peer_not_the_sum() {
+        let delay = Duration::from_millis(120);
+        let peers = 5;
+        let conns: Vec<Arc<Slow>> = (0..peers)
+            .map(|_| {
+                Arc::new(Slow {
+                    addr: HumdAddr {
+                        id: Hid::random_humd(),
+                        hints: vec![],
+                    },
+                    caps: PeerCapabilities::default(),
+                    delay,
+                    sent: AtomicU64::new(0),
+                })
+            })
+            .collect();
+
+        let sends: Vec<(Arc<dyn PeerConnection>, Tone)> = conns
+            .iter()
+            .map(|c| {
+                (
+                    c.clone() as Arc<dyn PeerConnection>,
+                    serde_json::json!({"chi": "chunk"}),
+                )
+            })
+            .collect();
+
+        let started = std::time::Instant::now();
+        let errors = fanout(sends, &SendStats::new()).await;
+        let elapsed = started.elapsed();
+
+        assert!(errors.is_empty(), "healthy peers must not report errors");
+        for c in &conns {
+            assert_eq!(c.sent.load(Ordering::SeqCst), 1, "every peer got the tone");
+        }
+        assert!(
+            elapsed < delay * 2,
+            "a serial fanout would take {}ms; took {elapsed:?}",
+            delay.as_millis() * peers
+        );
+    }
+
     #[tokio::test]
     async fn a_stalled_peer_times_out_rather_than_hanging() {
         let c = conn(false);
@@ -131,13 +275,21 @@ mod tests {
         let started = std::time::Instant::now();
         let got = tokio::time::timeout(
             Duration::from_secs(10),
-            send_bounded_with(&as_conn(&c), serde_json::json!({"chi": "chunk"}), &stats, short),
+            send_bounded_with(
+                &as_conn(&c),
+                serde_json::json!({"chi": "chunk"}),
+                &stats,
+                short,
+            ),
         )
         .await
         .expect("send_bounded must return, not hang");
 
         assert_eq!(got, Err(SendError::TimedOut));
-        assert!(started.elapsed() < Duration::from_secs(5), "returned promptly");
+        assert!(
+            started.elapsed() < Duration::from_secs(5),
+            "returned promptly"
+        );
     }
 
     #[tokio::test]
@@ -152,7 +304,10 @@ mod tests {
             Duration::from_millis(50),
         )
         .await;
-        assert!(c.closed.load(Ordering::SeqCst), "a stalled peer must be closed");
+        assert!(
+            c.closed.load(Ordering::SeqCst),
+            "a stalled peer must be closed"
+        );
         assert_eq!(stats.timed_out(), 1);
     }
 
@@ -160,13 +315,20 @@ mod tests {
     async fn a_healthy_peer_sends_and_is_not_counted() {
         let c = conn(false);
         let stats = SendStats::new();
-        let got =
-            send_bounded_with(&as_conn(&c), serde_json::json!({"chi": "chunk"}), &stats, SEND_TIMEOUT)
-                .await;
+        let got = send_bounded_with(
+            &as_conn(&c),
+            serde_json::json!({"chi": "chunk"}),
+            &stats,
+            SEND_TIMEOUT,
+        )
+        .await;
         assert_eq!(got, Ok(()));
         assert_eq!(*c.sends.lock(), 1);
         assert_eq!(stats.timed_out(), 0);
-        assert!(!c.closed.load(Ordering::SeqCst), "a working peer stays open");
+        assert!(
+            !c.closed.load(Ordering::SeqCst),
+            "a working peer stays open"
+        );
     }
 
     #[tokio::test]
@@ -188,17 +350,22 @@ mod tests {
             }
             fn close(&self) {}
         }
-        let c: Arc<dyn PeerConnection> =
-            Arc::new(Broken(
-                HumdAddr { id: Hid::random_humd(), hints: vec![] },
-                PeerCapabilities::default(),
-            ));
+        let c: Arc<dyn PeerConnection> = Arc::new(Broken(
+            HumdAddr {
+                id: Hid::random_humd(),
+                hints: vec![],
+            },
+            PeerCapabilities::default(),
+        ));
         let stats = SendStats::new();
-        let got = send_bounded_with(&c, serde_json::json!({"chi": "chunk"}), &stats, SEND_TIMEOUT).await;
-        assert!(
-            matches!(got, Err(SendError::Failed(_))),
-            "got {got:?}"
-        );
+        let got = send_bounded_with(
+            &c,
+            serde_json::json!({"chi": "chunk"}),
+            &stats,
+            SEND_TIMEOUT,
+        )
+        .await;
+        assert!(matches!(got, Err(SendError::Failed(_))), "got {got:?}");
         assert_eq!(stats.failed(), 1);
         assert_eq!(stats.timed_out(), 0, "a failure is not a stall");
     }
diff --git a/ensemble/src/tcp.rs b/ensemble/src/tcp.rs
index e9e6610..d771745 100644
--- a/ensemble/src/tcp.rs
+++ b/ensemble/src/tcp.rs
@@ -1,157 +1,103 @@
-//! Real-wire TCP transport for the ensemble.
-//!
-//! NDJSON over `tokio::net::TcpStream`. One framing rule: each tone is
-//! a single JSON value followed by `\n`. Mirrors the unix-socket shape
-//! humd's thrum module uses (`humd/src/thrumd/conn.rs`) — same parser, different socket.
-//!
-//! Auth lives one layer up: the ensemble's signed `chi:"hello"` exchange
-//! authenticates the peer once a connection is installed. This module
-//! only carries bytes.
-//!
-//! Shape:
-//! - [`TcpEndpoint`] — one live [`PeerConnection`] backed by a TCP
-//!   stream. Spawns a reader task that parses NDJSON into the receiver
-//!   mpsc; `send` serialises + writes under a `tokio::Mutex` so
-//!   concurrent senders can't interleave half-lines.
-//! - [`TcpListener`] — accepts inbound connections. The accepted
-//!   endpoint starts with a placeholder [`HumdAddr`]; the ensemble
-//!   drainer learns the real id once the peer's first hello arrives.
-//! - [`TcpTransport`] — [`Transport`] impl. `connect` dials the first
-//!   `tcp:host:port` hint in the [`HumdAddr`].
+//! NDJSON over TCP. Framing and the frame ceiling live in [`crate::framing`].
+//! Identity is settled one layer up by the signed `chi:"hello"` exchange.
 
 use std::net::SocketAddr;
 use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, Ordering};
 
-use anyhow::{anyhow, Result};
+use anyhow::{Result, anyhow};
 use async_trait::async_trait;
-use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
-use tokio::net::tcp::{OwnedReadHalf, OwnedWriteHalf};
 use tokio::net::TcpStream;
-use tokio::sync::{mpsc, Mutex};
+use tokio::net::tcp::OwnedWriteHalf;
+use tokio::sync::{Mutex, mpsc};
 
-use crate::{HumdAddr, Hid, PeerCapabilities, PeerConnection, Tone, Transport};
+use crate::framing;
+use crate::opening::Gate;
+use crate::{Hid, HumdAddr, PeerCapabilities, PeerConnection, Tone, Transport};
 
-/// Inbound-channel capacity — matches `InMemoryEndpoint`.
 const RECV_CAP: usize = 256;
 
-/// One TCP-backed peer link. Reader task drains NDJSON lines into the
-/// receiver mpsc; `send` writes a serialised tone + newline under a
-/// mutex (NDJSON framing requires atomic line writes).
 pub struct TcpEndpoint {
     peer: HumdAddr,
     caps: PeerCapabilities,
-    /// Serialises concurrent writers. `Option` so `close` can drop the
-    /// write half and subsequent sends fail cleanly.
-    writer: Mutex<Option<OwnedWriteHalf>>,
-    /// Inbound stream. Take-once.
+    writer: Arc<Mutex<Option<OwnedWriteHalf>>>,
     rx: parking_lot::Mutex<Option<mpsc::Receiver<Tone>>>,
+    closed: AtomicBool,
+    gate: Gate,
 }
 
 impl TcpEndpoint {
-    /// Dial a remote `host:port` and wrap the resulting stream.
-    pub async fn connect(
-        addr: &str,
-        peer: HumdAddr,
-        caps: PeerCapabilities,
-    ) -> Result<Arc<Self>> {
+    pub async fn connect(addr: &str, peer: HumdAddr, caps: PeerCapabilities) -> Result<Arc<Self>> {
         let stream = TcpStream::connect(addr).await?;
-        // Disable Nagle: tones are small and latency-sensitive, and the
-        // NDJSON framing means we never want a partial line stalling in
-        // the kernel waiting for more bytes.
         let _ = stream.set_nodelay(true);
         Ok(Self::from_stream(stream, peer, caps))
     }
 
-    /// Wrap an already-accepted stream. Used by `TcpListener::accept`.
-    pub fn from_stream(
-        stream: TcpStream,
-        peer: HumdAddr,
-        caps: PeerCapabilities,
-    ) -> Arc<Self> {
+    pub fn from_stream(stream: TcpStream, peer: HumdAddr, caps: PeerCapabilities) -> Arc<Self> {
         let _ = stream.set_nodelay(true);
         let (read_half, write_half) = stream.into_split();
         let (tx, rx) = mpsc::channel::<Tone>(RECV_CAP);
         let me = Arc::new(Self {
             peer,
             caps,
-            writer: Mutex::new(Some(write_half)),
+            writer: Arc::new(Mutex::new(Some(write_half))),
             rx: parking_lot::Mutex::new(Some(rx)),
+            closed: AtomicBool::new(false),
+            gate: Gate::new(),
         });
-        tokio::spawn(read_loop(read_half, tx));
+        tokio::spawn(framing::pump(read_half, tx, "tcp"));
         me
     }
+    async fn write(&self, tone: Tone) -> Result<()> {
+        let mut guard = self.writer.lock().await;
+        if self.closed.load(Ordering::SeqCst) {
+            return Err(anyhow!("tcp send: link closed"));
+        }
+        let w = guard
+            .as_mut()
+            .ok_or_else(|| anyhow!("tcp send: writer closed"))?;
+        framing::write_frame(w, &tone).await
+    }
 }
 
 #[async_trait]
 impl PeerConnection for TcpEndpoint {
-    fn peer(&self) -> &HumdAddr { &self.peer }
-    fn capabilities(&self) -> &PeerCapabilities { &self.caps }
+    fn peer(&self) -> &HumdAddr {
+        &self.peer
+    }
+    fn capabilities(&self) -> &PeerCapabilities {
+        &self.caps
+    }
 
     async fn send(&self, tone: Tone) -> Result<()> {
-        let mut line = serde_json::to_vec(&tone)
-            .map_err(|e| anyhow!("tcp send: serialize: {e}"))?;
-        line.push(b'\n');
-        let mut guard = self.writer.lock().await;
-        let w = guard
-            .as_mut()
-            .ok_or_else(|| anyhow!("tcp send: writer closed"))?;
-        w.write_all(&line)
-            .await
-            .map_err(|e| anyhow!("tcp send: write: {e}"))?;
-        Ok(())
+        self.gate.wait().await;
+        self.write(tone).await
+    }
+
+    async fn send_opening(&self, tone: Tone) -> Result<()> {
+        let result = self.write(tone).await;
+        self.gate.opened();
+        result
     }
 
     fn take_receiver(&self) -> Option<mpsc::Receiver<Tone>> {
         self.rx.lock().take()
     }
 
-    fn close(&self) {
-        // Drop the writer half and the receiver. Best-effort, idempotent.
-        // We can't await shutdown from a sync method, so spawn a task to
-        // gracefully close the write half if it's still present.
-        let writer_slot = {
-            let mut guard = self.writer.try_lock();
-            match guard {
-                Ok(ref mut g) => g.take(),
-                Err(_) => None, // contended — let the holder finish; their write will fail next
-            }
-        };
-        if let Some(mut w) = writer_slot {
-            tokio::spawn(async move {
-                let _ = w.shutdown().await;
-            });
-        }
-        let _ = self.rx.lock().take();
+    fn arm_opening(&self) {
+        self.gate.arm();
     }
-}
 
-/// Reader loop — parses NDJSON lines into `Tone`s and forwards. Exits on
-/// EOF, read error, or when the receiver is dropped.
-async fn read_loop(read_half: OwnedReadHalf, tx: mpsc::Sender<Tone>) {
-    let mut lines = BufReader::new(read_half).lines();
-    loop {
-        let line = match lines.next_line().await {
-            Ok(Some(l)) => l,
-            Ok(None) => break, // EOF
-            Err(e) => {
-                tracing::trace!(target: "ensemble.tcp", err = %e, "tcp.read.failed");
-                break;
-            }
-        };
-        if line.is_empty() {
-            continue;
-        }
-        let tone: Tone = match serde_json::from_str(&line) {
-            Ok(v) => v,
-            Err(e) => {
-                tracing::trace!(target: "ensemble.tcp", err = %e, "tcp.parse.failed");
-                continue;
+    fn close(&self) {
+        self.closed.store(true, Ordering::SeqCst);
+        self.gate.shut();
+        let writer = self.writer.clone();
+        tokio::spawn(async move {
+            if let Some(mut w) = writer.lock().await.take() {
+                let _ = tokio::io::AsyncWriteExt::shutdown(&mut w).await;
             }
-        };
-        if tx.send(tone).await.is_err() {
-            // Receiver dropped — endpoint closed.
-            break;
-        }
+        });
+        let _ = self.rx.lock().take();
     }
 }
 
@@ -195,11 +141,15 @@ impl TcpListener {
 pub struct TcpTransport;
 
 impl TcpTransport {
-    pub fn new() -> Self { Self }
+    pub fn new() -> Self {
+        Self
+    }
 }
 
 impl Default for TcpTransport {
-    fn default() -> Self { Self::new() }
+    fn default() -> Self {
+        Self::new()
+    }
 }
 
 #[async_trait]
@@ -210,7 +160,8 @@ impl Transport for TcpTransport {
             .iter()
             .find_map(|h| h.strip_prefix("tcp:"))
             .ok_or_else(|| anyhow!("tcp transport: no tcp: hint in HumdAddr"))?;
-        let endpoint = TcpEndpoint::connect(hint, addr.clone(), PeerCapabilities::default()).await?;
+        let endpoint =
+            TcpEndpoint::connect(hint, addr.clone(), PeerCapabilities::default()).await?;
         Ok(endpoint as Arc<dyn PeerConnection>)
     }
 }
diff --git a/ensemble/src/tls.rs b/ensemble/src/tls.rs
index 1b231b8..75a219c 100644
--- a/ensemble/src/tls.rs
+++ b/ensemble/src/tls.rs
@@ -1,61 +1,34 @@
-//! TLS-over-TCP transport for the ensemble.
+//! TLS-over-TCP transport for the ensemble. Sibling of [`crate::tcp`]:
+//! same framing, same contracts, wrapped in a [`tokio_rustls::TlsStream`].
 //!
-//! Sibling of [`crate::tcp`] — same NDJSON framing, same `Transport` +
-//! `PeerConnection` contract, but the underlying `TcpStream` is wrapped
-//! in a [`tokio_rustls::TlsStream`]. T2+ peers run over the public
-//! internet; plaintext is the loopback/LAN path only.
-//!
-//! ## Trust model — pinned fingerprints, no CA
-//!
-//! Self-signed cert on the server. The dialer carries the expected
-//! cert fingerprint (SHA-256 of the DER bytes) and accepts ONLY that
-//! one cert — see [`PinnedFingerprintVerifier`]. This matches the T2
-//! "known peers" model: trust is established out of band when you put
-//! a peer's fingerprint in `peers.json`; the TLS layer is just a
-//! confidentiality + integrity tunnel that proves the box on the other
-//! end really holds the private key for the fingerprint you wrote down.
-//! No CA, no chain validation, no SNI matching, no expiry handling —
-//! those belong to T4 with a real PKI.
-//!
-//! Shape:
-//! - [`TlsTcpEndpoint`] — one live [`PeerConnection`] over a TLS-wrapped
-//!   TCP stream. Reader task drains NDJSON into the receiver mpsc;
-//!   `send` writes a serialised tone + newline under a `tokio::Mutex`.
-//! - [`TlsTcpListener`] — accepts inbound TLS connections. The
-//!   accepted endpoint starts with a placeholder [`HumdAddr`]; the
-//!   ensemble drainer learns the real id once the peer's first hello
-//!   arrives — same as plaintext.
-//! - [`TlsTcpTransport`] — [`Transport`] impl. `connect` dials the
-//!   first `tls:host:port` hint in the [`HumdAddr`], using a
-//!   pre-configured client config that pins one fingerprint.
+//! Trust model: pinned fingerprints, no CA, no chain, no SNI, no expiry.
+//! The dialer accepts exactly one cert — the SHA-256 of its DER bytes
+//! written down out of band in `peers.json`. See [`PinnedFingerprintVerifier`].
 
 use std::net::SocketAddr;
 use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, Ordering};
 
-use anyhow::{anyhow, Result};
+use anyhow::{Result, anyhow};
 use async_trait::async_trait;
 use sha2::{Digest, Sha256};
-use tokio::io::{split, AsyncBufReadExt, AsyncWriteExt, BufReader, ReadHalf, WriteHalf};
+use tokio::io::{WriteHalf, split};
 use tokio::net::TcpStream;
-use tokio::sync::{mpsc, Mutex};
+use tokio::sync::{Mutex, mpsc};
 use tokio_rustls::{TlsAcceptor, TlsConnector, TlsStream};
 
 use rustls::client::danger::{HandshakeSignatureValid, ServerCertVerified, ServerCertVerifier};
 use rustls::pki_types::{CertificateDer, PrivateKeyDer, ServerName, UnixTime};
 use rustls::{ClientConfig, DigitallySignedStruct, ServerConfig, SignatureScheme};
 
-use crate::{HumdAddr, Hid, PeerCapabilities, PeerConnection, Tone, Transport};
+use crate::framing;
+use crate::opening::Gate;
+use crate::{Hid, HumdAddr, PeerCapabilities, PeerConnection, Tone, Transport};
 
-/// Inbound-channel capacity — matches `InMemoryEndpoint` / `TcpEndpoint`.
 const RECV_CAP: usize = 256;
 
-/// Hint prefix on a [`HumdAddr`] that signals "dial this `host:port`
-/// over TLS-pinned TCP." Sibling of `tcp:` so peers can advertise both
-/// plaintext and TLS endpoints and dialers pick the right one.
 pub const TLS_HINT: &str = "tls:";
 
-/// SHA-256 of a certificate's DER bytes. The bytes used for pinning in
-/// `peers.json` — write this on the server, paste it on the client.
 pub fn cert_fingerprint(cert: &CertificateDer<'_>) -> [u8; 32] {
     let mut h = Sha256::new();
     h.update(cert.as_ref());
@@ -67,22 +40,16 @@ pub fn cert_fingerprint(cert: &CertificateDer<'_>) -> [u8; 32] {
 
 // ── Endpoint ───────────────────────────────────────────────────────────────
 
-/// One TLS-over-TCP peer link. Same NDJSON framing as the plaintext
-/// path — the TLS layer is invisible to the framing code, we just hand
-/// it a `TlsStream` instead of a `TcpStream`.
 pub struct TlsTcpEndpoint {
     peer: HumdAddr,
     caps: PeerCapabilities,
-    /// Serialises concurrent writers. `Option` so `close` can drop the
-    /// write half and subsequent sends fail cleanly.
-    writer: Mutex<Option<WriteHalf<TlsStream<TcpStream>>>>,
-    /// Inbound stream. Take-once.
+    writer: Arc<Mutex<Option<WriteHalf<TlsStream<TcpStream>>>>>,
     rx: parking_lot::Mutex<Option<mpsc::Receiver<Tone>>>,
+    closed: AtomicBool,
+    gate: Gate,
 }
 
 impl TlsTcpEndpoint {
-    /// Wrap an already-handshaken TLS stream. Used by both the client
-    /// and listener paths once rustls has finished the handshake.
     pub fn from_stream(
         stream: TlsStream<TcpStream>,
         peer: HumdAddr,
@@ -93,12 +60,24 @@ impl TlsTcpEndpoint {
         let me = Arc::new(Self {
             peer,
             caps,
-            writer: Mutex::new(Some(write_half)),
+            writer: Arc::new(Mutex::new(Some(write_half))),
             rx: parking_lot::Mutex::new(Some(rx)),
+            closed: AtomicBool::new(false),
+            gate: Gate::new(),
         });
-        tokio::spawn(read_loop(read_half, tx));
+        tokio::spawn(framing::pump(read_half, tx, "tls"));
         me
     }
+    async fn write(&self, tone: Tone) -> Result<()> {
+        let mut guard = self.writer.lock().await;
+        if self.closed.load(Ordering::SeqCst) {
+            return Err(anyhow!("tls send: link closed"));
+        }
+        let w = guard
+            .as_mut()
+            .ok_or_else(|| anyhow!("tls send: writer closed"))?;
+        framing::write_frame(w, &tone).await
+    }
 
     /// Dial a remote `host:port` with a fingerprint-pinned TLS config.
     /// `expected_fingerprint` is the SHA-256 of the peer's cert DER —
@@ -135,68 +114,34 @@ impl PeerConnection for TlsTcpEndpoint {
     }
 
     async fn send(&self, tone: Tone) -> Result<()> {
-        let mut line = serde_json::to_vec(&tone)
-            .map_err(|e| anyhow!("tls send: serialize: {e}"))?;
-        line.push(b'\n');
-        let mut guard = self.writer.lock().await;
-        let w = guard
-            .as_mut()
-            .ok_or_else(|| anyhow!("tls send: writer closed"))?;
-        w.write_all(&line)
-            .await
-            .map_err(|e| anyhow!("tls send: write: {e}"))?;
-        Ok(())
+        self.gate.wait().await;
+        self.write(tone).await
+    }
+
+    async fn send_opening(&self, tone: Tone) -> Result<()> {
+        let result = self.write(tone).await;
+        self.gate.opened();
+        result
     }
 
     fn take_receiver(&self) -> Option<mpsc::Receiver<Tone>> {
         self.rx.lock().take()
     }
 
-    fn close(&self) {
-        // Drop the writer half and the receiver. Best-effort, idempotent.
-        let slot = match self.writer.try_lock() {
-            Ok(mut g) => g.take(),
-            Err(_) => None, // contended — holder's next write fails when peer drops
-        };
-        if let Some(mut w) = slot {
-            tokio::spawn(async move {
-                let _ = w.shutdown().await;
-            });
-        }
-        let _ = self.rx.lock().take();
+    fn arm_opening(&self) {
+        self.gate.arm();
     }
-}
 
-/// Reader loop — parses NDJSON lines off the TLS read half and forwards
-/// them. Identical control flow to the plaintext path; the TLS framing
-/// is fully encapsulated inside `TlsStream`.
-async fn read_loop(
-    read_half: ReadHalf<TlsStream<TcpStream>>,
-    tx: mpsc::Sender<Tone>,
-) {
-    let mut lines = BufReader::new(read_half).lines();
-    loop {
-        let line = match lines.next_line().await {
-            Ok(Some(l)) => l,
-            Ok(None) => break, // EOF
-            Err(e) => {
-                tracing::trace!(target: "ensemble.tls", err = %e, "tls.read.failed");
-                break;
-            }
-        };
-        if line.is_empty() {
-            continue;
-        }
-        let tone: Tone = match serde_json::from_str(&line) {
-            Ok(v) => v,
-            Err(e) => {
-                tracing::trace!(target: "ensemble.tls", err = %e, "tls.parse.failed");
-                continue;
+    fn close(&self) {
+        self.closed.store(true, Ordering::SeqCst);
+        self.gate.shut();
+        let writer = self.writer.clone();
+        tokio::spawn(async move {
+            if let Some(mut w) = writer.lock().await.take() {
+                let _ = tokio::io::AsyncWriteExt::shutdown(&mut w).await;
             }
-        };
-        if tx.send(tone).await.is_err() {
-            break;
-        }
+        });
+        let _ = self.rx.lock().take();
     }
 }
 
@@ -409,8 +354,8 @@ impl Transport for TlsTcpTransport {
             .iter()
             .find_map(|h| h.strip_prefix(TLS_FP_HINT))
             .ok_or_else(|| anyhow!("tls transport: no `tls-fp:` hint in HumdAddr"))?;
-        let fp_bytes = hex::decode(fp_hex)
-            .map_err(|e| anyhow!("tls transport: fingerprint hex: {e}"))?;
+        let fp_bytes =
+            hex::decode(fp_hex).map_err(|e| anyhow!("tls transport: fingerprint hex: {e}"))?;
         if fp_bytes.len() != 32 {
             return Err(anyhow!(
                 "tls transport: fingerprint must be 32 bytes (got {})",
@@ -419,13 +364,9 @@ impl Transport for TlsTcpTransport {
         }
         let mut fp = [0u8; 32];
         fp.copy_from_slice(&fp_bytes);
-        let endpoint = TlsTcpEndpoint::connect(
-            host_port,
-            addr.clone(),
-            PeerCapabilities::default(),
-            fp,
-        )
-        .await?;
+        let endpoint =
+            TlsTcpEndpoint::connect(host_port, addr.clone(), PeerCapabilities::default(), fp)
+                .await?;
         Ok(endpoint as Arc<dyn PeerConnection>)
     }
 }
diff --git a/ensemble/tests/gossip_integration.rs b/ensemble/tests/gossip_integration.rs
index 4af0ec3..22f11d8 100644
--- a/ensemble/tests/gossip_integration.rs
+++ b/ensemble/tests/gossip_integration.rs
@@ -1,9 +1,8 @@
-
 use std::time::Duration;
 
 use ensemble::{
-    gossip::{gossip_tone, mint_msg_id, GOSSIP_CHI},
     Ensemble, HumdKey, InMemoryEndpoint, PeerCapabilities,
+    gossip::{GOSSIP_CHI, gossip_tone, mint_msg_id},
 };
 use serde_json::json;
 use tokio::time::timeout;
@@ -22,14 +21,8 @@ async fn gossip_percolates_one_hop_and_dedupes_duplicates() {
         ..Default::default()
     };
 
-    let (a_to_b, b_to_a) = InMemoryEndpoint::pair(
-        a_id, caps.clone(),
-        b_id, caps.clone(),
-    );
-    let (b_to_c, c_to_b) = InMemoryEndpoint::pair(
-        b_id, caps.clone(),
-        c_id, caps.clone(),
-    );
+    let (a_to_b, b_to_a) = InMemoryEndpoint::pair(a_id, caps.clone(), b_id, caps.clone());
+    let (b_to_c, c_to_b) = InMemoryEndpoint::pair(b_id, caps.clone(), c_id, caps.clone());
 
     let ens_a = Ensemble::new(a_id);
     let ens_b = Ensemble::new(b_id);
@@ -55,12 +48,12 @@ async fn gossip_percolates_one_hop_and_dedupes_duplicates() {
 
     let x_key = HumdKey::generate();
     let x_id = x_key.hid();
-    let (x_to_b, b_to_x) = InMemoryEndpoint::pair(
-        x_id, caps.clone(),
-        b_id, caps.clone(),
-    );
+    let (x_to_b, b_to_x) = InMemoryEndpoint::pair(x_id, caps.clone(), b_id, caps.clone());
     ens_b.install(b_to_x, caps.clone(), &b_key);
+    let ens_x = Ensemble::new(x_id);
+    ens_x.install(x_to_b.clone(), caps.clone(), &x_key);
     tokio::time::sleep(Duration::from_millis(20)).await;
+    assert!(ens_b.handshake_done(&x_id), "B must complete X's handshake");
 
     while let Ok(Ok(_)) = timeout(Duration::from_millis(1), sub_c.recv()).await {}
 
diff --git a/ensemble/tests/iroh_integration.rs b/ensemble/tests/iroh_integration.rs
index 6fb4b04..f790a6f 100644
--- a/ensemble/tests/iroh_integration.rs
+++ b/ensemble/tests/iroh_integration.rs
@@ -1,9 +1,8 @@
-
 use std::time::Duration;
 
 use ed25519_dalek::SigningKey;
 use ensemble::{
-    Ensemble, HumdAddr, Hid, HumdKey, IrohTransport, PeerCapabilities, PeerConnection, Transport,
+    Ensemble, Hid, HumdAddr, HumdKey, IrohTransport, PeerCapabilities, PeerConnection, Transport,
 };
 use serde_json::json;
 
@@ -41,9 +40,9 @@ impl Skip {
     fn resolve(self) -> Option<()> {
         let forced = std::env::var(REQUIRE_ENV).is_ok_and(|v| v != "0" && !v.is_empty());
         match self {
-            Skip::Environmental(reason) if forced => panic!(
-                "{REQUIRE_ENV} is set, so this failure cannot be skipped: {reason}"
-            ),
+            Skip::Environmental(reason) if forced => {
+                panic!("{REQUIRE_ENV} is set, so this failure cannot be skipped: {reason}")
+            }
             Skip::Environmental(reason) => {
                 eprintln!("iroh_integration: SKIPPED — {reason}");
                 eprintln!(
@@ -106,8 +105,12 @@ async fn iroh_endpoint_routes_tones_both_ways() {
         );
     }
 
-    let server_key = HumdKey(SigningKey::from_bytes(&server.endpoint().secret_key().to_bytes()));
-    let client_key = HumdKey(SigningKey::from_bytes(&client.endpoint().secret_key().to_bytes()));
+    let server_key = HumdKey(SigningKey::from_bytes(
+        &server.endpoint().secret_key().to_bytes(),
+    ));
+    let client_key = HumdKey(SigningKey::from_bytes(
+        &client.endpoint().secret_key().to_bytes(),
+    ));
     assert_eq!(server_key.hid(), server_humd_id);
     assert_eq!(client_key.hid(), client_humd_id);
 
@@ -140,6 +143,7 @@ async fn iroh_endpoint_routes_tones_both_ways() {
             "chi": "ping",
             "rid": "iroh-pong-1",
             "to": client_humd_for_task.to_hex(),
+            "from": server_humd_id.to_hex(),
         });
         ensemble
             .route(pong)
@@ -150,10 +154,8 @@ async fn iroh_endpoint_routes_tones_both_ways() {
         Ok::<(), String>(())
     });
 
-    let mut server_humd_addr = HumdAddr::new(server_humd_id).with_hint(format!(
-        "iroh:{}",
-        hex::encode(server_node_id.as_bytes())
-    ));
+    let mut server_humd_addr = HumdAddr::new(server_humd_id)
+        .with_hint(format!("iroh:{}", hex::encode(server_node_id.as_bytes())));
     for hint in server_sockets {
         server_humd_addr = server_humd_addr.with_hint(hint);
     }
@@ -174,11 +176,9 @@ async fn iroh_endpoint_routes_tones_both_ways() {
         "chi": "perf-mark",
         "rid": "iroh-mark-1",
         "to": server_humd_id.to_hex(),
+        "from": client_humd_id.to_hex(),
     });
-    ensemble
-        .route(mark)
-        .await
-        .expect("route perf-mark");
+    ensemble.route(mark).await.expect("route perf-mark");
 
     let mut server_task = server_task;
     let got = tokio::select! {
@@ -211,7 +211,7 @@ async fn iroh_endpoint_routes_tones_both_ways() {
 }
 
 mod classify {
-    use super::{force_transport_tests, Skip, REQUIRE_ENV};
+    use super::{REQUIRE_ENV, Skip, force_transport_tests};
 
     #[test]
     fn sandbox_signatures_are_environmental() {
@@ -246,9 +246,8 @@ mod classify {
 
     #[test]
     fn a_real_failure_panics_even_without_the_env_var() {
-        let r = std::panic::catch_unwind(|| {
-            Skip::Real("bind: quic handshake failed".into()).resolve()
-        });
+        let r =
+            std::panic::catch_unwind(|| Skip::Real("bind: quic handshake failed".into()).resolve());
         assert!(r.is_err(), "a real failure must never resolve to a skip");
     }
 
diff --git a/ensemble/tests/tcp_integration.rs b/ensemble/tests/tcp_integration.rs
index 5ab1390..1cbe8ae 100644
--- a/ensemble/tests/tcp_integration.rs
+++ b/ensemble/tests/tcp_integration.rs
@@ -6,9 +6,7 @@
 
 use std::time::Duration;
 
-use ensemble::{
-    Ensemble, HumdAddr, HumdKey, PeerCapabilities, TcpEndpoint, TcpListener,
-};
+use ensemble::{Ensemble, HumdAddr, HumdKey, PeerCapabilities, TcpEndpoint, TcpListener};
 use serde_json::json;
 
 #[tokio::test]
@@ -82,7 +80,7 @@ async fn tcp_endpoint_routes_tone_across_wire() {
     // Route a ping from A → B. The first tone over the wire is A's
     // hello (absorbed by B's drainer), the second is this ping (fans
     // out to subscribers).
-    let ping = json!({"chi": "ping", "rid": "tcp-1", "to": b_id.to_hex()});
+    let ping = json!({"chi": "ping", "rid": "tcp-1", "to": b_id.to_hex(), "from": a_id.to_hex()});
     ensemble_a.route(ping).await.expect("route ping");
 
     // Wait for the ping on B's subscribe channel.
diff --git a/ensemble/tests/tls_integration.rs b/ensemble/tests/tls_integration.rs
index 614cc9e..fc8e0a0 100644
--- a/ensemble/tests/tls_integration.rs
+++ b/ensemble/tests/tls_integration.rs
@@ -13,8 +13,7 @@ use std::sync::Arc;
 use std::time::Duration;
 
 use ensemble::{
-    cert_fingerprint, Ensemble, HumdAddr, HumdKey, PeerCapabilities, TlsTcpEndpoint,
-    TlsTcpListener,
+    Ensemble, HumdAddr, HumdKey, PeerCapabilities, TlsTcpEndpoint, TlsTcpListener, cert_fingerprint,
 };
 use rcgen::generate_simple_self_signed;
 use rustls::pki_types::{CertificateDer, PrivateKeyDer, PrivatePkcs8KeyDer};
@@ -92,7 +91,7 @@ async fn tls_endpoint_routes_tone_across_pinned_wire() {
 
     // Route a ping from A → B. First tone is A's hello (drained by B);
     // second is this ping which fans out to subscribers.
-    let ping = json!({"chi": "ping", "rid": "tls-1", "to": b_id.to_hex()});
+    let ping = json!({"chi": "ping", "rid": "tls-1", "to": b_id.to_hex(), "from": a_id.to_hex()});
     ensemble_a.route(ping).await.expect("route ping");
 
     let got = tokio::time::timeout(Duration::from_secs(2), sub_b.recv())
diff --git a/hum-paths/src/config.rs b/hum-paths/src/config.rs
index bcd67f2..590d2aa 100644
--- a/hum-paths/src/config.rs
+++ b/hum-paths/src/config.rs
@@ -40,19 +40,14 @@ impl Default for HumdSection {
 
 // ── fs ────────────────────────────────────────────────────────────────────
 
-#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(rename_all = "lowercase")]
 pub enum FsMode {
+    #[default]
     Rw,
     Ro,
 }
 
-impl Default for FsMode {
-    fn default() -> Self {
-        FsMode::Rw
-    }
-}
-
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct FsRoot {
     /// Tilde-expanded at load time. Always stored as an absolute path.
diff --git a/hum-paths/src/lib.rs b/hum-paths/src/lib.rs
index db582f6..f508b16 100644
--- a/hum-paths/src/lib.rs
+++ b/hum-paths/src/lib.rs
@@ -72,6 +72,7 @@ pub const RUNTIME_INFO_BASENAME: &str = "runtime.json";
 pub const HUM_JSON_BASENAME: &str = "hum.json";
 pub const PEERS_JSON_BASENAME: &str = "peers.json";
 pub const ORCHFILE_BASENAME: &str = "Orchfile";
+pub const ROUTING_JSON_BASENAME: &str = "routing.json";
 /// Subdirectory of a hum source tree that holds hive crates.
 pub const HIVES_SUBDIR: &str = "hives";
 /// Subdirectory of a hum source tree that holds recipes (installable bundles).
@@ -109,6 +110,11 @@ pub fn thrum_sock_resolved() -> PathBuf {
 /// humd HTTP control socket.
 pub fn http_sock() -> PathBuf { runtime_dir().join(HTTP_SOCK_BASENAME) }
 
+/// Kademlia routing table, restored on boot.
+pub fn routing_json() -> PathBuf {
+    state_dir().join(ROUTING_JSON_BASENAME)
+}
+
 /// Penny lifetime counters.
 pub fn penny() -> PathBuf { runtime_dir().join(PENNY_BASENAME) }
 
@@ -237,10 +243,10 @@ pub fn daemon_logs(name: &str) -> DaemonLogs {
     #[cfg(target_os = "macos")]
     {
         let base = home().join("Library/Logs");
-        return DaemonLogs::Files {
+        DaemonLogs::Files {
             stdout: base.join(format!("sh.hum.{name}.out.log")),
             stderr: base.join(format!("sh.hum.{name}.err.log")),
-        };
+        }
     }
     #[cfg(not(target_os = "macos"))]
     DaemonLogs::Journald { unit: name.to_string() }
diff --git a/humd/src/lib.rs b/humd/src/lib.rs
index f70976a..5d60ae2 100644
--- a/humd/src/lib.rs
+++ b/humd/src/lib.rs
@@ -19,6 +19,7 @@ mod peer_transport;
 pub mod peers;
 mod penny;
 pub mod redial;
+mod routing_store;
 pub mod supervisor;
 pub mod thrumd;
 pub use identity::{key_path, load_or_mint_key, read_key};
@@ -39,9 +40,11 @@ pub struct DaemonConfig {
     pub http_path: PathBuf,
     pub mcp_addr: std::net::SocketAddr,
     pub penny_path: PathBuf,
+    pub routing_path: PathBuf,
     pub hum_cfg: hum_paths::config::HumConfig,
     pub cli_path: String,
     pub penny_persist_interval: Duration,
+    pub routing_persist_interval: Duration,
     pub thrum_override: Option<Thrum>,
     pub ensemble: Option<Arc<Ensemble>>,
     pub bind_mcp: bool,
@@ -73,9 +76,11 @@ impl DaemonConfig {
             http_path,
             mcp_addr: ([127, 0, 0, 1], mcp_port).into(),
             penny_path: hum_paths::penny(),
+            routing_path: hum_paths::routing_json(),
             hum_cfg: hum_paths::config::load(),
             cli_path: std::env::var("CLAUDE_CLI_PATH").unwrap_or_else(|_| "claude".into()),
             penny_persist_interval: Duration::from_secs(10),
+            routing_persist_interval: Duration::from_secs(30),
             thrum_override: None,
             ensemble: None,
             bind_mcp: true,
@@ -146,6 +151,11 @@ where
         }),
     };
 
+    if let Some(ens) = &ensemble_opt {
+        routing_store::restore_on_boot(ens, &cfg.routing_path);
+        routing_store::spawn_persister(ens.clone(), cfg.routing_path.clone(), cfg.routing_persist_interval);
+    }
+
     let mut peer_reach: Vec<String> = Vec::new();
     let mut supervisor: Option<supervisor::Supervisor> = None;
     if let (Some(ens), Some(key)) = (&ensemble_opt, &cfg.humd_key) {
diff --git a/humd/src/routing_store.rs b/humd/src/routing_store.rs
new file mode 100644
index 0000000..c32ec41
--- /dev/null
+++ b/humd/src/routing_store.rs
@@ -0,0 +1,139 @@
+use std::io;
+use std::path::{Path, PathBuf};
+use std::sync::Arc;
+use std::time::Duration;
+
+use ensemble::{Ensemble, HumdAddr};
+use tracing::{info, trace, warn};
+
+#[derive(serde::Serialize, serde::Deserialize)]
+struct OnDisk {
+    version: u32,
+    addrs: Vec<HumdAddr>,
+}
+
+pub fn load(path: &Path) -> Vec<HumdAddr> {
+    let bytes = match std::fs::read(path) {
+        Ok(b) => b,
+        Err(e) if e.kind() == io::ErrorKind::NotFound => {
+            trace!(target: "routing", "routing.load.absent");
+            return Vec::new();
+        }
+        Err(e) => {
+            warn!(target: "routing", "routing.load.read_failed err={e}");
+            return Vec::new();
+        }
+    };
+    match serde_json::from_slice::<OnDisk>(&bytes) {
+        Ok(disk) => disk.addrs,
+        Err(e) => {
+            warn!(target: "routing", "routing.load.parse_failed err={e}");
+            Vec::new()
+        }
+    }
+}
+
+pub fn save(ensemble: &Ensemble, path: &Path) -> io::Result<()> {
+    if let Some(dir) = path.parent() {
+        if !dir.as_os_str().is_empty() {
+            std::fs::create_dir_all(dir)?;
+        }
+    }
+    let disk = OnDisk {
+        version: 1,
+        addrs: ensemble.routing_snapshot(),
+    };
+    let bytes =
+        serde_json::to_vec(&disk).map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?;
+    let tmp = path.with_extension("json.tmp");
+    std::fs::write(&tmp, &bytes)?;
+    std::fs::rename(&tmp, path)?;
+    Ok(())
+}
+
+pub fn restore_on_boot(ensemble: &Ensemble, path: &Path) {
+    let addrs = load(path);
+    if addrs.is_empty() {
+        return;
+    }
+    let restored = ensemble.routing_restore(addrs);
+    info!(target: "routing", peers = restored, "routing.restored");
+}
+
+pub fn spawn_persister(ensemble: Arc<Ensemble>, path: PathBuf, interval: Duration) {
+    tokio::spawn(async move {
+        let mut tick = tokio::time::interval(interval);
+        tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
+        tick.tick().await;
+        loop {
+            tick.tick().await;
+            if let Err(e) = save(&ensemble, &path) {
+                warn!(target: "routing", "routing.persist.failed err={e}");
+            }
+        }
+    });
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn addr() -> HumdAddr {
+        HumdAddr::new(ensemble::Hid::random_humd()).with_hint("192.0.2.7:7777")
+    }
+
+    fn ens() -> Arc<Ensemble> {
+        Arc::new(Ensemble::new(ensemble::Hid::random_humd()))
+    }
+
+    #[test]
+    fn a_missing_file_restores_nothing() {
+        let dir = std::env::temp_dir().join(format!(
+            "routing-absent-{}",
+            ensemble::Hid::random_humd().short()
+        ));
+        assert!(load(&dir.join("routing.json")).is_empty());
+    }
+
+    #[test]
+    fn a_saved_table_survives_a_restart() {
+        let path = std::env::temp_dir()
+            .join(format!("routing-{}", ensemble::Hid::random_humd().short()))
+            .join("routing.json");
+
+        let before = ens();
+        let peer = addr();
+        assert_eq!(before.routing_restore(vec![peer.clone()]), 1);
+        save(&before, &path).expect("save must succeed");
+
+        let after = ens();
+        restore_on_boot(&after, &path);
+        assert_eq!(
+            after.kad_routing_table_len(),
+            1,
+            "the table must survive a restart"
+        );
+        assert!(
+            after
+                .kad_closest(&peer.id, 4)
+                .iter()
+                .any(|a| a.id == peer.id)
+        );
+
+        let _ = std::fs::remove_file(&path);
+    }
+
+    #[test]
+    fn a_corrupt_file_restores_nothing_rather_than_panicking() {
+        let path = std::env::temp_dir()
+            .join(format!(
+                "routing-bad-{}",
+                ensemble::Hid::random_humd().short()
+            ))
+            .join("routing.json");
+        std::fs::create_dir_all(path.parent().unwrap()).unwrap();
+        std::fs::write(&path, b"{not json").unwrap();
+        assert!(load(&path).is_empty());
+        let _ = std::fs::remove_file(&path);
+    }
+}
diff --git a/sim/src/lib.rs b/sim/src/lib.rs
index 1703f2b..98e960b 100644
--- a/sim/src/lib.rs
+++ b/sim/src/lib.rs
@@ -95,7 +95,7 @@ impl Sim {
 
     pub async fn spawn_humd(&self, id: Hid) -> Arc<SimHumd> {
         let thrum = Thrum::new();
-        let ensemble = Arc::new(Ensemble::new(id));
+        let ensemble = Arc::new(Ensemble::with_strict_auth(id, false));
         let (shutdown_tx, shutdown_rx) = oneshot::channel::<()>();
 
         let tmp = std::env::temp_dir().join(format!("sim-humd-{}", id.short()));
@@ -115,6 +115,8 @@ impl Sim {
             http_path: tmp.join(hum_paths::HTTP_SOCK_BASENAME),
             mcp_addr: ([127, 0, 0, 1], 0).into(),
             penny_path,
+            routing_path: tmp.join(hum_paths::ROUTING_JSON_BASENAME),
+            routing_persist_interval: Duration::from_secs(3600),
             hum_cfg: hum_paths::config::HumConfig::default(),
             cli_path: "noop".into(),
             penny_persist_interval: Duration::from_secs(3600),
@@ -172,6 +174,8 @@ impl Sim {
         self.wire(a, b)
     }
 
+    /// Unauthenticated mesh link. Use [`Sim::wire_signed`] when the test is
+    /// about identity.
     pub fn wire(&self, a: Hid, b: Hid) -> Result<()> {
         let humds = self.humds.read();
         let ha = humds
@@ -232,6 +236,8 @@ impl Sim {
             http_path: tmp.join(hum_paths::HTTP_SOCK_BASENAME),
             mcp_addr: ([127, 0, 0, 1], 0).into(),
             penny_path,
+            routing_path: tmp.join(hum_paths::ROUTING_JSON_BASENAME),
+            routing_persist_interval: Duration::from_secs(3600),
             hum_cfg: hum_paths::config::HumConfig::default(),
             cli_path: "noop".into(),
             penny_persist_interval: Duration::from_secs(3600),
diff --git a/sim/tests/liveness.rs b/sim/tests/liveness.rs
index 4c01758..16e905d 100644
--- a/sim/tests/liveness.rs
+++ b/sim/tests/liveness.rs
@@ -1,4 +1,3 @@
-
 use std::time::Duration;
 
 use ensemble::Liveness;
@@ -27,6 +26,21 @@ fn liveness(sim: &Sim, observer: ensemble::Hid, peer: ensemble::Hid) -> Option<L
         .map(|(_, l)| l)
 }
 
+/// Transport death propagates asynchronously: killing a link drops the
+/// sender, humd's drain loop ends, and only then does the lease flip to
+/// TransportClosed. Poll for it rather than assuming a fixed sleep is long
+/// enough — under CI load it is not.
+async fn await_dead(sim: &Sim, observer: ensemble::Hid, peer: ensemble::Hid) {
+    let deadline = std::time::Instant::now() + Duration::from_secs(5);
+    while liveness(sim, observer, peer) != Some(Liveness::Dead) {
+        assert!(
+            std::time::Instant::now() < deadline,
+            "a peer whose link was killed must go Dead"
+        );
+        tokio::time::sleep(Duration::from_millis(5)).await;
+    }
+}
+
 #[tokio::test(flavor = "multi_thread")]
 async fn a_talking_peer_is_live_and_survives_a_sweep() {
     let (sim, a, b) = pair().await;
@@ -56,16 +70,15 @@ async fn a_killed_peer_goes_dead() {
     assert_eq!(liveness(&sim, a, b), Some(Liveness::Live));
 
     sim.kill_peer(b, a).expect("kill b→a");
-    tokio::time::sleep(Duration::from_millis(40)).await;
-    assert_eq!(liveness(&sim, a, b), Some(Liveness::Dead));
+    await_dead(&sim, a, b).await;
 }
 
 #[tokio::test(flavor = "multi_thread")]
 async fn a_dead_peer_survives_until_swept() {
     let (sim, a, b) = pair().await;
     sim.kill_link(a, b).expect("kill");
-    tokio::time::sleep(Duration::from_millis(40)).await;
     assert_eq!(sim.peer_count(a), 1, "transport death does not self-evict");
+    await_dead(&sim, a, b).await;
 
     let evicted = sim.evict_expired(a, TTL).expect("sweep");
     assert_eq!(evicted, vec![b]);
@@ -109,7 +122,7 @@ async fn a_partitioned_peer_is_not_dead() {
 async fn a_rewired_peer_is_live_again() {
     let (sim, a, b) = pair().await;
     sim.kill_link(a, b).expect("kill");
-    tokio::time::sleep(Duration::from_millis(40)).await;
+    await_dead(&sim, a, b).await;
     assert_eq!(sim.evict_expired(a, TTL).expect("sweep"), vec![b]);
     assert_eq!(sim.peer_count(a), 0);
 
@@ -138,8 +151,8 @@ async fn eviction_only_touches_the_dead_peer() {
     sim.probe(a, b).await.expect("probe");
     sim.probe(a, c).await.expect("probe");
     sim.kill_link(b, a).expect("kill b");
-    tokio::time::sleep(Duration::from_millis(40)).await;
 
+    await_dead(&sim, a, b).await;
     assert_eq!(sim.evict_expired(a, TTL).expect("sweep"), vec![b]);
     assert_eq!(liveness(&sim, a, c), Some(Liveness::Live), "c is untouched");
     assert_eq!(sim.peer_count(a), 1, "only b was reaped");