Expose trusted broker caller context without changing legacy dispatch

.github/workflows/ci.yml

@@ -59,15 +59,17 @@ jobs:
       - uses: Swatinem/rust-cache@v2
       - uses: taiki-e/install-action@cargo-llvm-cov
       - name: Build loader fixtures
-        run: cargo build --locked --example module_clock --example module_clock_two --example module_wrong_target --no-default-features --features modules,macros
+        run: cargo build --locked --example module_clock --example module_clock_two --example module_wrong_target --example module_context --example module_context_eager --no-default-features --features modules,macros
       - name: Enforce 90% line coverage per source file
         shell: bash
         run: |
           set -euo pipefail
           module=$(find target/debug/examples -maxdepth 1 -type f -name 'libmodule_clock.so' -print -quit)
           module_two=$(find target/debug/examples -maxdepth 1 -type f -name 'libmodule_clock_two.so' -print -quit)
           wrong_target=$(find target/debug/examples -maxdepth 1 -type f -name 'libmodule_wrong_target.so' -print -quit)
-          for artifact in "$module" "$module_two" "$wrong_target"; do
+          context_module=$(find target/debug/examples -maxdepth 1 -type f -name 'libmodule_context.so' -print -quit)
+          context_eager_module=$(find target/debug/examples -maxdepth 1 -type f -name 'libmodule_context_eager.so' -print -quit)
+          for artifact in "$module" "$module_two" "$wrong_target" "$context_module" "$context_eager_module"; do
             if [ -z "$artifact" ] || [ ! -f "$artifact" ]; then
               echo "a module loader fixture is missing" >&2
               exit 1
@@ -91,7 +93,7 @@ jobs:
           # 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.
-          TINYBUS_TEST_MODULE="$PWD/$module" TINYBUS_TEST_MODULE_TWO="$private/$library" TINYBUS_TEST_WRONG_TARGET="$PWD/$wrong_target" 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$|crates/tinybus/src/connection\.rs$|crates/tinybus/src/test_support\.rs$' --fail-under-file-lines 90 -- --include-ignored
+          TINYBUS_TEST_MODULE="$PWD/$module" TINYBUS_TEST_MODULE_TWO="$private/$library" TINYBUS_TEST_WRONG_TARGET="$PWD/$wrong_target" TINYBUS_TEST_CONTEXT_MODULE="$PWD/$context_module" TINYBUS_TEST_CONTEXT_EAGER_MODULE="$PWD/$context_eager_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$|crates/tinybus/src/connection\.rs$|crates/tinybus/src/test_support\.rs$' --fail-under-file-lines 90 -- --include-ignored
 
   # The bus is meant to be embeddable on every platform OpenHuman ships on. The
   # socket adapter remains Unix-only, while the in-process module transport
@@ -128,7 +130,7 @@ jobs:
       - uses: dtolnay/rust-toolchain@stable
       - uses: Swatinem/rust-cache@v2
       - name: Build example module
-        run: cargo build --locked --example module_clock --example module_clock_two --example module_wrong_target --no-default-features --features modules,macros
+        run: cargo build --locked --example module_clock --example module_clock_two --example module_wrong_target --example module_context --example module_context_eager --no-default-features --features modules,macros
       - name: Test shared module test support
         run: cargo test --locked --no-default-features --features modules,macros,test-support test_support::tests
       - name: Exercise the real loader (Unix)
@@ -139,7 +141,9 @@ jobs:
           module=$(find target/debug/examples -maxdepth 1 -type f \( -name 'libmodule_clock.so' -o -name 'libmodule_clock.dylib' \) -print -quit)
           module_two=$(find target/debug/examples -maxdepth 1 -type f \( -name 'libmodule_clock_two.so' -o -name 'libmodule_clock_two.dylib' \) -print -quit)
           wrong_target=$(find target/debug/examples -maxdepth 1 -type f \( -name 'libmodule_wrong_target.so' -o -name 'libmodule_wrong_target.dylib' \) -print -quit)
