no_std support for the Transform API (#73)

CHANGELOG.md

@@ -29,6 +29,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- **`no_std` Transform API (Design 027)**: `.transform()` and `.transform_join()` are now available on `no_std + alloc` targets — no longer Tokio-only. Multi-input join fan-in moved out of `aimdb-core` into the new `JoinFanInRuntime` traits in `aimdb-executor`, with implementations in the Tokio (`mpsc::channel`, capacity 64), Embassy (`embassy_sync::Channel`, capacity 8), and WASM (`futures_channel::mpsc`, capacity 64) adapters. ([aimdb-core](aimdb-core/CHANGELOG.md), [aimdb-executor](aimdb-executor/CHANGELOG.md), [aimdb-tokio-adapter](aimdb-tokio-adapter/CHANGELOG.md), [aimdb-embassy-adapter](aimdb-embassy-adapter/CHANGELOG.md), [aimdb-wasm-adapter](aimdb-wasm-adapter/CHANGELOG.md))
+- **Task-model join handler**: New `JoinBuilder::on_triggers(FnOnce(JoinEventRx, Producer) -> impl Future)` API replaces the previous callback model. Eliminates per-event heap allocation and lets handler state borrow across `.await` points. **Breaking change** vs. the old `with_state().on_trigger(...)` form — see [aimdb-core](aimdb-core/CHANGELOG.md).
+- **Weather-mesh `DewPoint` demo**: All three weather stations (alpha, beta, gamma) now derive a `DewPoint` record from `Temperature` and `Humidity` via `transform_join`, demonstrating the API end-to-end on Tokio and Embassy.
+- Design document: 027 (`no_std` Support for Transform API)
 - **MCP public mode**: New `--public` flag restricts the MCP server to read-only tools for safe internet-facing deployments with SSRF protection ([tools/aimdb-mcp](tools/aimdb-mcp/CHANGELOG.md))
 - **MCP `--socket` flag**: Default socket path can be set at startup, simplifying single-instance workflows ([tools/aimdb-mcp](tools/aimdb-mcp/CHANGELOG.md))
 - **Context-Aware Deserializers (Design 026)**: Inbound connector deserializers can now receive a `RuntimeContext<R>` for platform-independent timestamps and logging
@@ -45,6 +49,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Changed
 
+- **aimdb-embassy-adapter**: `SpmcRing` subscriber-slot exhaustion now emits a `defmt::error!` with guidance to increase the `CONSUMERS` const generic. Counting rule: one slot per `.tap()`, `.link_to()`, and `transform_join` input.
+- **aimdb-codegen**: Generated join handler stubs updated to the new `on_triggers` task model (`async fn task_handler(JoinEventRx, Producer<...>)`).
 - **aimdb-core**: Breaking API changes to `InboundConnectorLink`, `Router`, and `RouterBuilder` to support `DeserializerKind` (see [aimdb-core/CHANGELOG.md](aimdb-core/CHANGELOG.md))
 - **aimdb-core**: Breaking API change — `ConnectorLink.serializer` now stores `SerializerKind` instead of `SerializerFn`
 - **aimdb-core**: `.with_serializer()` renamed to `.with_serializer_raw()` for the old single-argument pattern

aimdb-codegen/CHANGELOG.md

@@ -7,7 +7,21 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
-No changes yet.
+### Changed
+
+- **Generated join handler stubs** updated to match the new task-model `on_triggers` API (Design 027). Multi-input task handlers are now generated as:
+  ```rust
+  pub async fn task_handler(
+      mut _rx: aimdb_core::transform::JoinEventRx,
+      _producer: aimdb_core::Producer<Output, TokioAdapter>,
+  ) {
+      while let Ok(_trigger) = _rx.recv().await {
+          todo!("implement task_handler")
+      }
+  }
+  ```
+  Previously generated `fn task_handler(JoinTrigger, &mut (), &Producer<...>) -> Pin<Box<dyn Future>>` for the callback model.
+- `build_transform_call` for join tasks now emits `.on_triggers(handler)` instead of `.with_state(()).on_trigger(handler)`.
 
 ## [0.1.0] - 2026-03-11
 

aimdb-codegen/src/rust.rs

@@ -1589,8 +1589,7 @@ fn build_transform_call(task: &TaskDef, variant_ident: &syn::Ident) -> TokenStre
         quote! {
             .transform_join(|j| {
                 j #(#input_calls)*
-                    .with_state(())
-                    .on_trigger(#handler_ident)
+                    .on_triggers(#handler_ident)
             })
         }
     } else {
@@ -1612,7 +1611,7 @@ fn build_transform_call(task: &TaskDef, variant_ident: &syn::Ident) -> TokenStre
 ///
 /// | Inputs | Outputs | API                   | Generated stub            |
 /// |--------|---------|-----------------------|---------------------------|
-/// | N > 1  | ≥ 1     | `.transform_join()`   | `fn task_handler(JoinTrigger, &mut (), &Producer<O, R>)` |
+/// | N > 1  | ≥ 1     | `.transform_join()`   | `async fn task_handler(JoinEventRx, Producer<O, R>)` |
 /// | 1      | ≥ 1     | `.transform().map()`  | `fn task_transform(&Input) -> Option<Output>` |
 /// | 0      | ≥ 1     | `.source()`           | `async fn task(RuntimeContext, Producer<O, R>)` |
 /// | ≥ 1    | 0       | `.tap()`              | `async fn task(RuntimeContext, Consumer<I, R>)` |
@@ -1647,9 +1646,7 @@ pub fn generate_hub_tasks_rs(state: &ArchitectureState) -> String {
         }
 
         if n_in > 1 && n_out >= 1 {
-            // Multi-input → join handler
-            // Returns Pin<Box<dyn Future>> — the only concrete return type that satisfies
-            // the for<'a,'b> HRTB on on_trigger. `-> impl Future` does NOT work here.
+            // Multi-input → join handler (task model: owns event loop and state)
             let handler = format!("{}_handler", task.name);
             let inputs_doc = task
                 .inputs
@@ -1661,12 +1658,13 @@ pub fn generate_hub_tasks_rs(state: &ArchitectureState) -> String {
             fns.push_str(&format!(
                 "/// Join handler — match `trigger.index()` to identify which input fired:\n\
 /// {inputs_doc}\n\
-pub fn {handler}(\n\
-    _trigger: aimdb_core::transform::JoinTrigger,\n\
-    _state: &mut (),\n\
-    _producer: &aimdb_core::Producer<{out_t}, TokioAdapter>,\n\
-) -> std::pin::Pin<Box<dyn std::future::Future<Output = ()> + Send + 'static>> {{\n\
-    Box::pin(async move {{ todo!(\"implement {handler}\") }})\n\
+pub async fn {handler}(\n\
+    mut _rx: aimdb_core::transform::JoinEventRx,\n\
+    _producer: aimdb_core::Producer<{out_t}, TokioAdapter>,\n\
+) {{\n\
+    while let Ok(_trigger) = _rx.recv().await {{\n\
+        todo!(\"implement {handler}\")\n\
+    }}\n\
 }}\n\n"
             ));
         } else if n_in == 1 && n_out >= 1 {

aimdb-core/CHANGELOG.md

@@ -9,6 +9,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- **`no_std` support for the full Transform API (Design 027)**: `.transform()` and `.transform_join()` are now available on `no_std + alloc` targets. Multi-input join fan-in is no longer hardcoded to `tokio::sync::mpsc`; it uses the runtime-agnostic `JoinFanInRuntime` traits from `aimdb-executor`, implemented by Tokio, Embassy, and WASM adapters.
+- **`JoinEventRx`** — type-erased trigger receiver passed to the `on_triggers` handler. Call `.recv().await` in a loop to consume `JoinTrigger` events from all input forwarders.
+- **`transform_join` as an inherent method on `RecordRegistrar`** (gated `feature = "alloc"`, `R: JoinFanInRuntime`). Previously only exposed via the `impl_record_registrar_ext!` macro under `feature = "std"`.
 - **Context-Aware Deserializers (Design 026)**: Inbound connector deserializers can now receive a `RuntimeContext<R>` for platform-independent timestamps and logging during deserialization
   - New `ContextDeserializerFn` type alias for context-aware type-erased deserializer callbacks
   - New `DeserializerKind` enum (`Raw` / `Context`) to enforce mutual exclusivity between plain and context-aware deserializers
@@ -24,6 +27,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Changed
 
+- **Breaking — Join handler API redesign (Design 027 §Q4)**: `JoinBuilder::with_state(...).on_trigger(Fn(...) -> Pin<Box<dyn Future>>)` replaced with task-model `JoinBuilder::on_triggers(FnOnce(JoinEventRx, Producer) -> impl Future)`. The handler now owns the event loop, eliminating per-event heap allocation and allowing state to be borrowed across `.await` points.
+- **`transform.rs` split into `transform/{mod,single,join}.rs`** — internal reorganization to keep the `alloc`-only join path separate from the runtime-agnostic single-input path. `JoinBuilder`, `JoinPipeline`, `JoinTrigger`, `JoinEventRx` are now re-exported from `transform::join`.
+- `transform_join_raw` now requires `R: JoinFanInRuntime` (was `feature = "std"`).
+- `ExecutorError::QueueClosed` mapped to `DbError::RuntimeError` in `From<ExecutorError>`.
 - **Breaking**: `InboundConnectorLink::deserializer` field type changed from `DeserializerFn` to `DeserializerKind`
 - **Breaking**: `InboundConnectorLink::new()` now takes `DeserializerKind` instead of `DeserializerFn`
 - **Breaking**: `Router::route()` signature changed to accept an additional `ctx` parameter

aimdb-core/src/error.rs

@@ -697,6 +697,18 @@ impl From<aimdb_executor::ExecutorError> for DbError {
                     DbError::RuntimeError { _message: () }
                 }
             }
+            ExecutorError::QueueClosed => {
+                #[cfg(feature = "std")]
+                {
+                    DbError::RuntimeError {
+                        message: "join queue closed".to_string(),
+                    }
+                }
+                #[cfg(not(feature = "std"))]
+                {
+                    DbError::RuntimeError { _message: () }
+                }
+            }
         }
     }
 }

aimdb-core/src/ext_macros.rs

@@ -103,19 +103,6 @@ macro_rules! impl_record_registrar_ext {
                     $crate::transform::TransformBuilder<I, T, $runtime>,
                 ) -> $crate::transform::TransformPipeline<I, T, $runtime>;
 
-            /// Multi-input reactive transform (join).
-            ///
-            /// Derives this record from multiple input records. Panics if a `.source()` or
-            /// another `.transform()` is already registered.
-            #[cfg(feature = "std")]
-            fn transform_join<F>(
-                &'a mut self,
-                build_fn: F,
-            ) -> &'a mut $crate::RecordRegistrar<'a, T, $runtime>
-            where
-                F: FnOnce(
-                    $crate::transform::JoinBuilder<T, $runtime>,
-                ) -> $crate::transform::JoinPipeline<T, $runtime>;
         }
 
         #[cfg(feature = $feature)]
