Added stat version

crates/federation-net/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "federation-net"
-version = "0.1.0"
+version = "0.2.0"
 description = "Generic peer-to-peer networking engine built on Iroh"
 readme = "README.md"
 documentation = "https://docs.rs/federation-net"

crates/federation-net/src/config.rs

@@ -41,6 +41,9 @@ pub struct NetworkConfig {
     /// Auxiliary ALPN protocols on which this peer accepts raw byte streams
     /// (see [`crate::NetworkEngine::stream_acceptor`]).
     pub stream_protocols: Vec<Vec<u8>>,
+    /// Auxiliary stream protocols whose handshake is independent of the
+    /// application schema. Intended for compatibility/capability discovery.
+    pub schema_independent_stream_protocols: Vec<Vec<u8>>,
 }
 
 impl NetworkConfig {
@@ -65,6 +68,7 @@ pub struct NetworkConfigBuilder {
     max_concurrent_streams_per_peer: Option<usize>,
     rendezvous: Option<RendezvousConfig>,
     stream_protocols: Vec<Vec<u8>>,
+    schema_independent_stream_protocols: Vec<Vec<u8>>,
 }
 
 impl NetworkConfigBuilder {
@@ -124,6 +128,17 @@ impl NetworkConfigBuilder {
         self
     }
 
+    /// Declares an auxiliary ALPN protocol that validates the network and
+    /// federation transport version, but deliberately does not require the
+    /// application schema id to match.
+    ///
+    /// This is suitable for bounded, self-versioned capability discovery
+    /// protocols that must remain reachable across schema upgrades.
+    pub fn schema_independent_stream_protocol(mut self, alpn: impl Into<Vec<u8>>) -> Self {
+        self.schema_independent_stream_protocols.push(alpn.into());
+        self
+    }
+
     /// Validates the configuration and builds a [`NetworkConfig`].
     pub fn build(self) -> Result<NetworkConfig> {
         let data_dir = self
@@ -210,6 +225,27 @@ impl NetworkConfigBuilder {
                 )));
             }
         }
+        for (index, alpn) in self.schema_independent_stream_protocols.iter().enumerate() {
+            if alpn.is_empty() {
+                return Err(NetworkError::InvalidConfig(
+                    "schema-independent stream protocol ALPN must not be empty".into(),
+                ));
+            }
+            if alpn.as_slice() == ALPN {
+                return Err(NetworkError::InvalidConfig(
+                    "schema-independent stream protocol ALPN must differ from the engine ALPN"
+                        .into(),
+                ));
+            }
+            if self.schema_independent_stream_protocols[..index].contains(alpn)
+                || self.stream_protocols.contains(alpn)
+            {
+                return Err(NetworkError::InvalidConfig(format!(
+                    "duplicate stream protocol ALPN: {}",
+                    String::from_utf8_lossy(alpn)
+                )));
+            }
+        }
 
         Ok(NetworkConfig {
             data_dir,
@@ -221,6 +257,7 @@ impl NetworkConfigBuilder {
             max_concurrent_streams_per_peer,
             rendezvous: self.rendezvous,
             stream_protocols: self.stream_protocols,
+            schema_independent_stream_protocols: self.schema_independent_stream_protocols,
         })
     }
 }
@@ -326,5 +363,21 @@ mod tests {
                 .build()
                 .is_err()
         );
+        assert!(
+            base_builder()
+                .stream_protocol("dup/1")
+                .schema_independent_stream_protocol("dup/1")
+                .build()
+                .is_err()
+        );
+
+        let config = base_builder()
+            .schema_independent_stream_protocol("furumi/capabilities/1")
+            .build()
+            .expect("valid schema-independent stream protocol");
+        assert_eq!(
+            config.schema_independent_stream_protocols,
+            vec![b"furumi/capabilities/1".to_vec()]
+        );
     }
 }

crates/federation-net/src/engine.rs

@@ -383,6 +383,28 @@ impl<M: Message> Shared<M> {
         }
     }
 