-          for artifact in "$module" "$module_two" "$wrong_target"; do
+          context_module=$(find target/debug/examples -maxdepth 1 -type f \( -name 'libmodule_context.so' -o -name 'libmodule_context.dylib' \) -print -quit)
+          context_eager_module=$(find target/debug/examples -maxdepth 1 -type f \( -name 'libmodule_context_eager.so' -o -name 'libmodule_context_eager.dylib' \) -print -quit)
+          for artifact in "$module" "$module_two" "$wrong_target" "$context_module" "$context_eager_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
@@ -156,7 +160,7 @@ jobs:
             digest="$(shasum -a 256 "$private/$library" | awk '{print $1}')"
           fi
           printf '"%s" = "%s"\n' "$library" "$digest" > "$private/modules.toml"
-          TINYBUS_TEST_MODULE="$PWD/$module" TINYBUS_TEST_MODULE_TWO="$private/$library" TINYBUS_TEST_WRONG_TARGET="$PWD/$wrong_target" cargo test --locked --no-default-features --features modules,macros,test-support -- --ignored
+          TINYBUS_TEST_MODULE="$PWD/$module" TINYBUS_TEST_MODULE_TWO="$private/$library" TINYBUS_TEST_WRONG_TARGET="$PWD/$wrong_target" TINYBUS_TEST_CONTEXT_MODULE="$PWD/$context_module" TINYBUS_TEST_CONTEXT_EAGER_MODULE="$PWD/$context_eager_module" cargo test --locked --no-default-features --features modules,macros,test-support -- --ignored
       # Directory admission against this runner's real filesystem: a bundle
       # in a directory other accounts can write falls back to the verified
       # release cache, and on macOS the real /Applications is admitted.
@@ -190,13 +194,22 @@ jobs:
           $acl.AddAccessRule($rule)
           Set-Acl -LiteralPath $private -AclObject $acl
           Copy-Item target/debug/examples/module_clock.dll $private
+          Copy-Item target/debug/examples/module_context.dll $private
+          Copy-Item target/debug/examples/module_context_eager.dll $private
+          Copy-Item target/debug/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_wrong_target.dll') | ForEach-Object {
+            $digest = (Get-FileHash -LiteralPath (Join-Path $private $_) -Algorithm SHA256).Hash.ToLowerInvariant()
+            '"' + $_ + '" = "' + $digest + '"'
+          }
+          $fixtureAllowlist | Set-Content -LiteralPath (Join-Path $private modules.toml)
           $clockTwoDirectory = Join-Path $private module-clock-two
           New-Item -ItemType Directory -Force $clockTwoDirectory | Out-Null
           Set-Acl -LiteralPath $clockTwoDirectory -AclObject $acl
           Copy-Item target/debug/examples/module_clock_two.dll $clockTwoDirectory
           $clockTwoHash = (Get-FileHash -LiteralPath (Join-Path $clockTwoDirectory module_clock_two.dll) -Algorithm SHA256).Hash.ToLowerInvariant()
           '"module_clock_two.dll" = "' + $clockTwoHash + '"' | Set-Content -LiteralPath (Join-Path $clockTwoDirectory modules.toml) -NoNewline
-          Copy-Item target/debug/examples/module_wrong_target.dll $private
           $mixed = Join-Path $private mixed
           New-Item -ItemType Directory -Force $mixed | Out-Null
           Set-Acl -LiteralPath $mixed -AclObject $acl
@@ -207,6 +220,8 @@ jobs:
           $wrongTarget = (Get-ChildItem "$private/module_wrong_target.dll" | Select-Object -First 1).FullName
           $env:TEMP = $private
           $env:TMP = $private
