feat: make plugin-only Serve mode nodes discoverable/reachable by other mesh peers

crates/mesh-llm-host-runtime/src/mesh/host_role_claims.rs

@@ -0,0 +1,98 @@
+use super::node::Node;
+use super::peer_state::NodeRole;
+use std::collections::BTreeMap;
+
+#[derive(Debug, Clone, Copy, Eq, Hash, Ord, PartialEq, PartialOrd)]
+pub(crate) enum HostRoleClaim {
+    LocalModel,
+    PluginInference,
+}
+
+#[derive(Default)]
+pub(crate) struct HostRoleClaims(BTreeMap<HostRoleClaim, usize>);
+
+impl HostRoleClaims {
+    fn claim(&mut self, claim: HostRoleClaim) {
+        *self.0.entry(claim).or_default() += 1;
+    }
+
+    fn release(&mut self, claim: HostRoleClaim) -> bool {
+        let Some(count) = self.0.get_mut(&claim) else {
+            return false;
+        };
+        if *count > 1 {
+            *count -= 1;
+        } else {
+            self.0.remove(&claim);
+        }
+        true
+    }
+
+    fn is_empty(&self) -> bool {
+        self.0.is_empty()
+    }
+}
+
+impl Node {
+    pub async fn claim_host_role(&self, claim: HostRoleClaim, http_port: u16) {
+        let transitioned = {
+            let mut claims = self.host_role_claims.lock().await;
+            claims.claim(claim);
+            let mut role = self.role.lock().await;
+            if matches!(*role, NodeRole::Worker) {
+                *role = NodeRole::Host { http_port };
+                true
+            } else {
+                false
+            }
+        };
+        if transitioned {
+            self.regossip().await;
+        }
+    }
+
+    pub async fn release_host_role(&self, claim: HostRoleClaim) {
+        let transitioned = {
+            let mut claims = self.host_role_claims.lock().await;
+            if !claims.release(claim) || !claims.is_empty() {
+                false
+            } else {
+                let mut role = self.role.lock().await;
+                if matches!(*role, NodeRole::Host { .. }) {
+                    *role = NodeRole::Worker;
+                    true
+                } else {
+                    false
+                }
+            }
+        };
+        if transitioned {
+            self.regossip().await;
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[tokio::test]
+    async fn host_role_claims_are_reference_counted_across_sources() {
+        let node = Node::new_for_tests(NodeRole::Worker).await.unwrap();
+
+        node.claim_host_role(HostRoleClaim::LocalModel, 9337).await;
+        node.claim_host_role(HostRoleClaim::PluginInference, 9337)
+            .await;
+        assert_eq!(node.role().await, NodeRole::Host { http_port: 9337 });
+
+        node.release_host_role(HostRoleClaim::PluginInference).await;
+        assert_eq!(node.role().await, NodeRole::Host { http_port: 9337 });
+
+        node.claim_host_role(HostRoleClaim::LocalModel, 9337).await;
+        node.release_host_role(HostRoleClaim::LocalModel).await;
+        assert_eq!(node.role().await, NodeRole::Host { http_port: 9337 });
+
+        node.release_host_role(HostRoleClaim::LocalModel).await;
+        assert_eq!(node.role().await, NodeRole::Worker);
+    }
+}

crates/mesh-llm-host-runtime/src/mesh/mod.rs

@@ -85,6 +85,7 @@ mod connections;
 mod direct_path;
 mod gossip;
 mod heartbeat;
+mod host_role_claims;
 mod identity_persistence;
 mod lan_bootstrap;
 mod model_identity;
@@ -108,6 +109,7 @@ mod stun;
 
 use connection_reservation::*;
 use connections::*;
+pub(crate) use host_role_claims::{HostRoleClaim, HostRoleClaims};
 use model_identity::*;
 use node_identity::*;
 use operational_logging::{MeshOperationalEvent, record_mesh_operational_event};

crates/mesh-llm-host-runtime/src/mesh/node.rs

@@ -40,6 +40,7 @@ pub struct Node {
     pub(crate) local_mesh_requirements: crate::MeshRequirements,
     pub(crate) state: Arc<Mutex<MeshState>>,
     pub(crate) role: Arc<Mutex<NodeRole>>,
+    pub(crate) host_role_claims: Arc<Mutex<HostRoleClaims>>,
     pub(crate) models: Arc<Mutex<Vec<String>>>,
     pub(crate) model_source: Arc<Mutex<Option<String>>>,
     pub(crate) serving_models: Arc<Mutex<Vec<String>>>,
@@ -775,6 +776,7 @@ impl Node {
                 recent_mesh_rejections: VecDeque::new(),
             })),
             role: Arc::new(Mutex::new(role)),
+            host_role_claims: Arc::new(Mutex::new(HostRoleClaims::default())),
             models: Arc::new(Mutex::new(Vec::new())),
             model_source: Arc::new(Mutex::new(None)),
             serving_models: Arc::new(Mutex::new(Vec::new())),
@@ -950,6 +952,7 @@ impl Node {
                 recent_mesh_rejections: VecDeque::new(),
             })),
             role: Arc::new(Mutex::new(role)),
