replace shadow types by wire codec

3 files changed, 433 insertions, 428 deletions

diff --git a/src/transport/mod.rs b/src/transport/mod.rs
index 35e3499..14c5ee4 100644
--- a/src/transport/mod.rs
+++ b/src/transport/mod.rs
@@ -2,6 +2,7 @@ use anyhow::Result;
 use async_trait::async_trait;
 use automerge::ChangeHash;
 use automerge::sync::Message;
+use serde::{Deserialize, Serialize};
 use sha2::{Digest, Sha256};
 use std::boxed::Box;
 use std::future::Future;
@@ -14,6 +15,7 @@ pub mod mycelium;
 #[cfg(test)]
 pub(crate) mod sim;
 pub mod traffic;
+pub(crate) mod wire;
 
 /// Random identifier of one run of a node (from start to exit).
 pub type RunId = [u8; 16];
@@ -83,10 +85,12 @@ fn digest_of(sorted_heads: &[ChangeHash]) -> [u8; 32] {
 }
 
 /// PING and PONG. Addressed to one peer: `syncs_with_you` is about the recipient.
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct Keepalive {
     /// Only comparable between nodes on the same epoch.
+    #[serde(with = "crate::transport::wire::codec::heads")]
     pub heads: HeadsSummary,
+    #[serde(with = "crate::transport::wire::codec::epoch")]
     pub epoch: Epoch,
     /// See `SyncEnvelope::epoch_seq`.
     pub epoch_seq: u32,
@@ -109,12 +113,12 @@ pub enum ParsedMessage {
 
 /// Content-addressed bytes that any node holding them can serve, in chunks
 /// (`swarm::blob`).
-#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
 pub enum BlobId {
     /// An Automerge change, by its hash: e.g. an epoch's snapshot change.
-    Change(ChangeHash),
+    Change(#[serde(with = "crate::transport::wire::codec::change_hash")] ChangeHash),
     /// Bytes by their BLAKE3 hash: e.g. the content of a remake.
-    Content([u8; 32]),
+    Content(#[serde(with = "crate::transport::wire::codec::bytes32")] [u8; 32]),
 }
 
 impl std::fmt::Display for BlobId {
@@ -131,9 +135,10 @@ impl std::fmt::Display for BlobId {
 
 /// A blob request or one of its answers. Each is a message of its own, so no message
 /// is larger than one chunk.
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct BlobMessage {
     /// The sender's epoch (every message carries it).
+    #[serde(with = "crate::transport::wire::codec::epoch")]
     pub epoch: Epoch,
     /// See `SyncEnvelope::epoch_seq`.
     pub epoch_seq: u32,
@@ -141,20 +146,20 @@ pub struct BlobMessage {
     pub body: BlobBody,
 }
 
-#[derive(Clone)]
+#[derive(Clone, Serialize, Deserialize)]
 pub enum BlobBody {
     /// Asks for these chunks; for the manifest if empty.
-    Request {
-        chunks: Vec<u32>,
-    },
+    Request { chunks: Vec<u32> },
     /// Size and per-chunk BLAKE3 hashes; the last chunk may be shorter.
     Manifest {
         size: u64,
         chunk_size: u32,
+        #[serde(with = "crate::transport::wire::codec::hashes32")]
         hashes: Vec<[u8; 32]>,
     },
     Chunk {
         index: u32,
+        #[serde(with = "crate::transport::wire::codec::shared_bytes")]
         data: bytes::Bytes,
     },
     /// The sender doesn't have the blob.
@@ -234,52 +239,61 @@ impl ParsedMessage {
     }
 }
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct HelloRequest {
     /// The sender's swarm, if it knows it (a new node doesn't yet). A receiver in
     /// another swarm doesn't answer.
     pub swarm_id: Option<String>,
+    #[serde(with = "crate::transport::wire::codec::epoch")]
     pub epoch: Epoch,
     /// See `SyncEnvelope::epoch_seq`.
     pub epoch_seq: u32,
     /// Proof of the swarm key (`swarm::auth`); empty without one.
+    #[serde(with = "serde_bytes")]
     pub auth: Vec<u8>,
     /// The sender is ready (same meaning as `SyncEnvelope::ready`).
     pub ready: bool,
     /// Random per run of the sender; the reply echoes it.
+    #[serde(with = "crate::transport::wire::codec::run_id")]
     pub run_id: RunId,
 }
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct HelloReply {
     pub swarm_id: String,
     pub peers: Vec<String>,
+    #[serde(with = "crate::transport::wire::codec::epoch")]
     pub epoch: Epoch,
     /// See `SyncEnvelope::epoch_seq`.
     pub epoch_seq: u32,
     /// Proof of the swarm key (`swarm::auth`); empty without one.
+    #[serde(with = "serde_bytes")]
     pub auth: Vec<u8>,
     /// The sender is ready.
     pub ready: bool,
     /// The `run_id` of the HELLO answered. A reply to a HELLO from an earlier run of the
     /// recipient (delivered late, or queued while it was down) is ignored.
+    #[serde(with = "crate::transport::wire::codec::run_id")]
     pub run_id: RunId,
     /// How long before this HELLO the sender had last heard from the recipient, if ever
     /// (in milliseconds): tells a restarted node how long it was away.
     pub away_ms: Option<u64>,
     /// The sender's own run (as its HELLOs carry it): a new one means it restarted.
+    #[serde(with = "crate::transport::wire::codec::run_id")]
     pub sender_run_id: RunId,
 }
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct SyncEnvelope {
     pub swarm_id: String,
     /// Epoch of the document the message was generated from.
+    #[serde(with = "crate::transport::wire::codec::epoch")]
     pub epoch: Epoch,
     /// How many times the sender changed epoch in its current run, when it sent this
     /// (`Replica::epoch_seq`). Messages may arrive late and out of order: the receiver
     /// ignores an epoch older than the newest it has seen in the same session.
     pub epoch_seq: u32,
+    #[serde(with = "crate::transport::wire::codec::sync_message")]
     pub message: Message,
     /// The sender was ready when it generated the message: its document is the swarm's,
     /// not one it is still filling (joining, rejoining). Only such a peer can tell a
diff --git a/src/transport/mycelium.rs b/src/transport/mycelium.rs
index efab475..a7bd7d9 100644
--- a/src/transport/mycelium.rs
+++ b/src/transport/mycelium.rs
@@ -1,11 +1,8 @@
 use anyhow::{Context, Result, anyhow, bail};
 use async_trait::async_trait;
-use automerge::ChangeHash;
 use base64::engine::{Engine, general_purpose::STANDARD as BASE64};
 use reqwest::StatusCode;
-use serde::{Deserialize, Deserializer, Serialize, Serializer};
-use serde_bytes::{ByteBuf, Bytes};
-use std::borrow::Cow;
+use serde::{Deserialize, Serialize};
 use std::boxed::Box;
 use std::fs::File;
 use std::io::Write;
@@ -18,21 +15,14 @@ use thiserror::Error;
 use tokio::process::{Child, Command};
 use url::Url;
 
-use crate::transport::EncodedMessage;
-use crate::transport::{BlobBody, BlobId, BlobMessage};
-use crate::transport::{Epoch, HelloReply, HelloRequest, SyncEnvelope};
-use crate::transport::{HeadsSummary, Keepalive, MAX_LISTED_HEADS};
-use crate::transport::{MessageHandler, ParsedMessage, Transport};
+use crate::transport::wire;
+use crate::transport::{EncodedMessage, MessageHandler, ParsedMessage, Transport};
 
 const MYCELIUM_POLL_TIMEOUT_SECS: u64 = 300;
 
-const TOPIC_PREFIX: &str = "continuum.v1.";
-const TOPIC_HELLO: &str = "continuum.v1.hello";
-const TOPIC_HELLO_REPLY: &str = "continuum.v1.hello-reply";
-const TOPIC_SYNC: &str = "continuum.v1.sync";
-const TOPIC_PING: &str = "continuum.v1.ping";
-const TOPIC_PONG: &str = "continuum.v1.pong";
-const TOPIC_BLOB: &str = "continuum.v1.blob";
+/// The one Mycelium topic all of continuum's messages travel on; what a message is, its
+/// payload says (`wire`).
+const TOPIC: &str = "continuum.v1";
 
 const EMPTY_QUEUE_DELAY: Duration = Duration::from_millis(25);
 
@@ -490,7 +480,7 @@ impl MyceliumTransport {
                 .client
                 .get(self.messages_endpoint.clone())
                 .query(&[("timeout", 0)])
-                .query(&[("topic", encode_topic(TOPIC_PREFIX))])
+                .query(&[("topic", encode_topic(TOPIC))])
                 .send()
                 .await;
             match response {
@@ -626,7 +616,7 @@ impl Transport for MyceliumTransport {
                 .poll_client
                 .get(self.messages_endpoint.clone())
                 .query(&[("timeout", MYCELIUM_POLL_TIMEOUT_SECS)])
-                .query(&[("topic", encode_topic(TOPIC_PREFIX))])
+                .query(&[("topic", encode_topic(TOPIC))])
                 .send()
                 .await
             {
@@ -746,26 +736,11 @@ fn base64_decoded_len(encoded: &str) -> usize {
 }
 
 fn decode_payload(topic_encoded: &str, payload: &str) -> Result<ParsedMessage, TransportError> {
-    let topic: &str = &decode_topic(topic_encoded)?;
-    let bytes = BASE64.decode(payload)?;
-    match topic {
-        TOPIC_HELLO => {
-            let msg = bytes_to_hello_request(&bytes)?;
-            Ok(ParsedMessage::Hello(msg))
-        }
-        TOPIC_HELLO_REPLY => {
-            let msg = bytes_to_hello_reply(&bytes)?;
-            Ok(ParsedMessage::HelloReply(msg))
-        }
-        TOPIC_SYNC => {
-            let msg = bytes_to_sync_envelope(&bytes)?;
-            Ok(ParsedMessage::Sync(msg))
-        }
-        TOPIC_PING => Ok(ParsedMessage::Ping(bytes_to_keepalive(&bytes)?)),
-        TOPIC_PONG => Ok(ParsedMessage::Pong(bytes_to_keepalive(&bytes)?)),
-        TOPIC_BLOB => Ok(ParsedMessage::Blob(bytes_to_blob(&bytes)?)),
-        t => Err(TransportError::UnknownTopic(t.to_string())),
+    let topic = decode_topic(topic_encoded)?;
+    if topic != TOPIC {
+        return Err(TransportError::UnknownTopic(topic));
     }
+    Ok(wire::decode(&BASE64.decode(payload)?)?)
 }
 
 fn parse_message(message: &InboundMessage) -> Result<(String, ParsedMessage), TransportError> {
@@ -785,376 +760,17 @@ fn format_message(
     recipient: &str,
     message: &ParsedMessage,
 ) -> Result<(OutboundMessage, usize), TransportError> {
-    let (topic, payload_bytes) = match message {
-        ParsedMessage::Hello(req) => (TOPIC_HELLO, hello_request_to_bytes(req)?),
-        ParsedMessage::HelloReply(res) => (TOPIC_HELLO_REPLY, hello_reply_to_bytes(res)?),
-        ParsedMessage::Sync(env) => (TOPIC_SYNC, sync_envelope_to_bytes(env)?),
-        ParsedMessage::Ping(k) => (TOPIC_PING, keepalive_to_bytes(k)?),
-        ParsedMessage::Pong(k) => (TOPIC_PONG, keepalive_to_bytes(k)?),
-        ParsedMessage::Blob(blob) => (TOPIC_BLOB, blob_to_bytes(blob)?),
-    };
+    let payload_bytes = wire::encode(message)?;
     let outbound = OutboundMessage {
         dst: Destination {
             pk: recipient.to_owned(),
         },
-        topic: encode_topic(topic),
+        topic: encode_topic(TOPIC),
         payload: encode_payload(&payload_bytes),
     };
     Ok((outbound, payload_bytes.len()))
 }
 
-// =====================================================================
-// 1. WIRE SHADOWS
-// =====================================================================
-//
-// rmp_serde encodes structs as arrays. Fields added later are appended at the end and
-// marked #[serde(default)], so a message from a node that predates them still decodes
-// (an absent epoch is the genesis epoch).
-
-/// Epoch on the wire. The default (generation 0, empty id) is the genesis epoch.
-#[derive(Serialize, Deserialize, Default)]
-struct EpochShadow {
-    generation: u64,
-    id: ByteBuf,
-    /// The sender's epoch changes in its current run (`SyncEnvelope::epoch_seq`).
-    #[serde(default)]
-    seq: u32,
-}
-
-fn epoch_to_shadow(epoch: &Epoch, seq: u32) -> EpochShadow {
-    EpochShadow {
-        generation: epoch.generation,
-        id: ByteBuf::from(epoch.id.to_vec()),
-        seq,
-    }
-}
-
-fn epoch_from_shadow(shadow: &EpochShadow) -> Result<Epoch, TransportError> {
-    if shadow.generation == 0 && shadow.id.is_empty() {
-        return Ok(Epoch::GENESIS);
-    }
-    let id = shadow
-        .id
-        .as_slice()
-        .try_into()
-        .map_err(|_| TransportError::InvalidPayload("epoch id must be 32 bytes".into()))?;
-    Ok(Epoch {
-        generation: shadow.generation,
-        id,
-    })
-}
-
-/// Either `heads` (listed) or `digest` (32 bytes, `heads` empty) is set.
-#[derive(Serialize, Deserialize)]
-struct KeepaliveShadow {
-    heads: Vec<ByteBuf>,
-    digest: ByteBuf,
-    epoch: EpochShadow,
-    ready: bool,
-    syncs_with_you: bool,
-}
-
-fn keepalive_to_bytes(msg: &Keepalive) -> Result<Vec<u8>, rmp_serde::encode::Error> {
-    let (heads, digest) = match &msg.heads {
-        HeadsSummary::Heads(heads) => (
-            heads
-                .iter()
-                .map(|head| ByteBuf::from(head.0.to_vec()))
-                .collect(),
-            ByteBuf::new(),
-        ),
-        HeadsSummary::Digest(digest) => (Vec::new(), ByteBuf::from(digest.to_vec())),
-    };
-    rmp_serde::to_vec(&KeepaliveShadow {
-        heads,
-        digest,
-        epoch: epoch_to_shadow(&msg.epoch, msg.epoch_seq),
-        ready: msg.ready,
-        syncs_with_you: msg.syncs_with_you,
-    })
-}
-
-fn bytes_to_keepalive(bytes: &[u8]) -> Result<Keepalive, TransportError> {
-    let shadow: KeepaliveShadow = rmp_serde::from_slice(bytes)?;
-    let heads = if shadow.digest.is_empty() {
-        if shadow.heads.len() > MAX_LISTED_HEADS {
-            return Err(TransportError::InvalidPayload(format!(
-                "more than {MAX_LISTED_HEADS} heads listed"
-            )));
-        }
-        let heads = shadow
-            .heads
-            .iter()
-            .map(|head| ChangeHash::try_from(head.as_slice()))
-            .collect::<Result<Vec<_>, _>>()
-            .map_err(|_| TransportError::InvalidPayload("a head must be 32 bytes".into()))?;
-        HeadsSummary::Heads(heads)
-    } else {
-        let digest =
-            shadow.digest.as_slice().try_into().map_err(|_| {
-                TransportError::InvalidPayload("heads digest must be 32 bytes".into())
-            })?;
-        HeadsSummary::Digest(digest)
-    };
-    Ok(Keepalive {
-        heads,
-        epoch: epoch_from_shadow(&shadow.epoch)?,
-        epoch_seq: shadow.epoch.seq,
-        ready: shadow.ready,
-        syncs_with_you: shadow.syncs_with_you,
-    })
-}
-
-#[derive(Serialize, Deserialize)]
-pub struct HelloRequestShadow {
-    swarm_id: Option<String>,
-    epoch: EpochShadow,
-    auth: ByteBuf,
-    ready: bool,
-    run_id: ByteBuf,
-}
-
-#[derive(Serialize, Deserialize)]
-pub struct HelloReplyShadow<'a> {
-    // Cow allows zero-copy references when saving, but creates an owned String when loading
-    pub swarm_id: Cow<'a, str>,
-    #[serde(default)]
-    pub peers: Vec<String>,
-    #[serde(default)]
-    epoch: EpochShadow,
-    #[serde(default)]
-    auth: ByteBuf,
-    ready: bool,
-    run_id: ByteBuf,
-    away_ms: Option<u64>,
-    sender_run_id: ByteBuf,
-}
-
-#[derive(Serialize, Deserialize)]
-pub struct SyncEnvelopeShadow<'a> {
-    pub swarm_id: Cow<'a, str>,
-    #[serde(with = "automerge_msg_codec")]
-    pub message: automerge::sync::Message,
-    #[serde(default)]
-    epoch: EpochShadow,
-    /// Absent from nodes that predate it: never counts as ready.
-    #[serde(default)]
-    ready: bool,
-}
-
-// =====================================================================
-// 2. CUSTOM CODEC
-// =====================================================================
-mod automerge_msg_codec {
-    use super::*;
-    use automerge::sync::Message;
-
-    pub fn serialize<S>(msg: &Message, serializer: S) -> Result<S::Ok, S::Error>
-    where
-        S: Serializer,
-    {
-        let bytes = msg.clone().encode();
-        Bytes::new(&bytes).serialize(serializer)
-    }
-
-    pub fn deserialize<'de, D>(deserializer: D) -> Result<Message, D::Error>
-    where
-        D: Deserializer<'de>,
-    {
-        let byte_buf = ByteBuf::deserialize(deserializer)?;
-        Message::decode(&byte_buf)
-            .map_err(|e| serde::de::Error::custom(format!("Automerge decode failed: {:?}", e)))
-    }
-}
-
-// =====================================================================
-// 3. CLEAN CONVERSION INTERFACE
-// =====================================================================
-
-// --- HelloRequest ---
-pub fn hello_request_to_bytes(msg: &HelloRequest) -> Result<Vec<u8>, rmp_serde::encode::Error> {
-    rmp_serde::to_vec(&HelloRequestShadow {
-        swarm_id: msg.swarm_id.clone(),
-        epoch: epoch_to_shadow(&msg.epoch, msg.epoch_seq),
-        auth: ByteBuf::from(msg.auth.clone()),
-        ready: msg.ready,
-        run_id: ByteBuf::from(msg.run_id.to_vec()),
-    })
-}
-pub fn bytes_to_hello_request(bytes: &[u8]) -> Result<HelloRequest, TransportError> {
-    let shadow: HelloRequestShadow = rmp_serde::from_slice(bytes)?;
-    Ok(HelloRequest {
-        swarm_id: shadow.swarm_id,
-        epoch: epoch_from_shadow(&shadow.epoch)?,
-        epoch_seq: shadow.epoch.seq,
-        auth: shadow.auth.into_vec(),
-        ready: shadow.ready,
-        run_id: run_id_from_shadow(&shadow.run_id)?,
-    })
-}
-
-fn run_id_from_shadow(bytes: &ByteBuf) -> Result<crate::transport::RunId, TransportError> {
-    bytes
-        .as_slice()
-        .try_into()
-        .map_err(|_| TransportError::InvalidPayload("run id must be 16 bytes".into()))
-}
-
-// --- HelloReply ---
-pub fn hello_reply_to_bytes(msg: &HelloReply) -> Result<Vec<u8>, rmp_serde::encode::Error> {
-    let shadow = HelloReplyShadow {
-        swarm_id: Cow::Borrowed(&msg.swarm_id),
-        peers: msg.peers.clone(), // Vec must be cloned for serialization wrapper matching
-        epoch: epoch_to_shadow(&msg.epoch, msg.epoch_seq),
-        auth: ByteBuf::from(msg.auth.clone()),
-        ready: msg.ready,
-        run_id: ByteBuf::from(msg.run_id.to_vec()),
-        away_ms: msg.away_ms,
-        sender_run_id: ByteBuf::from(msg.sender_run_id.to_vec()),
-    };
-    rmp_serde::to_vec(&shadow)
-}
-pub fn bytes_to_hello_reply(bytes: &[u8]) -> Result<HelloReply, TransportError> {
-    let shadow: HelloReplyShadow = rmp_serde::from_slice(bytes)?;
-    Ok(HelloReply {
-        swarm_id: shadow.swarm_id.into_owned(),
-        peers: shadow.peers,
-        epoch: epoch_from_shadow(&shadow.epoch)?,
-        epoch_seq: shadow.epoch.seq,
-        auth: shadow.auth.into_vec(),
-        ready: shadow.ready,
-        run_id: run_id_from_shadow(&shadow.run_id)?,
-        away_ms: shadow.away_ms,
-        sender_run_id: run_id_from_shadow(&shadow.sender_run_id)?,
-    })
-}
-
-// --- SyncEnvelope ---
-pub fn sync_envelope_to_bytes(msg: &SyncEnvelope) -> Result<Vec<u8>, rmp_serde::encode::Error> {
-    let shadow = SyncEnvelopeShadow {
-        swarm_id: Cow::Borrowed(&msg.swarm_id),
-        message: msg.message.clone(), // Clone required by automerge_msg_codec ownership rules
-        epoch: epoch_to_shadow(&msg.epoch, msg.epoch_seq),
-        ready: msg.ready,
-    };
-    rmp_serde::to_vec(&shadow)
-}
-pub fn bytes_to_sync_envelope(bytes: &[u8]) -> Result<SyncEnvelope, TransportError> {
-    let shadow: SyncEnvelopeShadow = rmp_serde::from_slice(bytes)?;
-    Ok(SyncEnvelope {
-        swarm_id: shadow.swarm_id.into_owned(),
-        epoch: epoch_from_shadow(&shadow.epoch)?,
-        epoch_seq: shadow.epoch.seq,
-        message: shadow.message,
-        ready: shadow.ready,
-    })
-}
-
-// --- BlobMessage ---
-#[derive(Serialize, Deserialize)]
-enum BlobIdShadow {
-    Change(ByteBuf),
-    Content(ByteBuf),
-}
-
-#[derive(Serialize, Deserialize)]
-enum BlobBodyShadow {
-    Request {
-        chunks: Vec<u32>,
-    },
-    Manifest {
-        size: u64,
-        chunk_size: u32,
-        hashes: Vec<ByteBuf>,
-    },
-    Chunk {
-        index: u32,
-        data: ByteBuf,
-    },
-    Missing,
-}
-
-#[derive(Serialize, Deserialize)]
-struct BlobShadow {
-    epoch: EpochShadow,
-    id: BlobIdShadow,
-    body: BlobBodyShadow,
-}
-
-fn blob_to_bytes(msg: &BlobMessage) -> Result<Vec<u8>, rmp_serde::encode::Error> {
-    let id = match msg.id {
-        BlobId::Change(hash) => BlobIdShadow::Change(ByteBuf::from(hash.0.to_vec())),
-        BlobId::Content(hash) => BlobIdShadow::Content(ByteBuf::from(hash.to_vec())),
-    };
-    let body = match &msg.body {
-        BlobBody::Request { chunks } => BlobBodyShadow::Request {
-            chunks: chunks.clone(),
-        },
-        BlobBody::Manifest {
-            size,
-            chunk_size,
-            hashes,
-        } => BlobBodyShadow::Manifest {
-            size: *size,
-            chunk_size: *chunk_size,
-            hashes: hashes
-                .iter()
-                .map(|hash| ByteBuf::from(hash.to_vec()))
-                .collect(),
-        },
-        BlobBody::Chunk { index, data } => BlobBodyShadow::Chunk {
-            index: *index,
-            data: ByteBuf::from(data.to_vec()),
-        },
-        BlobBody::Missing => BlobBodyShadow::Missing,
-    };
-    rmp_serde::to_vec(&BlobShadow {
-        epoch: epoch_to_shadow(&msg.epoch, msg.epoch_seq),
-        id,
-        body,
-    })
-}
-
-fn hash32(bytes: &[u8], what: &str) -> Result<[u8; 32], TransportError> {
-    bytes
-        .try_into()
-        .map_err(|_| TransportError::InvalidPayload(format!("{what} must be 32 bytes")))
-}
-
-fn bytes_to_blob(bytes: &[u8]) -> Result<BlobMessage, TransportError> {
-    let shadow: BlobShadow = rmp_serde::from_slice(bytes)?;
-    let id = match shadow.id {
-        BlobIdShadow::Change(hash) => BlobId::Change(ChangeHash(hash32(&hash, "a change hash")?)),
-        BlobIdShadow::Content(hash) => BlobId::Content(hash32(&hash, "a content hash")?),
-    };
-    let body = match shadow.body {
-        BlobBodyShadow::Request { chunks } => BlobBody::Request { chunks },
-        BlobBodyShadow::Manifest {
-            size,
-            chunk_size,
-            hashes,
-        } => BlobBody::Manifest {
-            size,
-            chunk_size,
-            hashes: hashes
-                .iter()
-                .map(|hash| hash32(hash, "a chunk hash"))
-                .collect::<Result<_, _>>()?,
-        },
-        BlobBodyShadow::Chunk { index, data } => BlobBody::Chunk {
-            index,
-            data: data.into_vec().into(),
-        },
-        BlobBodyShadow::Missing => BlobBody::Missing,
-    };
-    Ok(BlobMessage {
-        epoch: epoch_from_shadow(&shadow.epoch)?,
-        epoch_seq: shadow.epoch.seq,
-        id,
-        body,
-    })
-}
-
 #[derive(Error, Debug)]
 pub enum TransportError {
     #[error("Mycelium delivered a message without a sender public key")]
@@ -1166,18 +782,12 @@ pub enum TransportError {
     #[error("UTF-8 decoding failed: {0}")]
     Utf8Decode(#[from] std::string::FromUtf8Error),
 
-    #[error("MessagePack serialization failed: {0}")]
-    MessagePackEncode(#[from] rmp_serde::encode::Error),
-
-    #[error("MessagePack deserialization failed: {0}")]
-    MessagePackDecode(#[from] rmp_serde::decode::Error),
+    #[error(transparent)]
+    Wire(#[from] wire::WireError),
 
     #[error("Unknown topic: {0}")]
     UnknownTopic(String),
 
-    #[error("Invalid payload: {0}")]
-    InvalidPayload(String),
-
     #[error("Network HTTP request failed: {0}")]
     Http(#[from] reqwest::Error),
 
@@ -1191,6 +801,10 @@ pub enum TransportError {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::transport::{
+        BlobBody, BlobId, BlobMessage, Epoch, HeadsSummary, Keepalive, MAX_LISTED_HEADS,
+    };
+    use automerge::ChangeHash;
 
     /// Mycelium splits messages into chunks of this size.
     const MYCELIUM_CHUNK: usize = 1300;
@@ -1211,14 +825,17 @@ mod tests {
             ready: true,
             syncs_with_you: true,
         };
-        let payload = keepalive_to_bytes(&keepalive).unwrap();
+        let payload = wire::encode(&ParsedMessage::Pong(keepalive.clone())).unwrap();
         // Mycelium's message data: a length byte, the topic, then our payload.
-        let size = 1 + TOPIC_PONG.len().max(TOPIC_PING.len()) + payload.len();
+        let size = 1 + TOPIC.len() + payload.len();
         assert!(
             size + 50 <= MYCELIUM_CHUNK,
             "keepalive with {MAX_LISTED_HEADS} heads takes {size} bytes"
         );
-        assert_eq!(bytes_to_keepalive(&payload).unwrap().heads, keepalive.heads);
+        let ParsedMessage::Pong(decoded) = wire::decode(&payload).unwrap() else {
+            panic!("not a PONG");
+        };
+        assert_eq!(decoded.heads, keepalive.heads);
     }
 
     #[test]
@@ -1228,7 +845,11 @@ mod tests {
             id: [7; 32],
         };
         let id = BlobId::Change(ChangeHash([9; 32]));
-        let chunk = BlobMessage {
+        let decode_blob = |payload: &[u8]| match wire::decode(payload).unwrap() {
+            ParsedMessage::Blob(blob) => blob,
+            other => panic!("not a blob: {other:?}"),
+        };
+        let chunk = ParsedMessage::Blob(BlobMessage {
             epoch,
             epoch_seq: 4,
             id,
@@ -1236,11 +857,11 @@ mod tests {
                 index: 5,
                 data: vec![1u8; 64 * 1024].into(),
             },
-        };
-        let bytes = blob_to_bytes(&chunk).unwrap();
+        });
+        let bytes = wire::encode(&chunk).unwrap();
         // A 64 KiB chunk costs only a few bytes of framing.
         assert!(bytes.len() < 64 * 1024 + 128, "{} bytes", bytes.len());
-        let decoded = bytes_to_blob(&bytes).unwrap();
+        let decoded = decode_blob(&bytes);
         assert_eq!(decoded.id, id);
         assert_eq!(decoded.epoch, epoch);
         assert_eq!(decoded.epoch_seq, 4);
@@ -1248,7 +869,7 @@ mod tests {
             matches!(decoded.body, BlobBody::Chunk { index: 5, ref data } if data.len() == 64 * 1024)
         );
 
-        let manifest = BlobMessage {
+        let manifest = ParsedMessage::Blob(BlobMessage {
             epoch,
             epoch_seq: 4,
             id,
@@ -1257,8 +878,8 @@ mod tests {
                 chunk_size: 64,
                 hashes: vec![[1; 32], [2; 32]],
             },
-        };
-        let decoded = bytes_to_blob(&blob_to_bytes(&manifest).unwrap()).unwrap();
+        });
+        let decoded = decode_blob(&wire::encode(&manifest).unwrap());
         assert!(matches!(
             decoded.body,
             BlobBody::Manifest { size: 100, chunk_size: 64, ref hashes } if hashes.len() == 2
@@ -1266,6 +887,22 @@ mod tests {
     }
 
     #[test]
+    fn messages_round_trip_through_the_api_body() {
+        let keepalive = ParsedMessage::Ping(Keepalive {
+            heads: HeadsSummary::of(vec![ChangeHash([1; 32])]),
+            epoch: Epoch::GENESIS,
+            epoch_seq: 0,
+            ready: true,
+            syncs_with_you: true,
+        });
+        let encoded = encode_message("peer", &keepalive).unwrap();
+        assert!(matches!(
+            decode_message(&encoded.body).unwrap(),
+            ParsedMessage::Ping(_)
+        ));
+    }
+
+    #[test]
     fn listed_heads_equal_their_digest() {
         let heads: Vec<_> = (0..3u8).map(|n| ChangeHash([n; 32])).collect();
         let listed = HeadsSummary::of(heads.clone());
diff --git a/src/transport/wire.rs b/src/transport/wire.rs
new file mode 100644
index 0000000..8a8cbb8
--- /dev/null
+++ b/src/transport/wire.rs
@@ -0,0 +1,354 @@
+//! The wire format of a message: one byte saying what it is (`MessageKind`), then the
+//! message itself in MessagePack, as derived from its type in `transport` (structs as
+//! arrays, fields in declaration order). The codecs below make hashes, IDs and bytes
+//! compact (MessagePack binaries, not arrays of numbers) and check their lengths.
+//!
+//! The wire format may change freely until the first release: there is no compatibility
+//! with older nodes to keep.
+
+use thiserror::Error;
+
+use super::{MessageKind, ParsedMessage};
+
+/// What the first byte of a payload says.
+fn tag(kind: MessageKind) -> u8 {
+    match kind {
+        MessageKind::Hello => 0,
+        MessageKind::HelloReply => 1,
+        MessageKind::Sync => 2,
+        MessageKind::Ping => 3,
+        MessageKind::Pong => 4,
+        MessageKind::Blob => 5,
+    }
+}
+
+#[derive(Error, Debug)]
+pub enum WireError {
+    #[error("empty payload")]
+    Empty,
+
+    #[error("unknown message kind {0}")]
+    UnknownKind(u8),
+
+    #[error("MessagePack serialization failed: {0}")]
+    Encode(#[from] rmp_serde::encode::Error),
+
+    #[error("MessagePack deserialization failed: {0}")]
+    Decode(#[from] rmp_serde::decode::Error),
+}
+
+/// The payload of `message`.
+pub(crate) fn encode(message: &ParsedMessage) -> Result<Vec<u8>, WireError> {
+    let mut payload = vec![tag(message.kind())];
+    match message {
+        ParsedMessage::Hello(hello) => rmp_serde::encode::write(&mut payload, hello)?,
+        ParsedMessage::HelloReply(reply) => rmp_serde::encode::write(&mut payload, reply)?,
+        ParsedMessage::Sync(envelope) => rmp_serde::encode::write(&mut payload, envelope)?,
+        ParsedMessage::Ping(keepalive) | ParsedMessage::Pong(keepalive) => {
+            rmp_serde::encode::write(&mut payload, keepalive)?
+        }
+        ParsedMessage::Blob(blob) => rmp_serde::encode::write(&mut payload, blob)?,
+    }
+    Ok(payload)
+}
+
+/// The message in `payload` (as made by [`encode`]).
+pub(crate) fn decode(payload: &[u8]) -> Result<ParsedMessage, WireError> {
+    let (&kind, body) = payload.split_first().ok_or(WireError::Empty)?;
+    let is = |wanted: MessageKind| kind == tag(wanted);
+    Ok(if is(MessageKind::Hello) {
+        ParsedMessage::Hello(rmp_serde::from_slice(body)?)
+    } else if is(MessageKind::HelloReply) {
+        ParsedMessage::HelloReply(rmp_serde::from_slice(body)?)
+    } else if is(MessageKind::Sync) {
+        ParsedMessage::Sync(rmp_serde::from_slice(body)?)
+    } else if is(MessageKind::Ping) {
+        ParsedMessage::Ping(rmp_serde::from_slice(body)?)
+    } else if is(MessageKind::Pong) {
+        ParsedMessage::Pong(rmp_serde::from_slice(body)?)
+    } else if is(MessageKind::Blob) {
+        ParsedMessage::Blob(rmp_serde::from_slice(body)?)
+    } else {
+        return Err(WireError::UnknownKind(kind));
+    })
+}
+
+/// Field codecs (`#[serde(with = "...")]`) for the message types.
+pub(crate) mod codec {
+    use serde::de::Error as _;
+    use serde::{Deserialize, Deserializer, Serialize, Serializer};
+    use serde_bytes::{ByteBuf, Bytes};
+
+    /// Exactly `N` bytes, or a decoding error naming `what`.
+    fn fixed<const N: usize, E: serde::de::Error>(bytes: &[u8], what: &str) -> Result<[u8; N], E> {
+        bytes
+            .try_into()
+            .map_err(|_| E::custom(format!("{what} must be {N} bytes")))
+    }
+
+    /// A run ID (`RunId`).
+    pub(crate) mod run_id {
+        use super::*;
+
+        pub(crate) fn serialize<S: Serializer>(value: &[u8; 16], s: S) -> Result<S::Ok, S::Error> {
+            Bytes::new(value).serialize(s)
+        }
+
+        pub(crate) fn deserialize<'de, D: Deserializer<'de>>(d: D) -> Result<[u8; 16], D::Error> {
+            fixed(&ByteBuf::deserialize(d)?, "a run id")
+        }
+    }
+
+    /// A 32-byte hash.
+    pub(crate) mod bytes32 {
+        use super::*;
+
+        pub(crate) fn serialize<S: Serializer>(value: &[u8; 32], s: S) -> Result<S::Ok, S::Error> {
+            Bytes::new(value).serialize(s)
+        }
+
+        pub(crate) fn deserialize<'de, D: Deserializer<'de>>(d: D) -> Result<[u8; 32], D::Error> {
+            fixed(&ByteBuf::deserialize(d)?, "a hash")
+        }
+    }
+
+    /// 32-byte hashes.
+    pub(crate) mod hashes32 {
+        use super::*;
+
+        pub(crate) fn serialize<S: Serializer>(
+            value: &[[u8; 32]],
+            s: S,
+        ) -> Result<S::Ok, S::Error> {
+            s.collect_seq(value.iter().map(|hash| Bytes::new(hash)))
+        }
+
+        pub(crate) fn deserialize<'de, D: Deserializer<'de>>(
+            d: D,
+        ) -> Result<Vec<[u8; 32]>, D::Error> {
+            Vec::<ByteBuf>::deserialize(d)?
+                .iter()
+                .map(|hash| fixed(hash, "a hash"))
+                .collect()
+        }
+    }
+
+    /// An Automerge change hash.
+    pub(crate) mod change_hash {
+        use super::*;
+        use automerge::ChangeHash;
+
+        pub(crate) fn serialize<S: Serializer>(
+            value: &ChangeHash,
+            s: S,
+        ) -> Result<S::Ok, S::Error> {
+            Bytes::new(&value.0).serialize(s)
+        }
+
+        pub(crate) fn deserialize<'de, D: Deserializer<'de>>(d: D) -> Result<ChangeHash, D::Error> {
+            Ok(ChangeHash(fixed(
+                &ByteBuf::deserialize(d)?,
+                "a change hash",
+            )?))
+        }
+    }
+
+    /// An epoch: generation and ID.
+    pub(crate) mod epoch {
+        use super::*;
+        use crate::document::replica::Epoch;
+
+        pub(crate) fn serialize<S: Serializer>(value: &Epoch, s: S) -> Result<S::Ok, S::Error> {
+            (value.generation, Bytes::new(&value.id)).serialize(s)
+        }
+
+        pub(crate) fn deserialize<'de, D: Deserializer<'de>>(d: D) -> Result<Epoch, D::Error> {
+            let (generation, id): (u64, ByteBuf) = Deserialize::deserialize(d)?;
+            Ok(Epoch {
+                generation,
+                id: fixed(&id, "an epoch id")?,
+            })
+        }
+    }
+
+    /// Heads as a keepalive carries them: listed (at most `MAX_LISTED_HEADS`, digest
+    /// empty), or as a digest (none listed).
+    pub(crate) mod heads {
+        use super::*;
+        use crate::transport::{HeadsSummary, MAX_LISTED_HEADS};
+        use automerge::ChangeHash;
+
+        pub(crate) fn serialize<S: Serializer>(
+            value: &HeadsSummary,
+            s: S,
+        ) -> Result<S::Ok, S::Error> {
+            let (listed, digest): (Vec<&Bytes>, &Bytes) = match value {
+                HeadsSummary::Heads(heads) => (
+                    heads.iter().map(|head| Bytes::new(&head.0)).collect(),
+                    Bytes::new(&[]),
+                ),
+                HeadsSummary::Digest(digest) => (Vec::new(), Bytes::new(digest)),
+            };
+            (listed, digest).serialize(s)
+        }
+
+        pub(crate) fn deserialize<'de, D: Deserializer<'de>>(
+            d: D,
+        ) -> Result<HeadsSummary, D::Error> {
+            let (listed, digest): (Vec<ByteBuf>, ByteBuf) = Deserialize::deserialize(d)?;
+            if !digest.is_empty() {
+                return Ok(HeadsSummary::Digest(fixed(&digest, "a heads digest")?));
+            }
+            if listed.len() > MAX_LISTED_HEADS {
+                return Err(D::Error::custom(format!(
+                    "more than {MAX_LISTED_HEADS} heads listed"
+                )));
+            }
+            listed
+                .iter()
+                .map(|head| -> Result<ChangeHash, D::Error> {
+                    Ok(ChangeHash(fixed(head, "a head")?))
+                })
+                .collect::<Result<Vec<_>, _>>()
+                .map(HeadsSummary::Heads)
+        }
+    }
+
+    /// An Automerge sync message, in Automerge's own encoding.
+    pub(crate) mod sync_message {
+        use super::*;
+        use automerge::sync::Message;
+
+        pub(crate) fn serialize<S: Serializer>(value: &Message, s: S) -> Result<S::Ok, S::Error> {
+            Bytes::new(&value.clone().encode()).serialize(s)
+        }
+
+        pub(crate) fn deserialize<'de, D: Deserializer<'de>>(d: D) -> Result<Message, D::Error> {
+            Message::decode(&ByteBuf::deserialize(d)?)
+                .map_err(|error| D::Error::custom(format!("Automerge decode failed: {error:?}")))
+        }
+    }
+
+    /// Shared bytes (`bytes::Bytes`).
+    pub(crate) mod shared_bytes {
+        use super::*;
+
+        pub(crate) fn serialize<S: Serializer>(
+            value: &bytes::Bytes,
+            s: S,
+        ) -> Result<S::Ok, S::Error> {
+            Bytes::new(value).serialize(s)
+        }
+
+        pub(crate) fn deserialize<'de, D: Deserializer<'de>>(
+            d: D,
+        ) -> Result<bytes::Bytes, D::Error> {
+            Ok(ByteBuf::deserialize(d)?.into_vec().into())
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::document::replica::Epoch;
+    use crate::transport::{
+        BlobBody, BlobId, BlobMessage, HeadsSummary, HelloReply, HelloRequest, Keepalive,
+        MAX_LISTED_HEADS,
+    };
+    use automerge::ChangeHash;
+
+    #[test]
+    fn handshakes_round_trip() {
+        let epoch = Epoch {
+            generation: 2,
+            id: [3; 32],
+        };
+        let hello = ParsedMessage::Hello(HelloRequest {
+            swarm_id: Some("swarm".into()),
+            epoch,
+            epoch_seq: 1,
+            auth: vec![1, 2, 3],
+            ready: true,
+            run_id: [4; 16],
+        });
+        let ParsedMessage::Hello(decoded) = decode(&encode(&hello).unwrap()).unwrap() else {
+            panic!("not a HELLO");
+        };
+        assert_eq!(decoded.swarm_id.as_deref(), Some("swarm"));
+        assert_eq!((decoded.epoch, decoded.epoch_seq), (epoch, 1));
+        assert_eq!(decoded.auth, vec![1, 2, 3]);
+        assert_eq!(decoded.run_id, [4; 16]);
+
+        let reply = ParsedMessage::HelloReply(HelloReply {
+            swarm_id: "swarm".into(),
+            peers: vec!["p".into()],
+            epoch,
+            epoch_seq: 1,
+            auth: Vec::new(),
+            ready: false,
+            run_id: [4; 16],
+            away_ms: Some(5),
+            sender_run_id: [6; 16],
+        });
+        let ParsedMessage::HelloReply(decoded) = decode(&encode(&reply).unwrap()).unwrap() else {
+            panic!("not a HELLO_REPLY");
+        };
+        assert_eq!(decoded.peers, vec!["p".to_owned()]);
+        assert_eq!(decoded.away_ms, Some(5));
+        assert_eq!(decoded.sender_run_id, [6; 16]);
+    }
+
+    #[test]
+    fn keepalives_keep_their_kind_and_heads() {
+        let keepalive = Keepalive {
+            heads: HeadsSummary::Digest([8; 32]),
+            epoch: Epoch::GENESIS,
+            epoch_seq: 0,
+            ready: true,
+            syncs_with_you: false,
+        };
+        let decoded = decode(&encode(&ParsedMessage::Pong(keepalive.clone())).unwrap()).unwrap();
+        let ParsedMessage::Pong(decoded) = decoded else {
+            panic!("not a PONG");
+        };
+        assert_eq!(decoded.heads, keepalive.heads);
+    }
+
+    #[test]
+    fn too_many_listed_heads_are_refused() {
+        let keepalive = Keepalive {
+            heads: HeadsSummary::Heads(
+                (0..=MAX_LISTED_HEADS as u8)
+                    .map(|n| ChangeHash([n; 32]))
+                    .collect(),
+            ),
+            epoch: Epoch::GENESIS,
+            epoch_seq: 0,
+            ready: true,
+            syncs_with_you: true,
+        };
+        assert!(decode(&encode(&ParsedMessage::Ping(keepalive)).unwrap()).is_err());
+    }
+
+    #[test]
+    fn unknown_kinds_and_bad_payloads_are_refused() {
+        assert!(matches!(decode(&[]), Err(WireError::Empty)));
+        assert!(matches!(decode(&[99]), Err(WireError::UnknownKind(99))));
+
+        let blob = ParsedMessage::Blob(BlobMessage {
+            epoch: Epoch::GENESIS,
+            epoch_seq: 0,
+            id: BlobId::Content([1; 32]),
+            body: BlobBody::Missing,
+        });
+        let payload = encode(&blob).unwrap();
+        assert!(matches!(decode(&payload).unwrap(), ParsedMessage::Blob(_)));
+        // A payload cut short doesn't decode.
+        assert!(decode(&payload[..payload.len() - 1]).is_err());
+        // Nor one of another kind.
+        let mut relabelled = payload.clone();
+        relabelled[0] = tag(MessageKind::Hello);
+        assert!(decode(&relabelled).is_err());
+    }
+}