+          $env:TINYBUS_TEST_CONTEXT_MODULE = (Get-Item "$private/module_context.dll").FullName
+          $env:TINYBUS_TEST_CONTEXT_EAGER_MODULE = (Get-Item "$private/module_context_eager.dll").FullName
           $env:TINYBUS_TEST_MODULE = $module
           $env:TINYBUS_TEST_MODULE_TWO = $moduleTwo
           $env:TINYBUS_TEST_WRONG_TARGET = $wrongTarget

Cargo.toml

@@ -13,6 +13,7 @@ resolver = "3"
 [workspace.package]
 version = "0.1.3"
 edition = "2024"
+rust-version = "1.85"
 license = "GPL-3.0-only"
 repository = "https://github.com/tinyhumansai/tinybus"
 

crates/tinybus-macros/Cargo.toml

@@ -2,6 +2,7 @@
 name = "tinybus-macros"
 version.workspace = true
 edition.workspace = true
+rust-version.workspace = true
 license.workspace = true
 repository.workspace = true
 description = "The #[interface] attribute for tinybus. Not useful on its own; use the `tinybus` crate."

crates/tinybus-macros/src/lib.rs

@@ -210,13 +210,15 @@ fn expand(
 /// Read `name = "…"` off the attribute.
 fn interface_name(args: &Punctuated<Meta, Comma>) -> syn::Result<String> {
     for arg in args {
-        if let Meta::NameValue(nv) = arg
-            && nv.path.is_ident("name")
-            && let syn::Expr::Lit(lit) = &nv.value
-            && let syn::Lit::Str(s) = &lit.lit
-        {
-            return Ok(s.value());
+        let Meta::NameValue(nv) = arg else { continue };
+        if !nv.path.is_ident("name") {
+            continue;
         }
+        let syn::Expr::Lit(lit) = &nv.value else {
+            continue;
+        };
+        let syn::Lit::Str(s) = &lit.lit else { continue };
+        return Ok(s.value());
     }
     Err(Error::new(
         proc_macro2::Span::call_site(),

crates/tinybus-module/Cargo.toml

@@ -2,6 +2,7 @@
 name = "tinybus-module"
 version.workspace = true
 edition.workspace = true
+rust-version.workspace = true
 license.workspace = true
 repository.workspace = true
 description = "Module-side runtime and export macro for trusted tinybus cdylibs."

crates/tinybus-module/src/lib.rs

@@ -3,11 +3,16 @@
 //! Each module owns a Tokio runtime. A statically linked `cdylib` has its own
 //! Tokio thread-locals, so attempting to borrow the host runtime is both
 //! incorrect and capable of silently blocking a host worker thread.
+//!
+//! Caller context depends on the trusted host's optional ABI-v1 routing callback.
+//! Its assertion is captured per frame at enqueue time; older host prefixes
+//! remain compatible but supply unverified context. Sender headers remain intact
+//! for protocol routing and streams even when they carry no caller authority.
 
 use std::ffi::c_void;
 use std::future::Future;
 use std::panic::{AssertUnwindSafe, catch_unwind};
-use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
 use std::sync::{Arc, Mutex as StdMutex, OnceLock};
 use std::time::Duration;
 
@@ -164,6 +169,12 @@ unsafe impl Send for HostCalls {}
 unsafe impl Sync for HostCalls {}
 
 impl HostCalls {
+    fn is_broker_routed(&self) -> bool {
+        self.0
+            .broker_routing
+            .is_some_and(|routing| unsafe { routing(self.0.host_ctx) } == 1)
+    }
+
     fn send(&self, bytes: &[u8]) -> i32 {
         unsafe { (self.0.send)(self.0.host_ctx, bytes.as_ptr(), bytes.len()) }
     }
@@ -244,9 +255,13 @@ impl tracing::field::Visit for LogVisitor {
     }
 }
 
+// Private envelope captures trusted-host routing before the frame is queued.
+type IncomingFrame = (Vec<u8>, bool);
+
 struct ModuleTransport {
     host: HostCalls,
-    inbound: Mutex<mpsc::Receiver<Vec<u8>>>,
+    inbound: Mutex<mpsc::Receiver<IncomingFrame>>,
+    provenance: Arc<AtomicBool>,
     detach_on_panic: bool,
 }
 
@@ -273,11 +288,16 @@ impl Transport for ModuleTransport {
 
     async fn recv(&self) -> Result<Option<Message>> {
         let bytes = self.inbound.lock().await.recv().await;
-        let Some(bytes) = bytes else {
+        let Some((bytes, brokered)) = bytes else {
             return Ok(None);
         };
         self.host.wake();
-        Ok(Some(serde_json::from_slice(&bytes)?))
+        let message = serde_json::from_slice(&bytes)?;
+        // The connection is the only reader and samples this slot immediately
+        // after recv. Keep header identity intact for stream ownership/routing;
+        // only contextual dispatch consumes the captured delivery authority.
+        self.provenance.store(brokered, Ordering::Release);
+        Ok(Some(message))
     }
 
     async fn close(&self) -> Result<()> {
@@ -290,8 +310,21 @@ impl Transport for ModuleTransport {
     }
 }
 
+fn module_connection(transport: Arc<ModuleTransport>) -> Connection {
+    if transport.host.0.broker_routing.is_some() {
+        let provenance = transport.provenance.clone();
+        // SAFETY: this private transport has one reader, publishes the queue's
+        // per-frame proof before returning, and no other writer touches its
+        // slot. The trusted host asserts routing at deliver, not at receive.
+        unsafe { Connection::__attach_module_with_provenance(transport, provenance) }
+    } else {
+        Connection::attach(transport)
+    }
+}
+
 struct RuntimeState {
-    inbound: StdMutex<Option<mpsc::Sender<Vec<u8>>>>,
+    inbound: StdMutex<Option<mpsc::Sender<IncomingFrame>>>,
+    host: HostCalls,
     runtime: StdMutex<Option<tokio::runtime::Runtime>>,
     reinitialize: Option<Reinitialize>,
 }
@@ -307,10 +340,11 @@ unsafe extern "C" fn deliver(ctx: *mut c_void, ptr: *const u8, len: usize) -> i3
             return TB_BAD_ARGUMENT;
         }
         let state = unsafe { &*(ctx.cast::<RuntimeState>()) };
+        let brokered = state.host.is_broker_routed();
         let bytes = unsafe { std::slice::from_raw_parts(ptr, len) }.to_vec();
         let sender = state.inbound.lock().expect("module inbound lock").clone();
         match sender {
-            Some(sender) => match sender.try_send(bytes) {
+            Some(sender) => match sender.try_send((bytes, brokered)) {
                 Ok(()) => TB_OK,
                 Err(mpsc::error::TrySendError::Full(_)) => TB_BACKPRESSURE,
                 Err(mpsc::error::TrySendError::Closed(_)) => TB_CLOSED,
@@ -398,6 +432,15 @@ unsafe fn write_module_vtable(
 ///
 /// This is public only for [`module_export!`] expansions. Module authors call
 /// the macro, not this function directly.
+///
+/// # Safety
+/// The host vtable must meet [`TbHostVtable::read_compatible`]'s requirements.
+/// A non-null `out` must be aligned and writable for its declared capacity,
+/// with a readable size field; writes are bounded to that capacity. Borrowed
+/// configuration bytes must remain readable during initialization.
+/// In particular an affirmative routing callback is a trusted host assertion
+/// that only real broker deliveries enter `deliver`, not an arbitrary frame
+/// injection capability. Artifact attestation does not establish this fact.
 #[doc(hidden)]
 pub unsafe fn start_module<F, Fut>(
     host: *const TbHostVtable,
@@ -424,6 +467,10 @@ where
 }
 
 /// Start a linked module without replacing the host's process-global hooks.
+///
+/// # Safety
+/// The caller must uphold the host, output, configuration, and routing contracts
+/// of [`start_module`].
 #[doc(hidden)]
 pub unsafe fn start_linked_module<F, Fut>(
     host: *const TbHostVtable,
@@ -475,10 +522,10 @@ where
         }
         // `size` is the frozen prefix field; do not copy the full vtable until
         // the host has proved that all v1 fields are present.
-        if unsafe { host.cast::<u32>().read() } < size_of::<TbHostVtable>() as u32 {
+        let Some(host) = (unsafe { TbHostVtable::read_compatible(host) }) else {
             return TB_BAD_ARGUMENT;
-        }
-        let host = HostCalls(unsafe { *host });
+        };
+        let host = HostCalls(host);
 
         // A cdylib carries its own statically linked `tracing` and `std` state;
         // this registration is global to the module's copy, not the embedding
@@ -531,16 +578,18 @@ where
             Err(_) => return TB_CLOSED,
         };
         let (inbound_tx, inbound_rx) = mpsc::channel(MODULE_QUEUE_CAPACITY);
-        let transport = Box::new(ModuleTransport {
+        let transport = Arc::new(ModuleTransport {
             host,
             inbound: Mutex::new(inbound_rx),
+            provenance: Arc::new(AtomicBool::new(false)),
             detach_on_panic,
         });
         let connection_slot = Arc::new(StdMutex::new(None));
         let task_connection_slot = connection_slot.clone();
 
         runtime.spawn(async move {
-            let outcome = match Connection::connect(transport).await {
+            let connection = module_connection(transport);
+            let outcome = match connection.handshake().await.map(|()| connection) {
                 Ok(connection) => {
                     connection.__set_panic_handler(std::sync::Arc::new(move || {
                         let location = panic_location
@@ -581,6 +630,7 @@ where
         let supports_reinitialize = reinitializer.is_some();
 
         let state = Box::new(RuntimeState {
+            host,
             inbound: StdMutex::new(Some(inbound_tx)),
             runtime: StdMutex::new(Some(runtime)),
             reinitialize: reinitializer,
@@ -600,10 +650,7 @@ unsafe fn parse_config<C: serde::de::DeserializeOwned>(
     if host.is_null() {
         return Err(TB_BAD_ARGUMENT);
     }
-    if unsafe { host.cast::<u32>().read() } < size_of::<TbHostVtable>() as u32 {
-        return Err(TB_BAD_ARGUMENT);
-    }
-    let host_ref = unsafe { &*host };
+    let host_ref = unsafe { TbHostVtable::read_compatible(host) }.ok_or(TB_BAD_ARGUMENT)?;
     if host_ref.config.len > 1024 * 1024
         || (host_ref.config.ptr.is_null() && host_ref.config.len != 0)
     {
@@ -622,6 +669,10 @@ unsafe fn parse_config<C: serde::de::DeserializeOwned>(
 /// This retains the original `FnOnce` contract for direct SDK callers. The
 /// export macro uses [`start_reconfigurable_module`] because a named setup
 /// function can safely be called again.
+///
+/// # Safety
+/// The caller must uphold the host, output, configuration, and routing contracts
+/// of [`start_module`].
 #[doc(hidden)]
 pub unsafe fn start_module_with_config<C, F, Fut>(
     host: *const TbHostVtable,
@@ -653,6 +704,10 @@ where
 }
 
 /// Initialize a configured module whose setup function may be called again.
+///
+/// # Safety
+/// The caller must uphold the host, output, configuration, and routing contracts
+/// of [`start_module`].
 #[doc(hidden)]
 pub unsafe fn start_reconfigurable_module<C, F, Fut>(
     host: *const TbHostVtable,
@@ -672,6 +727,10 @@ where
 }
 
 /// Start a configurable linked module without process-global hooks.
+///
+/// # Safety
+/// The caller must uphold the host, output, configuration, and routing contracts
+/// of [`start_module`].
 #[doc(hidden)]
 pub unsafe fn start_linked_reconfigurable_module<C, F, Fut>(
     host: *const TbHostVtable,