fix: retain native terminal ownership across cancellation

crates/tinybus-module/src/execution_context_tests.rs

@@ -27,6 +27,7 @@ fn owner() -> (ExecutionOwner, &'static AtomicUsize) {
         },
         ready,
         broker_routing: None,
+        native_terminal_state: None,
     });
     (ExecutionOwner::new(host), count)
 }

crates/tinybus-module/src/lib_lifecycle_tests.rs

@@ -23,6 +23,7 @@ fn host() -> TbHostVtable {
         },
         ready,
         broker_routing: None,
+        native_terminal_state: None,
     }
 }
 struct DropCounter(Arc<AtomicUsize>);

crates/tinybus-module/src/lib_tests.rs

@@ -74,6 +74,7 @@ fn host(config: &[u8]) -> TbHostVtable {
         },
         ready: host_ready,
         broker_routing: None,
+        native_terminal_state: None,
     }
 }
 
@@ -745,6 +746,7 @@ fn an_older_host_prefix_is_copied_without_reading_the_additive_tail() {
     let ptr = std::ptr::from_ref(&prefix).cast::<TbHostVtable>();
     let normalized = unsafe { TbHostVtable::read_compatible(ptr) }.unwrap();
     assert!(normalized.broker_routing.is_none());
+    assert!(normalized.native_terminal_state.is_none());
     assert_eq!(
         unsafe { parse_config::<serde_json::Value>(ptr) }.unwrap(),
         serde_json::json!({})
@@ -758,9 +760,12 @@ fn an_older_host_prefix_is_copied_without_reading_the_additive_tail() {
 #[test]
 fn an_incomplete_host_tail_never_reads_the_callback() {
     let mut current = host(b"{}");
-    current.size = size_of::<TbHostVtable>() as u32 - 1;
+    current.size = (std::mem::offset_of!(TbHostVtable, broker_routing)
+        + size_of::<Option<unsafe extern "C" fn(*mut c_void) -> u32>>()
+        - 1) as u32;
     let normalized = unsafe { TbHostVtable::read_compatible(&current) }.unwrap();
     assert!(normalized.broker_routing.is_none());
+    assert!(normalized.native_terminal_state.is_none());
     assert!(unsafe { TbHostVtable::read_compatible(std::ptr::null()) }.is_none());
     current.size = tinybus::module::abi::TB_HOST_VTABLE_BASE_SIZE - 1;
     assert!(unsafe { TbHostVtable::read_compatible(&current) }.is_none());

crates/tinybus/Cargo.toml

@@ -82,7 +82,7 @@ test-support = ["modules", "macros"]
 
 
 [dev-dependencies]
-tokio = { version = "1", features = ["macros", "rt-multi-thread", "sync", "time", "io-util"] }
+tokio = { version = "1", features = ["macros", "rt-multi-thread", "sync", "time", "io-util", "test-util"] }
 tempfile = "3"
 
 [[bin]]
@@ -135,3 +135,8 @@ required-features = ["modules"]
 name = "module_host_callback"
 crate-type = ["cdylib"]
 required-features = ["modules"]
+
+[[example]]
+name = "module_controlled_init"
+crate-type = ["cdylib"]
+required-features = ["modules"]

crates/tinybus/examples/module_controlled_init.rs

@@ -0,0 +1,139 @@
+//! Native ABI fixture with bounded, externally released test barriers.
+use std::ffi::c_void;
+use std::sync::{Arc, Condvar, Mutex};
+use tinybus::module::abi::*;
+
+#[repr(C)]
+#[derive(Clone, Copy)]
+struct Controls {
+    ctx: *mut c_void,
+    entered: unsafe extern "C" fn(*mut c_void, *const TbHostVtable),
+    joined: unsafe extern "C" fn(*mut c_void),
+}
+unsafe impl Send for Controls {}
+
+struct Instance {
+    controls: Controls,
+    host: TbHostVtable,
+    stop: Arc<(Mutex<bool>, Condvar)>,
+    worker: Option<std::thread::JoinHandle<()>>,
+}
+
+#[unsafe(no_mangle)]
+pub static TINYBUS_MODULE_ABI_V1: TbAbiDescriptor =
+    TbAbiDescriptor::current("controlled-init", env!("CARGO_PKG_VERSION"));
+
+#[unsafe(no_mangle)]
+pub extern "C" fn tinybus_module_manifest_v1() -> TbSlice {
+    tinybus_module::manifest_slice(tinybus_module::ManifestDeclaration {
+        name: "controlled-init",
+        version: env!("CARGO_PKG_VERSION"),
+        provides: &["ai.tinyhumans.ControlledInit"],
+        methods: &[],
+        signals: &[],
+        requires: &[],
+        optional: &[],
+        lazy: true,
+        worker_threads: 1,
+    })
+}
+
+unsafe extern "C" fn deliver(_: *mut c_void, _: *const u8, _: usize) -> i32 {
+    TB_OK
+}
+unsafe extern "C" fn reinitialize(ctx: *mut c_void, _: *const u8, _: usize) -> i32 {
+    let instance = unsafe { &*ctx.cast::<Instance>() };
+    unsafe {
+        (instance.controls.entered)(instance.controls.ctx, &instance.host);
+    }
+    TB_OK
+}
+unsafe extern "C" fn shutdown(ctx: *mut c_void, _: u64) -> i32 {
+    let mut instance = unsafe { Box::from_raw(ctx.cast::<Instance>()) };
+    *instance.stop.0.lock().unwrap() = true;
+    instance.stop.1.notify_all();
+    instance.worker.take().unwrap().join().unwrap();
+    unsafe {
+        (instance.controls.joined)(instance.controls.ctx);
+    }
+    TB_OK
+}
+
+/// # Safety
+/// Host pointers are valid for the ABI call; config contains test Controls.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn tinybus_module_init_v1(
+    host: *const TbHostVtable,
+    out: *mut TbModuleVtable,
+) -> i32 {
+    let host = unsafe { &*host };
+    let copied_config =
+        unsafe { std::slice::from_raw_parts(host.config.ptr, host.config.len) }.to_vec();
+    // The public loader serializes JSON. This declared test fixture accepts the
+    // same finite Controls bytes in a JSON array, without product-side seams.
+    let encoded;
+    let mut ready = false;
+    let bytes = if copied_config.len() == size_of::<Controls>() {
+        &copied_config
+    } else {
+        let Ok(value) = serde_json::from_slice::<serde_json::Value>(&copied_config) else {
+            return TB_BAD_ARGUMENT;
+        };
+        ready = value
+            .get("test_ready")
+            .and_then(serde_json::Value::as_bool)
+            .unwrap_or(false);
+        let Some(value) = value.get("test_controls") else {
+            return TB_BAD_ARGUMENT;
+        };
+        let Ok(value) = serde_json::from_value::<Vec<u8>>(value.clone()) else {
+            return TB_BAD_ARGUMENT;
+        };
+        encoded = value;
+        &encoded
+    };
+    if bytes.len() != size_of::<Controls>() {
+        return TB_BAD_ARGUMENT;
+    }
+    let controls = unsafe { bytes.as_ptr().cast::<Controls>().read_unaligned() };
+    // The host-side barrier blocks actual opaque C entry, retaining its config
+    // borrow through fault publication and initiating waiter abandonment.
+    unsafe {
+        (controls.entered)(controls.ctx, host);
+    }
+    assert_eq!(
+        unsafe { std::slice::from_raw_parts(host.config.ptr, host.config.len) },
+        copied_config
+    );
+    let stop = Arc::new((Mutex::new(false), Condvar::new()));
+    let worker_stop = stop.clone();
+    let worker = std::thread::spawn(move || {
+        let mut stopped = worker_stop.0.lock().unwrap();
+        while !*stopped {
+            stopped = worker_stop.1.wait(stopped).unwrap();
+        }
+    });
+    let instance = Box::new(Instance {
+        controls,
+        host: *host,
+        stop,
+        worker: Some(worker),
+    });
+    unsafe {
+        *out = TbModuleVtable {
+            size: size_of::<TbModuleVtable>() as u32,
+            _reserved: 0,
+            module_ctx: Box::into_raw(instance).cast(),
+            deliver,
+            shutdown,
+            reinitialize: Some(reinitialize),
+        };
+    }
+    if ready {
+        // Actual opaque initialization emits Ready; tests never fabricate it.
+        unsafe {
+            (host.ready)(host.host_ctx);
+        }
+    }
+    TB_OK
+}

crates/tinybus/src/module/abi.rs

@@ -3,6 +3,7 @@
 //! ABI revision one grows through optional vtable tails. Read the frozen prefix
 //! first and check its declared size before reading a tail; an older host has no
 //! broker-routing callback and therefore supplies no authenticated call context.
+//! Each optional field has its own complete-field size check.
 
 use std::ffi::c_void;
 
@@ -28,6 +29,22 @@ pub const TB_TIMEOUT: i32 = -4;
 /// A pointer, length, or frame was invalid.
 pub const TB_BAD_ARGUMENT: i32 = -5;
 
+/// Exactly this value permits native per-instance admission. Zero, unknown
+/// bits, or any combination with terminal bits must deny admission. A sampled
+/// value is not a reusable permit for a later operation.
+pub const TB_NATIVE_TERMINAL_LIVE: u32 = 1;
+/// Terminal stop intent; once committed, the instance can never become live.
+pub const TB_NATIVE_TERMINAL_STOP_INTENT: u32 = 2;
+/// Actual fault intent, with public fault state taking precedence over stop.
+pub const TB_NATIVE_TERMINAL_FAULT: u32 = 4;
+/// An accepted opaque shutdown callback actually returned TB_OK. Fault and
+/// stop intent remain set and continue to deny admission even with this bit.
+/// A legacy callback's TB_OK alone proves neither independent worker completion
+/// nor receiver backing destruction. For native terminal hooks, TB_OK must
+/// additionally require actual worker/resource finish without sticky uncertainty
+/// or fault; that stronger module contract is separate from this host evidence.
+pub const TB_NATIVE_TERMINAL_PHYSICALLY_STOPPED: u32 = 8;
+
 /// Revision-one module initialization entrypoint.
 pub type TbModuleInit = unsafe extern "C" fn(*const TbHostVtable, *mut TbModuleVtable) -> i32;
 
@@ -166,6 +183,25 @@ pub struct TbHostVtable {
     /// The SDK trusts this callback under the unsafe module-init host contract;
     /// it is not authentication of arbitrary user-provided C vtables.
     pub broker_routing: Option<unsafe extern "C" fn(*mut c_void) -> u32>,
+    /// Optional live query for this native module instance, read only when
+    /// `size` covers this entire field. Absence means unsupported and must deny
+    /// native admission. Only exactly [`TB_NATIVE_TERMINAL_LIVE`] permits it;
+    /// zero, unknown bits, combined values and terminal flags deny. Terminal
+    /// intent is monotonic and must never return to Live.
+    ///
+    /// Modules copy this callback and `host_ctx`, never retain the borrowed
+    /// vtable or config storage. The host must keep the copied callback and
+    /// context valid and thread-safe for every retained view use, including
+    /// independent workers after runtime shutdown and host destruction. A
+    /// destroyed host leaves a terminal tombstone, never recycles or reopens
+    /// this context. Custom unsafe hosts must truthfully uphold this contract;
+    /// the SDK cannot authenticate an arbitrary in-process C host.
+    ///
+    /// Query success linearizes admission only; it neither grants publication
+    /// authority nor cancels an operation already admitted before later intent.
+    /// The SDK's existing process-lifetime context retention is unbounded;
+    /// this optional callback adds no aggregate resource bound.
+    pub native_terminal_state: Option<unsafe extern "C" fn(*mut c_void) -> u32>,
 }
 
 /// Size required from an older host before reading its frozen prefix.
@@ -176,12 +212,16 @@ impl TbHostVtable {
     ///
     /// Returns `None` for a null pointer or a size below
     /// [`TB_HOST_VTABLE_BASE_SIZE`]. A prefix-only or incomplete-tail host is
-    /// accepted with `broker_routing: None`; only a full tail is copied.
+    /// accepted. Each absent or incomplete optional callback normalizes to
+    /// `None`, independently of any later tail field.
     ///
     /// # Safety
     /// A non-null `host` must be aligned and readable for its first size field
     /// and every byte that size declares. Its callbacks must be valid and
-    /// thread-safe, and their context must outlive the module runtime. An
+    /// thread-safe, and their context must outlive the module runtime. If
+    /// `native_terminal_state` is supplied, its copied callback and context
+    /// must remain valid for every retained view use, even after runtime
+    /// shutdown or host destruction, as documented on that field. An
     /// affirmative routing callback (returning exactly 1) must truthfully
     /// guarantee exclusive broker delivery at the time each frame is queued.
     #[doc(hidden)]
@@ -190,8 +230,17 @@ impl TbHostVtable {
             return None;
         }
         let base = unsafe { host.cast::<TbHostVtableV1Prefix>().read() };
-        let broker_routing = if base.size as usize >= size_of::<Self>() {
-            unsafe { std::ptr::addr_of!((*host).broker_routing).read() }
+        let query_size = size_of::<Option<unsafe extern "C" fn(*mut c_void) -> u32>>();
+        let broker_routing =
+            if base.size as usize >= std::mem::offset_of!(Self, broker_routing) + query_size {
+                unsafe { std::ptr::addr_of!((*host).broker_routing).read() }
+            } else {
+                None
+            };
+        let native_terminal_state = if base.size as usize
+            >= std::mem::offset_of!(Self, native_terminal_state) + query_size
+        {
+            unsafe { std::ptr::addr_of!((*host).native_terminal_state).read() }
         } else {
             None
         };
@@ -206,6 +255,7 @@ impl TbHostVtable {
             config: base.config,
             ready: base.ready,
             broker_routing,
+            native_terminal_state,
         })
     }
 }