From 28d0377d007d0ce127a538c57d8cd73a1c4e5570 Mon Sep 17 00:00:00 2001
From: Steven Enamakel <enamakel@tinyhumans.ai>
Date: Tue, 20 Oct 2026 21:01:35 +0300
Subject: [PATCH 01/29] feat(module): add bound ordinary stream callbacks

---
 crates/tinybus/Cargo.toml                     |   5 +
 .../examples/module_host_stream_callback.rs   |  94 ++++
 crates/tinybus/src/module/host_callbacks.rs   | 256 ++++++++++-
 .../src/module/host_callbacks_tests.rs        | 425 +++++++++++++++++-
 crates/tinybus/src/module/mod.rs              |   2 +-
 scripts/check-coverage.sh                     |   7 +-
 scripts/test-e2e.ps1                          |   4 +-
 scripts/test-e2e.sh                           |   7 +-
 8 files changed, 771 insertions(+), 29 deletions(-)
 create mode 100644 crates/tinybus/examples/module_host_stream_callback.rs

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/module/host_callbacks.rs b/crates/tinybus/src/module/host_callbacks.rs
index 56d636c1..63ed2c88 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
@@ -314,6 +355,65 @@ 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> {
+        let state = self.state.lifetime.lock().expect("callback lifetime lock");
+        if context.is_confidential()
+            || self.state.stream_endpoints.is_empty()
+            || !self.state.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"))
+    }
+
     /// Construct an outgoing proxy using this private host connection.
     ///
     /// The outgoing handle cannot expose the owned callback connection:
@@ -384,17 +484,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 +531,8 @@ impl HostCallbackBinding {
                 path.clone(),
                 GuardedInterface {
                     interface,
-                    endpoint: endpoint.clone(),
+                    confidential_endpoint,
+                    stream_endpoint,
                     state: self.state.clone(),
                 },
             )
