diff --git a/Cargo.lock b/Cargo.lock
index 82f91fc6..e09c7ec6 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -531,9 +531,9 @@ dependencies = [
 
 [[package]]
 name = "rustls-webpki"
-version = "0.103.15"
+version = "0.103.16"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f3c3cf1d8b1e7d4927e2d154c3fcb02979afb9939629c62cd9048d4f07b60ac2"
+checksum = "4443b1b6d39d7c663139942cc4558262562d2c84123bb6282ef177009ca8b281"
 dependencies = [
  "ring",
  "rustls-pki-types",
@@ -967,9 +967,9 @@ checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b"
 
 [[package]]
 name = "webpki-roots"
-version = "1.0.9"
+version = "1.0.10"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7dcd9d09a39985f5344844e66b0c530a33843579125f23e21e9f0f220850f22a"
+checksum = "1c1c8e20c8a38fc7d65abfe74eb7280e4aa667e8eec8bc84996d2874810748d9"
 dependencies = [
  "rustls-pki-types",
 ]
diff --git a/crates/tinybus/Cargo.toml b/crates/tinybus/Cargo.toml
index 69e5e311..ef2cf2b4 100644
--- a/crates/tinybus/Cargo.toml
+++ b/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"]
diff --git a/crates/tinybus/examples/module_host_stream_callback.rs b/crates/tinybus/examples/module_host_stream_callback.rs
new file mode 100644
index 00000000..53f04e56
--- /dev/null
+++ b/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,
+}
diff --git a/crates/tinybus/src/connection.rs b/crates/tinybus/src/connection.rs
index 867532ab..673f2ddb 100644
--- a/crates/tinybus/src/connection.rs
+++ b/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)
diff --git a/crates/tinybus/src/connection_tests.rs b/crates/tinybus/src/connection_tests.rs
index e8cabdc3..a23e2eda 100644
--- a/crates/tinybus/src/connection_tests.rs
+++ b/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(),
diff --git a/crates/tinybus/src/module/first_admitted_tests.rs b/crates/tinybus/src/module/first_admitted_tests.rs
index 989081c0..7c80df2c 100644
--- a/crates/tinybus/src/module/first_admitted_tests.rs
+++ b/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),
diff --git a/crates/tinybus/src/module/host.rs b/crates/tinybus/src/module/host.rs
index b67c87b6..495a44b3 100644
--- a/crates/tinybus/src/module/host.rs
+++ b/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
diff --git a/crates/tinybus/src/module/host_callbacks.rs b/crates/tinybus/src/module/host_callbacks.rs
index 56d636c1..8103fa9d 100644
--- a/crates/tinybus/src/module/host_callbacks.rs
+++ b/crates/tinybus/src/module/host_callbacks.rs
@@ -4,7 +4,10 @@ use super::*;
 use crate::message::Header;
 use crate::module::transport::AdmissionTermination;
 use crate::transport::memory::MemoryTransport;
-use crate::{CallContext, Connection, Interface, InterfaceName, MemberName, ObjectPath};
+use crate::{
+    CallContext, Connection, Interface, InterfaceName, MemberName, ObjectPath, StreamDescriptor,
+    StreamReader, StreamRef, StreamWriter,
+};
 use async_trait::async_trait;
 use tokio::sync::Notify;
 
@@ -19,6 +22,22 @@ pub struct HostCallbackEndpoint {
     pub members: Vec<MemberName>,
 }
 
