Agree a protocol by name and wire version

A peer's announcement already carried a protocol and a version; only the
name was being checked. Now both are, and a peer offering a protocol at a
version this build does not speak simply has no data plane — the control
plane keeps working, messages and signed state still flow, and the
difference is reported once rather than on every announcement.

The version compared is the *wire* version, not the software version, and
the trait says so: a plugin crate has its own version and it is nobody
else's business. Two peers on different releases work together for as
long as the bytes between them have not changed, and nothing in the
negotiation may be derived from anything that moves with a release. A
test pins that agreement turns on the name and version alone, with a
peer whose announcement carries a payload this build has never seen.

`PeerStatus` gained the protocols agreed with each peer, so an empty list
is visible as what it is: a session that is up, carrying control traffic,
with no protocol in common.

The forging test plugin was announcing version 1 while claiming to speak
the current one, so its payload was being set aside for the wrong reason.

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

crates/tsunagi/src/agent/network.rs

@@ -210,6 +210,9 @@ struct Runtime {
     opening: HashSet<(EndpointId, String)>,
     link_results_tx: mpsc::Sender<LinkOutcome>,
     link_results_rx: mpsc::Receiver<LinkOutcome>,
+    /// Peers already told about a protocol version that cannot match, so it
+    /// is said once rather than on every announcement.
+    reported_mismatch: HashSet<(EndpointId, String)>,
     /// Signed records, merged from every replica we have talked to.
     state: StateSet,
     /// Snapshots received while dispatching, handled on the next loop pass.
@@ -242,6 +245,7 @@ impl Runtime {
             opening: HashSet::new(),
             link_results_tx,
             link_results_rx,
+            reported_mismatch: HashSet::new(),
             state: StateSet::new(),
             pending_state: Vec::new(),
             own_version: 0,
@@ -887,12 +891,12 @@ impl Runtime {
 
     // ------------------------------------------------------------ data plane
 
-    /// Protocol ids this agent has a plugin for.
+    /// The protocols this agent speaks, with the version it speaks them at.
-    fn served_protocols(&self) -> Vec<String> {
+    fn served_protocols(&self) -> Vec<(String, u16)> {
         self.params
             .plugins
             .iter()
-            .map(|plugin| plugin.protocol_id().to_string())
+            .map(|plugin| (plugin.protocol_id().to_string(), plugin.protocol_version()))
             .collect()
     }
 
@@ -925,6 +929,10 @@ impl Runtime {
             });
         }
 
+        // Both the name and the version have to match. A peer offering
+        // `wg-quic` at a version this build does not speak is not a peer to
+        // carry traffic with, and a link opened anyway would fail later and
+        // say less about why.
         let wanted: Vec<(EndpointId, String)> = self
             .sessions
             .values()
@@ -934,9 +942,14 @@ impl Runtime {
                     .capabilities
                     .iter()
                     .filter(|capability| capability.enabled)
-                    .map(move |capability| (peer, capability.protocol.clone()))
+                    .map(move |capability| (peer, capability.protocol.clone(), capability.version))
             })
-            .filter(|(_, protocol)| served.contains(protocol))
+            .filter(|(_, protocol, version)| {
+                served
+                    .iter()
+                    .any(|(name, ours)| name == protocol && ours == version)
+            })
+            .map(|(peer, protocol, _)| (peer, protocol))
             .collect();
 
         for (peer, protocol) in wanted {
@@ -1271,6 +1284,26 @@ impl Runtime {
                 if plugin.protocol_id() != capability.protocol {
                     continue;
                 }
+                if plugin.protocol_version() != capability.version {
+                    // Said once per peer and protocol. A version that will
+                    // not match this build will not match it on the next
+                    // announcement either, and repeating it every round
+                    // would bury everything else.
+                    let key = (peer, capability.protocol.clone());
+                    if self.reported_mismatch.insert(key) {
+                        errors.push((
+                            capability.protocol.clone(),
+                            format!(
+                                "peer speaks {} version {} and this build speaks {}, so there \
+                                 is no data plane with it; the control plane is unaffected",
+                                capability.protocol,
+                                capability.version,
+                                plugin.protocol_version()
+                            ),
+                        ));
+                    }
+                    continue;
+                }
                 // The core hands the opaque payload over without interpreting it.
                 if let Err(err) = plugin.on_peer_capability(self.network_id, peer, capability) {
                     errors.push((plugin.protocol_id().to_string(), err.to_string()));
@@ -1290,6 +1323,7 @@ impl Runtime {
     // ----------------------------------------------------------------- status
 
     fn status(&self) -> NetworkStatus {
+        let served = self.served_protocols();
         let mut peers: Vec<PeerStatus> = self
             .sessions
             .values()
@@ -1299,6 +1333,17 @@ impl Runtime {
                     endpoint_id: session.peer,
                     role: session.role,
                     hostname: session.hostname.clone(),
+                    protocols: session
+                        .capabilities
+                        .iter()
+                        .filter(|capability| capability.enabled)
+                        .filter(|capability| {
+                            served.iter().any(|(name, ours)| {
+                                *name == capability.protocol && *ours == capability.version
+                            })
+                        })
+                        .map(|capability| capability.protocol.clone())
+                        .collect(),
                     capabilities: session.capabilities.clone(),
                     connected_for: session.established.elapsed(),
                     paths: snapshot.paths,

crates/tsunagi/src/agent/status.rs

@@ -83,6 +83,12 @@ pub struct PeerStatus {
     ///
     /// A mutable binding, not an identity.
     pub hostname: Option<String>,
+    /// Protocols agreed with this peer: both sides have them at the same
+    /// version.
+    ///
+    /// Empty means no data plane with that peer, which leaves the control
+    /// plane working — messages and state still flow.
+    pub protocols: Vec<String>,
     /// Capabilities the peer announced. Payloads stay opaque.
     pub capabilities: Vec<PluginCapability>,
     /// How long the session has been up.

crates/tsunagi/src/dataplane/mod.rs

@@ -84,6 +84,19 @@ impl PacketSink for DiscardPackets {
     }
 }
 
+/// One setting a protocol accepts.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct ProtocolOption {
+    /// The key, as written in `key=value`.
+    pub key: &'static str,
+    /// A word for the value, for the help line: `BYTES`, `NAME`.
+    pub value: &'static str,
+    /// What it does.
+    pub help: &'static str,
+    /// What happens when it is not given.
+    pub default: Option<&'static str>,
+}
+
 /// Errors a plugin may return. They are recorded, never fatal for the agent.
 #[derive(Debug, thiserror::Error)]
 #[non_exhaustive]
@@ -241,6 +254,29 @@ pub trait IpPlugin: Send + Sync + std::fmt::Debug + 'static {
     /// Must be non-empty and at most [`MAX_PROTOCOL_ID_LEN`] bytes.
     fn protocol_id(&self) -> &str;
 
+    /// The wire version of this protocol.
+    ///
+    /// Two peers carry traffic for each other only when they have the same
+    /// protocol at the same version. There is no negotiating a middle
+    /// ground: a protocol either speaks the same words at both ends or it
+    /// does not, and pretending otherwise produces a link that fails later
+    /// and less clearly.
+    ///
+    /// **Not the software version.** A plugin crate has its own version and
+    /// it is nobody else's business: two peers on different builds work
+    /// together for as long as the wire between them has not changed. This
+    /// number moves only when the bytes do, so it must never be derived from
+    /// `CARGO_PKG_VERSION` or anything else that moves with a release.
+    fn protocol_version(&self) -> u16;
+
+    /// The settings this protocol accepts, and what they mean.
+    ///
+    /// Declared rather than documented elsewhere, so the agent can list them
+    /// without knowing anything about the protocol.
+    fn options(&self) -> &'static [ProtocolOption] {
+        &[]
+    }
+
     /// Called once, when the agent starts, before any network is activated.
     ///
     /// The plugin keeps the context to ask for re-announcements and to report
@@ -370,6 +406,10 @@ impl IpPlugin for TestCapabilityPlugin {
         &self.protocol
     }
 
+    fn protocol_version(&self) -> u16 {
+        1
+    }
+
     fn local_capability(
         &self,
         _network: NetworkId,

crates/tsunagi/src/dataplane/wireguard/mod.rs

@@ -42,7 +42,7 @@ pub mod plugin;
 pub mod store;
 
 pub use crate::state::Ipv4Range;
-pub use announcement::{ValidatedAnnouncement, WgAnnouncement};
+pub use announcement::{ANNOUNCEMENT_VERSION, ValidatedAnnouncement, WgAnnouncement};
 // The interface, its addresses and how it is created belong to the system
 // level now: one agent has one interface, and no protocol owns it. Re-exported
 // here while callers are moved over.

crates/tsunagi/src/dataplane/wireguard/plugin.rs

@@ -244,6 +244,15 @@ pub struct WireguardPlugin {
 }
 
 impl WireguardPlugin {
+    /// The settings this protocol accepts.
+    pub const OPTIONS: &'static [crate::dataplane::ProtocolOption] =
+        &[crate::dataplane::ProtocolOption {
+            key: "keepalive",
+            value: "SECONDS",
+            help: "persistent keepalive interval; 0 turns it off",
+            default: Some("25"),
+        }];
+
     /// Opens the plugin's key store and starts its reconciliation task.
     ///
     /// Must be called from inside a tokio runtime; the plugin starts no
@@ -658,6 +667,14 @@ impl IpPlugin for WireguardPlugin {
         WIREGUARD_PROTOCOL
     }
 
+    fn protocol_version(&self) -> u16 {
+        super::announcement::ANNOUNCEMENT_VERSION
+    }
+
+    fn options(&self) -> &'static [crate::dataplane::ProtocolOption] {
+        Self::OPTIONS
+    }
+
     fn attach(&self, context: PluginContext) {
         if let Some(local) = context.local_endpoint_id() {
             let _ = self.worker.local_id.set(local);

crates/tsunagi/tests/wireguard.rs

@@ -942,6 +942,155 @@ async fn an_announcement_for_another_network_never_reaches_a_tunnel() {
     drop(attacker_dir);
 }
 
+#[tokio::test]
+async fn a_peer_at_another_protocol_version_gets_no_data_plane_and_keeps_the_control_plane() {
+    // Both sides must have the protocol at the same version. There is no
+    // middle ground to negotiate: either the words mean the same thing at
+    // both ends or they do not.
+    let discovery = SharedMemoryDiscovery::new();
+    let (name, secret) = network("wg-version");
+
+    let here = WgAgent::spawn(&discovery, "tvh").await;
+    let network_id = here.agent.join_network(&name, &secret).await.unwrap();
+
+    let ahead = Arc::new(FromTheFuture);
+    let dir = TempDir::new().unwrap();
+    let other = Agent::spawn(
+        config_with(dir.path(), &discovery).with_plugin(ahead.clone() as Arc<dyn IpPlugin>),
+    )
+    .await
+    .unwrap();
+
+    let mut events = here.agent.subscribe();
+    other.join_network(&name, &secret).await.unwrap();
+    wait_for_peers(&here.agent, network_id, 1).await;
+
+    let reason = wait_event(&mut events, |event| match event {
+        Event::PluginError { reason, .. } if reason.contains("version") => Some(reason.clone()),
+        _ => None,
+    })
+    .await;
+    assert!(
+        reason.contains("control plane is unaffected"),
+        "the message should say what still works: {reason}"
+    );
+
+    // No tunnel, and no attempt at one.
+    settle().await;
+    assert!(
+        here.plugin.overview(network_id).unwrap().peers.is_empty(),
+        "a version that cannot match must not become a tunnel"
+    );
+    let status = here.agent.network_status(network_id).await.unwrap();
+    assert_eq!(status.peers.len(), 1, "the session is up all the same");
+    assert!(
+        status.peers[0].protocols.is_empty(),
+        "nothing was agreed with it: {:?}",
+        status.peers[0].protocols
+    );
+
+    // And the control plane carries a message to it regardless.
+    assert_eq!(
+        here.agent
+            .broadcast(
+                network_id,
+                tsunagi::proto::ControlMessage::Ping {
+                    seq: 1,
+                    payload: b"still talking".to_vec(),
+                },
+            )
+            .await
+            .unwrap(),
+        1
+    );
+
+    other.shutdown().await;
+    here.shutdown().await;
+    drop(dir);
+}
+
+#[tokio::test]
+async fn agreement_turns_on_the_protocol_version_and_nothing_else() {
+    // A different build is not a different protocol. What is compared is the
+    // wire version and the name; everything else about a peer — its release,
+    // and whatever opaque payload its announcement carries — has no say.
+    let discovery = SharedMemoryDiscovery::new();
+    let (name, secret) = network("wg-same-wire");
+
+    let here = WgAgent::spawn(&discovery, "tsw").await;
+    let network_id = here.agent.join_network(&name, &secret).await.unwrap();
+
+    // Announces the right protocol at the right version, with a payload
+    // this build has never seen.
+    let stranger = Arc::new(ForgingPlugin {
+        payload: std::sync::Mutex::new(Some(b"from some other release".to_vec())),
+    });
+    let dir = TempDir::new().unwrap();
+    let other = Agent::spawn(
+        config_with(dir.path(), &discovery).with_plugin(stranger.clone() as Arc<dyn IpPlugin>),
+    )
+    .await
+    .unwrap();
+    other.join_network(&name, &secret).await.unwrap();
+    wait_for_peers(&here.agent, network_id, 1).await;
+
+    // Agreed, because the wire matches. The payload is rejected by the
+    // protocol afterwards, which is a separate matter and says nothing about
+    // whether the two agreed to talk.
+    wait_until("the protocol is agreed with it", || async {
+        let status = here.agent.network_status(network_id).await.ok()?;
+        status
+            .peers
+            .first()?
+            .protocols
+            .contains(&WIREGUARD_PROTOCOL.to_string())
+            .then_some(())
+    })
+    .await;
+
+    other.shutdown().await;
+    here.shutdown().await;
+    drop(dir);
+}
+
+/// A plugin claiming the same protocol at a version this build does not speak.
+#[derive(Debug)]
+struct FromTheFuture;
+
+impl IpPlugin for FromTheFuture {
+    fn protocol_id(&self) -> &str {
+        WIREGUARD_PROTOCOL
+    }
+
+    fn protocol_version(&self) -> u16 {
+        tsunagi::dataplane::wireguard::ANNOUNCEMENT_VERSION + 1
+    }
+
+    fn local_capability(
+        &self,
+        _network: NetworkId,
+    ) -> Result<Option<tsunagi::dataplane::PluginCapability>, tsunagi::dataplane::PluginError> {
+        Ok(Some(tsunagi::dataplane::PluginCapability {
+            protocol: WIREGUARD_PROTOCOL.to_string(),
+            version: self.protocol_version(),
+            enabled: true,
+            data: Vec::new(),
+        }))
+    }
+
+    fn on_peer_capability(
+        &self,
+        _network: NetworkId,
+        _peer: EndpointId,
+        _capability: &tsunagi::dataplane::PluginCapability,
+    ) -> Result<(), tsunagi::dataplane::PluginError> {
+        Ok(())
+    }
+
+    fn on_peer_gone(&self, _network: NetworkId, _peer: EndpointId) {}
+    fn on_network_deactivated(&self, _network: NetworkId) {}
+}
+
 /// A plugin that announces whatever bytes it is told to, under the WireGuard
 /// protocol id. Used to test what a hostile member can do.
 #[derive(Debug)]
@@ -954,6 +1103,12 @@ impl IpPlugin for ForgingPlugin {
         WIREGUARD_PROTOCOL
     }
 
+    /// The same version, so the announcement is looked at rather than
+    /// dismissed for the wrong reason.
+    fn protocol_version(&self) -> u16 {
+        tsunagi::dataplane::wireguard::ANNOUNCEMENT_VERSION
+    }
+
     fn local_capability(
         &self,
         _network: NetworkId,
@@ -964,7 +1119,9 @@ impl IpPlugin for ForgingPlugin {
         };
         Ok(payload.map(|data| tsunagi::dataplane::PluginCapability {
             protocol: WIREGUARD_PROTOCOL.to_string(),
-            version: 1,
+            // The version it actually speaks, so the payload is examined
+            // rather than set aside for the wrong reason.
+            version: tsunagi::dataplane::wireguard::ANNOUNCEMENT_VERSION,
             enabled: true,
             data,
         }))