@@ -451,7 +566,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 +590,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 +624,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 +663,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 +701,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 +828,7 @@ impl ModuleHost {
             state: Arc::new(CallbackState {
                 destination,
                 endpoints,
+                stream_endpoints,
                 signals,
                 module_transport: Mutex::new(None),
                 lifetime: Mutex::new(Lifetime {
@@ -655,7 +876,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..fd283335 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,294 @@ 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 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!(
+        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_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 +1825,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 +1903,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/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/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')

From 73d0dbef6ddeb06390b700a0c13933e8121378b4 Mon Sep 17 00:00:00 2001
From: Steven Enamakel <enamakel@tinyhumans.ai>
Date: Tue, 20 Oct 2026 23:37:48 +0300
Subject: [PATCH 02/29] test(stream): cover accepting an aborted stream after
 queued method call

Add a test asserting that a stream aborted before its handle-carrying method is dispatched still surfaces StreamAborted to a late receiver instead of a missing handle, and correct the ignore reason on the host stream callback test to name the right environment variable.

Auto-committed-on: dragonfly
---
 .../tinybus/src/module/host_callbacks_tests.rs |  2 +-
 crates/tinybus/src/stream/stream_tests.rs      | 18 ++++++++++++++++++
 2 files changed, 19 insertions(+), 1 deletion(-)

diff --git a/crates/tinybus/src/module/host_callbacks_tests.rs b/crates/tinybus/src/module/host_callbacks_tests.rs
index fd283335..7b40af90 100644
--- a/crates/tinybus/src/module/host_callbacks_tests.rs
+++ b/crates/tinybus/src/module/host_callbacks_tests.rs
@@ -877,7 +877,7 @@ impl Interface for ActualStreamCallback {
 }
 
 #[tokio::test]
-#[ignore = "requires TINYBUS_TEST_HOST_CALLBACK_MODULE built cdylib"]
+#[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(),
diff --git a/crates/tinybus/src/stream/stream_tests.rs b/crates/tinybus/src/stream/stream_tests.rs
index 4617bf02..566a4dcd 100644
--- a/crates/tinybus/src/stream/stream_tests.rs
+++ b/crates/tinybus/src/stream/stream_tests.rs
@@ -414,6 +414,24 @@ async fn a_dropped_writer_aborts_the_stream_instead_of_leaving_it_open() {
     );
 }
 
+#[tokio::test]
+async fn an_aborted_stream_can_still_be_accepted_after_its_method_call_was_queued() {
+    let (client, service) = bus().await;
+    let destination = BusName::new(SINK).unwrap();
+    let writer = client
+        .open_stream(&destination, StreamDescriptor::with_len(4))
+        .await
+        .unwrap();
+    let stream = writer.stream_ref();
+    writer.abort().await.unwrap();
+
+    // The writer can reach its terminal state before the method carrying this
+    // 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!(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;

From c5928d67246fd08fa7e2e21d2f74aa956249104c Mon Sep 17 00:00:00 2001
From: Steven Enamakel <enamakel@tinyhumans.ai>
Date: Tue, 20 Oct 2026 23:39:02 +0300
Subject: [PATCH 03/29] test(stream): cover buffered chunk before abort on late
 accept

The late-accept test now writes a chunk before aborting and asserts the
reader still receives that buffered data before observing the terminal
abort error, so the test reflects that a stream remains available as a
terminal failure until it is accepted.

Auto-committed-on: dragonfly
---
 crates/tinybus/src/stream/stream_tests.rs | 13 +++++++++----
 1 file changed, 9 insertions(+), 4 deletions(-)

diff --git a/crates/tinybus/src/stream/stream_tests.rs b/crates/tinybus/src/stream/stream_tests.rs
index 566a4dcd..442d0854 100644
--- a/crates/tinybus/src/stream/stream_tests.rs
+++ b/crates/tinybus/src/stream/stream_tests.rs
@@ -415,21 +415,26 @@ async fn a_dropped_writer_aborts_the_stream_instead_of_leaving_it_open() {
 }
 
 #[tokio::test]
-async fn an_aborted_stream_can_still_be_accepted_after_its_method_call_was_queued() {
+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 writer = client
+    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();
 
-    // The writer can reach its terminal state before the method carrying this
+    // 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!(matches!(reader.next_chunk().await, Err(Error::StreamAborted { .. })));
+    assert_eq!(reader.next_chunk().await.unwrap().unwrap(), b"a");
+    assert!(matches!(
+        reader.next_chunk().await,
+        Err(Error::StreamAborted { .. })
+    ));
 }
 
 #[tokio::test]

From 6c38f6c1bda454804fdca2f039d1a0e0cc73bb8a Mon Sep 17 00:00:00 2001
From: Steven Enamakel <enamakel@tinyhumans.ai>
Date: Tue, 20 Oct 2026 23:49:24 +0300
Subject: [PATCH 04/29] fix(stream): keep aborted streams readable until
 claimed

Abort now marks the stream finished and sealed while retaining an
unclaimed reader, so a method carrying the StreamRef that is dispatched
after Abort observes the terminal stream error instead of a missing
handle. Ownership is checked against the sender and the retained entry is
dropped once a reader has claimed it, bounding the state through the
existing closed-unread cap and idle reaper.

Auto-committed-on: dragonfly
---
 crates/tinybus/src/stream/mod.rs | 26 ++++++++++++++++++++++++--
 1 file changed, 24 insertions(+), 2 deletions(-)

diff --git a/crates/tinybus/src/stream/mod.rs b/crates/tinybus/src/stream/mod.rs
index b47a72dc..544cbe93 100644
--- a/crates/tinybus/src/stream/mod.rs
+++ b/crates/tinybus/src/stream/mod.rs
@@ -477,8 +477,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 +715,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?;

From 19e5d5ef0f1fe4dbfed9b4b360109b221c057cf7 Mon Sep 17 00:00:00 2001
From: Steven Enamakel <enamakel@tinyhumans.ai>
Date: Tue, 20 Oct 2026 23:51:16 +0300
Subject: [PATCH 05/29] test(stream): cover eviction of unclaimed aborted
 streams

Add a test asserting that an unclaimed aborted stream is evicted by the
same bounded closed-unread policy as an unclaimed completed stream, so
abort retention cannot grow without bound.

Auto-committed-on: dragonfly
---
 crates/tinybus/src/stream/stream_tests.rs | 28 +++++++++++++++++++++++
 1 file changed, 28 insertions(+)

diff --git a/crates/tinybus/src/stream/stream_tests.rs b/crates/tinybus/src/stream/stream_tests.rs
index 442d0854..852b954a 100644
--- a/crates/tinybus/src/stream/stream_tests.rs
+++ b/crates/tinybus/src/stream/stream_tests.rs
@@ -768,6 +768,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

From 7c951c244f9ae4b1a6c79016209f15e7bc64c497 Mon Sep 17 00:00:00 2001
From: Steven Enamakel <enamakel@tinyhumans.ai>
Date: Tue, 20 Oct 2026 23:52:18 +0300
Subject: [PATCH 06/29] chore(deps): bump rustls-webpki and webpki-roots

Update the locked versions of rustls-webpki to 0.103.16 and webpki-roots to
1.0.10, refreshing their checksums accordingly.

Auto-committed-on: dragonfly
---
 Cargo.lock | 8 ++++----
 1 file changed, 4 insertions(+), 4 deletions(-)

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",
 ]

From fb00a8c8ceda52b27aa9e98d5836cf10b0af2e3d Mon Sep 17 00:00:00 2001
From: Steven Enamakel <enamakel@tinyhumans.ai>
Date: Wed, 21 Oct 2026 15:14:56 +0300
Subject: [PATCH 07/29] test(host): add test for authenticated peer
 availability

Adds a new test case to verify that an authenticated peer is only available after a module is live and ready. This ensures correct lifecycle management for module interactions.

Auto-committed-on: dragonfly
---
 crates/tinybus/src/module/host_tests.rs | 47 +++++++++++++++++++++++++
 1 file changed, 47 insertions(+)

diff --git a/crates/tinybus/src/module/host_tests.rs b/crates/tinybus/src/module/host_tests.rs
index a5999da0..cadefc51 100644
--- a/crates/tinybus/src/module/host_tests.rs
+++ b/crates/tinybus/src/module/host_tests.rs
@@ -1053,6 +1053,53 @@ 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 sender: Option<BusName> = proxy.call("Sender", ()).await.unwrap();
+    let authenticated = host
+        .authenticated_peer("clock")
+        .expect("ready module has an authenticated peer");
+    assert!(authenticated.is_unique());
+    assert_eq!(sender.as_ref(), Some(&authenticated));
+
+    host.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;

From db8b4b7590de75be1d9b8d9f6a0ed815d8ae532f Mon Sep 17 00:00:00 2001
From: Steven Enamakel <enamakel@tinyhumans.ai>
Date: Wed, 21 Oct 2026 15:22:29 +0300
Subject: [PATCH 08/29] test(host): update host stop call in tests

This updates the host stop call within the host tests to use the inner field directly. This change aligns the test implementation with the current internal structure of the host.

Auto-committed-on: dragonfly
---
 crates/tinybus/src/module/host_tests.rs | 5 ++++-
 1 file changed, 4 insertions(+), 1 deletion(-)

diff --git a/crates/tinybus/src/module/host_tests.rs b/crates/tinybus/src/module/host_tests.rs
index cadefc51..a0cbbc6c 100644
--- a/crates/tinybus/src/module/host_tests.rs
+++ b/crates/tinybus/src/module/host_tests.rs
@@ -1095,7 +1095,10 @@ async fn authenticated_peer_is_available_only_for_a_live_ready_module() {
     assert!(authenticated.is_unique());
     assert_eq!(sender.as_ref(), Some(&authenticated));
 
-    host.stop("clock", Duration::from_secs(5)).await.unwrap();
+    host.inner
+        .stop("clock", Duration::from_secs(5))
+        .await
+        .unwrap();
     assert!(host.authenticated_peer("clock").is_none());
     broker_task.abort();
 }

From cde303d0006d9ba5f41bc14d55d45b1ea01e3e36 Mon Sep 17 00:00:00 2001
From: Steven Enamakel <enamakel@tinyhumans.ai>
Date: Wed, 21 Oct 2026 15:23:56 +0300
Subject: [PATCH 09/29] feat(host): add authenticated_peer method

Introduces `authenticated_peer` to return the broker-assigned unique peer name for a module only if it is live and ready. This ensures that peer names are only provided for modules that are fully operational and capable of serving requests.

Auto-committed-on: dragonfly
---
 crates/tinybus/src/module/host.rs       | 34 +++++++++++++++++++++++++
 crates/tinybus/src/module/host_tests.rs |  7 ++---
 2 files changed, 38 insertions(+), 3 deletions(-)

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_tests.rs b/crates/tinybus/src/module/host_tests.rs
index a0cbbc6c..43e729ab 100644
--- a/crates/tinybus/src/module/host_tests.rs
+++ b/crates/tinybus/src/module/host_tests.rs
@@ -1074,9 +1074,10 @@ async fn authenticated_peer_is_available_only_for_a_live_ready_module() {
     }
     .unwrap();
     assert!(host.authenticated_peer("clock").is_none());
-    assert!(host
-        .authenticated_peer("ai.tinyhumans.module.Clock")
-        .is_none());
+    assert!(
+        host.authenticated_peer("ai.tinyhumans.module.Clock")
+            .is_none()
+    );
 
     let connection = Connection::connect(bus.connect().await.unwrap())
         .await

From 1cf904e0e94fb70573cb8997117e6d78fd428e2a Mon Sep 17 00:00:00 2001
From: Steven Enamakel <enamakel@tinyhumans.ai>
Date: Wed, 21 Oct 2026 15:26:14 +0300
Subject: [PATCH 10/29] test(host): update authenticated peer test logic

The test for authenticated peer availability was updated to use the host's direct peer access instead of establishing a full connection and proxy. This change simplifies the test while still verifying the expected behavior regarding authenticated peers.

Auto-committed-on: dragonfly
---
 crates/tinybus/src/module/host_tests.rs | 15 +++------------
 1 file changed, 3 insertions(+), 12 deletions(-)

diff --git a/crates/tinybus/src/module/host_tests.rs b/crates/tinybus/src/module/host_tests.rs
index 43e729ab..03aaafc7 100644
--- a/crates/tinybus/src/module/host_tests.rs
+++ b/crates/tinybus/src/module/host_tests.rs
@@ -1079,22 +1079,13 @@ async fn authenticated_peer_is_available_only_for_a_live_ready_module() {
             .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 sender: Option<BusName> = proxy.call("Sender", ()).await.unwrap();
     let authenticated = host
         .authenticated_peer("clock")
         .expect("ready module has an authenticated peer");
     assert!(authenticated.is_unique());
-    assert_eq!(sender.as_ref(), Some(&authenticated));
+    // 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))

From 0b0111424f0527935d8e8818caafc13c5b126d34 Mon Sep 17 00:00:00 2001
From: Steven Enamakel <enamakel@tinyhumans.ai>
Date: Wed, 21 Oct 2026 15:27:01 +0300
Subject: [PATCH 11/29] test(host): add clock module test case

Adds a test case to verify that the authenticated peer for the Clock module is available after establishing a connection and proxy. This ensures the module can be correctly accessed via the bus.

Auto-committed-on: dragonfly
---
 crates/tinybus/src/module/host_tests.rs | 11 +++++++++++
 1 file changed, 11 insertions(+)

diff --git a/crates/tinybus/src/module/host_tests.rs b/crates/tinybus/src/module/host_tests.rs
index 03aaafc7..9eda9437 100644
--- a/crates/tinybus/src/module/host_tests.rs
+++ b/crates/tinybus/src/module/host_tests.rs
@@ -1079,6 +1079,17 @@ async fn authenticated_peer_is_available_only_for_a_live_ready_module() {
             .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");

From 50f3b8bf76a107f4206bcac8dfe9d8b835917fb8 Mon Sep 17 00:00:00 2001
From: Steven Enamakel <enamakel@tinyhumans.ai>
Date: Wed, 21 Oct 2026 15:30:03 +0300
Subject: [PATCH 12/29] test(host): add assertions to authentication test

Adds assertions to the host authentication test to verify that the authenticated peer matches the expected module and that the bus name is not registered as authenticated afterward. This ensures proper cleanup and state verification after a successful authentication flow.

Auto-committed-on: dragonfly
---
 crates/tinybus/src/module/host_tests.rs | 14 ++++++++++++++
 1 file changed, 14 insertions(+)

diff --git a/crates/tinybus/src/module/host_tests.rs b/crates/tinybus/src/module/host_tests.rs
index 9eda9437..5653582d 100644
--- a/crates/tinybus/src/module/host_tests.rs
+++ b/crates/tinybus/src/module/host_tests.rs
@@ -2500,9 +2500,23 @@ 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();
 }
 

From a16e37447f840a357fd9e0665d7a0b032eae32b5 Mon Sep 17 00:00:00 2001
From: Steven Enamakel <enamakel@tinyhumans.ai>
Date: Wed, 21 Oct 2026 15:33:15 +0300
Subject: [PATCH 13/29] test(host): update attestation assertion in host tests

This change updates the assertion in `authenticates_native_context` to clone the bus name when comparing the attestation result. This ensures correct ownership when asserting against the expected structure.

Auto-committed-on: dragonfly
---
 crates/tinybus/src/module/host_tests.rs | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/crates/tinybus/src/module/host_tests.rs b/crates/tinybus/src/module/host_tests.rs
index 5653582d..469dfc63 100644
--- a/crates/tinybus/src/module/host_tests.rs
+++ b/crates/tinybus/src/module/host_tests.rs
@@ -2489,7 +2489,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
         })
     );

From 4eb60da023a39a563f0ca091ae89455f8d517733 Mon Sep 17 00:00:00 2001
From: Steven Enamakel <enamakel@tinyhumans.ai>
Date: Wed, 21 Oct 2026 15:37:02 +0300
Subject: [PATCH 14/29] test(host): improve assertion formatting in host tests

This change refactors the assertion in `host_tests.rs` to improve readability by breaking the assertion across multiple lines. This does not change the test's behavior.

Auto-committed-on: dragonfly
---
 crates/tinybus/src/module/host_tests.rs | 8 +++++---
 1 file changed, 5 insertions(+), 3 deletions(-)

diff --git a/crates/tinybus/src/module/host_tests.rs b/crates/tinybus/src/module/host_tests.rs
index 469dfc63..2ff6d8b7 100644
--- a/crates/tinybus/src/module/host_tests.rs
+++ b/crates/tinybus/src/module/host_tests.rs
@@ -2505,9 +2505,11 @@ async fn authenticates_native_context(variable: &str) {
         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());
+    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));

From 2f2319ffd9ceff8942a4a78b883fe9b98d0cae5d Mon Sep 17 00:00:00 2001
From: Steven Enamakel <enamakel@tinyhumans.ai>
Date: Wed, 21 Oct 2026 15:40:05 +0300
Subject: [PATCH 15/29] test(host): add assertion for peer authentication in
 terminal state test

This adds an assertion to ensure that the peer is not authenticated when the host is in a terminal state before detachment. This helps verify the state transition logic under specific conditions.

Auto-committed-on: dragonfly
---
 crates/tinybus/src/module/host_tests.rs | 1 +
 1 file changed, 1 insertion(+)

diff --git a/crates/tinybus/src/module/host_tests.rs b/crates/tinybus/src/module/host_tests.rs
index 2ff6d8b7..99d336e7 100644
--- a/crates/tinybus/src/module/host_tests.rs
+++ b/crates/tinybus/src/module/host_tests.rs
@@ -2577,6 +2577,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())

