Refactor/034 phase2 134 mqtt knobs

.devcontainer/devcontainer.json

@@ -26,10 +26,7 @@
         "terminal.integrated.shell.linux": "/bin/bash",
         "rust-analyzer.cargo.features": "all",
         "rust-analyzer.check.command": "clippy",
-        "rust-analyzer.check.extraArgs": [
-          "--all-targets",
-          "--all-features"
-        ],
+        "rust-analyzer.check.allTargets": true,
         "rust-analyzer.cargo.buildScripts.enable": true,
         "rust-analyzer.procMacro.enable": true,
         "files.watcherExclude": {

CHANGELOG.md

@@ -31,6 +31,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- **Design 034 Phase 3 — sans-io KNX/IP tunneling engine shared by both transports (Issue #135, [review doc §3.7](docs/design/034-technical-debt-review.md)).** The entire tunneling lifecycle — CONNECT_REQUEST/RESPONSE handshake, TUNNELING_REQUEST/ACK sequence + pending-ACK bookkeeping, keepalive (CONNECTIONSTATE_REQUEST) scheduling, ACK-timeout sweeps, and reconnect-with-backoff — now lives **once**, in the new runtime-neutral `aimdb_knx_connector::tunnel` module (`no_std + alloc`, no tokio/embassy imports), driven as a poll-based state machine (events in, `Action`s out, `next_deadline()` for timer arming). `tokio_client.rs` (988 → ~530 lines incl. a new fake-gateway integration test) and `embassy_client.rs` (1,055 → ~450 lines) are reduced to socket shims; the previously untestable handshake/ACK/keepalive/reconnect paths now have 15 host-run unit tests plus a scripted localhost-UDP roundtrip test. Behavioral unifications: Embassy gains the 5 s CONNECT_RESPONSE timeout (previously waited forever), both shims reconnect on fatal socket errors, and the dead tokio-only per-publish ACK oneshot is dropped (it was always `None`); the CONNECT_REQUEST HPAI stays per-transport (`LocalEndpoint`: tokio = real bound address, Embassy = NAT mode). ([aimdb-knx-connector](aimdb-knx-connector/CHANGELOG.md))
+
 - **Design 034 Phase 2 — dyn-safe `RuntimeOps` capability trait (Issue #130, [review doc](docs/design/034-technical-debt-review.md)).** New object-safe trait in `aimdb-executor` (`name` / `now_nanos` / `unix_time` / boxed `sleep` / `log(LogLevel, …)`) so a runtime adapter can travel as `Arc<dyn RuntimeOps>` instead of a generic parameter — the groundwork for removing `R` from the record object graph (#131). Implemented by `TokioAdapter`, `EmbassyAdapter`, and `WasmAdapter`, each covered by a shared behavioral contract test. `BoxFuture`'s canonical definition moves to `aimdb-executor` (re-exported unchanged from `aimdb-core`). ([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))
 
 - **M17 — centralized Embassy connector spine: one audited home for the single-core `unsafe` ([Design 033](docs/design/033-M17-unify-connectors-drop-send.md)).** New `aimdb-embassy-adapter::connectors` module (features `connectors` / `connector-io`) collects the force-`Send` plumbing every Embassy connector used to hand-roll: session transports get `EmbassySessionClient`/`EmbassySessionServer`, `OneShotDialer`/`OneShotListener`/`OneShotCell`, and the framed `EmbassyConnection` + `Framer`; data-plane transports get the `EmbassySink`/`EmbassySource` bridges (over `EmbassySinkRaw`/`EmbassySourceRaw`) that ride core's existing `pump_sink`/`pump_source`, plus `into_box_future` for protocol tasks. The serial Embassy half is now thin sugar (just a COBS `Framer`) with **zero `unsafe`** (down from a 407-line hand-roll with 7 `unsafe impl`s); the MQTT and KNX Embassy halves dropped their hand-rolled publisher/router loops and `SendFutureWrapper` use to ride core's pumps (KNX inbound telegrams now flow through `pump_source`). All connector-crate `unsafe`/`SendFutureWrapper` is gone — confined to the adapter. The std/Tokio side, `aimdb-client`, the WebSocket server, examples, and tests are unchanged. (Chosen over Design 033's original "drop `Send` from the contract", which would have pushed `!Send` onto the std side; see the doc's Implementation Decision.) ([aimdb-embassy-adapter](aimdb-embassy-adapter/CHANGELOG.md), [aimdb-serial-connector](aimdb-serial-connector/CHANGELOG.md), [aimdb-mqtt-connector](aimdb-mqtt-connector/CHANGELOG.md), [aimdb-knx-connector](aimdb-knx-connector/CHANGELOG.md))
@@ -44,6 +46,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Changed (breaking)
 
+- **Design 034 Phase 3 — runtime type parameter `R` removed from the object graph (Issue #131, [review doc §3.2/§3.3](docs/design/034-technical-debt-review.md)).** The runtime is now a *value* (`Arc<dyn aimdb_executor::RuntimeOps>`, the #130 groundwork) instead of a type parameter; the only generic left on the user-facing object graph is the record type `T`. The full break inventory:
+  - **Types lose `R`:** `AimDb<R>` → `AimDb`, `AimDbBuilder<R>` → `AimDbBuilder` (the `NoRuntime` typestate is gone — a missing runtime is a `build()` error per the #133 contract), `TypedRecord<T, R>` → `TypedRecord<T>`, `RecordRegistrar<'a, T, R>` → `RecordRegistrar<'a, T>`, `TransformBuilder<I, O, R>` → `TransformBuilder<I, O>`, `JoinBuilder<O, R>` → `JoinBuilder<O>`, `RecordT<R>` → `RecordT`, and `ConnectorBuilder<R>` → `ConnectorBuilder` (its `build()` takes `&AimDb`). Turbofish updates: `get_typed_record_by_key::<T, R>` → `::<T>`, `as_typed::<T, R>` → `::<T>`.
+  - **`RuntimeContext` is a concrete struct** wrapping `Arc<dyn RuntimeOps>`. `ctx.time().now()` returns **`u64` nanoseconds** from an arbitrary monotonic epoch (was `R::Instant`); `sleep` takes a plain `core::time::Duration` (plus `sleep_millis`/`sleep_secs` helpers); `millis()`/`secs()`/`micros()`/`duration_since()`/`duration_as_nanos()` are deleted (durations are concrete, instants are integer math); the panicking `extract_from_any` is deleted.
+  - **`source`/`tap`/`transform`/`transform_join` are inherent methods** on `RecordRegistrar` — the per-adapter extension-trait wrappers and `aimdb-core`'s `ext_macros.rs` are gone, as are the `*_raw` variants (the inherent methods *are* the foundational API; closures keep the `(ctx, producer)` arg order, so user code mostly just drops an import). The adapter ext traits (`TokioRecordRegistrarExt`, `EmbassyRecordRegistrarExt`(+`Custom`), `WasmRecordRegistrarExt`) shrink to the one genuinely adapter-specific step: `.buffer(cfg)` construction.
+  - **`JoinFanInRuntime` deleted** (with `JoinQueue`/`JoinSender`/`JoinReceiver` and all three per-adapter `join_queue.rs` files): multi-input join fan-in now uses one bounded `async-channel` queue in core — the same primitive the session engine already uses on tokio, Embassy, and WASM. Capacity stays 64 on `std`/wasm32 and rises to 16 on embedded `no_std` (up from Embassy's 8); the queue now closes on every runtime when all forwarders exit (the Embassy queue previously never closed).
+  - **Runtime accessors:** `runtime_arc()` → `runtime_ops()` (returns `Arc<dyn RuntimeOps>`), new `runtime_ctx()`; the borrowed `AimDb::runtime()` and the type-erased `runtime_any()` are deleted (zero callers in aimdb/aimdb-pro — `&*db.runtime_ops()` covers the borrowed flavor; [follow-up doc §2.5](docs/design/035-review-followups-deferred.md)). `AimDbBuilder::on_start` closures receive `RuntimeContext` (was `Arc<R>`). Context-aware (de)serializers receive the concrete `RuntimeContext` (was `Arc<dyn Any + Send + Sync>`); `Router::route`'s ctx argument follows. The `RuntimeForProfiling` marker-trait workaround is deleted (profiling clocks ride `RuntimeOps::now_nanos`); the session client engine's clock is `Arc<dyn RuntimeOps>` (was `Arc<R: TimeOps>`).
+  - **Embassy network capability moves to construction:** the `EmbassyNetwork` runtime trait is deleted (a `dyn RuntimeOps` can't surface adapter-specific capabilities) along with `EmbassyAdapter::new_with_network` — `EmbassyAdapter` is a stateless unit type with **zero `unsafe`**. The Embassy MQTT/KNX connector builders take the `embassy_net::Stack` at construction (`MqttConnectorBuilder::new(url, stack)` / `KnxConnectorBuilder::new(url, stack)`), wrapped in the new force-`Send + Sync` `aimdb_embassy_adapter::connectors::NetStack` so the single-core `unsafe` stays in the one audited module.
+  - **Downstream follows mechanically:** `aimdb-sync` (`AimDb` handles), `aimdb-persistence` (`.persist()`/`.with_persistence()` ext traits de-genericized), `aimdb-data-contracts::log_tap`, `aimdb-codegen` templates (generated `configure_schema(builder: &mut AimDbBuilder)`), and all examples/tools.
+  - Acceptance: `grep -rn "extract_from_any|runtime_any|RuntimeForProfiling|JoinFanInRuntime"` over the workspace sources returns nothing; the remaining `dyn Any` in core is data-plane only (`SerializerFn`/`produce_any`/`JoinTrigger`/`TopicProviderAny` — the #131 §6 stretch, tracked as a follow-up). ([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), [aimdb-mqtt-connector](aimdb-mqtt-connector/CHANGELOG.md), [aimdb-knx-connector](aimdb-knx-connector/CHANGELOG.md), [aimdb-persistence](aimdb-persistence/CHANGELOG.md), [aimdb-sync](aimdb-sync/CHANGELOG.md))
+
 - **Design 034 Phase 2 — MQTT knobs move out of core; `ConnectorConfig` pruned (Issue #134, [review doc §3.6](docs/design/034-technical-debt-review.md)).** Core's generic link builders drop `with_qos`/`with_retain` (`with_timeout_ms` stays, de-MQTT'd); the knobs now live in `aimdb-mqtt-connector` as the `MqttLinkExt` (qos, outbound + inbound) and `MqttOutboundLinkExt` (retain, publish-side only) extension traits, pushing the **same** `("qos", …)`/`("retain", …)` option keys the MQTT clients have always read — wire behavior unchanged; importing the trait makes the MQTT intent explicit at the call site (generic escape hatch: `with_config(key, value)`). `ConnectorConfig` loses its never-read typed `qos`/`retain` fields and the speculative Kafka/HTTP/shmem interpretation docs — it keeps `timeout_ms` + `protocol_options`, and core now documents no protocol that lacks an in-tree connector. ([aimdb-core](aimdb-core/CHANGELOG.md), [aimdb-mqtt-connector](aimdb-mqtt-connector/CHANGELOG.md))
 
 - **Design 034 Phase 2 — panic-free builder validation: `build()` reports every configuration mistake at once (Issue #133, [review doc §3.4](docs/design/034-technical-debt-review.md)).** Builder methods never panic on user mistakes anymore. Conflicting `.source()`/`.transform()`/`.link_from()` registrations, missing serializers/deserializers, invalid connector URLs, unregistered schemes, and key-reused-with-different-type are *recorded* (the conflicting registration is skipped) and `build()` returns one `DbError::InvalidConfiguration { errors: Vec<ConfigError> }` carrying **all** findings — each with the record key and, where applicable, the connector URL. The worst panic — "requires a buffer" firing at spawn time inside a connector factory closure — is now a build()-time check, which also makes `.buffer()` after `.link_to()`/`.link_from()` legal (order-independent). Duplicate keys and dependency-graph cycles fold into the same collected report (previously distinct `DuplicateRecordKey`/`CyclicDependency` returns from `build()`). Remaining `panic!`/`expect`s in the builder path are internal invariants and say "this is a bug in aimdb-core". ([aimdb-core](aimdb-core/CHANGELOG.md))

aimdb-client/tests/pump_client.rs

@@ -33,7 +33,7 @@ struct Msg {
 /// is robust against subscription-registration timing (a fresh subscriber may
 /// only see values produced after it attaches).
 async fn mirror_reaches(
-    db: &Arc<AimDb<TokioAdapter>>,
+    db: &Arc<AimDb>,
     key: &str,
     want: &serde_json::Value,
     mut push: impl FnMut(),

aimdb-codegen/src/lib.rs

@@ -6,7 +6,7 @@
 //! - **Mermaid diagram** — `.aimdb/architecture.mermaid`, a read-only graph
 //!   projection of the architecture (see [`generate_mermaid`])
 //! - **Rust source** — `src/generated_schema.rs`, compilable AimDB schema
-//!   using the actual 0.5.x API (see [`generate_rust`])
+//!   using the current AimDB API (see [`generate_rust`])
 //!
 //! # Usage
 //!

aimdb-codegen/src/rust.rs

@@ -1,7 +1,7 @@
 //! Rust source code generator
 //!
 //! Converts an [`ArchitectureState`] into compilable Rust source that uses the
-//! actual AimDB 0.5.x API: `#[derive(RecordKey)]`, `BufferCfg`, and
+//! current AimDB API: `#[derive(RecordKey)]`, `BufferCfg`, and
 //! `AimDbBuilder::configure()`.
 //!
 //! Uses [`quote`] for quasi-quoting token streams and [`prettyplease`] for
@@ -383,7 +383,7 @@ pub fn generate_tasks_rs(state: &ArchitectureState, binary_name: &str) -> Option
             let fn_name = format_ident!("{}", task.name);
 
             // Build parameter list
-            let mut params: Vec<TokenStream> = vec![quote! { ctx: RuntimeContext<TokioAdapter> }];
+            let mut params: Vec<TokenStream> = vec![quote! { ctx: RuntimeContext }];
             for input in &task.inputs {
                 let arg_name = format_ident!("{}", to_snake_case(&input.record));
                 let value_type = format_ident!("{}Value", input.record);
@@ -418,7 +418,6 @@ pub fn generate_tasks_rs(state: &ArchitectureState, binary_name: &str) -> Option
 
     let file_tokens = quote! {
         use aimdb_core::{Consumer, DbResult, Producer, RuntimeContext};
-        use aimdb_tokio_adapter::TokioAdapter;
         use #common_crate::*;
 
         #(#task_fns)*
@@ -557,7 +556,6 @@ fn emit_imports(state: &ArchitectureState) -> TokenStream {
         use aimdb_core::builder::AimDbBuilder;
         use aimdb_core::RecordKey;
         use aimdb_data_contracts::{#(#contract_traits),*};
-        use aimdb_executor::RuntimeAdapter;
         use serde::{Deserialize, Serialize};
     }
 }
@@ -731,7 +729,7 @@ fn emit_configure_schema(state: &ArchitectureState) -> TokenStream {
         /// addresses. Producers, consumers, serializers, and deserializers contain
         /// business logic and must be provided by application code — they are not
         /// generated here.
-        pub fn configure_schema<R: RuntimeAdapter + 'static>(builder: &mut AimDbBuilder<R>) {
+        pub fn configure_schema(builder: &mut AimDbBuilder) {
             #(#record_blocks)*
         }
     }
@@ -1250,7 +1248,6 @@ pub fn generate_hub_schema_rs(state: &ArchitectureState) -> String {
     let file_tokens = quote! {
         use aimdb_core::buffer::BufferCfg;
         use aimdb_core::builder::AimDbBuilder;
-        use aimdb_executor::RuntimeAdapter;
         use #common_crate::*;
 
         #configure_fn
@@ -1688,7 +1685,7 @@ pub fn {handler}(input: &{in_t}) -> Option<{out_t}> {{\n\
             // Pure source
             fns.push_str(&format!(
                 "pub async fn {}(\n\
-    _ctx: aimdb_core::RuntimeContext<TokioAdapter>,\n\
+    _ctx: aimdb_core::RuntimeContext,\n\
     _producer: aimdb_core::Producer<{out_t}>,\n\
 ) {{\n\
     todo!(\"implement {}\")\n\
@@ -1699,7 +1696,7 @@ pub fn {handler}(input: &{in_t}) -> Option<{out_t}> {{\n\
             // Pure sink / tap
             fns.push_str(&format!(
                 "pub async fn {}(\n\
-    _ctx: aimdb_core::RuntimeContext<TokioAdapter>,\n\
+    _ctx: aimdb_core::RuntimeContext,\n\
     _consumer: aimdb_core::Consumer<{in_t}>,\n\
 ) {{\n\
     todo!(\"implement {}\")\n\
@@ -1727,7 +1724,6 @@ pub async fn {task_name}() {{\n\
 // This file is scaffolded once — it will not be overwritten on subsequent runs.\n\
 // Regenerate signatures: delete this file, then run `aimdb generate --hub`.\n\
 \n\
-use aimdb_tokio_adapter::TokioAdapter;\n\
 use {common_crate}::*;\n\
 \n\
 {fns}"
@@ -1832,10 +1828,6 @@ url = "mqtt://ota/cmd/{variant}"
             out.contains("use aimdb_core::RecordKey;"),
             "Missing RecordKey import:\n{out}"
         );
-        assert!(
-            out.contains("use aimdb_executor::RuntimeAdapter;"),
-            "Missing RuntimeAdapter import:\n{out}"
-        );
         assert!(
             out.contains("use serde::{Deserialize, Serialize};"),
             "Missing serde import:\n{out}"
@@ -1937,9 +1929,7 @@ url = "mqtt://ota/cmd/{variant}"
     fn configure_schema_function_present() {
         let out = generated();
         assert!(
-            out.contains(
-                "pub fn configure_schema<R: RuntimeAdapter + 'static>(builder: &mut AimDbBuilder<R>)"
-            ),
+            out.contains("pub fn configure_schema(builder: &mut AimDbBuilder)"),
             "Missing configure_schema function:\n{out}"
         );
     }