feat(connection): add non-replacing interface export

crates/tinybus/Cargo.toml

@@ -138,6 +138,11 @@ name = "module_host_callback"
 crate-type = ["cdylib"]
 required-features = ["modules"]
 
+[[example]]
+name = "module_host_stream_callback"
+crate-type = ["cdylib"]
+required-features = ["modules"]
+
 [[example]]
 name = "module_controlled_init"
 crate-type = ["cdylib"]

crates/tinybus/examples/module_host_stream_callback.rs

@@ -0,0 +1,94 @@
+//! Actual eager native fixture exercising confidential callbacks and the
+//! separately declared ordinary stream callback capability.
+use async_trait::async_trait;
+use serde::Deserialize;
+use serde_json::Value;
+use tinybus::{
+    BusName, Connection, Error, Interface, InterfaceName, MemberName, Result, StreamDescriptor,
+    StreamRef,
+};
+
+#[derive(Deserialize)]
+struct Config {
+    host: BusName,
+}
+
+struct CallbackService(Connection, BusName);
+
+#[async_trait]
+impl Interface for CallbackService {
+    fn name(&self) -> InterfaceName {
+        "org.example.EagerCallback".parse().unwrap()
+    }
+    fn members(&self) -> Vec<MemberName> {
+        vec!["RoundTrip".parse().unwrap()]
+    }
+    async fn call(&self, member: &MemberName, args: Value) -> Result<Value> {
+        if member.as_str() != "RoundTrip" || args != serde_json::json!([]) {
+            return Err(Error::failed("unknown callback fixture request"));
+        }
+        let callback = self
+            .0
+            .proxy(self.1.as_str(), "/callback", "org.example.Callback")?;
+
+        // Ordinary traffic cannot enter a confidential-only member.
+        if callback.call::<Value>("Observe", ()).await.is_ok() {
+            return Err(Error::failed("ordinary confidential callback was accepted"));
+        }
+        if callback.call::<Value>("Undeclared", ()).await.is_ok() {
+            return Err(Error::failed("undeclared callback member was accepted"));
+        }
+
+        let mut outgoing = self
+            .0
+            .open_stream(&self.1, StreamDescriptor::with_len(12))
+            .await?;
+        let outgoing_ref = outgoing.stream_ref();
+        let write_outgoing = tokio::spawn(async move {
+            outgoing.write(b"module->host").await?;
+            outgoing.finish().await
+        });
+        let incoming_ref: StreamRef = callback.call("Transfer", (outgoing_ref,)).await?;
+        write_outgoing
+            .await
+            .map_err(|_| Error::failed("module stream writer task failed"))??;
+        let mut incoming = self.0.accept_stream_from(&incoming_ref, &self.1)?;
+        if incoming.read_to_end_capped(12).await? != b"host->module" {
+            return Err(Error::failed("host callback stream payload did not match"));
+        }
+
+        callback.call_confidential::<Value>("Observe", ()).await?;
+        if callback
+            .call::<Value>(
+                "Transfer",
+                (StreamRef {
+                    id: "undeclared-test-handle".into(),
+                    content_type: None,
+                    len: Some(0),
+                },),
+            )
+            .await
+            .is_ok()
+        {
+            return Err(Error::failed("revoked stream callback was accepted"));
+        }
+        Ok(Value::Null)
+    }
+}
+
+async fn setup(connection: Connection, config: Config) -> Result<()> {
+    connection
+        .serve_at(
+            "/module".parse()?,
+            CallbackService(connection.clone(), config.host),
+        )
+        .await?;
+    connection.request_name("org.example.EagerCallback").await
+}
+
+tinybus_module::module_export! {
+    setup = setup, config = Config,
+    worker_threads = 1,
+    provides = ["org.example.EagerCallback"],
+    methods = ["RoundTrip"], signals = [], requires = [], optional = [], lazy = false,
+}

crates/tinybus/src/connection.rs

@@ -456,6 +456,28 @@ impl Connection {
         Ok(())
     }
 
+    /// Export `interface` at `path` only if no interface is already served
+    /// there.
+    ///
+    /// Registration is atomic with respect to other calls to this method: a
+    /// collision leaves every existing interface at that path untouched. Use
+    /// [`Connection::serve_at`] when replacement is intentional.
+    pub async fn try_serve_at(&self, path: ObjectPath, interface: impl Interface) -> Result<()> {
+        if self
+            .inner
+            .objects
+            .write()
+            .await
+            .insert_if_absent(path, Arc::new(interface))
+        {
+            Ok(())
+        } else {
+            Err(Error::failed(
+                "interface is already served at this object path",
+            ))
+        }
+    }
+
     /// Stop exporting everything at `path`.
     pub async fn unserve(&self, path: &ObjectPath) -> bool {
         self.inner.objects.write().await.remove(path)

crates/tinybus/src/connection_tests.rs

@@ -39,10 +39,114 @@ impl Interface for Echo {
     }
 }
 
+struct ReplacementEcho;
+
+#[async_trait]
+impl Interface for ReplacementEcho {
+    fn name(&self) -> InterfaceName {
+        InterfaceName::new("ai.tinyhumans.Test").unwrap()
+    }
+
+    fn members(&self) -> Vec<MemberName> {
+        vec![MemberName::new("Echo").unwrap()]
+    }
+
+    async fn call(&self, _member: &MemberName, _args: Value) -> Result<Value> {
+        Ok(Value::String("replacement".to_string()))
+    }
+}
+
 fn path() -> ObjectPath {
     ObjectPath::new("/ai/tinyhumans/Test").unwrap()
 }
 