From 65c6d6bc7cd2e5a6a0107901f613789e1935074e Mon Sep 17 00:00:00 2001
From: Steven Enamakel <enamakel@tinyhumans.ai>
Date: Wed, 21 Oct 2026 15:42:19 +0300
Subject: [PATCH 16/29] test(host): add test case for peer availability on
 module state change

This adds a test case to verify that an authenticated peer becomes unavailable when the module's info is disabled and then unavailable again when the module is faulted. This ensures correct peer management during module state transitions.

Auto-committed-on: dragonfly
---
 crates/tinybus/src/module/host_tests.rs | 14 ++++++++++++++
 1 file changed, 14 insertions(+)

diff --git a/crates/tinybus/src/module/host_tests.rs b/crates/tinybus/src/module/host_tests.rs
index 99d336e7..4863096c 100644
--- a/crates/tinybus/src/module/host_tests.rs
+++ b/crates/tinybus/src/module/host_tests.rs
@@ -1094,6 +1094,20 @@ async fn authenticated_peer_is_available_only_for_a_live_ready_module() {
         .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.

From d4284fa812a2fb44546c015660aaa6838bbb000c Mon Sep 17 00:00:00 2001
From: Steven Enamakel <enamakel@tinyhumans.ai>
Date: Wed, 21 Oct 2026 16:35:19 +0300
Subject: [PATCH 17/29] feat(connection): add try_serve_at for non-replacing
 interface export

Adds Connection::try_serve_at, which exports an interface at a path only
when that interface name is not already served there, returning an error
instead of replacing the live implementation. The check and insert happen
under the object-tree write lock so concurrent registrations cannot both
claim the same address.

Auto-committed-on: dragonfly
---
 crates/tinybus/src/connection.rs       | 20 +++++++++
 crates/tinybus/src/connection_tests.rs | 57 ++++++++++++++++++++++++++
 crates/tinybus/src/service/tree.rs     | 22 ++++++++++
 3 files changed, 99 insertions(+)

diff --git a/crates/tinybus/src/connection.rs b/crates/tinybus/src/connection.rs
index 867532ab..a162620c 100644
--- a/crates/tinybus/src/connection.rs
+++ b/crates/tinybus/src/connection.rs
@@ -456,6 +456,26 @@ impl Connection {
         Ok(())
     }
 
+    /// Export `interface` at `path` only if the same interface is not already
+    /// served there.
+    ///
+    /// Registration is atomic with respect to other calls to this method: a
+    /// collision leaves the existing implementation 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..896fe46d 100644
--- a/crates/tinybus/src/connection_tests.rs
+++ b/crates/tinybus/src/connection_tests.rs
@@ -39,10 +39,67 @@ 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, Value::String("original".to_string()));
+}
+
+#[tokio::test]
+async fn concurrent_try_serve_at_calls_have_exactly_one_winner() {
+    let (client, service) = brokered_pair().await;
+    let (first, second) = tokio::join!(
+        service.try_serve_at(path(), ReplacementEcho),
+        service.try_serve_at(path(), Echo),
+    );
+    assert_ne!(first.is_ok(), second.is_ok());
+
+    let result: Value = client
+        .proxy("ai.tinyhumans.Test", 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, Value::String("original".to_string()));
+    }
+}
+
 fn call(member: &str, body: Value) -> Message {
     Message::method_call(
         BusName::new("ai.tinyhumans.Test").unwrap(),
diff --git a/crates/tinybus/src/service/tree.rs b/crates/tinybus/src/service/tree.rs
index 4bd8bb6d..5d6209ec 100644
--- a/crates/tinybus/src/service/tree.rs
+++ b/crates/tinybus/src/service/tree.rs
@@ -42,6 +42,28 @@ impl ObjectTree {
         entry.push(interface);
     }
 
+    /// Export `interface` at `path` only when that interface name 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 {
+        let name = interface.name();
+        if self
+            .objects
+            .get(&path)
+            .is_some_and(|existing| existing.iter().any(|item| item.name() == name))
+        {
+            return false;
+        }
+        self.objects.entry(path).or_default().push(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()

From 07f529263d939ebb70a5c3c3124cd1e488797b44 Mon Sep 17 00:00:00 2001
From: Steven Enamakel <enamakel@tinyhumans.ai>
Date: Wed, 21 Oct 2026 16:39:33 +0300
Subject: [PATCH 18/29] style(tinybus): reformat error and tidy connection
 tests

Reformatted a multi-line error construction in the connection code and
updated the connection tests to use a dedicated object path for the
concurrent serve case and to compare against the expected JSON value.

Auto-committed-on: dragonfly
---
 crates/tinybus/src/connection.rs       | 4 +++-
 crates/tinybus/src/connection_tests.rs | 9 +++++----
 2 files changed, 8 insertions(+), 5 deletions(-)

diff --git a/crates/tinybus/src/connection.rs b/crates/tinybus/src/connection.rs
index a162620c..b20bf3e5 100644
--- a/crates/tinybus/src/connection.rs
+++ b/crates/tinybus/src/connection.rs
@@ -472,7 +472,9 @@ impl Connection {
         {
             Ok(())
         } else {
-            Err(Error::failed("interface is already served at this object path"))
+            Err(Error::failed(
+                "interface is already served at this object path",
+            ))
         }
     }
 
diff --git a/crates/tinybus/src/connection_tests.rs b/crates/tinybus/src/connection_tests.rs
index 896fe46d..5bdc0b86 100644
--- a/crates/tinybus/src/connection_tests.rs
+++ b/crates/tinybus/src/connection_tests.rs
@@ -75,20 +75,21 @@ async fn try_serve_at_refuses_a_collision_without_replacing_the_live_interface()
         .call("Echo", ("original",))
         .await
         .unwrap();
-    assert_eq!(result, Value::String("original".to_string()));
+    assert_eq!(result, serde_json::json!(["original"]));
 }
 
 #[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(path(), ReplacementEcho),
-        service.try_serve_at(path(), Echo),
+        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", path(), "ai.tinyhumans.Test")
+        .proxy("ai.tinyhumans.Test", concurrent_path, "ai.tinyhumans.Test")
         .unwrap()
         .call("Echo", ("original",))
         .await

From a30dd2b51a9f06a9c1a6506043a00fcc9654284e Mon Sep 17 00:00:00 2001
From: Steven Enamakel <enamakel@tinyhumans.ai>
Date: Wed, 21 Oct 2026 16:41:07 +0300
Subject: [PATCH 19/29] test(tinybus): cover serve_at replacement behaviour

Add a test asserting that serve_at still replaces an existing interface
and routes calls to the replacement, guarding the behaviour alongside the
new collision-refusing try_serve_at path.

Auto-committed-on: dragonfly
---
 crates/tinybus/src/connection_tests.rs | 14 ++++++++++++++
 1 file changed, 14 insertions(+)

diff --git a/crates/tinybus/src/connection_tests.rs b/crates/tinybus/src/connection_tests.rs
index 5bdc0b86..bfdb85d5 100644
--- a/crates/tinybus/src/connection_tests.rs
+++ b/crates/tinybus/src/connection_tests.rs
@@ -78,6 +78,20 @@ async fn try_serve_at_refuses_a_collision_without_replacing_the_live_interface()
     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;

From a0d62d1a6dc518d048c799a7ae51483a56dfb47b Mon Sep 17 00:00:00 2001
From: Steven Enamakel <enamakel@tinyhumans.ai>
Date: Wed, 21 Oct 2026 16:48:29 +0300
Subject: [PATCH 20/29] fix(tinybus): refuse try_serve_at when any interface
 occupies the path

try_serve_at previously rejected a registration only when an interface with
the same name was already served at the path, so a second, differently named
interface could be added alongside it. It now treats any occupied path as a
collision and leaves every existing interface untouched, matching the
documented contract.

Auto-committed-on: dragonfly
---
 crates/tinybus/src/connection.rs       |  6 ++---
 crates/tinybus/src/connection_tests.rs | 37 +++++++++++++++++++++++++-
 crates/tinybus/src/service/tree.rs     | 11 +++-----
 3 files changed, 42 insertions(+), 12 deletions(-)

diff --git a/crates/tinybus/src/connection.rs b/crates/tinybus/src/connection.rs
index b20bf3e5..673f2ddb 100644
--- a/crates/tinybus/src/connection.rs
+++ b/crates/tinybus/src/connection.rs
@@ -456,11 +456,11 @@ impl Connection {
         Ok(())
     }
 
-    /// Export `interface` at `path` only if the same interface is not already
-    /// served there.
+    /// 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 the existing implementation untouched. Use
+    /// 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
diff --git a/crates/tinybus/src/connection_tests.rs b/crates/tinybus/src/connection_tests.rs
index bfdb85d5..4fbcaa9a 100644
--- a/crates/tinybus/src/connection_tests.rs
+++ b/crates/tinybus/src/connection_tests.rs
@@ -111,10 +111,45 @@ async fn concurrent_try_serve_at_calls_have_exactly_one_winner() {
     if first.is_ok() {
         assert_eq!(result, Value::String("replacement".to_string()));
     } else {
-        assert_eq!(result, Value::String("original".to_string()));
+        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/service/tree.rs b/crates/tinybus/src/service/tree.rs
index 5d6209ec..4ad8ca32 100644
--- a/crates/tinybus/src/service/tree.rs
+++ b/crates/tinybus/src/service/tree.rs
@@ -42,7 +42,7 @@ impl ObjectTree {
         entry.push(interface);
     }
 
-    /// Export `interface` at `path` only when that interface name is vacant.
+    /// 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
@@ -52,15 +52,10 @@ impl ObjectTree {
         path: ObjectPath,
         interface: Arc<dyn Interface>,
     ) -> bool {
-        let name = interface.name();
-        if self
-            .objects
-            .get(&path)
-            .is_some_and(|existing| existing.iter().any(|item| item.name() == name))
-        {
+        if self.objects.contains_key(&path) {
             return false;
         }
-        self.objects.entry(path).or_default().push(interface);
+        self.objects.insert(path, vec![interface]);
         true
     }
 

From 7fbf62d536fb31502136752be062a78d6e22dc33 Mon Sep 17 00:00:00 2001
From: Steven Enamakel <enamakel@tinyhumans.ai>
Date: Wed, 21 Oct 2026 16:48:35 +0300
Subject: [PATCH 21/29] test(tinybus): reformat serve_at call in occupied path
 test

Collapse the multi-line serve_at call in the try_serve_at test into a single line. No behaviour change.

Auto-committed-on: dragonfly
---
 crates/tinybus/src/connection_tests.rs | 5 +----
 1 file changed, 1 insertion(+), 4 deletions(-)

diff --git a/crates/tinybus/src/connection_tests.rs b/crates/tinybus/src/connection_tests.rs
index 4fbcaa9a..a23e2eda 100644
--- a/crates/tinybus/src/connection_tests.rs
+++ b/crates/tinybus/src/connection_tests.rs
@@ -135,10 +135,7 @@ impl Interface for DifferentInterface {
 #[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();
+    service.serve_at(path(), DifferentInterface).await.unwrap();
     assert!(service.try_serve_at(path(), ReplacementEcho).await.is_err());
 
     let result: Value = client

From 8fcff92de9e4eb1a94fa267759ae7d246b3fe801 Mon Sep 17 00:00:00 2001
From: Steven Enamakel <enamakel@tinyhumans.ai>
Date: Thu, 22 Oct 2026 01:19:21 +0000
Subject: [PATCH 22/29] test(stream): cover abort waking a blocked write and
 guard stream authorization

Add a test asserting that aborting a stream wakes a chunk write blocked on a
full window and still lets the reader drain the unclaimed chunk before seeing
the abort. Also extend the host callback tests to check that the binding guard
authorizes streams for the module peer only, and stops doing so once revoked.

Auto-committed-on: cuttlefish
Co-authored-by: Medulla <medulla@tinyhumans.ai>
---
 .../src/module/host_callbacks_tests.rs        |  6 ++
 crates/tinybus/src/stream/stream_tests.rs     | 72 +++++++++++++++++++
 2 files changed, 78 insertions(+)

diff --git a/crates/tinybus/src/module/host_callbacks_tests.rs b/crates/tinybus/src/module/host_callbacks_tests.rs
index 7b40af90..03c236df 100644
--- a/crates/tinybus/src/module/host_callbacks_tests.rs
+++ b/crates/tinybus/src/module/host_callbacks_tests.rs
@@ -724,6 +724,11 @@ async fn callback_streams_reject_other_peers_nonconfidential_calls_and_revoked_b
     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,
@@ -762,6 +767,7 @@ async fn callback_streams_reject_other_peers_nonconfidential_calls_and_revoked_b
     );
 
     binding.revoke();
+    assert!(!guard.authorizes_stream(&context));
     assert!(
         binding
             .accept_stream_from_module(&context, &stream)
diff --git a/crates/tinybus/src/stream/stream_tests.rs b/crates/tinybus/src/stream/stream_tests.rs
index 852b954a..cde7680a 100644
--- a/crates/tinybus/src/stream/stream_tests.rs
+++ b/crates/tinybus/src/stream/stream_tests.rs
@@ -437,6 +437,78 @@ async fn an_aborted_stream_remains_available_as_a_terminal_failure_until_accepte
     ));
 }
 
+#[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;

From fffc0e835d23fc74214de72ed8ec35f9c21bce60 Mon Sep 17 00:00:00 2001
From: Steven Enamakel <enamakel@tinyhumans.ai>
Date: Thu, 22 Oct 2026 01:20:16 +0000
Subject: [PATCH 23/29] refactor(tinybus): share stream authority check between
 callback guards

Moved the stream authority check onto the shared callback state so the
host callback guard can expose an authorizes_stream predicate alongside
the existing confidential authorizes check. The binding now delegates to
the same helper instead of duplicating the logic.

Auto-committed-on: cuttlefish
Co-authored-by: Medulla <medulla@tinyhumans.ai>
---
 crates/tinybus/src/module/host_callbacks.rs | 43 ++++++++++++++-------
 1 file changed, 28 insertions(+), 15 deletions(-)

diff --git a/crates/tinybus/src/module/host_callbacks.rs b/crates/tinybus/src/module/host_callbacks.rs
index 63ed2c88..8103fa9d 100644
--- a/crates/tinybus/src/module/host_callbacks.rs
+++ b/crates/tinybus/src/module/host_callbacks.rs
@@ -204,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.
@@ -219,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
@@ -229,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 {
@@ -398,20 +424,7 @@ impl HostCallbackBinding {
     }
 
     fn authorized_stream_module(&self, context: &CallContext) -> Result<BusName> {
-        let state = self.state.lifetime.lock().expect("callback lifetime lock");
-        if context.is_confidential()
-            || self.state.stream_endpoints.is_empty()
-            || !self.state.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"))
+        self.state.authorized_stream_module(context)
     }
 
     /// Construct an outgoing proxy using this private host connection.

From 2eb9386b4dea19c25b4287d882cb209f5d2756df Mon Sep 17 00:00:00 2001
From: Steven Enamakel <enamakel@tinyhumans.ai>
Date: Thu, 22 Oct 2026 01:20:42 +0000
Subject: [PATCH 24/29] fix(stream): wake writers parked on a full window when
 a stream is aborted

A watch channel now signals abort to writers blocked on a full chunk window, so they fail with StreamAborted instead of hanging until their deadline. This preserves an unclaimed reader so it can still drain buffered chunks and observe the terminal outcome.

Auto-committed-on: cuttlefish
Co-authored-by: Medulla <medulla@tinyhumans.ai>
---
 crates/tinybus/src/stream/mod.rs | 31 ++++++++++++++++++++++++++++---
 1 file changed, 28 insertions(+), 3 deletions(-)

diff --git a/crates/tinybus/src/stream/mod.rs b/crates/tinybus/src/stream/mod.rs
index 544cbe93..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

From 73ce53098cf9f796a9d88a4253ca7c89dadbbebd Mon Sep 17 00:00:00 2001
From: Steven Enamakel <enamakel@tinyhumans.ai>
Date: Thu, 22 Oct 2026 01:20:54 +0000
Subject: [PATCH 25/29] test(stream): collapse multi-line dispatch calls in
 abort test

Reformatted two registry dispatch calls in the abort test to single lines, matching rustfmt output. No behaviour change.

Auto-committed-on: cuttlefish
Co-authored-by: Medulla <medulla@tinyhumans.ai>
---
 crates/tinybus/src/stream/stream_tests.rs | 10 ++--------
 1 file changed, 2 insertions(+), 8 deletions(-)

diff --git a/crates/tinybus/src/stream/stream_tests.rs b/crates/tinybus/src/stream/stream_tests.rs
index cde7680a..ef54ed2f 100644
--- a/crates/tinybus/src/stream/stream_tests.rs
+++ b/crates/tinybus/src/stream/stream_tests.rs
@@ -469,10 +469,7 @@ async fn abort_wakes_a_pending_write_and_retains_the_unclaimed_reader() {
         "Write",
         serde_json::json!([&id, 0, crate::stream::base64::encode(b"first")]),
     );
-    registry
-        .dispatch(&first.header, first.body)
-        .await
-        .unwrap();
+    registry.dispatch(&first.header, first.body).await.unwrap();
 
     let stream = registry.inbound.lock().unwrap().get(&id).unwrap().clone();
     let second = message(
@@ -491,10 +488,7 @@ async fn abort_wakes_a_pending_write_and_retains_the_unclaimed_reader() {
     assert!(!pending_write.is_finished());
 
     let abort = message("Abort", serde_json::json!([&id]));
-    registry
-        .dispatch(&abort.header, abort.body)
-        .await
-        .unwrap();
+    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")

From 5e85165b0619424fc82885ed6a23a90b5b97108f Mon Sep 17 00:00:00 2001
From: Steven Enamakel <enamakel@tinyhumans.ai>
Date: Thu, 22 Oct 2026 01:21:47 +0000
Subject: [PATCH 26/29] test(tinybus): make bundle fixture dirs owner-only

The marked_bundle helper now restricts the artifact directory and its
ancestors to mode 0700 so the fixture matches the permissions the
download cache expects.

Auto-committed-on: cuttlefish
Co-authored-by: Medulla <medulla@tinyhumans.ai>
---
 crates/tinybus/src/module/first_admitted_tests.rs | 4 ++++
 1 file changed, 4 insertions(+)

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),

From a557d2b540056563b4f6e32dbc08b87f3749ec8f Mon Sep 17 00:00:00 2001
From: Steven Enamakel <enamakel@tinyhumans.ai>
Date: Thu, 22 Oct 2026 01:22:33 +0000
Subject: [PATCH 27/29] test(module): use private temp dirs for artifact
 loading tests

The artifact loading tests now create their temporary directories with
tempfile::tempdir instead of nesting them inside the current directory, and on
Unix a helper tightens the permissions of the created tree to 0o700. This keeps
the fixtures out of the working tree and ensures the loader sees directories
that are not group or world writable.

Auto-committed-on: cuttlefish
Co-authored-by: Medulla <medulla@tinyhumans.ai>
---
 crates/tinybus/src/module/host_tests.rs | 28 ++++++++++++++++++++-----
 1 file changed, 23 insertions(+), 5 deletions(-)

diff --git a/crates/tinybus/src/module/host_tests.rs b/crates/tinybus/src/module/host_tests.rs
index 4863096c..2811184f 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,9 @@ 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 directory = tempfile::tempdir().unwrap();
+    #[cfg(unix)]
+    make_private_tree(directory.path(), directory.path());
     let extension = if cfg!(windows) {
         "dll"
     } else if cfg!(target_os = "macos") {
@@ -582,7 +596,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 +605,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 +675,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 +684,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 +726,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 +743,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") {

From 47de96709b34fa4f64ad91f055f9c90616f8a309 Mon Sep 17 00:00:00 2001
From: Steven Enamakel <enamakel@tinyhumans.ai>
Date: Thu, 22 Oct 2026 01:33:51 +0000
Subject: [PATCH 28/29] test(module): create temp dirs inside the working
 directory

The lazy manifest test now builds its temporary directory inside the
current working directory instead of the system temp location, so the
test works in environments where the default temp path is not usable.

Auto-committed-on: cuttlefish
Co-authored-by: Medulla <medulla@tinyhumans.ai>
---
 crates/tinybus/src/module/host_tests.rs | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/crates/tinybus/src/module/host_tests.rs b/crates/tinybus/src/module/host_tests.rs
index 2811184f..caf36b7c 100644
--- a/crates/tinybus/src/module/host_tests.rs
+++ b/crates/tinybus/src/module/host_tests.rs
@@ -523,7 +523,7 @@ 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().unwrap();
+    let directory = tempfile::tempdir_in(std::env::current_dir().unwrap()).unwrap();
     #[cfg(unix)]
     make_private_tree(directory.path(), directory.path());
     let extension = if cfg!(windows) {

From dfb2f1ca5ec510f6c0cbd51bd972f01bb71212d5 Mon Sep 17 00:00:00 2001
From: Steven Enamakel <enamakel@tinyhumans.ai>
Date: Thu, 22 Oct 2026 01:34:17 +0000
Subject: [PATCH 29/29] test(module): create lazy manifest temp dir under
 canonical temp root

The lazy manifest test now builds its temporary directory inside the
canonicalized system temp directory instead of the current working
directory, avoiding failures when the working directory is not writable
or is itself a symlink.

Auto-committed-on: cuttlefish
Co-authored-by: Medulla <medulla@tinyhumans.ai>
---
 crates/tinybus/src/module/host_tests.rs | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/crates/tinybus/src/module/host_tests.rs b/crates/tinybus/src/module/host_tests.rs
index caf36b7c..860b25b8 100644
--- a/crates/tinybus/src/module/host_tests.rs
+++ b/crates/tinybus/src/module/host_tests.rs
@@ -523,7 +523,8 @@ 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) {