+            host_role_claims: Arc::new(Mutex::new(HostRoleClaims::default())),
             models: Arc::new(Mutex::new(Vec::new())),
             model_source: Arc::new(Mutex::new(None)),
             serving_models: Arc::new(Mutex::new(Vec::new())),
@@ -1154,6 +1157,7 @@ impl Node {
         self.role.lock().await.clone()
     }
 
+    #[cfg(test)]
     pub async fn set_role(&self, role: NodeRole) {
         *self.role.lock().await = role;
     }

crates/mesh-llm-host-runtime/src/mesh/tests/connections.rs

@@ -418,6 +418,7 @@ async fn make_test_node_with_requirements(
             recent_mesh_rejections: VecDeque::new(),
         })),
         role: Arc::new(Mutex::new(role)),
+        host_role_claims: Arc::new(Mutex::new(HostRoleClaims::default())),
         models: Arc::new(Mutex::new(Vec::new())),
         model_source: Arc::new(Mutex::new(None)),
         serving_models: Arc::new(Mutex::new(Vec::new())),
@@ -505,6 +506,65 @@ async fn make_test_node_with_requirements(
     Ok(node)
 }
 
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn host_role_claim_transitions_regossip_to_connected_peer() -> Result<()> {
+    let host = make_test_node(super::NodeRole::Worker).await?;
+    let peer = make_test_node(super::NodeRole::Worker).await?;
+    host.set_mesh_id("host-role-claim-regossip-test".to_string())
+        .await;
+    peer.set_mesh_id("host-role-claim-regossip-test".to_string())
+        .await;
+    host.start_accepting();
+    peer.start_accepting();
+
+    let host_id = host.id();
+    peer.join(&host.invite_token().await).await?;
+    wait_for_peer(&peer, host_id).await;
+    wait_for_peer(&host, peer.id()).await;
+
+    host.claim_host_role(super::HostRoleClaim::PluginInference, 9337)
+        .await;
+    tokio::time::timeout(std::time::Duration::from_secs(5), async {
+        loop {
+            if peer
+                .peers()
+                .await
+                .into_iter()
+                .find(|candidate| candidate.id == host_id)
+                .is_some_and(|candidate| {
+                    candidate.role == super::NodeRole::Host { http_port: 9337 }
+                })
+            {
+                break;
+            }
+            tokio::time::sleep(std::time::Duration::from_millis(25)).await;
+        }
+    })
+    .await
+    .expect("connected peer should receive host promotion gossip");
+
+    host.release_host_role(super::HostRoleClaim::PluginInference)
+        .await;
+    tokio::time::timeout(std::time::Duration::from_secs(5), async {
+        loop {
+            if peer
+                .peers()
+                .await
+                .into_iter()
+                .find(|candidate| candidate.id == host_id)
+                .is_some_and(|candidate| candidate.role == super::NodeRole::Worker)
+            {
+                break;
+            }
+            tokio::time::sleep(std::time::Duration::from_millis(25)).await;
+        }
+    })
+    .await
+    .expect("connected peer should receive host demotion gossip");
+
+    Ok(())
+}
+
 #[tokio::test]
 async fn set_serving_models_preserves_existing_known_descriptor_capabilities_when_adding_model()
 -> Result<()> {

crates/mesh-llm-host-runtime/src/plugin/transport.rs

@@ -27,6 +27,8 @@ use tokio::sync::{Mutex, mpsc, oneshot};
 const PLUGIN_ENVELOPE_PREFIX_READ_TIMEOUT: Duration =
     Duration::from_secs((super::health::HEALTH_CHECK_INTERVAL_SECS * 4) + 30);
 const PLUGIN_ENVELOPE_BODY_READ_TIMEOUT: Duration = Duration::from_secs(30);
+#[cfg(test)]
+const PLUGIN_ENVELOPE_TEST_READ_TIMEOUT: Duration = Duration::from_secs(1);
 const PLUGIN_MESH_STREAM_RESPONSE_TIMEOUT: Duration =
     Duration::from_secs(super::REQUEST_TIMEOUT_SECS);
 
@@ -419,31 +421,41 @@ pub(crate) async fn read_envelope(stream: &mut LocalStream) -> Result<super::pro
 
 #[cfg(test)]
 pub(crate) async fn read_envelope_from_reader<R>(stream: &mut R) -> Result<super::proto::Envelope>
+where
+    R: tokio::io::AsyncRead + Unpin,
+{
+    read_envelope_from_reader_with_timeouts(
+        stream,
+        PLUGIN_ENVELOPE_TEST_READ_TIMEOUT,
+        PLUGIN_ENVELOPE_TEST_READ_TIMEOUT,
+    )
+    .await
+}
+
+#[cfg(test)]
+async fn read_envelope_from_reader_with_timeouts<R>(
+    stream: &mut R,
+    prefix_timeout: Duration,
+    body_timeout: Duration,
+) -> Result<super::proto::Envelope>
 where
     R: tokio::io::AsyncRead + Unpin,
 {
     let mut len_buf = [0u8; 4];
     tokio::time::timeout(
-        PLUGIN_ENVELOPE_PREFIX_READ_TIMEOUT,
+        prefix_timeout,
         AsyncReadExt::read_exact(stream, &mut len_buf),
     )
     .await
-    .map_err(|_| {
-        anyhow!("timeout reading plugin frame prefix after {PLUGIN_ENVELOPE_PREFIX_READ_TIMEOUT:?}")
-    })??;
+    .map_err(|_| anyhow!("timeout reading plugin frame prefix after {prefix_timeout:?}"))??;
     let len = u32::from_le_bytes(len_buf) as usize;
     if len > 16 * 1024 * 1024 {
         bail!("Plugin frame too large");
     }
     let mut body = vec![0u8; len];
-    tokio::time::timeout(
-        PLUGIN_ENVELOPE_BODY_READ_TIMEOUT,
-        AsyncReadExt::read_exact(stream, &mut body),
-    )
-    .await
-    .map_err(|_| {
-        anyhow!("timeout reading plugin frame body after {PLUGIN_ENVELOPE_BODY_READ_TIMEOUT:?}")
-    })??;
+    tokio::time::timeout(body_timeout, AsyncReadExt::read_exact(stream, &mut body))
+        .await
+        .map_err(|_| anyhow!("timeout reading plugin frame body after {body_timeout:?}"))??;
     Ok(prost::Message::decode(body.as_slice())?)
 }
 

crates/mesh-llm-host-runtime/src/runtime/mod.rs

@@ -18,6 +18,7 @@ mod model_lifecycle;
 pub(crate) mod model_reconciliation;
 mod operational_logging;
 mod options;
+mod plugin_host_role;
 mod proxy;
 mod publication;
 mod release_attestation;

crates/mesh-llm-host-runtime/src/runtime/plugin_host_role.rs

@@ -0,0 +1,35 @@
+use crate::{mesh, plugin};
+use std::time::Duration;
+
+const WATCH_INTERVAL: Duration = Duration::from_secs(5);
+
+/// Promotes this node to `NodeRole::Host` whenever a loaded plugin is
+/// advertising at least one inference model, and releases the plugin's host
+/// claim when that stops being true.
+pub(super) fn spawn(node: mesh::Node, plugin_manager: plugin::PluginManager, http_port: u16) {
+    tokio::spawn(async move {
+        let mut plugin_claimed = false;
+        loop {
+            tokio::time::sleep(WATCH_INTERVAL).await;
+            let has_plugin_models = match plugin_manager.inference_models().await {
+                Ok(models) => !models.is_empty(),
+                Err(error) => {
+                    tracing::warn!(
+                        %error,
+                        "plugin host-role watcher: failed to read plugin inference models, skipping this tick"
+                    );
+                    continue;
+                }
+            };
+            if has_plugin_models && !plugin_claimed {
+                node.claim_host_role(mesh::HostRoleClaim::PluginInference, http_port)
+                    .await;
+                plugin_claimed = true;
+            } else if !has_plugin_models && plugin_claimed {
+                node.release_host_role(mesh::HostRoleClaim::PluginInference)
+                    .await;
+                plugin_claimed = false;
+            }
+        }
+    });
+}

crates/mesh-llm-host-runtime/src/runtime/run_auto.rs

@@ -1,4 +1,5 @@
 use super::daemon_startup::{check_mode_conflicts, resolve_effective_mode};
+use super::plugin_host_role;
 use super::startup_identity::{emit_private_mesh_name_warning, handle_public_identity_transition};
 use super::status::mesh_guardrail_mode_to_openai;
 use super::{
@@ -1351,6 +1352,33 @@ pub(super) async fn run_auto(ctx: RunAutoContext) -> Result<()> {
 
     let tunnel_mgr =
         tunnel::Manager::start(node.clone(), channels.rpc, channels.http, channels.stage).await?;
+    // Both halves of inbound reachability are established here for any node
+    // that can serve, rather than only as a side effect of a local model
+    // finishing load.
+    //
+    // `set_http_port` is what lets a plugin-only node (no local model ever
+    // loads) accept inbound requests at all: the api proxy it points at is
+    // already bound and already answers correctly with no models loaded, so a
+    // tunneled request arriving before any model is ready gets a normal "not
+    // available" response instead of being silently dropped (the previous
+    // behavior whenever this was still 0 — see `network/tunnel.rs`'s
+    // `port == 0` early-return). The three call sites in `startup_handles.rs`
+    // remain and are now redundant-but-harmless — same node, same `api_port`,
+    // for the lifetime of the process.
+    //
+    // `plugin_host_role::spawn` is the other half: whether peers actually
+    // route here.
+    //
+    // Both are gated on `!is_client`. A client node has no compute to offer
+    // and never advertises `Host`, so nothing selects it as a route target;
+    // leaving its inbound HTTP tunnel terminated at the `port == 0` check
+    // keeps it exactly as reachable as it was before this change — not at
+    // all — instead of turning it into a mesh-internal request relay for any
+    // admitted peer that dials it.
+    if !is_client {
+        tunnel_mgr.set_http_port(api_port);
+        plugin_host_role::spawn(node.clone(), plugin_manager.clone(), api_port);
+    }
 
     // Election publishes per-model targets
     let (target_tx, target_rx) = tokio::sync::watch::channel(election::ModelTargets::default());