+#[tokio::test]
+async fn try_serve_at_refuses_a_collision_without_replacing_the_live_interface() {
+    let (client, service) = brokered_pair().await;
+    let error = service
+        .try_serve_at(path(), ReplacementEcho)
+        .await
+        .unwrap_err();
+    assert_eq!(error.wire_name(), Error::FAILED);
+
+    let result: Value = client
+        .proxy("ai.tinyhumans.Test", path(), "ai.tinyhumans.Test")
+        .unwrap()
+        .call("Echo", ("original",))
+        .await
+        .unwrap();
+    assert_eq!(result, serde_json::json!(["original"]));
+}
+
+#[tokio::test]
+async fn serve_at_keeps_its_existing_replacement_behavior() {
+    let (client, service) = brokered_pair().await;
+    service.serve_at(path(), ReplacementEcho).await.unwrap();
+
+    let result: Value = client
+        .proxy("ai.tinyhumans.Test", path(), "ai.tinyhumans.Test")
+        .unwrap()
+        .call("Echo", ())
+        .await
+        .unwrap();
+    assert_eq!(result, Value::String("replacement".to_string()));
+}
+
+#[tokio::test]
+async fn concurrent_try_serve_at_calls_have_exactly_one_winner() {
+    let (client, service) = brokered_pair().await;
+    let concurrent_path = ObjectPath::new("/ai/tinyhumans/TestConcurrent").unwrap();
+    let (first, second) = tokio::join!(
+        service.try_serve_at(concurrent_path.clone(), ReplacementEcho),
+        service.try_serve_at(concurrent_path.clone(), Echo),
+    );
+    assert_ne!(first.is_ok(), second.is_ok());
+
+    let result: Value = client
+        .proxy("ai.tinyhumans.Test", concurrent_path, "ai.tinyhumans.Test")
+        .unwrap()
+        .call("Echo", ("original",))
+        .await
+        .unwrap();
+    if first.is_ok() {
+        assert_eq!(result, Value::String("replacement".to_string()));
+    } else {
+        assert_eq!(result, serde_json::json!(["original"]));
+    }
+}
+
+struct DifferentInterface;
+
+#[async_trait]
+impl Interface for DifferentInterface {
+    fn name(&self) -> InterfaceName {
+        InterfaceName::new("ai.tinyhumans.Other").unwrap()
+    }
+
+    fn members(&self) -> Vec<MemberName> {
+        vec![MemberName::new("Echo").unwrap()]
+    }
+
+    async fn call(&self, _member: &MemberName, _args: Value) -> Result<Value> {
+        Ok(Value::String("other".to_string()))
+    }
+}
+
+#[tokio::test]
+async fn try_serve_at_refuses_any_occupied_object_path() {
+    let (client, service) = brokered_pair().await;
+    service.serve_at(path(), DifferentInterface).await.unwrap();
+    assert!(service.try_serve_at(path(), ReplacementEcho).await.is_err());
+
+    let result: Value = client
+        .proxy("ai.tinyhumans.Test", path(), "ai.tinyhumans.Other")
+        .unwrap()
+        .call("Echo", ())
+        .await
+        .unwrap();
+    assert_eq!(result, Value::String("other".to_string()));
+}
+
 fn call(member: &str, body: Value) -> Message {
     Message::method_call(
         BusName::new("ai.tinyhumans.Test").unwrap(),

crates/tinybus/src/module/first_admitted_tests.rs

@@ -183,6 +183,10 @@ fn a_bundle_marker_is_never_honoured_from_the_download_cache() {
 fn marked_bundle(bundled: &std::path::Path) -> std::path::PathBuf {
     let dir = artifact_dir(bundled, "demo", "1.0.0", ASSET.host_key).unwrap();
     std::fs::create_dir_all(&dir).unwrap();
+    use std::os::unix::fs::PermissionsExt;
+    for path in dir.ancestors().take_while(|path| path.starts_with(bundled)) {
+        std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o700)).unwrap();
+    }
     std::fs::write(
         dir.join(format!("{}.sha256", ASSET.archive)),
         format!("{}\n", ASSET.sha256),

crates/tinybus/src/module/host.rs

@@ -404,6 +404,40 @@ impl ModuleHost {
         modules
     }
 
+    /// Return the broker-assigned unique peer name for a live, ready module.
+    ///
+    /// `module_name` is the module's sanitized descriptor name as returned by
+    /// [`ModuleInfo::name`], not its well-known bus name. The returned name is
+    /// the identity the broker stamps onto ordinary calls from that exact
+    /// loaded instance. This lookup is not an attestation or confidential-call
+    /// grant; confidential delivery continues to require the existing pinned
+    /// admission and callback-binding checks.
+    ///
+    /// Returns `None` for unknown, disabled, initializing, stopped, faulted,
+    /// or lifecycle-transitioning modules. A caller must not retain a name as
+    /// authority beyond the invocation that needs it.
+    #[must_use]
+    pub fn authenticated_peer(&self, module_name: &str) -> Option<BusName> {
+        let loaded = self.inner.loaded.lock().expect("module list lock");
+        let module = loaded
+            .iter()
+            .find(|module| module.info.name == module_name)?;
+        if !module.info.enabled
+            || module.lifecycle_busy
+            || module.transport.native_lifecycle_pending()
+            || !module.transport.is_ready()
+            || module.transport.is_faulted()
+            || module.transport.init_failed()
+        {
+            return None;
+        }
+        matches!(
+            module.snapshot().state,
+            ModuleState::Ready | ModuleState::Serving
+        )
+        .then(|| module.unique_name.clone())
+    }
+
     /// Load one newly installed module.
     ///
     /// A verified operator allowlist admits confidential delivery to the exact