+    fn validate_stream_handshake(
+        &self,
+        alpn: &[u8],
+        handshake: &Handshake,
+    ) -> Option<HandshakeErrorCode> {
+        if handshake.protocol_version != PROTOCOL_VERSION {
+            Some(HandshakeErrorCode::UnsupportedProtocolVersion)
+        } else if handshake.network_id != self.config.network_id {
+            Some(HandshakeErrorCode::NetworkMismatch)
+        } else if !self
+            .config
+            .schema_independent_stream_protocols
+            .iter()
+            .any(|protocol| protocol.as_slice() == alpn)
+            && handshake.schema_id != self.config.schema_id
+        {
+            Some(HandshakeErrorCode::SchemaMismatch)
+        } else {
+            None
+        }
+    }
+
     /// Drives an established connection: accepts message streams until the
     /// connection closes, then cleans up the registry entry.
     async fn connection_loop(
@@ -627,7 +649,7 @@ impl<M: Message> Shared<M> {
         .and_then(|res| res);
 
         let verdict = match &handshake {
-            Ok(handshake) => self.validate_handshake(handshake),
+            Ok(handshake) => self.validate_stream_handshake(alpn, handshake),
             Err(_) => Some(HandshakeErrorCode::InvalidHandshake),
         };
         if let Some(code) = verdict {
@@ -948,6 +970,7 @@ impl<M: Message> NetworkEngine<M> {
         let stream_acceptors = config
             .stream_protocols
             .iter()
+            .chain(config.schema_independent_stream_protocols.iter())
             .map(|alpn| {
                 let (sender, receiver) = mpsc::channel(STREAM_ACCEPT_QUEUE);
                 (
@@ -982,6 +1005,13 @@ impl<M: Message> NetworkEngine<M> {
             };
             router_builder = router_builder.accept(alpn.clone(), handler);
         }
+        for alpn in &shared.config.schema_independent_stream_protocols {
+            let handler = StreamProtocol::<M> {
+                shared: Arc::downgrade(&shared),
+                alpn: alpn.clone(),
+            };
+            router_builder = router_builder.accept(alpn.clone(), handler);
+        }
         let router = router_builder.spawn();
         *lock(&shared.router) = Some(router);
         if let Some(rendezvous) = shared.config.rendezvous.clone() {

crates/federation-net/src/lib.rs

@@ -72,6 +72,7 @@ pub use engine::{
 pub use error::{NetworkError, Result};
 pub use event::{ConnectionDirection, NetworkEvent, NetworkEventReceiver};
 pub use protocol::{ALPN, NetworkId, PROTOCOL_VERSION, SchemaId};
+pub use rendezvous::RENDEZVOUS_RECORD_VERSION;
 pub use rendezvous::{DEFAULT_RENDEZVOUS_ENTRY_TTL, DEFAULT_RENDEZVOUS_INTERVAL, RendezvousConfig};
 pub use ticket::{PeerTicket, TICKET_VERSION};
 

crates/federation-net/src/rendezvous.rs

@@ -38,7 +38,8 @@ pub const DEFAULT_RENDEZVOUS_INTERVAL: Duration = Duration::from_secs(60);
 pub const DEFAULT_RENDEZVOUS_ENTRY_TTL: Duration = Duration::from_secs(30 * 60);
 
 /// Version of the rendezvous record wire format.
-const RECORD_VERSION: u16 = 1;
+/// Version of the Mainline-DHT rendezvous record envelope.
+pub const RENDEZVOUS_RECORD_VERSION: u16 = 1;
 /// Domain separation context for deriving the record signing key.
 const KEY_DERIVATION_CONTEXT: &str = "federation-net:rendezvous:v1";
 /// BEP44 caps mutable values at 1000 bytes; stay safely below.
@@ -104,7 +105,7 @@ pub(crate) fn now_ms() -> u64 {
 
 fn decode_record(bytes: &[u8]) -> Option<Vec<RendezvousEntry>> {
     let record: RendezvousRecord = postcard::from_bytes(bytes).ok()?;
-    (record.version == RECORD_VERSION).then_some(record.entries)
+    (record.version == RENDEZVOUS_RECORD_VERSION).then_some(record.entries)
 }
 
 /// Merges entries from every record instance seen this round with our own,
@@ -144,7 +145,7 @@ fn merge_entries(
 fn encode_record_capped(mut entries: Vec<RendezvousEntry>) -> Result<Vec<u8>> {
     loop {
         let record = RendezvousRecord {
-            version: RECORD_VERSION,
+            version: RENDEZVOUS_RECORD_VERSION,
             entries,
         };
         let encoded = postcard::to_stdvec(&record).map_err(|err| {
@@ -279,7 +280,7 @@ mod tests {
     fn malformed_and_wrong_version_records_are_ignored() {
         assert!(decode_record(b"garbage").is_none());
         let record = RendezvousRecord {
-            version: RECORD_VERSION + 1,
+            version: RENDEZVOUS_RECORD_VERSION + 1,
             entries: vec![],
         };
         let encoded = postcard::to_stdvec(&record).expect("encode");

crates/federation-net/tests/integration.rs

@@ -12,6 +12,7 @@ use federation_net::{
 const TEST_TIMEOUT: Duration = Duration::from_secs(120);
 /// Timeout used when waiting for a single event.
 const EVENT_TIMEOUT: Duration = Duration::from_secs(30);
+const CAPABILITY_ALPN: &[u8] = b"test/capabilities/1";
 
 #[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
 enum TestMessage {
@@ -207,6 +208,64 @@ async fn schema_mismatch_is_rejected() {
     .expect("test timed out");
 }
 
+#[tokio::test]
+async fn schema_independent_stream_crosses_schema_versions() {
+    let _net = NET_LOCK.lock().await;
+    tokio::time::timeout(TEST_TIMEOUT, async {
+        let dir_a = tempfile::tempdir().expect("tempdir");
+        let dir_b = tempfile::tempdir().expect("tempdir");
+        let config_a = NetworkConfig::builder()
+            .data_dir(dir_a.path())
+            .network_id(NetworkId::from_name("same-network"))
+            .schema_id(SchemaId::from_name("schema-one"))
+            .schema_independent_stream_protocol(CAPABILITY_ALPN)
+            .build()
+            .expect("config a");
+        let config_b = NetworkConfig::builder()
+            .data_dir(dir_b.path())
+            .network_id(NetworkId::from_name("same-network"))
+            .schema_id(SchemaId::from_name("schema-two"))
+            .schema_independent_stream_protocol(CAPABILITY_ALPN)
+            .build()
+            .expect("config b");
+        let (engine_a, _events_a) = Engine::start(config_a).await.expect("engine a");
+        let (engine_b, _events_b) = Engine::start(config_b).await.expect("engine b");
+        let mut acceptor = engine_a
+            .stream_acceptor(CAPABILITY_ALPN)
+            .expect("capability acceptor");
+        let address = engine_a.ticket().await.expect("ticket").endpoint_addr;
+
+        let accepting = tokio::spawn(async move {
+            let mut stream = acceptor.accept().await.expect("incoming stream");
+            let value = stream.recv.read_to_end(16).await.expect("read request");
+            assert_eq!(value, b"versions?");
+            stream.send.write_all(b"v2").await.expect("write response");
+            stream.send.finish().expect("finish response");
+            let _ = stream.send.stopped().await;
+        });
+        let mut stream = engine_b
+            .open_stream(address, CAPABILITY_ALPN)
+            .await
+            .expect("schema-independent stream");
+        stream
+            .send
+            .write_all(b"versions?")
+            .await
+            .expect("write request");
+        stream.send.finish().expect("finish request");
+        assert_eq!(
+            stream.recv.read_to_end(16).await.expect("read response"),
+            b"v2"
+        );
+        accepting.await.expect("accept task");
+
+        engine_a.shutdown().await.expect("shutdown a");
+        engine_b.shutdown().await.expect("shutdown b");
+    })
+    .await
+    .expect("test timed out");
+}
+
 #[tokio::test]
 async fn shutdown_disconnects_peers() {
     let _net = NET_LOCK.lock().await;

crates/music-dht/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "music-dht"
-version = "0.2.0"
+version = "0.3.0"
 description = "Distributed music library search: a Kademlia-style DHT on top of federation-net"
 readme = "README.md"
 documentation = "https://docs.rs/music-dht"

crates/music-dht/src/capabilities.rs

@@ -0,0 +1,227 @@
+//! Runtime publication of application and protocol versions.
+//!
+//! Capability manifests are informational. They let applications explain
+//! interoperability problems and suggest an update, but do not authorize a
+//! peer or trigger any update action.
+
+use std::collections::BTreeMap;
+
+use federation_net::ByteStream;
+use serde::{Deserialize, Serialize};
+use tokio::io::{AsyncRead, AsyncReadExt};
+
+use crate::error::{MusicDhtError, Result};
+
+/// Auxiliary ALPN used to query a peer's version manifest.
+pub const CAPABILITIES_ALPN: &[u8] = b"furumi/capabilities/1";
+/// Capability protocol and manifest envelope version.
+pub const CAPABILITIES_PROTOCOL_VERSION: u16 = 1;
+/// Maximum accepted capability JSON line.
+pub const MAX_CAPABILITIES_LINE: usize = 64 * 1024;
+/// Maximum protocol entries accepted from one peer.
+pub const MAX_PROTOCOL_ENTRIES: usize = 64;
+
+/// Stable protocol identifier for federation-net.
+pub const FEDERATION_NET_ID: &str = "federation_net";
+/// Stable protocol identifier for endpoint tickets.
+pub const TICKET_ID: &str = "ticket";
+/// Stable protocol identifier for Mainline-DHT rendezvous records.
+pub const RENDEZVOUS_ID: &str = "rendezvous";
+/// Stable protocol identifier for music-dht.
+pub const MUSIC_DHT_ID: &str = "music_dht";
+/// Stable protocol identifier for rich catalog streams.
+pub const CATALOG_ID: &str = "catalog";
+/// Stable protocol identifier for personal-device synchronization.
+pub const DEVICE_SYNC_ID: &str = "device_sync";
+/// Stable protocol identifier for Jam playback control.
+pub const JAM_ID: &str = "jam";
+
+/// Application and protocol versions published by one peer.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct CapabilityManifest {
+    /// Manifest schema version.
+    pub manifest_version: u16,
+    /// Application family, for example `furumi`.
+    pub application: String,
+    /// User-visible application release.
+    pub application_version: String,
+    /// Supported/current protocol version by stable identifier.
+    pub protocols: BTreeMap<String, u16>,
+}
+
+impl CapabilityManifest {
+    /// Creates a manifest containing every protocol owned by Frid.
+    pub fn frid(application: impl Into<String>, application_version: impl Into<String>) -> Self {
+        let mut protocols = BTreeMap::new();
+        protocols.insert(
+            FEDERATION_NET_ID.to_string(),
+            federation_net::PROTOCOL_VERSION,
+        );
+        protocols.insert(TICKET_ID.to_string(), federation_net::TICKET_VERSION);
+        protocols.insert(
+            RENDEZVOUS_ID.to_string(),
+            federation_net::RENDEZVOUS_RECORD_VERSION,
+        );
+        protocols.insert(MUSIC_DHT_ID.to_string(), crate::DHT_PROTOCOL_VERSION);
+        protocols.insert(
+            CATALOG_ID.to_string(),
+            crate::catalog::CATALOG_PROTOCOL_VERSION,
+        );
+        protocols.insert(
+            DEVICE_SYNC_ID.to_string(),
+            crate::device_sync::DEVICE_SYNC_PROTOCOL_VERSION,
+        );
+        protocols.insert(JAM_ID.to_string(), crate::jam::JAM_PROTOCOL_VERSION);
+        Self {
+            manifest_version: CAPABILITIES_PROTOCOL_VERSION,
+            application: application.into(),
+            application_version: application_version.into(),
+            protocols,
+        }
+    }
+
+    /// Adds an application-owned protocol to the manifest.
+    pub fn with_protocol(mut self, id: impl Into<String>, version: u16) -> Self {
+        self.protocols.insert(id.into(), version);
+        self
+    }
+
+    /// Removes a shared protocol that this particular application does not
+    /// expose, while retaining the canonical versions for the others.
+    pub fn without_protocol(mut self, id: &str) -> Self {
+        self.protocols.remove(id);
+        self
+    }
+
+    /// Validates bounds and identifiers received from a peer.
+    pub fn validate(&self) -> Result<()> {
+        if self.manifest_version != CAPABILITIES_PROTOCOL_VERSION {
+            return Err(protocol_error(format!(
+                "unsupported capability manifest version {}",
+                self.manifest_version
+            )));
+        }
+        if self.application.trim().is_empty()
+            || self.application.len() > 64
+            || self.application_version.len() > 64
+            || self.protocols.len() > MAX_PROTOCOL_ENTRIES
+        {
+            return Err(protocol_error("invalid capability manifest bounds"));
+        }
+        for (id, version) in &self.protocols {
+            if id.is_empty()
+                || id.len() > 64
+                || !id
+                    .bytes()
+                    .all(|byte| byte.is_ascii_lowercase() || byte.is_ascii_digit() || byte == b'_')
+                || *version == 0
+            {
+                return Err(protocol_error("invalid capability protocol entry"));
+            }
+        }
+        Ok(())
+    }
+}
+
+/// One request/response message on [`CAPABILITIES_ALPN`].
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(tag = "type", rename_all = "snake_case")]
+pub enum CapabilityMessage {
+    /// Requests the peer's current manifest.
+    Get {
+        /// Request protocol version.
+        version: u16,
+    },
+    /// Returns a manifest.
+    Manifest {
+        /// Published version manifest.
+        manifest: CapabilityManifest,
+    },
+    /// Refuses a malformed or unsupported request.
+    Error {
+        /// Human-readable diagnostic.
+        message: String,
+    },
+}
+
+/// Writes one bounded JSON-lines capability message.
+pub async fn write_message(stream: &mut ByteStream, message: &CapabilityMessage) -> Result<()> {
+    let mut bytes = serde_json::to_vec(message).map_err(protocol_error)?;
+    if bytes.len() > MAX_CAPABILITIES_LINE {
+        return Err(protocol_error("capability message is too large"));
+    }
+    bytes.push(b'\n');
+    stream.send.write_all(&bytes).await.map_err(network_error)?;
+    Ok(())
+}
+
+/// Reads one bounded JSON-lines capability message.
+pub async fn read_message(stream: &mut ByteStream) -> Result<CapabilityMessage> {
+    read_message_from(&mut stream.recv).await
+}
+
+/// Reads one bounded JSON-lines capability message from an async reader.
+pub async fn read_message_from<R: AsyncRead + Unpin>(reader: &mut R) -> Result<CapabilityMessage> {
+    let mut bytes = Vec::new();
+    let mut byte = [0_u8; 1];
+    loop {
+        let read = reader.read(&mut byte).await.map_err(network_error)?;
+        if read == 0 || byte[0] == b'\n' {
+            break;
+        }
+        bytes.push(byte[0]);
+        if bytes.len() > MAX_CAPABILITIES_LINE {
+            return Err(protocol_error("capability message is too large"));
+        }
+    }
+    if bytes.is_empty() {
+        return Err(protocol_error("capability message is empty"));
+    }
+    let message: CapabilityMessage = serde_json::from_slice(&bytes).map_err(protocol_error)?;
+    if let CapabilityMessage::Manifest { manifest } = &message {
+        manifest.validate()?;
+    }
+    Ok(message)
+}
+
+fn protocol_error(error: impl std::fmt::Display) -> MusicDhtError {
+    MusicDhtError::Protocol(error.to_string())
+}
+
+fn network_error(error: impl std::fmt::Display) -> MusicDhtError {
+    MusicDhtError::Network(error.to_string())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn frid_manifest_contains_every_shared_protocol() {
+        let manifest = CapabilityManifest::frid("furumi", "1.2.3");
+        for id in [
+            FEDERATION_NET_ID,
+            TICKET_ID,
+            RENDEZVOUS_ID,
+            MUSIC_DHT_ID,
+            CATALOG_ID,
+            DEVICE_SYNC_ID,
+            JAM_ID,
+        ] {
+            assert!(manifest.protocols.contains_key(id), "missing {id}");
+        }
+        manifest.validate().unwrap();
+    }
+
+    #[test]
+    fn application_protocol_is_additive() {
+        let manifest = CapabilityManifest::frid("furumi", "1.2.3").with_protocol("audio", 1);
+        assert_eq!(manifest.protocols.get("audio"), Some(&1));
+    }
+
+    #[test]
+    fn invalid_protocol_identifier_is_rejected() {
+        let manifest = CapabilityManifest::frid("furumi", "1.2.3").with_protocol("Audio Stream", 1);
+        assert!(manifest.validate().is_err());
+    }
+}

crates/music-dht/src/catalog.rs

@@ -8,6 +8,8 @@ use serde::{Deserialize, Serialize};
 
 /// ALPN of the Furumi catalog stream protocol.
 pub const CATALOG_ALPN: &[u8] = b"furumi-fd/catalog/1";
+/// Version of the Furumi catalog stream protocol.
+pub const CATALOG_PROTOCOL_VERSION: u16 = 1;
 
 /// Request sent by a catalog client.
 #[derive(Debug, Clone, Default, Serialize, Deserialize)]

crates/music-dht/src/config.rs

@@ -55,6 +55,9 @@ pub struct MusicDhtConfig {
     /// Auxiliary ALPN protocols on which this peer accepts raw byte streams
     /// (see [`crate::MusicDhtService::stream_acceptor`]).
     pub stream_protocols: Vec<Vec<u8>>,
+    /// Auxiliary stream protocols that remain reachable across DHT schema
+    /// upgrades, such as bounded capability discovery.
+    pub schema_independent_stream_protocols: Vec<Vec<u8>>,
 }
 
 impl MusicDhtConfig {
@@ -80,6 +83,7 @@ pub struct MusicDhtConfigBuilder {
     dial_timeout: Option<Duration>,
     rendezvous: Option<RendezvousConfig>,
     stream_protocols: Vec<Vec<u8>>,
+    schema_independent_stream_protocols: Vec<Vec<u8>>,
 }
 
 impl MusicDhtConfigBuilder {
@@ -146,6 +150,13 @@ impl MusicDhtConfigBuilder {
         self
     }
 
+    /// Declares a self-versioned auxiliary protocol that does not require the
+    /// local and remote DHT schema ids to match.
+    pub fn schema_independent_stream_protocol(mut self, alpn: impl Into<Vec<u8>>) -> Self {
+        self.schema_independent_stream_protocols.push(alpn.into());
+        self
+    }
+
     /// Validates and builds the configuration.
     pub fn build(self) -> Result<MusicDhtConfig> {
         let data_dir = self
@@ -171,6 +182,7 @@ impl MusicDhtConfigBuilder {
             dial_timeout: self.dial_timeout.unwrap_or(DEFAULT_DIAL_TIMEOUT),
             rendezvous: self.rendezvous,
             stream_protocols: self.stream_protocols,
+            schema_independent_stream_protocols: self.schema_independent_stream_protocols,
         };
         for (name, value) in [
             ("republish_interval", config.republish_interval),

crates/music-dht/src/jam.rs

@@ -0,0 +1,286 @@
+//! Ephemeral shared playback control between independent Furumi peers.
+//!
+//! Jam is separate from personal-device sync. Possession of a [`JamInvite`]
+//! capability authorizes playback control for one host process, but never
+//! grants likes, playlists, history, or trusted-device membership.
+
+use serde::{Deserialize, Serialize};
+use tokio::io::{AsyncRead, AsyncReadExt};
+
+use crate::device_sync::{PlaybackCommand, PlaybackSnapshot};
+use crate::error::{MusicDhtError, Result};
+
+/// Dedicated stream protocol for Jam control version 1.
+pub const JAM_ALPN_V1: &[u8] = b"furumi/jam/1";
+/// Current Jam stream protocol.
+pub const JAM_ALPN: &[u8] = JAM_ALPN_V1;
+/// Current Jam wire version.
+pub const JAM_PROTOCOL_VERSION: u16 = 1;
+/// Maximum accepted JSON message.
+pub const MAX_JAM_LINE: usize = 8 * 1024 * 1024;
+/// Recommended timeout for a host with no participant polls.
+pub const DEFAULT_JAM_IDLE_TTL_MS: i64 = 30 * 60 * 1_000;
+/// Maximum commands accepted in one participant poll.
+pub const MAX_JAM_COMMANDS_PER_POLL: usize = 128;
+
+/// Long-lived host capability encoded as `frid://j/<base64url-json>`.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct JamInvite {
+    /// Invite schema version.
+    pub v: u16,
+    /// Host transport ticket.
+    #[serde(rename = "t")]
+    pub ticket: String,
+    /// Runtime Jam session id.
+    #[serde(rename = "j")]
+    pub jam_id: String,
+    /// Runtime capability secret.
+    #[serde(rename = "s")]
+    pub secret: String,
+    /// Host player/device id.
+    #[serde(rename = "d")]
+    pub host_device_id: String,
+    /// Host display name.
+    #[serde(rename = "n")]
+    pub host_name: String,
+}
+
+impl JamInvite {
+    /// Encodes this invite as an opaque `frid://j/...` capability.
+    pub fn to_uri(&self) -> Result<String> {
+        validate_invite(self)?;
+        let bytes = serde_json::to_vec(self).map_err(protocol_err)?;
+        Ok(format!("frid://j/{}", base64url_encode(&bytes)))
+    }
+
+    /// Parses and validates an opaque Jam capability.
+    pub fn from_uri(uri: &str) -> Result<Self> {
+        let encoded = uri
+            .trim()
+            .strip_prefix("frid://j/")
+            .ok_or_else(|| MusicDhtError::Protocol("expected frid://j invite".to_string()))?;
+        let bytes = base64url_decode(encoded)?;
+        let invite: Self = serde_json::from_slice(&bytes).map_err(protocol_err)?;
+        validate_invite(&invite)?;
+        Ok(invite)
+    }
+}
+
+/// Display identity scoped to one runtime Jam.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct JamParticipant {
+    /// Participant runtime id.
+    pub participant_id: String,
+    /// User-visible name.
+    pub name: String,
+    /// Last successful host exchange in Unix milliseconds.
+    pub last_seen_ms: i64,
+}
+
+/// Deduplicated playback command submitted by a participant.
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct JamCommand {
+    /// Unique command id generated by the participant.
+    pub command_id: String,
+    /// Participant runtime id.
+    pub participant_id: String,
+    /// Command payload shared with personal-device and web control.
+    pub command: PlaybackCommand,
+    /// Client timestamp for diagnostics.
+    pub sent_at_ms: i64,
+}
+
+/// Top-level JSON-lines Jam message.
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(tag = "type", rename_all = "snake_case")]
+pub enum JamWireMessage {
+    /// Participant poll: submits commands and requests host state.
+    Poll {
+        /// Protocol version.
+        version: u16,
+        /// Jam id from the capability.
+        jam_id: String,
+        /// Capability secret.
+        secret: String,
+        /// Participant display identity.
+        participant: JamParticipant,
+        /// Commands not yet acknowledged by the host.
+        #[serde(default)]
+        commands: Vec<JamCommand>,
+    },
+    /// Host response to a poll.
+    Snapshot {
+        /// Whether the capability was accepted.
+        accepted: bool,
+        /// Optional refusal reason.
+        #[serde(default)]
+        error: Option<String>,
+        /// IDs of commands accepted by this host runtime.
+        #[serde(default)]
+        acknowledged_command_ids: Vec<String>,
+        /// Current host playback state.
+        #[serde(default)]
+        playback: Option<PlaybackSnapshot>,
+        /// Currently visible participants.
+        #[serde(default)]
+        participants: Vec<JamParticipant>,
+        /// Host response time in Unix milliseconds.
+        host_time_ms: i64,
+    },
+    /// Explicit best-effort participant departure.
+    Leave {
+        /// Protocol version.
+        version: u16,
+        /// Jam id from the capability.
+        jam_id: String,
+        /// Capability secret.
+        secret: String,
+        /// Participant runtime id.
+        participant_id: String,
+    },
+}
+
+/// Reads a bounded Jam message from an async reader.
+pub async fn read_message_from<R: AsyncRead + Unpin>(reader: &mut R) -> Result<JamWireMessage> {
+    let mut line = Vec::new();
+    let mut byte = [0_u8; 1];
+    loop {
+        let read = reader.read(&mut byte).await.map_err(network_err)?;
+        if read == 0 || byte[0] == b'\n' {
+            break;
+        }
+        line.push(byte[0]);
+        if line.len() > MAX_JAM_LINE {
+            return Err(MusicDhtError::Protocol(
+                "Jam protocol line is too large".to_string(),
+            ));
+        }
+    }
+    if line.is_empty() {
+        return Err(MusicDhtError::Protocol(
+            "Jam protocol message is empty".to_string(),
+        ));
+    }
+    serde_json::from_slice(&line).map_err(protocol_err)
+}
+
+fn validate_invite(invite: &JamInvite) -> Result<()> {
+    if invite.v != JAM_PROTOCOL_VERSION {
+        return Err(MusicDhtError::Protocol(format!(
+            "unsupported Jam invite version {}",
+            invite.v
+        )));
+    }
+    if invite.ticket.trim().is_empty()
+        || invite.jam_id.trim().is_empty()
+        || invite.secret.len() < 16
+        || invite.host_device_id.trim().is_empty()
+        || invite.host_name.trim().is_empty()
+    {
+        return Err(MusicDhtError::Protocol("incomplete Jam invite".to_string()));
+    }
+    Ok(())
+}
+
+fn base64url_encode(bytes: &[u8]) -> String {
+    const TABLE: &[u8; 64] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-_";
+    let mut out = String::new();
+    let mut i = 0;
+    while i < bytes.len() {
+        let b0 = bytes[i];
+        let b1 = bytes.get(i + 1).copied().unwrap_or(0);
+        let b2 = bytes.get(i + 2).copied().unwrap_or(0);
+        out.push(TABLE[(b0 >> 2) as usize] as char);
+        out.push(TABLE[(((b0 & 3) << 4) | (b1 >> 4)) as usize] as char);
+        if i + 1 < bytes.len() {
+            out.push(TABLE[(((b1 & 15) << 2) | (b2 >> 6)) as usize] as char);
+        }
+        if i + 2 < bytes.len() {
+            out.push(TABLE[(b2 & 63) as usize] as char);
+        }
+        i += 3;
+    }
+    out
+}
+
+fn base64url_decode(value: &str) -> Result<Vec<u8>> {
+    fn decode(byte: u8) -> Option<u8> {
+        match byte {
+            b'A'..=b'Z' => Some(byte - b'A'),
+            b'a'..=b'z' => Some(byte - b'a' + 26),
+            b'0'..=b'9' => Some(byte - b'0' + 52),
+            b'-' => Some(62),
+            b'_' => Some(63),
+            _ => None,
+        }
+    }
+    let bytes = value.as_bytes();
+    if bytes.len() % 4 == 1 {
+        return Err(invalid_base64());
+    }
+    let mut out = Vec::with_capacity(bytes.len() * 3 / 4);
+    let mut i = 0;
+    while i < bytes.len() {
+        let a = decode(bytes[i]).ok_or_else(invalid_base64)?;
+        let b = decode(*bytes.get(i + 1).ok_or_else(invalid_base64)?).ok_or_else(invalid_base64)?;
+        let c = bytes.get(i + 2).and_then(|byte| decode(*byte));
+        let d = bytes.get(i + 3).and_then(|byte| decode(*byte));
+        out.push((a << 2) | (b >> 4));
+        if let Some(c) = c {
+            out.push((b << 4) | (c >> 2));
+            if let Some(d) = d {
+                out.push((c << 6) | d);
+            }
+        }
+        i += 4;
+    }
+    Ok(out)
+}
+
+fn invalid_base64() -> MusicDhtError {
+    MusicDhtError::Protocol("invalid base64url Jam invite".to_string())
+}
+
+fn protocol_err(err: impl std::fmt::Display) -> MusicDhtError {
+    MusicDhtError::Protocol(err.to_string())
+}
+
+fn network_err(err: impl std::fmt::Display) -> MusicDhtError {
+    MusicDhtError::Network(err.to_string())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn invite() -> JamInvite {
+        JamInvite {
+            v: JAM_PROTOCOL_VERSION,
+            ticket: "endpoint-ticket".to_string(),
+            jam_id: "jam_123".to_string(),
+            secret: "0123456789abcdef0123456789abcdef".to_string(),
+            host_device_id: "dev_host".to_string(),
+            host_name: "Living room".to_string(),
+        }
+    }
+
+    #[test]
+    fn jam_invite_round_trips() {
+        let invite = invite();
+        let uri = invite.to_uri().unwrap();
+        assert!(uri.starts_with("frid://j/"));
+        assert_eq!(JamInvite::from_uri(&uri).unwrap(), invite);
+    }
+
+    #[test]
+    fn personal_device_invites_are_not_jam_capabilities() {
+        assert!(JamInvite::from_uri("frid://i/abcd").is_err());
+    }
+
+    #[test]
+    fn weak_capability_is_rejected() {
+        let mut invite = invite();
+        invite.secret = "short".to_string();
+        assert!(invite.to_uri().is_err());
+    }
+}

crates/music-dht/src/lib.rs

@@ -76,12 +76,14 @@
 #![warn(missing_docs)]
 #![forbid(unsafe_code)]
 
+pub mod capabilities;
 pub mod catalog;
 mod config;
 mod database;
 pub mod device_sync;
 mod dht;
 mod error;
+pub mod jam;
 mod message;
 mod node;
 mod normalization;

crates/music-dht/src/service.rs

@@ -28,7 +28,7 @@ use crate::routing::{NodeContact, NodeId};
 ///
 /// The historical value is retained because changing it would partition
 /// existing deployments.
-pub const SCHEMA_NAME: &str = "music-dht-poc-v3";
+pub const SCHEMA_NAME: &str = "music-dht-v5";
 
 /// Capacity of the application event channel.
 const EVENT_CHANNEL_CAPACITY: usize = 256;
@@ -285,6 +285,9 @@ impl MusicDhtService {
         for alpn in &config.stream_protocols {
             engine_builder = engine_builder.stream_protocol(alpn.clone());
         }
+        for alpn in &config.schema_independent_stream_protocols {
+            engine_builder = engine_builder.schema_independent_stream_protocol(alpn.clone());
+        }
         let engine_config = engine_builder
             .build()
             .map_err(|err| MusicDhtError::Network(err.to_string()))?;