+/// A finite set of ordinary (non-confidential) stream-carrying methods for
+/// one exact host callback interface.
+///
+/// Stream bytes use ordinary TinyBus `Write` calls, so these members cannot
+/// be confidential callbacks. Their authority is restricted to the exact
+/// loaded module and callback instance that declared them.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct HostStreamCallbackEndpoint {
+    /// Fixed object path; no subtree matching.
+    pub path: ObjectPath,
+    /// Fixed interface; no alias matching.
+    pub interface: InterfaceName,
+    /// Permitted stream-carrying methods, between one and 64 distinct names.
+    pub members: Vec<MemberName>,
+}
+
 /// A finite set of recipient-directed signals a host callback may emit.
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub struct HostCallbackSignal {
@@ -43,6 +62,7 @@ struct Lifetime {
 pub(crate) struct CallbackState {
     pub(crate) destination: BusName,
     endpoints: Vec<HostCallbackEndpoint>,
+    stream_endpoints: Vec<HostStreamCallbackEndpoint>,
     signals: Vec<HostCallbackSignal>,
     module_transport: Mutex<Option<std::sync::Weak<ModuleTransport>>>,
     lifetime: Mutex<Lifetime>,
@@ -53,6 +73,27 @@ pub(crate) struct CallbackState {
 }
 
 impl CallbackState {
+    fn required_endpoint_count(&self) -> usize {
+        let mut interfaces = Vec::new();
+        for (path, interface) in self
+            .endpoints
+            .iter()
+            .map(|endpoint| (&endpoint.path, &endpoint.interface))
+            .chain(
+                self.stream_endpoints
+                    .iter()
+                    .map(|endpoint| (&endpoint.path, &endpoint.interface)),
+            )
+        {
+            if !interfaces.iter().any(|(known_path, known_interface)| {
+                *known_path == path && *known_interface == interface
+            }) {
+                interfaces.push((path, interface));
+            }
+        }
+        interfaces.len()
+    }
+
     /// Physical cleanup waits offruntime, without a watchdog or native exclusion.
     pub(crate) fn wait_drained(&self) {
         let mut state = self
@@ -163,6 +204,23 @@ impl CallbackState {
             .clone()
             .ok_or_else(|| Error::failed("directed signal recipient is unavailable"))
     }
+
+    fn authorized_stream_module(&self, context: &CallContext) -> Result<BusName> {
+        let state = self.lifetime.lock().expect("callback lifetime lock");
+        if context.is_confidential()
+            || self.stream_endpoints.is_empty()
+            || !self.module_live()
+            || !state.active
+            || state.module.is_none()
+            || context.authenticated_sender() != state.module.as_ref()
+        {
+            return Err(Error::failed("callback stream authority is unavailable"));
+        }
+        state
+            .module
+            .clone()
+            .ok_or_else(|| Error::failed("callback stream recipient is unavailable"))
+    }
 }
 
 /// Read-only liveness and authenticated module identity for a callback worker.
@@ -178,7 +236,10 @@ impl HostCallbackGuard {
         self.0.module_peer()
     }
 
-    /// Whether this binding remains active for this authenticated sender.
+    /// Whether this confidential callback binding remains active for this sender.
+    ///
+    /// Ordinary stream callbacks are checked separately with
+    /// [`HostCallbackGuard::authorizes_stream`].
     pub fn authorizes(&self, context: &CallContext) -> bool {
         let state = self.0.lifetime.lock().expect("callback lifetime lock");
         state.active
@@ -188,6 +249,12 @@ impl HostCallbackGuard {
             && context.authenticated_sender() == state.module.as_ref()
     }
 
+    /// Whether this binding remains active for an ordinary stream callback
+    /// from its exact authenticated module sender.
+    pub fn authorizes_stream(&self, context: &CallContext) -> bool {
+        self.0.authorized_stream_module(context).is_ok()
+    }
+
     /// Number of callbacks whose user futures have not yet completed or dropped.
     /// A caller timeout does not remove an executing callback from this count.
     pub fn running(&self) -> usize {
@@ -314,6 +381,52 @@ impl HostCallbackBinding {
     pub fn unique_name(&self) -> &BusName {
         &self.state.destination
     }
+
+    /// Accept a stream opened by this binding's currently authorized module.
+    ///
+    /// The raw connection remains private. The callback context supplies the
+    /// exact authenticated sender, and the stream is accepted only from that
+    /// sender while this binding remains active. This is for an explicitly
+    /// declared ordinary stream callback; stream payloads are not confidential.
+    ///
+    /// # Errors
+    /// Returns an error if the context is confidential, is not for this live
+    /// binding, or the stream was not opened by its bound module.
+    pub fn accept_stream_from_module(
+        &self,
+        context: &CallContext,
+        stream: &StreamRef,
+    ) -> Result<StreamReader> {
+        let sender = self.authorized_stream_module(context)?;
+        self.connection.accept_stream_from(stream, &sender)
+    }
+
+    /// Open a stream to this binding's currently authorized module.
+    ///
+    /// This is the callback-side counterpart of
+    /// [`HostCallbackBinding::accept_stream_from_module`]. It does not expose
+    /// the binding's connection or permit a caller-supplied destination.
+    /// Like all streams, its payload uses ordinary TinyBus calls.
+    ///
+    /// # Errors
+    /// Returns an error if the context is confidential, is not for this live
+    /// binding, or the destination refuses the stream.
+    pub async fn open_stream_to_module(
+        &self,
+        context: &CallContext,
+        descriptor: StreamDescriptor,
+        timeout: Duration,
+    ) -> Result<StreamWriter> {
+        let recipient = self.authorized_stream_module(context)?;
+        self.connection
+            .open_stream_with_timeout(&recipient, descriptor, timeout)
+            .await
+    }
+
+    fn authorized_stream_module(&self, context: &CallContext) -> Result<BusName> {
+        self.state.authorized_stream_module(context)
+    }
+
     /// Construct an outgoing proxy using this private host connection.
     ///
     /// The outgoing handle cannot expose the owned callback connection:
@@ -384,17 +497,31 @@ impl HostCallbackBinding {
     /// Every declared member must be implemented; duplicate registration fails.
     pub async fn serve_at(&self, path: ObjectPath, interface: impl Interface) -> Result<()> {
         let name = interface.name();
-        let endpoint = self
+        let confidential_endpoint = self
             .state
             .endpoints
             .iter()
             .find(|endpoint| endpoint.path == path && endpoint.interface == name)
-            .ok_or_else(|| Error::failed("callback endpoint is not declared"))?;
+            .cloned();
+        let stream_endpoint = self
+            .state
+            .stream_endpoints
+            .iter()
+            .find(|endpoint| endpoint.path == path && endpoint.interface == name)
+            .cloned();
+        if confidential_endpoint.is_none() && stream_endpoint.is_none() {
+            return Err(Error::failed("callback endpoint is not declared"));
+        }
         let provided = interface.members();
-        if !endpoint
-            .members
+        if confidential_endpoint
             .iter()
-            .all(|member| provided.contains(member))
+            .flat_map(|endpoint| endpoint.members.iter())
+            .chain(
+                stream_endpoint
+                    .iter()
+                    .flat_map(|endpoint| endpoint.members.iter()),
+            )
+            .any(|member| !provided.contains(member))
         {
             return Err(Error::failed("callback endpoint members are missing"));
         }
@@ -417,7 +544,8 @@ impl HostCallbackBinding {
                 path.clone(),
                 GuardedInterface {
                     interface,
-                    endpoint: endpoint.clone(),
+                    confidential_endpoint,
+                    stream_endpoint,
                     state: self.state.clone(),
                 },
             )
@@ -451,7 +579,8 @@ impl Drop for HostCallbackBinding {
 
 struct GuardedInterface<I> {
     interface: I,
-    endpoint: HostCallbackEndpoint,
+    confidential_endpoint: Option<HostCallbackEndpoint>,
+    stream_endpoint: Option<HostStreamCallbackEndpoint>,
     state: Arc<CallbackState>,
 }
 
@@ -474,10 +603,26 @@ impl Drop for Running {
 #[async_trait]
 impl<I: Interface> Interface for GuardedInterface<I> {
     fn name(&self) -> InterfaceName {
-        self.endpoint.interface.clone()
+        self.confidential_endpoint
+            .as_ref()
+            .map(|endpoint| endpoint.interface.clone())
+            .or_else(|| {
+                self.stream_endpoint
+                    .as_ref()
+                    .map(|endpoint| endpoint.interface.clone())
+            })
+            .expect("declared callback interface")
     }
     fn members(&self) -> Vec<MemberName> {
-        self.endpoint.members.clone()
+        self.confidential_endpoint
+            .iter()
+            .flat_map(|endpoint| endpoint.members.iter().cloned())
+            .chain(
+                self.stream_endpoint
+                    .iter()
+                    .flat_map(|endpoint| endpoint.members.iter().cloned()),
+            )
+            .collect()
     }
     async fn call(&self, _: &MemberName, _: serde_json::Value) -> Result<serde_json::Value> {
         Err(Error::failed("authenticated scoped callback required"))
@@ -492,12 +637,21 @@ impl<I: Interface> Interface for GuardedInterface<I> {
         // check by carrying an earlier route admission through an await.
         let _running = {
             let mut state = self.state.lifetime.lock().expect("callback lifetime lock");
+            let confidential_member = self
+                .confidential_endpoint
+                .as_ref()
+                .is_some_and(|endpoint| endpoint.members.contains(member));
+            let stream_member = self
+                .stream_endpoint
+                .as_ref()
+                .is_some_and(|endpoint| endpoint.members.contains(member));
             if !self.state.module_live()
-                || !context.is_confidential()
+                || confidential_member == stream_member
+                || confidential_member && !context.is_confidential()
+                || stream_member && context.is_confidential()
                 || !state.active
                 || state.module.is_none()
                 || context.authenticated_sender() != state.module.as_ref()
-                || !self.endpoint.members.contains(member)
             {
                 return Err(Error::failed("scoped callback authority is unavailable"));
             }
@@ -522,10 +676,30 @@ impl ModuleHost {
         &self,
         endpoints: Vec<HostCallbackEndpoint>,
     ) -> Result<HostCallbackBinding> {
-        self.prepare_host_callbacks_with_signals(endpoints, Vec::new())
+        self.prepare_host_callbacks_with_stream_endpoints(endpoints, Vec::new())
             .await
     }
 
+    /// Create an exact callback binding with confidential endpoints and a
+    /// separate allowlist of ordinary stream-carrying methods.
+    ///
+    /// Stream endpoints are never treated as confidential: their handles and
+    /// payload use ordinary TinyBus calls. They remain restricted to the
+    /// binding's exact destination, active lifetime, and verified module
+    /// sender. Members cannot overlap the confidential endpoint declarations.
+    pub async fn prepare_host_callbacks_with_stream_endpoints(
+        &self,
+        endpoints: Vec<HostCallbackEndpoint>,
+        stream_endpoints: Vec<HostStreamCallbackEndpoint>,
+    ) -> Result<HostCallbackBinding> {
+        self.prepare_host_callbacks_with_signals_and_stream_endpoints(
+            endpoints,
+            Vec::new(),
+            stream_endpoints,
+        )
+        .await
+    }
+
     /// Create a private callback recipient with a finite directed-signal set.
     ///
     /// Every signal declaration must share a path and interface with one of
@@ -540,8 +714,67 @@ impl ModuleHost {
         endpoints: Vec<HostCallbackEndpoint>,
         signals: Vec<HostCallbackSignal>,
     ) -> Result<HostCallbackBinding> {
-        if endpoints.is_empty()
-            || endpoints.len() > 32
+        self.prepare_host_callbacks_with_signals_and_stream_endpoints(
+            endpoints,
+            signals,
+            Vec::new(),
+        )
+        .await
+    }
+
+    /// Create a callback binding with confidential methods, directed signals,
+    /// and separately declared ordinary stream methods.
+    pub async fn prepare_host_callbacks_with_signals_and_stream_endpoints(
+        &self,
+        endpoints: Vec<HostCallbackEndpoint>,
+        signals: Vec<HostCallbackSignal>,
+        stream_endpoints: Vec<HostStreamCallbackEndpoint>,
+    ) -> Result<HostCallbackBinding> {
+        let mut declared_interfaces = Vec::new();
+        for (path, interface) in endpoints
+            .iter()
+            .map(|endpoint| (&endpoint.path, &endpoint.interface))
+            .chain(
+                stream_endpoints
+                    .iter()
+                    .map(|endpoint| (&endpoint.path, &endpoint.interface)),
+            )
+        {
+            if !declared_interfaces
+                .iter()
+                .any(|(known_path, known_interface)| {
+                    *known_path == path && *known_interface == interface
+                })
+            {
+                declared_interfaces.push((path, interface));
+            }
+        }
+        let all_endpoint_count = declared_interfaces.len();
+        let invalid_stream_endpoint =
+            stream_endpoints.iter().any(|endpoint| {
+                endpoint.members.is_empty()
+                    || endpoint.members.len() > 64
+                    || endpoint
+                        .members
+                        .iter()
+                        .enumerate()
+                        .any(|(i, member)| endpoint.members[..i].contains(member))
+            }) || stream_endpoints.iter().enumerate().any(|(i, endpoint)| {
+                stream_endpoints[..i].iter().any(|previous| {
+                    previous.path == endpoint.path && previous.interface == endpoint.interface
+                }) || endpoints.iter().any(|confidential| {
+                    confidential.path == endpoint.path
+                        && confidential.interface == endpoint.interface
+                        && endpoint
+                            .members
+                            .iter()
+                            .any(|member| confidential.members.contains(member))
+                })
+            });
+        if all_endpoint_count == 0
+            || all_endpoint_count > 32
+            || invalid_stream_endpoint
+            || endpoints.is_empty() && !signals.is_empty()
             || endpoints.iter().any(|endpoint| {
                 endpoint.members.is_empty()
                     || endpoint.members.len() > 64
@@ -608,6 +841,7 @@ impl ModuleHost {
             state: Arc::new(CallbackState {
                 destination,
                 endpoints,
+                stream_endpoints,
                 signals,
                 module_transport: Mutex::new(None),
                 lifetime: Mutex::new(Lifetime {
@@ -655,7 +889,10 @@ impl ModuleHost {
                 .lifetime
                 .lock()
                 .expect("callback lifetime lock");
-            if state.used || state.revoked || state.served.len() != binding.state.endpoints.len() {
+            if state.used
+                || state.revoked
+                || state.served.len() != binding.state.required_endpoint_count()
+            {
                 return Err(Error::failed(
                     "callback binding is already used or endpoints are missing",
                 ));
diff --git a/crates/tinybus/src/module/host_callbacks_tests.rs b/crates/tinybus/src/module/host_callbacks_tests.rs
index e753b508..03c236df 100644
--- a/crates/tinybus/src/module/host_callbacks_tests.rs
+++ b/crates/tinybus/src/module/host_callbacks_tests.rs
@@ -3,6 +3,34 @@ use crate::message::Message;
 use crate::router::{MatchRule, Router};
 use std::sync::atomic::AtomicUsize;
 
+async fn active_stream_binding() -> (
+    Arc<ModuleHost>,
+    tokio::task::JoinHandle<Result<()>>,
+    HostCallbackBinding,
+    crate::Connection,
+    crate::transport::memory::MemoryBus,
+) {
+    let bus = crate::transport::memory::MemoryBus::new();
+    let broker = Broker::new();
+    let broker_task = broker.spawn(bus.clone());
+    let host = Arc::new(ModuleHost::new(broker));
+    let binding = host
+        .prepare_host_callbacks_with_stream_endpoints(vec![endpoint()], vec![stream_endpoint()])
+        .await
+        .unwrap();
+    let module = crate::Connection::connect(bus.connect().await.unwrap())
+        .await
+        .unwrap();
+    let module_peer = module.unique_name().unwrap();
+    {
+        let mut lifetime = binding.state.lifetime.lock().unwrap();
+        lifetime.active = true;
+        lifetime.used = true;
+        lifetime.module = Some(module_peer);
+    }
+    (host, broker_task, binding, module, bus)
+}
+
 fn endpoint() -> HostCallbackEndpoint {
     HostCallbackEndpoint {
         path: "/callback".parse().unwrap(),
@@ -10,6 +38,13 @@ fn endpoint() -> HostCallbackEndpoint {
         members: vec!["Observe".parse().unwrap()],
     }
 }
+fn stream_endpoint() -> HostStreamCallbackEndpoint {
+    HostStreamCallbackEndpoint {
+        path: "/callback".parse().unwrap(),
+        interface: "org.example.Callback".parse().unwrap(),
+        members: vec!["Transfer".parse().unwrap()],
+    }
+}
 fn signal_declaration() -> HostCallbackSignal {
     HostCallbackSignal {
         path: "/callback".parse().unwrap(),
@@ -21,6 +56,7 @@ fn state(source: BusName, destination: BusName) -> Arc<CallbackState> {
     Arc::new(CallbackState {
         destination,
         endpoints: vec![endpoint()],
+        stream_endpoints: vec![],
         signals: vec![signal_declaration()],
         module_transport: Mutex::new(None),
         lifetime: Mutex::new(Lifetime {
@@ -38,6 +74,85 @@ fn state(source: BusName, destination: BusName) -> Arc<CallbackState> {
     })
 }
 
+struct MixedCallback;
+#[async_trait]
+impl Interface for MixedCallback {
+    fn name(&self) -> InterfaceName {
+        endpoint().interface
+    }
+    fn members(&self) -> Vec<MemberName> {
+        vec![
+            endpoint().members[0].clone(),
+            stream_endpoint().members[0].clone(),
+        ]
+    }
+    async fn call(&self, member: &MemberName, _: serde_json::Value) -> Result<serde_json::Value> {
+        Ok(serde_json::Value::String(member.to_string()))
+    }
+}
+
+#[tokio::test]
+async fn mixed_callback_dispatch_keeps_confidential_and_stream_members_separate() {
+    let module_peer: BusName = ":1.8".parse().unwrap();
+    let callback_peer: BusName = ":1.9".parse().unwrap();
+    let mut state = state(module_peer.clone(), callback_peer);
+    Arc::get_mut(&mut state)
+        .unwrap()
+        .stream_endpoints
+        .push(stream_endpoint());
+    let guarded = GuardedInterface {
+        interface: MixedCallback,
+        confidential_endpoint: Some(endpoint()),
+        stream_endpoint: Some(stream_endpoint()),
+        state: state.clone(),
+    };
+    let confidential = CallContext::brokered(Some(&module_peer), true);
+    let ordinary = CallContext::brokered(Some(&module_peer), false);
+    let wrong_sender: BusName = ":1.10".parse().unwrap();
+    let wrong_ordinary = CallContext::brokered(Some(&wrong_sender), false);
+    let secret_member = &endpoint().members[0];
+    let stream_member = &stream_endpoint().members[0];
+
+    assert!(
+        guarded
+            .call_with_context(secret_member, serde_json::json!([]), &confidential)
+            .await
+            .is_ok()
+    );
+    assert!(
+        guarded
+            .call_with_context(secret_member, serde_json::json!([]), &ordinary)
+            .await
+            .is_err()
+    );
+    assert!(
+        guarded
+            .call_with_context(stream_member, serde_json::json!([]), &ordinary)
+            .await
+            .is_ok()
+    );
+    assert!(
+        guarded
+            .call_with_context(stream_member, serde_json::json!([]), &confidential)
+            .await
+            .is_err()
+    );
+    assert!(
+        guarded
+            .call_with_context(stream_member, serde_json::json!([]), &wrong_ordinary)
+            .await
+            .is_err()
+    );
+
+    state.revoke();
+    assert!(
+        guarded
+            .call_with_context(stream_member, serde_json::json!([]), &ordinary)
+            .await
+            .is_err()
+    );
+}
+
 #[test]
 fn directed_signals_target_only_the_live_bound_module_and_declared_member() {
     let mut router = Router::default();
@@ -284,7 +399,8 @@ async fn revoke_after_route_selection_blocks_dispatch_and_direct_unverified_disp
     assert!(resolves(&router, &message).is_ok());
     let (entered, mut receiver) = tokio::sync::mpsc::channel(1);
     let guarded = GuardedInterface {
-        endpoint: endpoint(),
+        confidential_endpoint: Some(endpoint()),
+        stream_endpoint: None,
         state: state.clone(),
         interface: BlockingCallback {
             entered,
@@ -341,7 +457,8 @@ async fn revoke_during_await_preserves_running_charge_until_actual_completion()
     let release = Arc::new(tokio::sync::Semaphore::new(0));
     let published = Arc::new(AtomicUsize::new(0));
     let guarded = GuardedInterface {
-        endpoint: endpoint(),
+        confidential_endpoint: Some(endpoint()),
+        stream_endpoint: None,
         state: state.clone(),
         interface: BlockingCallback {
             entered,
@@ -385,6 +502,18 @@ async fn endpoint_declarations_registration_and_broker_ownership_are_checked() {
             .await
             .is_err()
     );
+    assert!(
+        host.prepare_host_callbacks_with_stream_endpoints(
+            vec![endpoint()],
+            vec![HostStreamCallbackEndpoint {
+                path: endpoint().path,
+                interface: endpoint().interface,
+                members: endpoint().members,
+            }],
+        )
+        .await
+        .is_err()
+    );
     let mut duplicate = endpoint();
     duplicate.members.push(duplicate.members[0].clone());
     assert!(host.prepare_host_callbacks(vec![duplicate]).await.is_err());
@@ -545,10 +674,300 @@ async fn restricted_outgoing_handle_preserves_typed_calls_and_metadata() {
     binding.close().await.unwrap();
 }
 
+#[tokio::test]
+async fn a_bound_callback_can_receive_and_send_streams_without_exposing_its_connection() {
+    let (_host, broker_task, binding, module, _) = active_stream_binding().await;
+    let module_peer = module.unique_name().unwrap();
+    let context = CallContext::brokered(Some(&module_peer), false);
+
+    let mut module_writer = module
+        .open_stream(binding.unique_name(), StreamDescriptor::with_len(5))
+        .await
+        .unwrap();
+    let mut host_reader = binding
+        .accept_stream_from_module(&context, &module_writer.stream_ref())
+        .unwrap();
+    let module_send = tokio::spawn(async move {
+        module_writer.write(b"hello").await.unwrap();
+        module_writer.finish().await.unwrap();
+    });
+    assert_eq!(host_reader.read_to_end_capped(5).await.unwrap(), b"hello");
+    module_send.await.unwrap();
+
+    let mut host_writer = binding
+        .open_stream_to_module(
+            &context,
+            StreamDescriptor::with_len(5),
+            Duration::from_secs(2),
+        )
+        .await
+        .unwrap();
+    let mut module_reader = module
+        .accept_stream_from(&host_writer.stream_ref(), binding.unique_name())
+        .unwrap();
+    host_writer.write(b"world").await.unwrap();
+    host_writer.finish().await.unwrap();
+    assert_eq!(module_reader.read_to_end_capped(5).await.unwrap(), b"world");
+
+    binding.close().await.unwrap();
+    broker_task.abort();
+}
+
+#[tokio::test]
+async fn callback_streams_reject_other_peers_nonconfidential_calls_and_revoked_bindings() {
+    let (_host, broker_task, binding, module, bus) = active_stream_binding().await;
+    let module_peer = module.unique_name().unwrap();
+    let context = CallContext::brokered(Some(&module_peer), false);
+    let outsider = crate::Connection::connect(bus.connect().await.unwrap())
+        .await
+        .unwrap();
+    let outsider_peer = outsider.unique_name().unwrap();
+    let wrong_peer = CallContext::brokered(Some(&outsider_peer), false);
+    let confidential_call = CallContext::brokered(Some(&module_peer), true);
+    let guard = binding.guard();
+    assert!(!guard.authorizes(&context));
+    assert!(guard.authorizes_stream(&context));
+    assert!(!guard.authorizes_stream(&wrong_peer));
+    assert!(!guard.authorizes_stream(&confidential_call));
+    let stream = StreamRef {
+        id: "not-opened".to_owned(),
+        content_type: None,
+        len: Some(1),
+    };
+
+    assert!(
+        binding
+            .accept_stream_from_module(&wrong_peer, &stream)
+            .is_err()
+    );
+    assert!(
+        binding
+            .open_stream_to_module(
+                &wrong_peer,
+                StreamDescriptor::with_len(1),
+                Duration::from_secs(1),
+            )
+            .await
+            .is_err()
+    );
+    assert!(
+        binding
+            .accept_stream_from_module(&confidential_call, &stream)
+            .is_err()
+    );
+    assert!(
+        binding
+            .open_stream_to_module(
+                &confidential_call,
+                StreamDescriptor::with_len(1),
+                Duration::from_secs(1),
+            )
+            .await
+            .is_err()
+    );
+
+    binding.revoke();
+    assert!(!guard.authorizes_stream(&context));
+    assert!(
+        binding
+            .accept_stream_from_module(&context, &stream)
+            .is_err()
+    );
+    assert!(
+        binding
+            .open_stream_to_module(
+                &context,
+                StreamDescriptor::with_len(1),
+                Duration::from_secs(1),
+            )
+            .await
+            .is_err()
+    );
+    binding.close().await.unwrap();
+    broker_task.abort();
+}
+
+#[tokio::test]
+async fn callback_stream_acceptance_rejects_a_handle_opened_by_another_peer() {
+    let (_host, broker_task, binding, module, bus) = active_stream_binding().await;
+    let module_peer = module.unique_name().unwrap();
+    let context = CallContext::brokered(Some(&module_peer), false);
+    let outsider = crate::Connection::connect(bus.connect().await.unwrap())
+        .await
+        .unwrap();
+    let mut outsider_writer = outsider
+        .open_stream(binding.unique_name(), StreamDescriptor::with_len(1))
+        .await
+        .unwrap();
+
+    assert!(
+        binding
+            .accept_stream_from_module(&context, &outsider_writer.stream_ref())
+            .is_err()
+    );
+
+    outsider_writer.write(b"x").await.unwrap();
+    outsider_writer.finish().await.unwrap();
+    binding.close().await.unwrap();
+    broker_task.abort();
+}
+
+struct ActualStreamCallback {
+    binding: Arc<HostCallbackBinding>,
+    observe_entered: Arc<Notify>,
+    observe_release: Arc<Notify>,
+    transfer_dispatches: Arc<AtomicUsize>,
+    undeclared_dispatches: Arc<AtomicUsize>,
+}
+
+#[async_trait]
+impl Interface for ActualStreamCallback {
+    fn name(&self) -> InterfaceName {
+        endpoint().interface
+    }
+    fn members(&self) -> Vec<MemberName> {
+        vec![
+            endpoint().members[0].clone(),
+            stream_endpoint().members[0].clone(),
+            "Undeclared".parse().unwrap(),
+        ]
+    }
+    async fn call(&self, _: &MemberName, _: serde_json::Value) -> Result<serde_json::Value> {
+        Err(Error::failed("callback context is required"))
+    }
+    async fn call_with_context(
+        &self,
+        member: &MemberName,
+        args: serde_json::Value,
+        context: &CallContext,
+    ) -> Result<serde_json::Value> {
+        if member == &endpoint().members[0] {
+            self.observe_entered.notify_one();
+            self.observe_release.notified().await;
+            return Ok(serde_json::Value::Null);
+        }
+        if member.as_str() == "Undeclared" {
+            self.undeclared_dispatches.fetch_add(1, Ordering::SeqCst);
+            return Ok(serde_json::Value::Null);
+        }
+        self.transfer_dispatches.fetch_add(1, Ordering::SeqCst);
+        if member != &stream_endpoint().members[0] {
+            return Err(Error::failed("callback member is not declared"));
+        }
+
+        let (stream,): (StreamRef,) =
+            serde_json::from_value(args).map_err(|_| Error::failed("invalid stream request"))?;
+        let mut reader = self.binding.accept_stream_from_module(context, &stream)?;
+        let received = reader.read_to_end_capped(12).await?;
+        if received != b"module->host" {
+            return Err(Error::failed(
+                "module callback stream payload did not match",
+            ));
+        }
+        let stream_result = self
+            .binding
+            .open_stream_to_module(
+                context,
+                StreamDescriptor::with_len(12),
+                Duration::from_secs(2),
+            )
+            .await;
+        let mut writer = stream_result?;
+        let reply = writer.stream_ref();
+        tokio::spawn(async move {
+            let _ = writer.write(b"host->module").await;
+            let _ = writer.finish().await;
+        });
+        serde_json::to_value(reply).map_err(|_| Error::failed("stream reply encoding failed"))
+    }
+}
+
+#[tokio::test]
+#[ignore = "requires TINYBUS_TEST_HOST_STREAM_CALLBACK_MODULE built cdylib"]
+async fn admitted_module_uses_only_declared_stream_callbacks_and_revoke_blocks_reentry() {
+    let artifact = std::path::PathBuf::from(
+        std::env::var_os("TINYBUS_TEST_HOST_STREAM_CALLBACK_MODULE").unwrap(),
+    );
+    let hash = crate::module::hash::file_hex(std::fs::File::open(&artifact).unwrap()).unwrap();
+    let (_directory, artifact) = super::super::tests::staged_module(&artifact, &hash);
+    let bus = crate::transport::memory::MemoryBus::new();
+    let broker = Broker::new();
+    let broker_task = broker.spawn(bus.clone());
+    let host = Arc::new(ModuleHost::new(broker));
+    let binding = Arc::new(
+        host.prepare_host_callbacks_with_stream_endpoints(
+            vec![endpoint()],
+            vec![stream_endpoint()],
+        )
+        .await
+        .unwrap(),
+    );
+    let observe_entered = Arc::new(Notify::new());
+    let observe_release = Arc::new(Notify::new());
+    let transfer_dispatches = Arc::new(AtomicUsize::new(0));
+    let undeclared_dispatches = Arc::new(AtomicUsize::new(0));
+    binding
+        .serve_at(
+            endpoint().path,
+            ActualStreamCallback {
+                binding: binding.clone(),
+                observe_entered: observe_entered.clone(),
+                observe_release: observe_release.clone(),
+                transfer_dispatches: transfer_dispatches.clone(),
+                undeclared_dispatches: undeclared_dispatches.clone(),
+            },
+        )
+        .await
+        .unwrap();
+
+    let loaded = host
+        .load_file_with_host_callbacks(
+            artifact,
+            serde_json::json!({"host": binding.unique_name()}),
+            &binding,
+        )
+        .await
+        .unwrap();
+    let client = crate::Connection::connect(bus.connect().await.unwrap())
+        .await
+        .unwrap();
+    let module = client
+        .proxy(
+            "org.example.EagerCallback",
+            "/module",
+            "org.example.EagerCallback",
+        )
+        .unwrap();
+    let round_trip =
+        tokio::spawn(async move { module.call::<serde_json::Value>("RoundTrip", ()).await });
+    if tokio::time::timeout(Duration::from_secs(10), observe_entered.notified())
+        .await
+        .is_err()
+    {
+        panic!("the admitted module did not complete its stream round trip");
+    }
+    binding.revoke();
+    observe_release.notify_one();
+    let round_trip_result = tokio::time::timeout(Duration::from_secs(10), round_trip)
+        .await
+        .expect("module should observe revocation")
+        .unwrap();
+    let result = round_trip_result.unwrap();
+    assert!(result.is_null());
+    assert_eq!(transfer_dispatches.load(Ordering::SeqCst), 1);
+    assert_eq!(undeclared_dispatches.load(Ordering::SeqCst), 0);
+    host.inner
+        .stop(&loaded.name, Duration::from_secs(2))
+        .await
+        .unwrap();
+    binding.close().await.unwrap();
+    broker_task.abort();
+}
+
 #[tokio::test]
 async fn rollback_after_attach_releases_the_actual_initialization_observer() {
     let broker = Broker::new();
-    let host = ModuleHost::new(broker);
+    let host = Arc::new(ModuleHost::new(broker));
     let binding = host.prepare_host_callbacks(vec![endpoint()]).await.unwrap();
     let manifest = super::super::tests::manifest();
     let path = Path::new("observer-fixture.so");
@@ -1412,7 +1831,8 @@ fn runtime_drop_destroys_provider_future_before_offruntime_native_cleanup() {
                     guard: HostCallbackGuard(account.clone()),
                     observed: observed.clone(),
                 },
-                endpoint: endpoint(),
+                confidential_endpoint: Some(endpoint()),
+                stream_endpoint: None,
                 state: account.clone(),
             };
             let context = CallContext::brokered(account.module_peer().as_ref(), true);
@@ -1489,7 +1909,8 @@ async fn completing_provider_future_is_destroyed_before_its_running_account() {
             guard: HostCallbackGuard(account.clone()),
             observed: observed.clone(),
         },
-        endpoint: endpoint(),
+        confidential_endpoint: Some(endpoint()),
+        stream_endpoint: None,
         state: account.clone(),
     };
     let context = CallContext::brokered(account.module_peer().as_ref(), true);
diff --git a/crates/tinybus/src/module/host_tests.rs b/crates/tinybus/src/module/host_tests.rs
index a5999da0..860b25b8 100644
--- a/crates/tinybus/src/module/host_tests.rs
+++ b/crates/tinybus/src/module/host_tests.rs
@@ -19,6 +19,18 @@ static BLOCKING_REINIT_RECEIVER: OnceLock<StdMutex<Option<std::sync::mpsc::Recei
     OnceLock::new();
 static BLOCKING_REINIT_STARTED: AtomicBool = AtomicBool::new(false);
 
+#[cfg(unix)]
+fn make_private_tree(root: &Path, path: &Path) {
+    use std::os::unix::fs::PermissionsExt;
+
+    for component in path
+        .ancestors()
+        .take_while(|component| component.starts_with(root))
+    {
+        std::fs::set_permissions(component, std::fs::Permissions::from_mode(0o700)).unwrap();
+    }
+}
+
 struct FakeModule {
     tx: std::sync::mpsc::SyncSender<Vec<u8>>,
 }
@@ -511,7 +523,10 @@ async fn cancelling_a_waiting_stop_clears_its_pending_transition() {
 
 #[tokio::test]
 async fn a_lazy_manifest_registers_an_unmapped_library_and_the_first_call_loads_it() {
-    let directory = tempfile::tempdir_in(std::env::current_dir().unwrap()).unwrap();
+    let temp_root = std::fs::canonicalize(std::env::temp_dir()).unwrap();
+    let directory = tempfile::tempdir_in(temp_root).unwrap();
+    #[cfg(unix)]
+    make_private_tree(directory.path(), directory.path());
     let extension = if cfg!(windows) {
         "dll"
     } else if cfg!(target_os = "macos") {
@@ -582,7 +597,7 @@ async fn a_lazy_manifest_registers_an_unmapped_library_and_the_first_call_loads_
 async fn a_bundle_marker_equal_to_the_pin_admits_the_extraction_without_its_archive() {
     use crate::module::{ReleaseAsset, ReleasePlan, artifact_dir, load_first_admitted};
 
-    let bundled = tempfile::tempdir_in(std::env::current_dir().unwrap()).unwrap();
+    let bundled = tempfile::tempdir().unwrap();
     let user_cache = tempfile::tempdir().unwrap();
     let asset = ReleaseAsset {
         host_key: "macos-15-arm64",
@@ -591,6 +606,8 @@ async fn a_bundle_marker_equal_to_the_pin_admits_the_extraction_without_its_arch
     };
     let dir = artifact_dir(bundled.path(), "clock", "0.1.0", asset.host_key).unwrap();
     std::fs::create_dir_all(&dir).unwrap();
+    #[cfg(unix)]
+    make_private_tree(bundled.path(), &dir);
     let artifact = dir.join(format!(
         "clock.{}",
         crate::module::cache::library_extension()
@@ -659,7 +676,7 @@ async fn a_bundle_marker_equal_to_the_pin_admits_the_extraction_without_its_arch
 async fn a_bundle_marker_without_an_allowlist_is_refused_rather_than_loaded_unhashed() {
     use crate::module::{ReleaseAsset, ReleasePlan, artifact_dir, load_first_admitted};
 
-    let bundled = tempfile::tempdir_in(std::env::current_dir().unwrap()).unwrap();
+    let bundled = tempfile::tempdir().unwrap();
     let user_cache = tempfile::tempdir().unwrap();
     let asset = ReleaseAsset {
         host_key: "macos-15-arm64",
@@ -668,6 +685,8 @@ async fn a_bundle_marker_without_an_allowlist_is_refused_rather_than_loaded_unha
     };
     let dir = artifact_dir(bundled.path(), "clock", "0.1.0", asset.host_key).unwrap();
     std::fs::create_dir_all(&dir).unwrap();
+    #[cfg(unix)]
+    make_private_tree(bundled.path(), &dir);
     let artifact = dir.join(format!(
         "clock.{}",
         crate::module::cache::library_extension()
@@ -708,7 +727,7 @@ async fn a_bundle_marker_without_an_allowlist_is_refused_rather_than_loaded_unha
 
 #[test]
 fn an_invalid_lazy_sidecar_refuses_the_artifact_instead_of_loading_it_eagerly() {
-    let directory = tempfile::tempdir_in(std::env::current_dir().unwrap()).unwrap();
+    let directory = tempfile::tempdir().unwrap();
     let artifact = directory.path().join(if cfg!(windows) {
         "clock.dll"
     } else if cfg!(target_os = "macos") {
@@ -725,7 +744,7 @@ fn an_invalid_lazy_sidecar_refuses_the_artifact_instead_of_loading_it_eagerly()
 
 #[test]
 fn a_sidecar_without_lazy_init_refuses_the_artifact_instead_of_loading_it_eagerly() {
-    let directory = tempfile::tempdir_in(std::env::current_dir().unwrap()).unwrap();
+    let directory = tempfile::tempdir().unwrap();
     let artifact = directory.path().join(if cfg!(windows) {
         "clock.dll"
     } else if cfg!(target_os = "macos") {
@@ -1053,6 +1072,73 @@ async fn the_sender_on_a_module_frame_is_stamped_by_the_broker_like_any_other_pe
     broker_task.abort();
 }
 
+#[tokio::test]
+async fn authenticated_peer_is_available_only_for_a_live_ready_module() {
+    let _test_guard = FAKE_MODULE_TEST_LOCK.lock().await;
+    let bus = MemoryBus::new();
+    let broker = Broker::new();
+    let broker_task = broker.spawn(bus.clone());
+    let host = ModuleHost::new(broker);
+
+    assert!(host.authenticated_peer("clock").is_none());
+    let mut lazy_manifest = manifest();
+    lazy_manifest.lazy_init = true;
+    unsafe {
+        host.attach_raw(
+            "clock.so",
+            TbAbiDescriptor::current("clock", "0.1.0"),
+            lazy_manifest,
+            lazy_echo_init,
+        )
+    }
+    .unwrap();
+    assert!(host.authenticated_peer("clock").is_none());
+    assert!(
+        host.authenticated_peer("ai.tinyhumans.module.Clock")
+            .is_none()
+    );
+
+    let connection = Connection::connect(bus.connect().await.unwrap())
+        .await
+        .unwrap();
+    let proxy = connection
+        .proxy(
+            "ai.tinyhumans.module.Clock",
+            "/ai/tinyhumans/module/Clock",
+            "ai.tinyhumans.module.Clock",
+        )
+        .unwrap();
+    let _: Option<BusName> = proxy.call("Sender", ()).await.unwrap();
+    let authenticated = host
+        .authenticated_peer("clock")
+        .expect("ready module has an authenticated peer");
+    assert!(authenticated.is_unique());
+    {
+        let mut loaded = host.inner.loaded.lock().unwrap();
+        loaded[0].info.enabled = false;
+    }
+    assert!(host.authenticated_peer("clock").is_none());
+    {
+        let mut loaded = host.inner.loaded.lock().unwrap();
+        loaded[0].info.enabled = true;
+        loaded[0].info.state = ModuleState::Faulted {
+            reason: "fixture".into(),
+        };
+    }
+    assert!(host.authenticated_peer("clock").is_none());
+    host.inner.loaded.lock().unwrap()[0].info.state = ModuleState::Ready;
+    // The fake's Sender method reports its caller (this test connection),
+    // not the module's broker-assigned identity. The compiled-module proof
+    // verifies that an outbound module call carries this identity.
+
+    host.inner
+        .stop("clock", Duration::from_secs(5))
+        .await
+        .unwrap();
+    assert!(host.authenticated_peer("clock").is_none());
+    broker_task.abort();
+}
+
 #[tokio::test]
 async fn one_wedged_module_does_not_stall_another_modules_traffic() {
     let _test_guard = FAKE_MODULE_TEST_LOCK.lock().await;
@@ -2436,7 +2522,7 @@ async fn authenticates_native_context(variable: &str) {
     assert_eq!(
         proxy.attestation().await.unwrap(),
         Some(crate::attest::Attestation {
-            name: info.manifest.bus_name,
+            name: info.manifest.bus_name.clone(),
             sha256: hash
         })
     );
@@ -2447,9 +2533,25 @@ async fn authenticates_native_context(variable: &str) {
         .await
         .unwrap()
         .unwrap();
+    assert_eq!(
+        modules.authenticated_peer(&info.name),
+        Some(module.clone()),
+        "the peer returned by the host API must match the compiled module's broker-stamped sender"
+    );
+    assert!(
+        modules
+            .authenticated_peer(info.manifest.bus_name.as_str())
+            .is_none()
+    );
     let callback: (crate::BusName, crate::BusName) =
         proxy.call_confidential("Callback", ()).await.unwrap();
     assert_eq!(callback, (sender, module));
+    modules
+        .inner
+        .stop(&info.name, Duration::from_secs(5))
+        .await
+        .unwrap();
+    assert!(modules.authenticated_peer(&info.name).is_none());
     task.abort();
 }
 
@@ -2508,6 +2610,7 @@ async fn fixed_unique_module_calls_are_refused_in_terminal_states_before_detach(
         // Keep the real transport attached while setting its terminal state:
         // this probes the pre-detach window without racing the reader task.
         host.inner.loaded.lock().unwrap()[0].info.state = state;
+        assert!(host.authenticated_peer("clock").is_none());
         assert!(
             host.inner
                 .unavailable_for(&client.unique_name().unwrap())
diff --git a/crates/tinybus/src/module/mod.rs b/crates/tinybus/src/module/mod.rs
index 6d78c1a9..a0b80c84 100644
--- a/crates/tinybus/src/module/mod.rs
+++ b/crates/tinybus/src/module/mod.rs
@@ -49,7 +49,7 @@ pub use github::CachedRelease;
 #[cfg(feature = "modules")]
 pub use host::callbacks::{
     HostCallbackBinding, HostCallbackEndpoint, HostCallbackGuard, HostCallbackProxy,
-    HostCallbackSignal,
+    HostCallbackSignal, HostStreamCallbackEndpoint,
 };
 #[cfg(feature = "modules")]
 pub use host::{ModuleHost, ModuleInfo, ModuleState};
diff --git a/crates/tinybus/src/service/tree.rs b/crates/tinybus/src/service/tree.rs
index 4bd8bb6d..4ad8ca32 100644
--- a/crates/tinybus/src/service/tree.rs
+++ b/crates/tinybus/src/service/tree.rs
@@ -42,6 +42,23 @@ impl ObjectTree {
         entry.push(interface);
     }
 
+    /// Export `interface` at `path` only when the complete path is vacant.
+    ///
+    /// Unlike [`Self::insert`], this never replaces a live implementation. The
+    /// caller holds the connection's object-tree write lock across this check
+    /// and insert, so concurrent registrations cannot both claim one address.
+    pub(crate) fn insert_if_absent(
+        &mut self,
+        path: ObjectPath,
+        interface: Arc<dyn Interface>,
+    ) -> bool {
+        if self.objects.contains_key(&path) {
+            return false;
+        }
+        self.objects.insert(path, vec![interface]);
+        true
+    }
+
     /// Stop exporting everything at `path`. Returns whether anything went away.
     pub fn remove(&mut self, path: &ObjectPath) -> bool {
         self.objects.remove(path).is_some()
diff --git a/crates/tinybus/src/stream/mod.rs b/crates/tinybus/src/stream/mod.rs
index b47a72dc..876f9e7a 100644
--- a/crates/tinybus/src/stream/mod.rs
+++ b/crates/tinybus/src/stream/mod.rs
@@ -54,7 +54,7 @@ use std::time::{Duration, Instant};
 
 use serde::{Deserialize, Serialize};
 use serde_json::Value;
-use tokio::sync::{Mutex, mpsc};
+use tokio::sync::{Mutex, mpsc, watch};
 
 use crate::error::{Error, Result};
 use crate::message::Header;
@@ -206,6 +206,10 @@ struct Inbound {
     /// died mid-stream would leave a reader parked on a channel that can never
     /// close, which is precisely the hang this project exists to not have.
     outcome: Arc<std::sync::Mutex<Option<Outcome>>>,
+    /// Wakes a writer parked on a full window when the stream is aborted.
+    /// Unlike dropping the reader, this preserves an unclaimed reader so it can
+    /// still consume buffered chunks and observe the terminal outcome.
+    aborted: watch::Sender<bool>,
     last_activity: std::sync::Mutex<Instant>,
 }
 
@@ -218,7 +222,12 @@ impl Inbound {
     fn finish(&self, outcome: Outcome) {
         let mut slot = self.outcome.lock().expect("stream outcome lock");
         if slot.is_none() {
+            let aborted = matches!(&outcome, Outcome::Aborted(_));
             *slot = Some(outcome);
+            drop(slot);
+            if aborted {
+                self.aborted.send_replace(true);
+            }
         }
     }
 
@@ -332,6 +341,7 @@ impl StreamRegistry {
             chunks: std::sync::Mutex::new(Some(chunks)),
             reader: std::sync::Mutex::new(Some(reader)),
             outcome: Arc::new(std::sync::Mutex::new(None)),
+            aborted: watch::channel(false).0,
             last_activity: std::sync::Mutex::new(Instant::now()),
         });
 
@@ -400,6 +410,7 @@ impl StreamRegistry {
         let (id, seq, data): (String, u64, String) =
             serde_json::from_value(body).map_err(|e| Error::bad_arguments(member.clone(), e))?;
         let stream = self.lookup(&id, header)?;
+        let mut aborted = stream.aborted.subscribe();
         let limits = self.limits();
         let chunk = base64::decode(&data)?;
         if chunk.len() > MAX_CHUNK_LEN {
@@ -436,8 +447,22 @@ impl StreamRegistry {
         stream.touch();
         // The window is the whole flow-control story: this await is where a
         // sender that has run ahead of the reader waits, and the reply it is
-        // waiting on carries the sender's own deadline.
-        let delivered = chunks.send(chunk).await;
+        // waiting on carries the sender's own deadline. Abort also wakes this
+        // send while preserving an unclaimed reader for terminal observation.
+        if *aborted.borrow_and_update() {
+            return Err(Error::StreamAborted {
+                reason: "the sender aborted the stream".to_string(),
+            });
+        }
+        let delivered = tokio::select! {
+            biased;
+            _ = aborted.changed() => {
+                return Err(Error::StreamAborted {
+                    reason: "the sender aborted the stream".to_string(),
+                });
+            }
+            delivered = chunks.send(chunk) => delivered,
+        };
         drop(gate);
         if delivered.is_err() {
             // The reader was dropped. Tell the sender now rather than letting
@@ -477,8 +502,26 @@ impl StreamRegistry {
     fn abort(&self, header: &Header, member: &MemberName, body: Value) -> Result<Value> {
         let (id,): (String,) =
             serde_json::from_value(body).map_err(|e| Error::bad_arguments(member.clone(), e))?;
-        let stream = self.lookup(&id, header)?;
-        self.kill(&id, &stream, "the sender aborted the stream");
+        let mut streams = self.inbound.lock().expect("stream registry lock");
+        let stream = streams
+            .get(&id)
+            .cloned()
+            .ok_or_else(|| Error::UnknownStream { id: id.clone() })?;
+        if stream.owner != header.sender {
+            return Err(Error::UnknownStream { id });
+        }
+
+        stream.finish(Outcome::Aborted("the sender aborted the stream"));
+        stream.touch();
+        stream.seal();
+        // A method carrying the StreamRef can be dispatched after Abort. Keep
+        // an unclaimed reader so that call observes the terminal stream error
+        // instead of a missing handle. The existing closed-unread per-peer
+        // cap and idle reaper bound this retained state. An already-claimed
+        // reader needs no registry entry and is woken by sealing the stream.
+        if stream.reader.lock().expect("stream reader lock").is_none() {
+            streams.remove(&id);
+        }
         Ok(Value::Null)
     }
 
@@ -697,6 +740,10 @@ impl StreamWriter {
     }
 
     /// Abandon the stream, telling the receiver not to wait for the rest.
+    ///
+    /// If the receiver has not accepted the handle yet, it can still accept
+    /// the aborted stream and observe its terminal error. Until accepted, the
+    /// stream uses the same bounded closed-unread slot as a completed stream.
     pub async fn abort(mut self) -> Result<()> {
         self.finished = true;
         self.call("Abort", serde_json::json!([self.id])).await?;
diff --git a/crates/tinybus/src/stream/stream_tests.rs b/crates/tinybus/src/stream/stream_tests.rs
index 4617bf02..ef54ed2f 100644
--- a/crates/tinybus/src/stream/stream_tests.rs
+++ b/crates/tinybus/src/stream/stream_tests.rs
@@ -414,6 +414,95 @@ async fn a_dropped_writer_aborts_the_stream_instead_of_leaving_it_open() {
     );
 }
 
+#[tokio::test]
+async fn an_aborted_stream_remains_available_as_a_terminal_failure_until_accepted() {
+    let (client, service) = bus().await;
+    let destination = BusName::new(SINK).unwrap();
+    let mut writer = client
+        .open_stream(&destination, StreamDescriptor::with_len(4))
+        .await
+        .unwrap();
+    let stream = writer.stream_ref();
+    writer.write_chunk(b"a").await.unwrap();
+    writer.abort().await.unwrap();
+
+    // A stream can reach its terminal state before the method carrying its
+    // handle is dispatched. The late receiver must observe an aborted stream,
+    // rather than a missing handle that masks the sender's original failure.
+    let mut reader = service.accept_stream(&stream).unwrap();
+    assert_eq!(reader.next_chunk().await.unwrap().unwrap(), b"a");
+    assert!(matches!(
+        reader.next_chunk().await,
+        Err(Error::StreamAborted { .. })
+    ));
+}
+
+#[tokio::test]
+async fn abort_wakes_a_pending_write_and_retains_the_unclaimed_reader() {
+    let registry = Arc::new(crate::stream::StreamRegistry::new());
+    registry.set_limits(StreamLimits {
+        window_chunks: 1,
+        ..StreamLimits::default()
+    });
+    let owner: BusName = ":1.77".parse().unwrap();
+    let message = |member: &str, body: Value| {
+        let mut message = Message::method_call(
+            BusName::new(SINK).unwrap(),
+            ObjectPath::new(STREAM_PATH).unwrap(),
+            STREAM_INTERFACE.parse().unwrap(),
+            MemberName::new(member).unwrap(),
+            body,
+        );
+        message.header.sender = Some(owner.clone());
+        message
+    };
+
+    let open = message("Open", serde_json::json!([StreamDescriptor::default()]));
+    let id = registry
+        .dispatch(&open.header, open.body)
+        .await
+        .unwrap()
+        .as_str()
+        .unwrap()
+        .to_owned();
+    let first = message(
+        "Write",
+        serde_json::json!([&id, 0, crate::stream::base64::encode(b"first")]),
+    );
+    registry.dispatch(&first.header, first.body).await.unwrap();
+
+    let stream = registry.inbound.lock().unwrap().get(&id).unwrap().clone();
+    let second = message(
+        "Write",
+        serde_json::json!([&id, 1, crate::stream::base64::encode(b"second")]),
+    );
+    let registry_for_write = Arc::clone(&registry);
+    let pending_write = tokio::spawn(async move {
+        registry_for_write
+            .dispatch(&second.header, second.body)
+            .await
+    });
+    while stream.next_seq.load(std::sync::atomic::Ordering::Relaxed) < 2 {
+        tokio::task::yield_now().await;
+    }
+    assert!(!pending_write.is_finished());
+
+    let abort = message("Abort", serde_json::json!([&id]));
+    registry.dispatch(&abort.header, abort.body).await.unwrap();
+    let write_result = tokio::time::timeout(Duration::from_secs(1), pending_write)
+        .await
+        .expect("abort must wake a chunk write blocked on the full window")
+        .unwrap();
+    assert!(matches!(write_result, Err(Error::StreamAborted { .. })));
+
+    let mut reader = registry.take_reader(&id).unwrap();
+    assert_eq!(reader.next_chunk().await.unwrap().unwrap(), b"first");
+    assert!(matches!(
+        reader.next_chunk().await,
+        Err(Error::StreamAborted { .. })
+    ));
+}
+
 #[tokio::test]
 async fn writing_to_a_stream_nobody_will_read_fails_instead_of_hanging_forever() {
     let (client, service) = bus().await;
@@ -745,6 +834,34 @@ async fn a_closed_stream_nobody_collects_is_evicted_oldest_first() {
     assert_eq!(kept.next_chunk().await.unwrap().unwrap(), b"data");
 }
 
+#[tokio::test]
+async fn an_unclaimed_aborted_stream_uses_the_bounded_closed_unread_slot() {
+    let (client, service) = bus().await;
+    service.set_stream_limits(StreamLimits {
+        max_streams_per_peer: 1,
+        ..StreamLimits::default()
+    });
+    let destination = BusName::new(SINK).unwrap();
+    let mut abandoned = client
+        .open_stream(&destination, StreamDescriptor::with_len(4))
+        .await
+        .unwrap();
+    let abandoned_ref = abandoned.stream_ref();
+    abandoned.write_chunk(b"a").await.unwrap();
+    abandoned.abort().await.unwrap();
+
+    // The next Open evicts an unclaimed terminal stream by the same policy as
+    // an unclaimed completed stream; abort retention cannot grow without bound.
+    let _next = client
+        .open_stream(&destination, StreamDescriptor::default())
+        .await
+        .unwrap();
+    assert!(matches!(
+        service.accept_stream(&abandoned_ref),
+        Err(Error::UnknownStream { .. })
+    ));
+}
+
 #[tokio::test]
 async fn a_closed_stream_still_counts_for_nothing_against_the_live_limit() {
     // Closing frees the slot: a peer that finishes its transfers can keep
diff --git a/scripts/check-coverage.sh b/scripts/check-coverage.sh
index 9f1cd67a..edde89b7 100755
--- a/scripts/check-coverage.sh
+++ b/scripts/check-coverage.sh
@@ -1,7 +1,7 @@
 #!/usr/bin/env bash
 set -euo pipefail
 cd "$(dirname "${BASH_SOURCE[0]}")/.."
-cargo build --locked --example module_clock --example module_clock_two --example module_wrong_target --example module_context --example module_context_eager --example module_host_callback --example module_controlled_init --example module_terminal --no-default-features --features modules,macros,__native-fixture-support
+cargo build --locked --example module_clock --example module_clock_two --example module_wrong_target --example module_context --example module_context_eager --example module_host_callback --example module_host_stream_callback --example module_controlled_init --example module_terminal --no-default-features --features modules,macros,__native-fixture-support
 set -euo pipefail
 target_dir=$(cargo metadata --format-version 1 --no-deps | python3 -c 'import json,sys; print(json.load(sys.stdin)["target_directory"])')
 module=$(find "$target_dir/debug/examples" -maxdepth 1 -type f -name 'libmodule_clock.so' -print -quit)
@@ -10,9 +10,10 @@ wrong_target=$(find "$target_dir/debug/examples" -maxdepth 1 -type f -name 'libm
 context_module=$(find "$target_dir/debug/examples" -maxdepth 1 -type f -name 'libmodule_context.so' -print -quit)
 context_eager_module=$(find "$target_dir/debug/examples" -maxdepth 1 -type f -name 'libmodule_context_eager.so' -print -quit)
 host_callback_module=$(find "$target_dir/debug/examples" -maxdepth 1 -type f \( -name 'libmodule_host_callback.so' -o -name 'libmodule_host_callback.dylib' \) -print -quit)
+host_stream_callback_module=$(find "$target_dir/debug/examples" -maxdepth 1 -type f \( -name 'libmodule_host_stream_callback.so' -o -name 'libmodule_host_stream_callback.dylib' \) -print -quit)
 controlled_init_module=$(find "$target_dir/debug/examples" -maxdepth 1 -type f \( -name 'libmodule_controlled_init.so' -o -name 'libmodule_controlled_init.dylib' \) -print -quit)
 terminal_module=$(find "$target_dir/debug/examples" -maxdepth 1 -type f \( -name 'libmodule_terminal.so' -o -name 'libmodule_terminal.dylib' \) -print -quit)
-for artifact in "$module" "$module_two" "$wrong_target" "$context_module" "$context_eager_module" "$host_callback_module" "$controlled_init_module" "$terminal_module"; do
+for artifact in "$module" "$module_two" "$wrong_target" "$context_module" "$context_eager_module" "$host_callback_module" "$host_stream_callback_module" "$controlled_init_module" "$terminal_module"; do
   if [ -z "$artifact" ] || [ ! -f "$artifact" ]; then
     echo "a module loader fixture is missing" >&2
     exit 1
@@ -50,7 +51,7 @@ export TINYBUS_TEST_TERMINAL_MODULE="$terminal_private/$terminal_library"
 # process/network branches are not meaningful per-file coverage. The
 # test-support module is integration-test glue whose platform-specific
 # dynamic-loader and timeout paths run in the dedicated OS jobs instead.
-export TINYBUS_TEST_MODULE="$module" TINYBUS_TEST_MODULE_TWO="$private/$library" TINYBUS_TEST_WRONG_TARGET="$wrong_target" TINYBUS_TEST_CONTEXT_MODULE="$context_module" TINYBUS_TEST_CONTEXT_EAGER_MODULE="$context_eager_module" TINYBUS_TEST_HOST_CALLBACK_MODULE="$host_callback_module" TINYBUS_TEST_CONTROLLED_INIT_MODULE="$controlled_init_module"
+export TINYBUS_TEST_MODULE="$module" TINYBUS_TEST_MODULE_TWO="$private/$library" TINYBUS_TEST_WRONG_TARGET="$wrong_target" TINYBUS_TEST_CONTEXT_MODULE="$context_module" TINYBUS_TEST_CONTEXT_EAGER_MODULE="$context_eager_module" TINYBUS_TEST_HOST_CALLBACK_MODULE="$host_callback_module" TINYBUS_TEST_HOST_STREAM_CALLBACK_MODULE="$host_stream_callback_module" TINYBUS_TEST_CONTROLLED_INIT_MODULE="$controlled_init_module"
 cargo llvm-cov --workspace --all-features --ignore-filename-regex 'crates/tinybus/src/ports/(listener|transport)\.rs$|crates/tinybus/src/module/github\.rs$|crates/tinybus/src/module/(host|mod)\.rs$|crates/tinybus/src/bin/(tinybus\.rs|tinybus/unix\.rs)$|crates/tinybus/src/connection\.rs$|crates/tinybus/src/test_support\.rs$' --no-report -- --include-ignored --skip terminal_runtime_tests::compiled_native_
 
 terminal_tests=$(cargo test --locked -p tinybus-module --all-features --features tinybus/macros -- --list --ignored | sed -n 's/^\(terminal_runtime_tests::compiled_native_.*\): test$/\1/p')
diff --git a/scripts/test-e2e.ps1 b/scripts/test-e2e.ps1
index 8dfb9467..612fb284 100644
--- a/scripts/test-e2e.ps1
+++ b/scripts/test-e2e.ps1
@@ -22,12 +22,13 @@ Copy-Item (Join-Path $examples module_clock.dll) $private
 Copy-Item (Join-Path $examples module_context.dll) $private
 Copy-Item (Join-Path $examples module_context_eager.dll) $private
 Copy-Item (Join-Path $examples module_host_callback.dll) $private
+Copy-Item (Join-Path $examples module_host_stream_callback.dll) $private
 Copy-Item (Join-Path $examples module_controlled_init.dll) $private
 Copy-Item (Join-Path $examples module_terminal.dll) $private
 Copy-Item (Join-Path $examples module_wrong_target.dll) $private
 # Admit every intentional fixture by its actual built bytes so later
 # checks, including the wrong-target gate, can exercise their contract.
-$fixtureAllowlist = @('module_clock.dll', 'module_context.dll', 'module_context_eager.dll', 'module_host_callback.dll', 'module_controlled_init.dll', 'module_terminal.dll', 'module_wrong_target.dll') | ForEach-Object {
+$fixtureAllowlist = @('module_clock.dll', 'module_context.dll', 'module_context_eager.dll', 'module_host_callback.dll', 'module_host_stream_callback.dll', 'module_controlled_init.dll', 'module_terminal.dll', 'module_wrong_target.dll') | ForEach-Object {
   $digest = (Get-FileHash -LiteralPath (Join-Path $private $_) -Algorithm SHA256).Hash.ToLowerInvariant()
   '"' + $_ + '" = "' + $digest + '"'
 }
@@ -53,6 +54,7 @@ $env:TINYBUS_TEST_CONTEXT_EAGER_MODULE = (Get-Item "$private/module_context_eage
 $env:TINYBUS_TEST_CONTROLLED_INIT_MODULE = (Get-Item "$private/module_controlled_init.dll").FullName
 $env:TINYBUS_TEST_TERMINAL_MODULE = (Get-Item "$private/module_terminal.dll").FullName
 $env:TINYBUS_TEST_HOST_CALLBACK_MODULE = (Get-Item "$private/module_host_callback.dll").FullName
+$env:TINYBUS_TEST_HOST_STREAM_CALLBACK_MODULE = (Get-Item "$private/module_host_stream_callback.dll").FullName
 $env:TINYBUS_TEST_MODULE = $module
 $env:TINYBUS_TEST_MODULE_TWO = $moduleTwo
 $env:TINYBUS_TEST_WRONG_TARGET = $wrongTarget
diff --git a/scripts/test-e2e.sh b/scripts/test-e2e.sh
index 8973c7c5..d50796f1 100755
--- a/scripts/test-e2e.sh
+++ b/scripts/test-e2e.sh
@@ -1,7 +1,7 @@
 #!/usr/bin/env bash
 set -euo pipefail
 cd "$(dirname "${BASH_SOURCE[0]}")/.."
-cargo build --locked --example module_clock --example module_clock_two --example module_wrong_target --example module_context --example module_context_eager --example module_host_callback --example module_controlled_init --example module_terminal --no-default-features --features modules,macros,__native-fixture-support
+cargo build --locked --example module_clock --example module_clock_two --example module_wrong_target --example module_context --example module_context_eager --example module_host_callback --example module_host_stream_callback --example module_controlled_init --example module_terminal --no-default-features --features modules,macros,__native-fixture-support
 cargo test --locked --no-default-features --features modules,macros,test-support test_support::tests
 case "${RUNNER_OS:-$(uname -s)}" in
   Windows|MINGW*|MSYS*|CYGWIN*) pwsh -NoProfile -File scripts/test-e2e.ps1 ;;
@@ -14,9 +14,10 @@ wrong_target=$(find "$target_dir/debug/examples" -maxdepth 1 -type f \( -name 'l
 context_module=$(find "$target_dir/debug/examples" -maxdepth 1 -type f \( -name 'libmodule_context.so' -o -name 'libmodule_context.dylib' \) -print -quit)
 context_eager_module=$(find "$target_dir/debug/examples" -maxdepth 1 -type f \( -name 'libmodule_context_eager.so' -o -name 'libmodule_context_eager.dylib' \) -print -quit)
 host_callback_module=$(find "$target_dir/debug/examples" -maxdepth 1 -type f \( -name 'libmodule_host_callback.so' -o -name 'libmodule_host_callback.dylib' \) -print -quit)
+host_stream_callback_module=$(find "$target_dir/debug/examples" -maxdepth 1 -type f \( -name 'libmodule_host_stream_callback.so' -o -name 'libmodule_host_stream_callback.dylib' \) -print -quit)
 controlled_init_module=$(find "$target_dir/debug/examples" -maxdepth 1 -type f \( -name 'libmodule_controlled_init.so' -o -name 'libmodule_controlled_init.dylib' \) -print -quit)
 terminal_module=$(find "$target_dir/debug/examples" -maxdepth 1 -type f \( -name 'libmodule_terminal.so' -o -name 'libmodule_terminal.dylib' \) -print -quit)
-for artifact in "$module" "$module_two" "$wrong_target" "$context_module" "$context_eager_module" "$host_callback_module" "$controlled_init_module" "$terminal_module"; do
+for artifact in "$module" "$module_two" "$wrong_target" "$context_module" "$context_eager_module" "$host_callback_module" "$host_stream_callback_module" "$controlled_init_module" "$terminal_module"; do
   if [ -z "$artifact" ] || [ ! -f "$artifact" ]; then
     echo "a module example artifact is missing; the loader tests would not exercise it" >&2
     exit 1
@@ -47,7 +48,7 @@ else
 fi
 printf '"%s" = "%s"\n' "$terminal_library" "$terminal_digest" > "$terminal_private/modules.toml"
 export TINYBUS_TEST_TERMINAL_MODULE="$terminal_private/$terminal_library"
-export TINYBUS_TEST_MODULE="$module" TINYBUS_TEST_MODULE_TWO="$private/$library" TINYBUS_TEST_WRONG_TARGET="$wrong_target" TINYBUS_TEST_CONTEXT_MODULE="$context_module" TINYBUS_TEST_CONTEXT_EAGER_MODULE="$context_eager_module" TINYBUS_TEST_HOST_CALLBACK_MODULE="$host_callback_module" TINYBUS_TEST_CONTROLLED_INIT_MODULE="$controlled_init_module"
+export TINYBUS_TEST_MODULE="$module" TINYBUS_TEST_MODULE_TWO="$private/$library" TINYBUS_TEST_WRONG_TARGET="$wrong_target" TINYBUS_TEST_CONTEXT_MODULE="$context_module" TINYBUS_TEST_CONTEXT_EAGER_MODULE="$context_eager_module" TINYBUS_TEST_HOST_CALLBACK_MODULE="$host_callback_module" TINYBUS_TEST_HOST_STREAM_CALLBACK_MODULE="$host_stream_callback_module" TINYBUS_TEST_CONTROLLED_INIT_MODULE="$controlled_init_module"
 cargo test --locked --no-default-features --workspace --features modules,macros,test-support,__native-fixture-support -- --ignored --skip terminal_runtime_tests::compiled_native_
 
 terminal_tests=$(cargo test --locked -p tinybus-module --features __native-fixture-support,tinybus/macros -- --list --ignored | sed -n 's/^\(terminal_runtime_tests::compiled_native_.*\): test$/\1/p')