@@ -190,18 +177,6 @@ macro_rules! impl_record_registrar_ext {
                 self.transform_raw::<I, F>(input_key, build_fn)
             }
 
-            #[cfg(feature = "std")]
-            fn transform_join<F>(
-                &'a mut self,
-                build_fn: F,
-            ) -> &'a mut $crate::RecordRegistrar<'a, T, $runtime>
-            where
-                F: FnOnce(
-                    $crate::transform::JoinBuilder<T, $runtime>,
-                ) -> $crate::transform::JoinPipeline<T, $runtime>,
-            {
-                self.transform_join_raw(build_fn)
-            }
         }
     };
 
@@ -261,6 +236,7 @@ macro_rules! impl_record_registrar_ext {
                 F: FnOnce(
                     $crate::transform::TransformBuilder<I, T, $runtime>,
                 ) -> $crate::transform::TransformPipeline<I, T, $runtime>;
+
         }
 
         #[cfg(all($(feature = $feature),+))]
@@ -334,6 +310,7 @@ macro_rules! impl_record_registrar_ext {
             {
                 self.transform_raw::<I, F>(input_key, build_fn)
             }
+
         }
     };
 }

aimdb-core/src/lib.rs

@@ -16,6 +16,9 @@
 
 #![cfg_attr(not(feature = "std"), no_std)]
 
+#[cfg(feature = "alloc")]
+extern crate alloc;
+
 pub mod buffer;
 pub mod builder;
 pub mod connector;
@@ -73,6 +76,6 @@ pub use record_id::{RecordId, RecordKey, StringKey};
 pub use graph::{DependencyGraph, EdgeType, GraphEdge, GraphNode, RecordGraphInfo, RecordOrigin};
 
 // Transform API exports
-#[cfg(feature = "std")]
-pub use transform::{JoinBuilder, JoinPipeline, JoinTrigger};
+#[cfg(feature = "alloc")]
+pub use transform::{JoinBuilder, JoinEventRx, JoinPipeline, JoinTrigger};
 pub use transform::{TransformBuilder, TransformPipeline};