diff --git a/aimdb-codegen/src/rust.rs b/aimdb-codegen/src/rust.rs
index bd2c42b7..6b9940ec 100644
--- a/aimdb-codegen/src/rust.rs
+++ b/aimdb-codegen/src/rust.rs
@@ -251,7 +251,7 @@ pub fn generate_main_rs(state: &ArchitectureState, binary_name: &str) -> Option<
         .filter_map(|c| match c.protocol.as_str() {
             "mqtt" => Some(quote! { use aimdb_mqtt_connector::MqttConnector; }),
             "knx" => Some(quote! {
-                use aimdb_knx_connector::{Channels, KnxConnector};
+                use aimdb_knx_connector::KnxConnector;
                 use aimdb_tokio_adapter::net::{TokioDelay, TokioNet};
             }),
             "ws" => Some(quote! { use aimdb_websocket_connector::WebSocketConnector; }),
@@ -269,18 +269,13 @@ pub fn generate_main_rs(state: &ArchitectureState, binary_name: &str) -> Option<
             let default = &c.default;
             let ctor: TokenStream = match c.protocol.as_str() {
                 "mqtt" => quote! { MqttConnector::new(&#var_ident) },
-                // The adapter owns the socket and the clock; the channels are
-                // the binary's, in a block-scoped `static`.
+                // The adapter owns the socket and the clock.
                 "knx" => quote! {
-                    {
-                        static KNX_CHANNELS: Channels = Channels::new();
-                        KnxConnector::new(
-                            TokioNet::udp(std::net::Ipv4Addr::UNSPECIFIED),
-                            TokioDelay,
-                            &#var_ident,
-                            &KNX_CHANNELS,
-                        )
-                    }
+                    KnxConnector::new(
+                        TokioNet::udp(std::net::Ipv4Addr::UNSPECIFIED),
+                        TokioDelay,
+                        &#var_ident,
+                    )
                 },
                 "ws" => quote! {
                     WebSocketConnector::new()
@@ -1411,7 +1406,7 @@ pub fn generate_hub_main_rs(state: &ArchitectureState) -> String {
         }
         if has_knx {
             v.push(quote! {
-                use aimdb_knx_connector::{Channels, KnxConnector};
+                use aimdb_knx_connector::KnxConnector;
                 use aimdb_tokio_adapter::net::{TokioDelay, TokioNet};
             });
         }
@@ -1456,15 +1451,11 @@ pub fn generate_hub_main_rs(state: &ArchitectureState) -> String {
         }
         if has_knx {
             v.push(quote! {
-                .with_connector({
-                    static KNX_CHANNELS: Channels = Channels::new();
-                    KnxConnector::new(
-                        TokioNet::udp(std::net::Ipv4Addr::UNSPECIFIED),
-                        TokioDelay,
-                        &knx_gateway,
-                        &KNX_CHANNELS,
-                    )
-                })
+                .with_connector(KnxConnector::new(
+                    TokioNet::udp(std::net::Ipv4Addr::UNSPECIFIED),
+                    TokioDelay,
+                    &knx_gateway,
+                ))
             });
         }
         if has_ws {
diff --git a/aimdb-knx-connector/Cargo.toml b/aimdb-knx-connector/Cargo.toml
index 34863fd4..d50e4329 100644
--- a/aimdb-knx-connector/Cargo.toml
+++ b/aimdb-knx-connector/Cargo.toml
@@ -21,12 +21,12 @@ default = ["aimdb-core/alloc"]
 # feature here — which is what lets a third runtime (FreeRTOS/lwIP) use this
 # crate with no edit to it.
 #
-# `embassy-sync` is not an Embassy dependency in the executor sense: it is
-# no_std, no_alloc, and pulls no executor, and its `Channel` is this connector's
-# public queue type (`Channels`) on every runtime.
+# `embassy-sync` no longer backs a public type here (the connection task pulls
+# from the record buffers directly); it is dropped together with the old
+# connector SPI.
 connector = [
     "aimdb-core/alloc",
-    "aimdb-core/connector-session", # `pump_sink`/`pump_source`/`Source`/`Payload`
+    "aimdb-core/connector-session", # `DatagramBinder`/`Datagram`/`Delay`
     "embassy-sync",
 ]
 
@@ -49,10 +49,9 @@ embassy-runtime = ["connector"]
 # the **final binary** may choose one: a library enabling it would hand every
 # downstream binary a duplicate-symbol link error with no way to opt out.
 #
-# So this stays off by default and out of `connector`/`std`. This crate's own
-# tests get the impl through a dev-dependency; a std binary that instantiates a
-# `CriticalSectionRawMutex` channel and has no other impl in its graph can
-# either depend on `critical-section` with `features = ["std"]` directly (the
+# So this stays off by default and out of `connector`/`std`. A std binary that
+# instantiates a `CriticalSectionRawMutex` channel and has no other impl in its
+# graph can either depend on `critical-section` with `features = ["std"]` directly (the
 # documented way) or enable this feature. On Embassy the HAL (cortex-m /
 # embassy-rp / ...) already provides one.
 critical-section-std-impl = ["critical-section/std"]
@@ -92,9 +91,6 @@ heapless = { workspace = true }
 
 [dev-dependencies]
 tokio = { workspace = true, features = ["full"] }
-# Test binaries are binaries: selecting the impl here lets `CriticalSectionRawMutex`
-# link for this crate's tests without imposing that choice on consumers.
-critical-section = { version = "1.2", features = ["std"] }
 tokio-test = "0.4"
 aimdb-tokio-adapter = { path = "../aimdb-tokio-adapter", features = [
     "tokio-runtime",
diff --git a/aimdb-knx-connector/README.md b/aimdb-knx-connector/README.md
index 78add83b..9587c430 100644
--- a/aimdb-knx-connector/README.md
+++ b/aimdb-knx-connector/README.md
@@ -45,7 +45,7 @@ We're working with upstream to get these changes merged. Once published, the pat
 ## Quick Start (Tokio)
 
 ```rust
-use aimdb_knx_connector::{Channels, KnxConnector};
+use aimdb_knx_connector::KnxConnector;
 use aimdb_tokio_adapter::net::{TokioDelay, TokioNet};
 use aimdb_tokio_adapter::TokioAdapter;
 use std::net::Ipv4Addr;
@@ -55,11 +55,6 @@ struct LightState {
     is_on: bool,
 }
 
-// The connector's queues. `'static` because the connection task and the pumps
-// are spawned as `'static` futures; a `StaticCell` supplies this on an MCU.
-// One pair per connector — do not share it between two `KnxConnector`s.
-static CHANNELS: Channels = Channels::new();
-
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let db = AimDbBuilder::new()
@@ -70,7 +65,6 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
             TokioNet::udp(Ipv4Addr::UNSPECIFIED),
             TokioDelay,
             "knx://192.168.1.19:3671",
-            &CHANNELS,
         ))
         .configure::<LightState>(|reg| {
             reg.buffer(BufferCfg::SingleLatest)
@@ -96,14 +90,11 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
 
 ## Quick Start (Embassy)
 
-The same constructor — only the binder and clock change, and the channels come
-from a `static` instead of being sized at runtime:
+The same constructor — only the binder and clock change:
 
 ```rust
 use aimdb_embassy_adapter::net::{EmbassyDelay, EmbassyNet};
-use aimdb_knx_connector::{Channels, KnxConnector};
-
-static CHANNELS: Channels<32> = Channels::new();
+use aimdb_knx_connector::KnxConnector;
 
 let binder = EmbassyNet::udp(stack, rx_meta, rx_buf, tx_meta, tx_buf);
 
@@ -113,12 +104,10 @@ let builder = AimDbBuilder::new()
         binder,
         EmbassyDelay,
         "knx://192.168.1.19:3671",
-        &CHANNELS,
     ));
 ```
 
-Enable `features = ["connector"], default-features = false` — no
-`critical-section-std-impl`, since the HAL already provides an impl. See
+Enable `features = ["connector"], default-features = false`. See
 `examples/embassy-knx-connector-demo/` for the full wiring.
 
 ## Group Address Format
diff --git a/aimdb-knx-connector/src/client.rs b/aimdb-knx-connector/src/client.rs
index 6b01384e..f13d33bf 100644
--- a/aimdb-knx-connector/src/client.rs
+++ b/aimdb-knx-connector/src/client.rs
@@ -4,21 +4,22 @@
 //! The clock stays `RuntimeOps::now_nanos`, a plain call; only *sleeping* goes
 //! through [`Delay`], so nothing is boxed per loop iteration.
 //!
-//! The `embassy-sync` and `embassy-futures` types below are executor-independent
-//! — neither pulls an executor, and both build on std — so they back this task
-//! on either runtime.
+//! The task dispatches inbound telegrams into their records through an
+//! [`InboundDispatch`](aimdb_core::InboundDispatch) and pulls outbound values
+//! from an [`OutboundRoutes`] while the tunnel is connected. `embassy-futures`'
+//! select is executor-independent, so the task runs on either runtime.
 
-use alloc::string::{String, ToString};
 use alloc::sync::Arc;
 use alloc::vec::Vec;
-use core::future::Future;
+use core::fmt::Write as _;
+use core::future::poll_fn;
 use core::net::SocketAddr;
+use core::task::Poll;
 use core::time::Duration;
 
-use aimdb_core::session::{
-    Datagram, DatagramBinder, Delay, Payload, TransportError, TransportResult,
-};
-use aimdb_core::{log_debug, log_error, log_warn, RuntimeOps};
+use aimdb_core::connector::TopicBuf;
+use aimdb_core::session::{Datagram, DatagramBinder, Delay, TransportError, TransportResult};
+use aimdb_core::{log_debug, log_error, log_warn, OutboundRoutes, RouteId, RuntimeOps};
 
 use crate::tunnel::{
     drain_actions, GroupWrite, LocalEndpoint, Millis, TunnelConfig, TunnelEngine, TunnelIo,
@@ -30,22 +31,24 @@ const BIND_RETRY: Duration = Duration::from_secs(5);
 /// Per-datagram receive buffer; a KNXnet/IP frame fits comfortably.
 const RECV_BUF: usize = 512;
 
-/// Where parsed telegrams go — an `embassy_sync` channel on either runtime.
+/// Where parsed telegrams go: an [`InboundDispatch`](aimdb_core::InboundDispatch)
+/// in the connector.
 ///
-/// Non-blocking by contract: a full sink drops rather than stalling the
-/// protocol loop.
+/// Non-blocking by contract: delivery never stalls the protocol loop.
 pub trait TelegramSink {
-    /// Enqueue one `(group-address, payload)`. `false` if it was dropped.
-    fn try_send(&self, topic: String, payload: Payload) -> bool;
+    /// Deliver one telegram for group address `topic`.
+    fn deliver(&self, topic: &str, payload: &[u8]);
 }
 
-/// Where outbound commands come from, the dual of [`TelegramSink`].
-pub trait CommandSource {
-    /// Yield the next command. Parks forever once no producer remains, so the
-    /// select arm goes quiet instead of ending the task.
-    fn recv(&mut self) -> impl Future<Output = GroupWrite> + Send + '_;
+impl TelegramSink for aimdb_core::InboundDispatch {
+    fn deliver(&self, topic: &str, payload: &[u8]) {
+        self.dispatch(topic, payload);
+    }
 }
 
+/// The longest group address, `31/7/255`, is 8 bytes.
+const GROUP_ADDRESS_LEN: usize = 16;
+
 /// The socket-side glue for [`drain_actions`], written once against
 /// [`Datagram`] instead of once per runtime.
 struct NeutralIo<'a, U, S> {
@@ -77,8 +80,10 @@ where
 
     fn forward(&mut self, addr: GroupAddress, payload: Vec<u8>) {
         log_debug!("KNX telegram: {} ({} bytes)", addr, payload.len());
-        if !self.sink.try_send(addr.to_string(), Payload::from(payload)) {
-            log_warn!("KNX inbound: dropping telegram for {} (sink full)", addr);
+        let mut storage = [0u8; GROUP_ADDRESS_LEN];
+        let mut topic = TopicBuf::new(&mut storage);
+        if write!(topic, "{addr}").is_ok() {
+            self.sink.deliver(topic.as_str(), &payload);
         }
     }
 
@@ -87,20 +92,44 @@ where
     }
 }
 
+/// Hand the value `outbound` staged for route `id` to the engine as a group
+/// write. An invalid group address or an oversize payload is logged, counted
+/// as rejected in the route's `RouteStats`, and skipped.
+fn send_staged(engine: &mut TunnelEngine, outbound: &mut OutboundRoutes, id: RouteId, now: Millis) {
+    let command = {
+        let Some(msg) = outbound.take_staged() else {
+            return;
+        };
+        GroupWrite::try_new(msg.topic, msg.payload.as_slice())
+    };
+    match command {
+        Ok(command) => {
+            let _ = engine.handle_command(command, now);
+        }
+        Err(_e) => {
+            log_warn!(
+                "KNX outbound: skipping a value for route '{}': {:?}",
+                outbound.routes()[id].default_topic,
+                _e
+            );
+            outbound.reject(id);
+        }
+    }
+}
+
 /// Drive the engine over one socket's lifetime; returns when it asks for a reset.
-async fn drive_connection<U, D, S, C>(
+async fn drive_connection<U, D, S>(
     engine: &mut TunnelEngine,
     socket: &mut U,
     gateway: SocketAddr,
     runtime: &Arc<dyn RuntimeOps>,
     delay: &D,
     sink: &S,
-    commands: &mut C,
+    outbound: &mut OutboundRoutes,
 ) where
     U: Datagram + Send,
     D: Delay,
     S: TelegramSink + Sync,
-    C: CommandSource,
 {
     // Executor-independent despite the name: `embassy-futures` has no
     // dependencies and its select is pure `core::task`.
@@ -129,9 +158,12 @@ async fn drive_connection<U, D, S, C>(
     // one — unlike the `tokio::select!` this task replaces, which picked among
     // the ready arms at random. With a fixed order, sustained inbound traffic
     // means the first arm is ready on every pass and the command arm is never
-    // reached, so outbound `GroupWrite`s stall until the channel drops them.
+    // reached, so outbound values stall in their record buffers.
     // Swapping the two contended arms each pass restores that fairness.
     let mut inbound_first = true;
+    // Latched on `Ready(None)` (every route closed, or none): passing it
+    // through would resolve the arm on every pass and the loop would spin.
+    let mut outbound_done = false;
 
     loop {
         engine.poll(now_ms());
@@ -151,34 +183,37 @@ async fn drive_connection<U, D, S, C>(
         let deadline = delay.sleep(Duration::from_millis(sleep_ms));
         let mut recv_buf = [0u8; RECV_BUF];
 
-        // Only drain commands while connected: during connect and backoff the
-        // arm stays pending, so commands queue and flush once the handshake
-        // completes — as both hand-written clients do.
+        // Only pull values while connected: during connect and backoff the
+        // arm stays pending, so values wait in their record buffers and go out
+        // once the handshake completes. A value leaves its buffer only when
+        // the arm resolves, so a losing arm takes nothing.
         let connected = engine.is_connected();
-        let cmd_arm = async {
-            if connected {
-                commands.recv().await
-            } else {
-                core::future::pending().await
+        let cmd_arm = poll_fn(|cx| {
+            if !connected || outbound_done {
+                return Poll::Pending;
             }
-        };
+            match outbound.poll_stage(cx) {
+                Poll::Ready(Some(id)) => Poll::Ready(id),
+                Poll::Ready(None) => {
+                    outbound_done = true;
+                    Poll::Pending
+                }
+                Poll::Pending => Poll::Pending,
+            }
+        });
 
         // The deadline arm stays last in both orders: it only ever asks for a
         // `poll` the loop top would reach anyway.
         if inbound_first {
             match select3(socket.recv_from(&mut recv_buf), cmd_arm, deadline).await {
                 Either3::First(r) => apply_inbound(engine, &recv_buf, r, now_ms()),
-                Either3::Second(cmd) => {
-                    let _ = engine.handle_command(cmd, now_ms());
-                }
+                Either3::Second(id) => send_staged(engine, outbound, id, now_ms()),
                 // Woken for the engine deadline; `poll` at the loop top fires it.
                 Either3::Third(()) => {}
             }
         } else {
             match select3(cmd_arm, socket.recv_from(&mut recv_buf), deadline).await {
-                Either3::First(cmd) => {
-                    let _ = engine.handle_command(cmd, now_ms());
-                }
+                Either3::First(id) => send_staged(engine, outbound, id, now_ms()),
                 Either3::Second(r) => apply_inbound(engine, &recv_buf, r, now_ms()),
                 Either3::Third(()) => {}
             }
@@ -192,18 +227,17 @@ async fn drive_connection<U, D, S, C>(
 /// Binds a socket, advertises its real local endpoint when the stack exposes
 /// one, drives the shared [`TunnelEngine`] over that socket's lifetime, then
 /// rebinds after the engine's backoff.
-pub async fn connection_task<B, D, S, C>(
+pub async fn connection_task<B, D, S>(
     binder: B,
     gateway: SocketAddr,
     runtime: Arc<dyn RuntimeOps>,
     delay: D,
     sink: S,
-    mut commands: C,
+    mut outbound: OutboundRoutes,
 ) where
     B: DatagramBinder,
     D: Delay,
     S: TelegramSink + Sync,
-    C: CommandSource,
 {
     let now_ms = || runtime.now_nanos() / 1_000_000;
     let mut engine = TunnelEngine::new(TunnelConfig::default(), now_ms());
@@ -267,7 +301,7 @@ pub async fn connection_task<B, D, S, C>(
             &runtime,
             &delay,
             &sink,
-            &mut commands,
+            &mut outbound,
         )
         .await;
 
@@ -280,45 +314,15 @@ pub async fn connection_task<B, D, S, C>(
     }
 }
 
-/// Channel bridges over `embassy_sync`, which is executor-independent, so the
-/// same types back the task on both runtimes.
-#[cfg(feature = "connector")]
-pub mod shared_channel {
-    use super::{CommandSource, GroupWrite, Payload, TelegramSink};
-    use alloc::string::String;
-    use core::future::Future;
-    use embassy_sync::blocking_mutex::raw::CriticalSectionRawMutex;
-    use embassy_sync::channel::{Receiver, Sender};
-
-    /// Sending half of the inbound-telegram channel.
-    pub struct ChannelSink<'a, const N: usize>(
-        pub Sender<'a, CriticalSectionRawMutex, (String, Payload), N>,
-    );
-
-    impl<const N: usize> TelegramSink for ChannelSink<'_, N> {
-        fn try_send(&self, topic: String, payload: Payload) -> bool {
-            self.0.try_send((topic, payload)).is_ok()
-        }
-    }
-
-    /// Receiving half of the outbound-command channel.
-    pub struct ChannelCommands<'a, const N: usize>(
-        pub Receiver<'a, CriticalSectionRawMutex, GroupWrite, N>,
-    );
-
-    impl<const N: usize> CommandSource for ChannelCommands<'_, N> {
-        fn recv(&mut self) -> impl Future<Output = GroupWrite> + Send + '_ {
-            self.0.receive()
-        }
-    }
-}
-
 #[cfg(all(test, feature = "std"))]
 mod tests {
     use super::*;
+    use aimdb_core::buffer::BufferCfg;
     use aimdb_core::session::TransportError;
+    use aimdb_core::AimDbBuilder;
     use aimdb_tokio_adapter::net::{TokioDelay, TokioNet};
-    use aimdb_tokio_adapter::TokioAdapter;
+    use aimdb_tokio_adapter::{TokioAdapter, TokioRecordRegistrarExt};
+    use core::future::Future;
     use core::pin::Pin;
     use std::net::Ipv4Addr;
     use std::sync::atomic::{AtomicUsize, Ordering};
@@ -326,22 +330,25 @@ mod tests {
 
     /// Collects forwarded telegrams; `Sync`, as [`TelegramSink`] requires.
     #[derive(Default)]
-    struct VecSink(Mutex<Vec<(String, Payload)>>);
+    struct VecSink(Mutex<Vec<(String, Vec<u8>)>>);
 
     impl TelegramSink for VecSink {
-        fn try_send(&self, topic: String, payload: Payload) -> bool {
-            self.0.lock().expect("sink mutex").push((topic, payload));
-            true
+        fn deliver(&self, topic: &str, payload: &[u8]) {
+            self.0
+                .lock()
+                .expect("sink mutex")
+                .push((topic.into(), payload.into()));
         }
     }
 
-    /// No outbound producer: the command arm never fires.
-    struct NoCommands;
-
-    impl CommandSource for NoCommands {
-        fn recv(&mut self) -> impl Future<Output = GroupWrite> + Send + '_ {
-            core::future::pending()
-        }
+    /// No outbound routes: the command arm never fires.
+    async fn no_outbound() -> OutboundRoutes {
+        let (db, _runner) = AimDbBuilder::new()
+            .runtime(Arc::new(TokioAdapter))
+            .build()
+            .await
+            .expect("build db");
+        OutboundRoutes::new(&db, "knx").expect("outbound routes")
     }
 
     fn runtime() -> Arc<dyn RuntimeOps> {
@@ -354,15 +361,15 @@ mod tests {
     /// `Pin<Box<dyn Future<Output = ()> + Send + 'static>>`. Everything that
     /// declares `+ Send` on a return type does so to make this line compile for
     /// a *generic* task.
-    #[test]
-    fn unified_task_is_boxable_as_the_runners_send_future() {
+    #[tokio::test]
+    async fn unified_task_is_boxable_as_the_runners_send_future() {
         let task = connection_task(
             TokioNet::udp(Ipv4Addr::LOCALHOST),
             "127.0.0.1:3671".parse().expect("gateway addr"),
             runtime(),
             TokioDelay,
             VecSink::default(),
-            NoCommands,
+            no_outbound().await,
         );
         let _boxed: Pin<Box<dyn Future<Output = ()> + Send + 'static>> = Box::pin(task);
     }
@@ -384,7 +391,7 @@ mod tests {
             runtime(),
             TokioDelay,
             VecSink::default(),
-            NoCommands,
+            no_outbound().await,
         ));
 
         let mut buf = [0u8; 128];
@@ -430,7 +437,7 @@ mod tests {
             runtime(),
             TokioDelay,
             VecSink::default(),
-            NoCommands,
+            no_outbound().await,
         ));
 
         let mut buf = [0u8; 128];
@@ -451,39 +458,10 @@ mod tests {
         task.abort();
     }
 
-    /// The unified task on Tokio, moving real telegrams through the *same*
-    /// `embassy_sync` channel types the MCU uses: a full handshake, an inbound
-    /// telegram with its ACK, and an outbound command.
-    #[tokio::test]
-    async fn shared_embassy_channels_carry_telegrams_on_tokio() {
-        use super::shared_channel::{ChannelCommands, ChannelSink};
-        use embassy_sync::blocking_mutex::raw::CriticalSectionRawMutex;
-        use embassy_sync::channel::Channel;
-
-        const N: usize = 8;
-        // Leaked for `'static` borrows, as a `StaticCell` gives on the MCU.
-        let inbound: &'static Channel<CriticalSectionRawMutex, (String, Payload), N> =
-            Box::leak(Box::new(Channel::new()));
-        let commands: &'static Channel<CriticalSectionRawMutex, GroupWrite, N> =
-            Box::leak(Box::new(Channel::new()));
-
-        let gateway = tokio::net::UdpSocket::bind("127.0.0.1:0")
-            .await
-            .expect("bind gateway");
-        let gateway_addr = gateway.local_addr().expect("gateway addr");
-
-        let task = tokio::spawn(connection_task(
-            TokioNet::udp(Ipv4Addr::LOCALHOST),
-            gateway_addr,
-            runtime(),
-            TokioDelay,
-            ChannelSink::<N>(inbound.sender()),
-            ChannelCommands::<N>(commands.receiver()),
-        ));
-
-        let mut buf = [0u8; 1024];
-
-        // Handshake.
+    /// Answer the client's CONNECT_REQUEST with channel 7; returns the
+    /// client's address.
+    async fn accept_connect(gateway: &tokio::net::UdpSocket) -> SocketAddr {
+        let mut buf = [0u8; 128];
         let (_, client_addr) = tokio::time::timeout(RECV_TIMEOUT, gateway.recv_from(&mut buf))
             .await
             .expect("no CONNECT_REQUEST")
@@ -498,8 +476,63 @@ mod tests {
             .send_to(&connect_response, client_addr)
             .await
             .expect("send CONNECT_RESPONSE");
+        client_addr
+    }
+
+    /// Wait for a TUNNELING_REQUEST and check it writes 1 to group 1/0/8.
+    async fn expect_write_to_1_0_8(gateway: &tokio::net::UdpSocket) {
+        let mut buf = [0u8; 128];
+        let (len, _) = tokio::time::timeout(RECV_TIMEOUT, gateway.recv_from(&mut buf))
+            .await
+            .expect("no TUNNELING_REQUEST")
+            .expect("recv_from");
+        assert_eq!(u16::from_be_bytes([buf[2], buf[3]]), 0x0420);
+        assert_eq!(&buf[16..18], &[0x08, 0x08], "cEMI destination = 1/0/8");
+        assert_eq!(buf[len - 1], 0x81, "APCI GroupValueWrite | value 1");
+    }
+
+    /// A db whose `knx` connector tunnels to `gateway`: record `in` reads group
+    /// 1/0/7 and record `out` writes group 1/0/8.
+    async fn knx_db(gateway: SocketAddr) -> (aimdb_core::AimDb, aimdb_core::builder::AimDbRunner) {
+        let mut builder = AimDbBuilder::new()
+            .runtime(Arc::new(TokioAdapter))
+            .with_connector(crate::KnxConnector::new(
+                TokioNet::udp(Ipv4Addr::LOCALHOST),
+                TokioDelay,
+                format!("knx://{gateway}"),
+            ));
+        builder.configure::<u8>("in", |reg| {
+            reg.buffer(BufferCfg::SingleLatest)
+                .link_from("knx://1/0/7")
+                .with_deserializer(|_ctx, data: &[u8]| {
+                    data.first().copied().ok_or_else(|| String::from("empty"))
+                })
+                .finish();
+        });
+        builder.configure::<u8>("out", |reg| {
+            reg.buffer(BufferCfg::SingleLatest)
+                .link_to("knx://1/0/8")
+                .with_serializer(|_ctx, v: &u8| Ok(vec![*v]))
+                .finish();
+        });
+        builder.build().await.expect("build db")
+    }
+
+    /// The connector end to end: a full handshake, an inbound telegram
+    /// dispatched into its record with its ACK, and a produced value on the
+    /// wire as a group write.
+    #[tokio::test]
+    async fn telegrams_round_trip_through_records() {
+        let gateway = tokio::net::UdpSocket::bind("127.0.0.1:0")
+            .await
+            .expect("bind gateway");
+        let (db, runner) = knx_db(gateway.local_addr().expect("gateway addr")).await;
+        let mut inbound = db.subscribe::<u8>("in").expect("subscribe");
+        let task = tokio::spawn(runner.run());
 
-        // Inbound telegram -> ACK on the wire, payload on the shared channel.
+        let client_addr = accept_connect(&gateway).await;
+
+        // Inbound telegram -> ACK on the wire, value in the record.
         let cemi = [
             0x29, 0x00, 0xBC, 0xE0, 0x00, 0x00, 0x08, 0x07, 0x01, 0x00, 0x81,
         ];
@@ -513,6 +546,7 @@ mod tests {
             .await
             .expect("send telegram");
 
+        let mut buf = [0u8; 128];
         tokio::time::timeout(RECV_TIMEOUT, gateway.recv_from(&mut buf))
             .await
             .expect("no TUNNELING_ACK")
@@ -520,29 +554,32 @@ mod tests {
         assert_eq!(u16::from_be_bytes([buf[2], buf[3]]), 0x0421);
         assert_eq!(buf[8], 42, "sequence echoed");
 
-        let (topic, payload) = tokio::time::timeout(RECV_TIMEOUT, inbound.receive())
+        let value = tokio::time::timeout(RECV_TIMEOUT, inbound.recv())
             .await
-            .expect("no telegram reached the embassy-sync channel");
-        assert_eq!(topic, "1/0/7");
-        assert_eq!(&payload[..], &[0x01]);
-
-        // Outbound: a command through the shared channel reaches the wire.
-        let mut data = heapless::Vec::new();
-        data.push(0x01).expect("push");
-        commands
-            .send(GroupWrite {
-                group_addr: "1/0/8".parse().expect("group address"),
-                data,
-            })
-            .await;
+            .expect("no telegram reached the record")
+            .expect("recv");
+        assert_eq!(value, 0x01);
 
-        let (len, _) = tokio::time::timeout(RECV_TIMEOUT, gateway.recv_from(&mut buf))
+        // Outbound: a produced value reaches the wire.
+        db.producer::<u8>("out").expect("producer").produce(1);
+        expect_write_to_1_0_8(&gateway).await;
+
+        task.abort();
+    }
+
+    /// A value produced while the tunnel is still connecting waits in its
+    /// record buffer and goes out once the handshake completes.
+    #[tokio::test]
+    async fn a_value_produced_during_connect_is_sent_after_the_handshake() {
+        let gateway = tokio::net::UdpSocket::bind("127.0.0.1:0")
             .await
-            .expect("no TUNNELING_REQUEST")
-            .expect("recv_from");
-        assert_eq!(u16::from_be_bytes([buf[2], buf[3]]), 0x0420);
-        assert_eq!(&buf[16..18], &[0x08, 0x08], "cEMI destination = 1/0/8");
-        assert_eq!(buf[len - 1], 0x81, "APCI GroupValueWrite | value 1");
+            .expect("bind gateway");
+        let (db, runner) = knx_db(gateway.local_addr().expect("gateway addr")).await;
+        db.producer::<u8>("out").expect("producer").produce(1);
+        let task = tokio::spawn(runner.run());
+
+        accept_connect(&gateway).await;
+        expect_write_to_1_0_8(&gateway).await;
 
         task.abort();
     }
@@ -652,7 +689,7 @@ mod tests {
             runtime(),
             TokioDelay,
             VecSink::default(),
-            NoCommands,
+            no_outbound().await,
         ));
 
         // Must exceed the two `BIND_RETRY` sleeps the task waits out. `RECV_TIMEOUT`
@@ -704,7 +741,7 @@ mod tests {
             runtime(),
             TokioDelay,
             VecSink::default(),
-            NoCommands,
+            no_outbound().await,
         ));
 
         // Cycle 1: the socket knows its address, so the HPAI is explicit.
diff --git a/aimdb-knx-connector/src/connector.rs b/aimdb-knx-connector/src/connector.rs
index f11bf947..9e9edfd5 100644
--- a/aimdb-knx-connector/src/connector.rs
+++ b/aimdb-knx-connector/src/connector.rs
@@ -2,9 +2,9 @@
 //!
 //! Generic over core's [`DatagramBinder`](aimdb_core::session::DatagramBinder)
 //! and [`Delay`](aimdb_core::session::Delay), so the adapter owns the UDP
-//! socket and the clock while this crate owns the tunnelling protocol.
-//! The channels between the pumps and the connection task are `embassy_sync`,
-//! which is executor-independent, so one wiring serves both runtimes.
+//! socket and the clock while this crate owns the tunnelling protocol. The one
+//! connection task dispatches inbound telegrams into their records and pulls
+//! outbound values from their record buffers itself; core runs no pump for it.
 
 use alloc::boxed::Box;
 use alloc::string::String;
@@ -16,119 +16,31 @@ use core::net::SocketAddr;
 use core::pin::Pin;
 
 use aimdb_core::connector::{ConnectorBuilder, ConnectorUrl};
-use aimdb_core::session::{pump_sink, pump_source, Payload};
-use aimdb_core::transport::{Connector, ConnectorConfig, PublishError};
-use aimdb_core::{log_info, AimDb, DbError, DbResult, RuntimeOps};
+use aimdb_core::{log_info, AimDb, DbError, DbResult, InboundDispatch, OutboundRoutes, RuntimeOps};
 
-use embassy_sync::blocking_mutex::raw::CriticalSectionRawMutex;
-use embassy_sync::channel::Channel;
-
-use crate::client::{
-    connection_task,
-    shared_channel::{ChannelCommands, ChannelSink},
-};
-use crate::tunnel::GroupWrite;
+use crate::client::connection_task;
 
 type BoxFuture = Pin<Box<dyn Future<Output = ()> + Send + 'static>>;
 
 /// Default KNXnet/IP tunnelling port.
 const DEFAULT_PORT: u16 = 3671;
 
-/// Capacity of the command and telegram channels.
-///
-/// A const generic rather than a builder setter: the MCU allocates these in a
-/// `static`, where the size must be a constant.
-pub const DEFAULT_QUEUE: usize = 32;
-
-/// The inbound-telegram channel type.
-pub type TelegramChannel<const N: usize> = Channel<CriticalSectionRawMutex, (String, Payload), N>;
-/// The outbound-command channel type.
-pub type CommandChannel<const N: usize> = Channel<CriticalSectionRawMutex, GroupWrite, N>;
-
-/// Outbound half: `pump_sink` hands each serialized record here.
-struct KnxSink<'a, const N: usize> {
-    commands: &'a CommandChannel<N>,
-}
-
-impl<const N: usize> Connector for KnxSink<'_, N> {
-    fn publish(
-        &self,
-        destination: &str,
-        _config: &ConnectorConfig,
-        payload: &[u8],
-    ) -> Pin<Box<dyn Future<Output = Result<(), PublishError>> + Send + '_>> {
-        // Validation shared with the connection task (same checks, same order).
-        let command = GroupWrite::try_new(destination, payload);
-        Box::pin(async move {
-            self.commands.send(command?).await;
-            Ok(())
-        })
-    }
-}
-
-/// Inbound source drained by `pump_source`.
-struct KnxSource<'a, const N: usize> {
-    telegrams: &'a TelegramChannel<N>,
-}
-
-impl<const N: usize> aimdb_core::session::Source for KnxSource<'_, N> {
-    fn next(&mut self) -> aimdb_core::session::BoxFut<'_, Option<(String, Payload)>> {
-        Box::pin(async move { Some(self.telegrams.receive().await) })
-    }
-}
-
 /// KNX/IP tunnelling connector over an adapter's datagram transport.
-///
-/// `N` sizes both the command and telegram channels.
-pub struct KnxConnector<B, D, const N: usize = DEFAULT_QUEUE> {
+pub struct KnxConnector<B, D> {
     binder: B,
     delay: D,
     gateway_url: String,
-    channels: &'static Channels<N>,
-}
-
-/// The channel pair, held for the process lifetime.
-///
-/// `'static` because the connection task and the pumps are spawned as
-/// `'static` futures; a `StaticCell` supplies this on the MCU and a `static`
-/// item on a host, matching design 037's allocate-at-build model.
-pub struct Channels<const N: usize = DEFAULT_QUEUE> {
-    telegrams: TelegramChannel<N>,
-    commands: CommandChannel<N>,
 }
 
-impl<const N: usize> Default for Channels<N> {
-    fn default() -> Self {
-        Self::new()
-    }
-}
-
-impl<const N: usize> Channels<N> {
-    /// A fresh, empty channel pair.
-    pub const fn new() -> Self {
-        Self {
-            telegrams: Channel::new(),
-            commands: Channel::new(),
-        }
-    }
-}
-
-impl<B, D, const N: usize> KnxConnector<B, D, N> {
+impl<B, D> KnxConnector<B, D> {
     /// Connect to the KNX/IP gateway at `gateway_url` (`knx://host:port`).
     ///
-    /// `binder` and `delay` come from an adapter; `channels` is the caller's
-    /// `'static` channel pair.
-    pub fn new(
-        binder: B,
-        delay: D,
-        gateway_url: impl Into<String>,
-        channels: &'static Channels<N>,
-    ) -> Self {
+    /// `binder` and `delay` come from an adapter.
+    pub fn new(binder: B, delay: D, gateway_url: impl Into<String>) -> Self {
         Self {
             binder,
             delay,
             gateway_url: gateway_url.into(),
-            channels,
         }
     }
 
@@ -153,7 +65,7 @@ impl<B, D, const N: usize> KnxConnector<B, D, N> {
     }
 }
 
-impl<B, D, const N: usize> ConnectorBuilder for KnxConnector<B, D, N>
+impl<B, D> ConnectorBuilder for KnxConnector<B, D>
 where
     B: aimdb_core::session::DatagramBinder + Clone + Send + Sync + 'static,
     D: aimdb_core::session::Delay + Clone + Send + Sync + 'static,
@@ -167,32 +79,20 @@ where
             log_info!("Creating KNX connector for gateway {}", gateway);
 
             let runtime: Arc<dyn RuntimeOps> = db.runtime_ops();
-            let channels = self.channels;
+            // Built here, so routes are subscribed before the task first runs:
+            // a command produced while the tunnel connects waits in its record
+            // buffer and goes out once the handshake completes.
+            let inbound = InboundDispatch::new(db, "knx", &aimdb_core::ExactGrammar)?;
+            let outbound = OutboundRoutes::new(db, "knx")?;
             let task: BoxFuture = Box::pin(connection_task(
                 self.binder.clone(),
                 gateway,
                 runtime,
                 self.delay.clone(),
-                ChannelSink::<N>(channels.telegrams.sender()),
-                ChannelCommands::<N>(channels.commands.receiver()),
+                inbound,
+                outbound,
             ));
-
-            let mut futures: Vec<BoxFuture> = vec![task];
-            futures.extend(pump_source(
-                db,
-                db.inbound_router("knx", &aimdb_core::ExactGrammar)?,
-                KnxSource::<N> {
-                    telegrams: &channels.telegrams,
-                },
-            ));
-            futures.extend(pump_sink(
-                db,
-                "knx",
-                Arc::new(KnxSink::<N> {
-                    commands: &channels.commands,
-                }),
-            ));
-            Ok(futures)
+            Ok(vec![task])
         })
     }
 
@@ -222,8 +122,6 @@ mod tests {
     use aimdb_tokio_adapter::{TokioAdapter, TokioRecordRegistrarExt};
     use std::net::Ipv4Addr;
 
-    static CHANNELS: Channels<8> = Channels::new();
-
     async fn db() -> AimDb {
         let mut builder = AimDbBuilder::new().runtime(Arc::new(TokioAdapter));
         builder.configure::<u64>("light", |reg| {
@@ -236,11 +134,10 @@ mod tests {
     #[tokio::test]
     async fn an_unparsable_gateway_fails_the_build() {
         let db = db().await;
-        let connector = KnxConnector::<_, _, 8>::new(
+        let connector = KnxConnector::new(
             TokioNet::udp(Ipv4Addr::LOCALHOST),
             TokioDelay,
             "knx://not-an-ip:3671",
-            &CHANNELS,
         );
         let Err(err) = connector.build(&db).await else {
             panic!("a hostname must be rejected: it is never resolved");
@@ -251,22 +148,18 @@ mod tests {
         );
     }
 
-    /// A db with one inbound and one outbound `knx` route, so both pumps have
-    /// something to contribute.
+    /// A db with one inbound and one outbound `knx` route.
     ///
     /// A connector must be registered or the builder rejects the routes ("no
-    /// connector registered for scheme 'knx'"). It gets its own channel pair:
-    /// two connectors sharing one would split the command queue between them.
-    /// Nothing here is ever driven — `build` only collects futures.
+    /// connector registered for scheme 'knx'"). Nothing here is ever driven —
+    /// `build` only collects futures.
     async fn routed_db() -> AimDb {
-        static REGISTERED: Channels<8> = Channels::new();
         let mut builder = AimDbBuilder::new()
             .runtime(Arc::new(TokioAdapter))
-            .with_connector(KnxConnector::<_, _, 8>::new(
+            .with_connector(KnxConnector::new(
                 TokioNet::udp(Ipv4Addr::LOCALHOST),
                 TokioDelay,
                 "knx://127.0.0.1:3671",
-                &REGISTERED,
             ));
         builder.configure::<u64>("switch", |reg| {
             reg.buffer(BufferCfg::SingleLatest)
@@ -286,90 +179,51 @@ mod tests {
     }
 
     /// The connector registers under the `knx` scheme and contributes exactly
-    /// the connection task, one `pump_source`, and one publisher per outbound
-    /// route.
-    ///
-    /// The count is asserted, not just non-emptiness: `pump_source` yields one
-    /// future unconditionally and `pump_sink` one per outbound route, so a pump
-    /// silently dropping out of `build` is the failure this catches, and
-    /// `!is_empty()` would not — the connection task alone satisfies that.
+    /// one future, the connection task, which dispatches inbound telegrams and
+    /// pulls outbound values itself.
     #[tokio::test]
-    async fn build_yields_the_connection_task_and_pumps() {
-        static CH: Channels<8> = Channels::new();
-        let db = routed_db().await;
-        let connector = KnxConnector::<_, _, 8>::new(
-            TokioNet::udp(Ipv4Addr::LOCALHOST),
-            TokioDelay,
-            "knx://127.0.0.1:3671",
-            &CH,
-        );
-        assert_eq!(ConnectorBuilder::scheme(&connector), "knx");
-
-        let futures = connector.build(&db).await.expect("build");
-        assert_eq!(
-            futures.len(),
-            3,
-            "connection task + pump_source + one publisher for `knx://1/0/6`"
-        );
-    }
-
-    /// With no outbound route, `pump_sink` contributes nothing and the count
-    /// drops to two — the half of the contract the routed test cannot show.
-    #[tokio::test]
-    async fn an_inbound_only_db_yields_no_publisher() {
-        static CH: Channels<8> = Channels::new();
-        let db = db().await;
-        let connector = KnxConnector::<_, _, 8>::new(
-            TokioNet::udp(Ipv4Addr::LOCALHOST),
-            TokioDelay,
-            "knx://127.0.0.1:3671",
-            &CH,
-        );
+    async fn build_yields_only_the_connection_task() {
+        let connector = || {
+            KnxConnector::new(
+                TokioNet::udp(Ipv4Addr::LOCALHOST),
+                TokioDelay,
+                "knx://127.0.0.1:3671",
+            )
+        };
+        assert_eq!(ConnectorBuilder::scheme(&connector()), "knx");
 
-        let futures = connector.build(&db).await.expect("build");
-        assert_eq!(
-            futures.len(),
-            2,
-            "connection task + pump_source, and no publisher"
-        );
+        // With inbound and outbound routes, and with none.
+        for db in [routed_db().await, db().await] {
+            let futures = connector().build(&db).await.expect("build");
+            assert_eq!(futures.len(), 1, "the connection task alone");
+        }
     }
 
-    /// A second KNX connector is a configuration error, whether or not it
-    /// shares the first one's channels.
-    ///
-    /// Sharing is the louder mistake — both tasks would drain one command
-    /// queue — but separate channels are broken too: each connector collects
-    /// *all* `knx://` routes, so every `link_to` gets two publishers and
-    /// nothing says which gateway a route belongs to.
+    /// A second KNX connector is a configuration error: each connector
+    /// collects *all* `knx://` routes, so every route would be served twice
+    /// and nothing says which gateway it belongs to.
     #[tokio::test]
     async fn a_second_knx_connector_fails_the_build() {
-        static A: Channels<8> = Channels::new();
-        static B: Channels<8> = Channels::new();
-
-        for (second_channels, case) in [(&A, "shared channels"), (&B, "separate channels")] {
-            let builder = AimDbBuilder::new()
-                .runtime(Arc::new(TokioAdapter))
-                .with_connector(KnxConnector::<_, _, 8>::new(
-                    TokioNet::udp(Ipv4Addr::LOCALHOST),
-                    TokioDelay,
-                    "knx://127.0.0.1:3671",
-                    &A,
-                ))
-                .with_connector(KnxConnector::<_, _, 8>::new(
-                    TokioNet::udp(Ipv4Addr::LOCALHOST),
-                    TokioDelay,
-                    "knx://127.0.0.2:3671",
-                    second_channels,
-                ));
+        let builder = AimDbBuilder::new()
+            .runtime(Arc::new(TokioAdapter))
+            .with_connector(KnxConnector::new(
+                TokioNet::udp(Ipv4Addr::LOCALHOST),
+                TokioDelay,
+                "knx://127.0.0.1:3671",
+            ))
+            .with_connector(KnxConnector::new(
+                TokioNet::udp(Ipv4Addr::LOCALHOST),
+                TokioDelay,
+                "knx://127.0.0.2:3671",
+            ));
 
-            let Err(err) = builder.build().await else {
-                panic!("a second KNX connector must be rejected ({case})");
-            };
-            assert!(
-                format!("{err}").contains("More than one connector registered for scheme 'knx'"),
-                "unexpected error ({case}): {err}"
-            );
-        }
+        let Err(err) = builder.build().await else {
+            panic!("a second KNX connector must be rejected");
+        };
+        assert!(
+            format!("{err}").contains("More than one connector registered for scheme 'knx'"),
+            "unexpected error: {err}"
+        );
     }
 
     /// The whole wiring against a real UDP gateway: the task binds, advertises
@@ -381,18 +235,16 @@ mod tests {
     /// unspecified bind takes has its own test in `client`.
     #[tokio::test]
     async fn the_wired_connector_reaches_a_gateway() {
-        static CH: Channels<8> = Channels::new();
         let gateway = tokio::net::UdpSocket::bind("127.0.0.1:0")
             .await
             .expect("bind gateway");
         let addr = gateway.local_addr().expect("gateway addr");
 
         let db = db().await;
-        let connector = KnxConnector::<_, _, 8>::new(
+        let connector = KnxConnector::new(
             TokioNet::udp(Ipv4Addr::LOCALHOST),
             TokioDelay,
             format!("knx://{addr}"),
-            &CH,
         );
         let futures = connector.build(&db).await.expect("build");
         let driving: Vec<_> = futures.into_iter().map(tokio::spawn).collect();
diff --git a/aimdb-knx-connector/src/lib.rs b/aimdb-knx-connector/src/lib.rs
index c717b5e7..bd7f2312 100644
--- a/aimdb-knx-connector/src/lib.rs
+++ b/aimdb-knx-connector/src/lib.rs
@@ -17,8 +17,8 @@
 //! - `std`: `connector` plus core's `std`, knx-pico's std error impls, and the
 //!   back-compat DPT re-exports. Lifts `no_std`; adds no runtime.
 //! - `critical-section-std-impl`: **final binaries only** — selects
-//!   `critical-section`'s std impl, which `Channels` needs to link on a host.
-//!   An Embassy HAL already provides one.
+//!   `critical-section`'s std impl for a host binary that needs one. An
+//!   Embassy HAL already provides one.
 //! - `tracing`: Debug logging support (std)
 //! - `defmt`: Debug logging support (no_std)
 //!
@@ -48,7 +48,7 @@
 //! ```no_run
 //! use aimdb_core::buffer::BufferCfg;
 //! use aimdb_core::AimDbBuilder;
-//! use aimdb_knx_connector::{Channels, KnxConnector};
+//! use aimdb_knx_connector::KnxConnector;
 //! use aimdb_tokio_adapter::net::{TokioDelay, TokioNet};
 //! use aimdb_tokio_adapter::{TokioAdapter, TokioRecordRegistrarExt};
 //! use std::net::Ipv4Addr;
@@ -62,14 +62,12 @@
 //! # async fn demo() -> Result<(), Box<dyn std::error::Error>> {
 //! let runtime = Arc::new(TokioAdapter::new()?);
 //!
-//! static CHANNELS: Channels = Channels::new();
 //! let mut builder = AimDbBuilder::new()
 //!     .runtime(runtime)
 //!     .with_connector(KnxConnector::new(
 //!         TokioNet::udp(Ipv4Addr::UNSPECIFIED),
 //!         TokioDelay,
 //!         "knx://192.168.1.19:3671",
-//!         &CHANNELS,
 //!     ));
 //! builder.configure::<LightState>("light.state", |reg| {
 //!     reg.buffer(BufferCfg::SingleLatest)
@@ -100,14 +98,14 @@
 //! ```rust,ignore
 //! use aimdb_core::AimDbBuilder;
 //! use aimdb_embassy_adapter::EmbassyAdapter;
-//! use aimdb_knx_connector::connector::{Channels, KnxConnector};
+//! use aimdb_knx_connector::KnxConnector;
 //! use alloc::sync::Arc;
 //!
 //! let runtime = Arc::new(EmbassyAdapter::new());
 //!
 //! let db = AimDbBuilder::new()
 //!     .runtime(runtime)
-//!     .with_connector(KnxConnector::new(binder, EmbassyDelay, gateway, &CHANNELS))
+//!     .with_connector(KnxConnector::new(binder, EmbassyDelay, gateway))
 //!     .configure::<SensorData>(|reg| {
 //!         reg.buffer_sized::<16, 2>(EmbassyBufferType::SpmcRing)
 //!            .source(sensor_producer)
@@ -181,4 +179,4 @@ pub mod client;
 pub mod connector;
 
 #[cfg(feature = "connector")]
-pub use connector::{Channels, KnxConnector};
+pub use connector::KnxConnector;
diff --git a/aimdb-knx-connector/src/tunnel.rs b/aimdb-knx-connector/src/tunnel.rs
index ba04cae5..d3941546 100644
--- a/aimdb-knx-connector/src/tunnel.rs
+++ b/aimdb-knx-connector/src/tunnel.rs
@@ -43,8 +43,8 @@ const MAX_FRAME: usize = 12 + MAX_CEMI;
 /// A wire datagram to send to the gateway — stack-allocated.
 pub type Frame = heapless::Vec<u8, MAX_FRAME>;
 
-/// Outbound GroupValueWrite command, handed to the engine by the transport's
-/// command channel (fed by the `Connector::publish` side).
+/// Outbound GroupValueWrite command, handed to the engine by the connection
+/// task for each value it pulls from an outbound route.
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct GroupWrite {
     pub group_addr: GroupAddress,
@@ -79,8 +79,8 @@ pub enum Action {
     /// number of a tracked TUNNELING_REQUEST so a failed send can stop its
     /// ACK tracking (see `TunnelIo::send`); `None` for everything else.
     Send { frame: Frame, await_ack: Option<u8> },
-    /// Deliver a parsed inbound telegram toward `pump_source`
-    /// (`try_send`, drop-on-full — never stall the protocol loop).
+    /// Deliver a parsed inbound telegram into its records (synchronous;
+    /// never stalls the protocol loop).
     Telegram {
         addr: GroupAddress,
         payload: Vec<u8>,
@@ -544,8 +544,8 @@ pub(crate) trait TunnelIo {
     /// generic over this trait and its future has to be boxable as the
     /// runner's `Send` future. An impl whose socket future is `!Send` wraps it.
     fn send(&mut self, frame: &[u8]) -> impl core::future::Future<Output = bool> + Send;
-    /// Forward a parsed telegram toward `pump_source`. Non-blocking:
-    /// drop + log on a full channel rather than stalling the protocol loop.
+    /// Forward a parsed telegram into its records. Non-blocking: never
+    /// stalls the protocol loop.
     fn forward(&mut self, addr: GroupAddress, payload: Vec<u8>);
     /// An outbound telegram's ACK never arrived (log-only, see
     /// [`Action::AckTimeout`]).
diff --git a/aimdb-knx-connector/tests/shared_channel_on_std.rs b/aimdb-knx-connector/tests/shared_channel_on_std.rs
deleted file mode 100644
index c4bcbd24..00000000
--- a/aimdb-knx-connector/tests/shared_channel_on_std.rs
+++ /dev/null
@@ -1,53 +0,0 @@
-//! The MCU's channel and select types must work in a **linked** std binary, so
-//! one connection task can serve both runtimes.
-//!
-//! `CriticalSectionRawMutex` is the only `Sync` raw mutex `embassy-sync` offers
-//! — `NoopRawMutex` is `!Sync` and cannot back a shared channel at all — and
-//! using it pulls in `_critical_section_1_0_acquire`/`_release`, which nothing
-//! defines on std. Selecting an impl is the final binary's call, so the library
-//! does not make it: a test binary is a binary, and gets the std impl through
-//! this crate's `critical-section` dev-dependency. These tests fail to *link*,
-//! not to compile, if that ever comes undone.
-#![cfg(feature = "std")]
-
-use embassy_sync::blocking_mutex::raw::CriticalSectionRawMutex;
-use embassy_sync::channel::Channel;
-
-type Cmd = (u8, u16);
-
-/// A send and receive through the channel, exercising the critical section.
-#[tokio::test]
-async fn embassy_channel_carries_a_value_on_std() {
-    let channel: Channel<CriticalSectionRawMutex, Cmd, 4> = Channel::new();
-
-    channel.send((1, 0x0a0b)).await;
-    assert_eq!(channel.receive().await, (1, 0x0a0b));
-}
-
-/// The channel is `Send + Sync`, so a spawned task can enqueue while the
-/// protocol loop drains — what the unified connection task relies on.
-#[tokio::test]
-async fn embassy_channel_is_shareable_across_tasks_on_std() {
-    static CHANNEL: Channel<CriticalSectionRawMutex, Cmd, 4> = Channel::new();
-
-    let producer = tokio::spawn(async {
-        CHANNEL.send((2, 0x0c0d)).await;
-    });
-
-    assert_eq!(CHANNEL.receive().await, (2, 0x0c0d));
-    producer.await.expect("producer task");
-}
-
-/// `embassy-futures` drives the same select on std as on the MCU.
-#[tokio::test]
-async fn embassy_select_resolves_on_std() {
-    use embassy_futures::select::{select, Either};
-
-    let channel: Channel<CriticalSectionRawMutex, Cmd, 4> = Channel::new();
-    channel.send((3, 0x0e0f)).await;
-
-    match select(channel.receive(), core::future::pending::<()>()).await {
-        Either::First(cmd) => assert_eq!(cmd, (3, 0x0e0f)),
-        Either::Second(()) => panic!("pending future must never win"),
-    }
-}
diff --git a/aimdb-knx-connector/tests/topic_provider_tests.rs b/aimdb-knx-connector/tests/topic_provider_tests.rs
deleted file mode 100644
index 2441dfa7..00000000
--- a/aimdb-knx-connector/tests/topic_provider_tests.rs
+++ /dev/null
@@ -1,590 +0,0 @@
-//! Integration tests for KNX TopicProvider and TopicResolver functionality
-//!
-//! These tests verify the dynamic group address routing features:
-//! - **TopicProvider**: Outbound (AimDB → KNX) dynamic group address selection per-value
-//! - **TopicResolver**: Inbound (KNX → AimDB) late-binding group address resolution at startup
-//!
-//! The tests use mock data and don't require a running KNX/IP gateway.
-
-#![cfg(feature = "std")]
-
-use aimdb_core::buffer::BufferCfg;
-use aimdb_core::connector::TopicProvider;
-use aimdb_core::{AimDbBuilder, Producer, RuntimeContext};
-use aimdb_knx_connector::Channels;
-use aimdb_tokio_adapter::net::{TokioDelay, TokioNet};
-use aimdb_tokio_adapter::{TokioAdapter, TokioRecordRegistrarExt};
-use std::net::Ipv4Addr;
-use std::sync::atomic::{AtomicU32, Ordering};
-use std::sync::Arc;
-
-/// The connector's channel pair, shared by the registration tests — none of
-/// them runs a connection task, so one pair serves all three.
-static CHANNELS: Channels = Channels::new();
-
-// ============================================================================
-// Test Types
-// ============================================================================
-
-/// Dimmer value with room identifier for dynamic group address routing
-#[derive(Clone, Debug)]
-struct DimmerValue {
-    room_id: String,
-    level: u8, // 0-255
-}
-
-impl DimmerValue {
-    fn new(room_id: &str, level: u8) -> Self {
-        Self {
-            room_id: room_id.into(),
-            level,
-        }
-    }
-
-    fn to_knx_bytes(&self) -> Vec<u8> {
-        vec![self.level]
-    }
-}
-
-/// Switch state with zone identifier
-#[derive(Clone, Debug)]
-struct SwitchState {
-    zone_id: String,
-    is_on: bool,
-}
-
-impl SwitchState {
-    fn new(zone_id: &str, is_on: bool) -> Self {
-        Self {
-            zone_id: zone_id.into(),
-            is_on,
-        }
-    }
-
-    #[allow(dead_code)]
-    fn to_knx_bytes(&self) -> Vec<u8> {
-        vec![if self.is_on { 1 } else { 0 }]
-    }
-
-    fn from_knx_bytes(data: &[u8]) -> Result<Self, String> {
-        if data.is_empty() {
-            return Err("Empty data".into());
-        }
-        Ok(SwitchState {
-            zone_id: "unknown".into(),
-            is_on: data[0] != 0,
-        })
-    }
-}
-
-/// Temperature setpoint for HVAC control
-#[derive(Clone, Debug)]
-struct TemperatureSetpoint {
-    hvac_zone: u8, // Zone number 1-16
-    setpoint: f32, // Temperature in Celsius
-}
-
-impl TemperatureSetpoint {
-    fn new(hvac_zone: u8, setpoint: f32) -> Self {
-        Self {
-            hvac_zone,
-            setpoint,
-        }
-    }
-
-    fn to_knx_bytes(&self) -> Vec<u8> {
-        // KNX DPT 9.001 encoding (simplified for test)
-        let value = (self.setpoint * 100.0) as i16;
-        vec![(value >> 8) as u8, value as u8]
-    }
-}
-
-// ============================================================================
-// TopicProvider Implementations (KNX Group Addresses)
-// ============================================================================
-
-/// Dynamic group address provider based on room ID
-///
-/// Routes dimmer commands to different KNX group addresses based on room.
-/// In a real building, each room has its own dimmer actuator address.
-struct RoomBasedGroupAddressProvider {
-    /// Base group address (main/middle)
-    base_main: u8,
-    base_middle: u8,
-}
-
-impl RoomBasedGroupAddressProvider {
-    fn new(base_main: u8, base_middle: u8) -> Self {
-        Self {
-            base_main,
-            base_middle,
-        }
-    }
-
-    fn room_to_sub(&self, room_id: &str) -> u8 {
-        // Map room names to sub-addresses
-        match room_id {
-            "living" => 1,
-            "bedroom" => 2,
-            "kitchen" => 3,
-            "bathroom" => 4,
-            "office" => 5,
-            _ => 0, // Default/unknown
-        }
-    }
-}
-
-impl TopicProvider<DimmerValue> for RoomBasedGroupAddressProvider {
-    fn topic(&self, value: &DimmerValue) -> Option<String> {
-        let sub = self.room_to_sub(&value.room_id);
-        Some(format!(
-            "knx://{}/{}/{}",
-            self.base_main, self.base_middle, sub
-        ))
-    }
-}
-
-/// HVAC zone-based group address provider
-///
-/// Routes temperature setpoints to zone-specific group addresses.
-struct HvacZoneProvider;
-
-impl TopicProvider<TemperatureSetpoint> for HvacZoneProvider {
-    fn topic(&self, value: &TemperatureSetpoint) -> Option<String> {
-        // HVAC zones mapped to group addresses 5/0/1 through 5/0/16
-        if value.hvac_zone >= 1 && value.hvac_zone <= 16 {
-            Some(format!("knx://5/0/{}", value.hvac_zone))
-        } else {
-            None // Invalid zone, use fallback
-        }
-    }
-}
-
-/// Switch provider with emergency override
-///
-/// Demonstrates conditional routing: emergency signals go to broadcast address.
-struct SwitchWithEmergencyProvider;
-
-impl TopicProvider<SwitchState> for SwitchWithEmergencyProvider {
-    fn topic(&self, value: &SwitchState) -> Option<String> {
-        if value.zone_id == "emergency" {
-            // Emergency signals go to broadcast group
-            Some("knx://0/0/0".into())
-        } else if value.zone_id.starts_with("zone-") {
-            // Parse zone number
-            let zone_num: u8 = value.zone_id[5..].parse().unwrap_or(0);
-            Some(format!("knx://1/1/{}", zone_num))
-        } else {
-            None // Use default from link_to()
-        }
-    }
-}
-
-// ============================================================================
-// Unit Tests for KNX TopicProvider
-// ============================================================================
-
-#[test]
-fn test_room_based_group_address_provider() {
-    let provider = RoomBasedGroupAddressProvider::new(1, 0);
-
-    let living = DimmerValue::new("living", 128);
-    assert_eq!(provider.topic(&living), Some("knx://1/0/1".into()));
-
-    let bedroom = DimmerValue::new("bedroom", 255);
-    assert_eq!(provider.topic(&bedroom), Some("knx://1/0/2".into()));
-
-    let kitchen = DimmerValue::new("kitchen", 64);
-    assert_eq!(provider.topic(&kitchen), Some("knx://1/0/3".into()));
-
-    // Unknown room uses sub-address 0
-    let unknown = DimmerValue::new("garage", 100);
-    assert_eq!(provider.topic(&unknown), Some("knx://1/0/0".into()));
-}
-
-#[test]
-fn test_hvac_zone_provider() {
-    let provider = HvacZoneProvider;
-
-    let zone1 = TemperatureSetpoint::new(1, 21.0);
-    assert_eq!(provider.topic(&zone1), Some("knx://5/0/1".into()));
-
-    let zone16 = TemperatureSetpoint::new(16, 22.5);
-    assert_eq!(provider.topic(&zone16), Some("knx://5/0/16".into()));
-
-    // Invalid zone returns None (fallback)
-    let invalid = TemperatureSetpoint::new(0, 20.0);
-    assert_eq!(provider.topic(&invalid), None);
-
-    let too_high = TemperatureSetpoint::new(17, 20.0);
-    assert_eq!(provider.topic(&too_high), None);
-}
-
-#[test]
-fn test_switch_with_emergency_provider() {
-    let provider = SwitchWithEmergencyProvider;
-
-    // Emergency goes to broadcast
-    let emergency = SwitchState::new("emergency", true);
-    assert_eq!(provider.topic(&emergency), Some("knx://0/0/0".into()));
-
-    // Normal zones
-    let zone1 = SwitchState::new("zone-1", true);
-    assert_eq!(provider.topic(&zone1), Some("knx://1/1/1".into()));
-
-    let zone5 = SwitchState::new("zone-5", false);
-    assert_eq!(provider.topic(&zone5), Some("knx://1/1/5".into()));
-
-    // Unknown zone uses fallback
-    let unknown = SwitchState::new("lobby", true);
-    assert_eq!(provider.topic(&unknown), None);
-}
-
-// ============================================================================
-// Unit Tests for KNX TopicResolver
-// ============================================================================
-
-#[test]
-fn test_topic_resolver_from_config_file() {
-    // Simulate reading group address from a config file or service
-    std::env::set_var("KNX_SWITCH_GROUP", "1/2/3");
-
-    let resolver = || {
-        std::env::var("KNX_SWITCH_GROUP")
-            .ok()
-            .map(|addr| format!("knx://{}", addr))
-    };
-
-    assert_eq!(resolver(), Some("knx://1/2/3".into()));
-
-    std::env::remove_var("KNX_SWITCH_GROUP");
-}
-
-#[test]
-fn test_topic_resolver_with_discovery() {
-    use std::sync::Mutex;
-
-    // Simulate KNX device discovery
-    let discovered_devices = Arc::new(Mutex::new(vec![
-        ("dimmer-1", "1/0/1"),
-        ("dimmer-2", "1/0/2"),
-    ]));
-
-    let devices_clone = discovered_devices.clone();
-    let resolver = move || {
-        devices_clone
-            .lock()
-            .unwrap()
-            .iter()
-            .find(|(name, _)| *name == "dimmer-1")
-            .map(|(_, addr)| format!("knx://{}", addr))
-    };
-
-    assert_eq!(resolver(), Some("knx://1/0/1".into()));
-}
-
-// ============================================================================
-// Integration Test: TopicProvider with AimDbBuilder
-// ============================================================================
-
-/// Test that TopicProvider can be configured without connector (verifies API)
-#[tokio::test]
-async fn test_knx_topic_provider_registration_api() {
-    let runtime = Arc::new(TokioAdapter::new().unwrap());
-    let produced_count = Arc::new(AtomicU32::new(0));
-    let produced_count_clone = produced_count.clone();
-
-    let mut builder = AimDbBuilder::new().runtime(runtime);
-
-    // Register without link_to (no connector needed)
-    builder.configure::<DimmerValue>("knx.dimmer.living", |reg| {
-        let counter = produced_count_clone.clone();
-        reg.buffer(BufferCfg::SingleLatest).source(
-            move |_ctx: RuntimeContext, producer: Producer<DimmerValue>| {
-                let counter = counter.clone();
-                async move {
-                    let dimmer = DimmerValue::new("living", 200);
-                    producer.produce(dimmer);
-                    counter.fetch_add(1, Ordering::SeqCst);
-                }
-            },
-        );
-    });
-
-    // Build should succeed
-    assert!(builder.build().await.is_ok());
-}
-
-/// Test TopicProvider with KnxConnector registration
-///
-/// This test verifies the full configuration API including link_to + with_topic_provider
-/// works correctly at compile time. Runtime requires actual KNX gateway.
-#[tokio::test]
-async fn test_knx_topic_provider_with_connector_registration() {
-    let runtime = Arc::new(TokioAdapter::new().unwrap());
-
-    let mut builder = AimDbBuilder::new().runtime(runtime).with_connector(
-        aimdb_knx_connector::KnxConnector::new(
-            TokioNet::udp(Ipv4Addr::UNSPECIFIED),
-            TokioDelay,
-            "knx://192.168.1.10:3671",
-            &CHANNELS,
-        ),
-    );
-
-    // Register dimmer with dynamic group address provider
-    builder.configure::<DimmerValue>("knx.dimmer.living", |reg| {
-        reg.buffer(BufferCfg::SingleLatest)
-            .source(
-                |_ctx: RuntimeContext, producer: Producer<DimmerValue>| async move {
-                    let dimmer = DimmerValue::new("living", 200);
-                    producer.produce(dimmer);
-                },
-            )
-            .link_to("knx://1/0/0") // Fallback group address
-            .with_topic_provider(RoomBasedGroupAddressProvider::new(1, 0))
-            .with_serializer(|_ctx, dimmer: &DimmerValue| Ok(dimmer.to_knx_bytes()))
-            .finish();
-    });
-
-    // Build succeeds with connector registered
-    assert!(builder.build().await.is_ok());
-}
-
-#[tokio::test]
-async fn test_knx_topic_resolver_with_connector_registration() {
-    let runtime = Arc::new(TokioAdapter::new().unwrap());
-
-    std::env::set_var("KNX_SWITCH_INPUT", "1/2/10");
-
-    let mut builder = AimDbBuilder::new().runtime(runtime).with_connector(
-        aimdb_knx_connector::KnxConnector::new(
-            TokioNet::udp(Ipv4Addr::UNSPECIFIED),
-            TokioDelay,
-            "knx://192.168.1.10:3671",
-            &CHANNELS,
-        ),
-    );
-
-    // Register switch with dynamic group address resolver
-    builder.configure::<SwitchState>("knx.switch.zone1", |reg| {
-        reg.buffer(BufferCfg::SingleLatest)
-            .link_from("knx://1/2/0") // Fallback group address
-            .with_topic_resolver(|| {
-                std::env::var("KNX_SWITCH_INPUT")
-                    .ok()
-                    .map(|addr| format!("knx://{}", addr))
-            })
-            .with_deserializer(|_ctx, data: &[u8]| SwitchState::from_knx_bytes(data))
-            .finish();
-    });
-
-    // Build should succeed
-    assert!(builder.build().await.is_ok());
-
-    std::env::remove_var("KNX_SWITCH_INPUT");
-}
-
-#[tokio::test]
-async fn test_hvac_zone_routing() {
-    let runtime = Arc::new(TokioAdapter::new().unwrap());
-
-    let mut builder = AimDbBuilder::new().runtime(runtime).with_connector(
-        aimdb_knx_connector::KnxConnector::new(
-            TokioNet::udp(Ipv4Addr::UNSPECIFIED),
-            TokioDelay,
-            "knx://192.168.1.10:3671",
-            &CHANNELS,
-        ),
-    );
-
-    // HVAC setpoint with zone-based routing
-    builder.configure::<TemperatureSetpoint>("knx.hvac.setpoint", |reg| {
-        reg.buffer(BufferCfg::SingleLatest)
-            .source(
-                |_ctx: RuntimeContext, producer: Producer<TemperatureSetpoint>| async move {
-                    // Different zones get routed to different group addresses
-                    for zone in 1..=4 {
-                        let setpoint = TemperatureSetpoint::new(zone, 21.0 + zone as f32 * 0.5);
-                        producer.produce(setpoint);
-                    }
-                },
-            )
-            .link_to("knx://5/0/0") // Fallback for invalid zones
-            .with_topic_provider(HvacZoneProvider)
-            .with_serializer(|_ctx, sp: &TemperatureSetpoint| Ok(sp.to_knx_bytes()))
-            .finish();
-    });
-
-    assert!(builder.build().await.is_ok());
-}
-
-// ============================================================================
-// Test TopicProvider as a typed trait object with KNX Types
-// ============================================================================
-
-#[test]
-fn test_knx_topic_provider_as_trait_object() {
-    use aimdb_core::connector::TopicProvider;
-    use std::sync::Arc;
-
-    // Providers are stored as Arc<dyn TopicProvider<T>> and stay typed
-    // end-to-end — a wrong-type call is unrepresentable.
-    let provider: Arc<dyn TopicProvider<DimmerValue>> =
-        Arc::new(RoomBasedGroupAddressProvider::new(1, 0));
-
-    let dimmer = DimmerValue::new("living", 128);
-    assert_eq!(provider.topic(&dimmer), Some("knx://1/0/1".into()));
-}
-
-// ============================================================================
-// Test: Simulate Connector Group Address Resolution Logic
-// ============================================================================
-//
-// These tests simulate EXACTLY what the fused outbound reader does internally
-// while it still holds the typed value:
-// ```rust
-// let dest = topic.as_ref().and_then(|p| p.topic(&value));
-// // ...later, in the pump:
-// let dest = msg.dest.unwrap_or_else(|| default_group_addr.clone());
-// ```
-
-/// Simulates the fused reader's group address resolution for outbound telegrams
-fn resolve_group_address_like_connector<T>(
-    default_group_addr: &str,
-    topic_provider: Option<&dyn aimdb_core::connector::TopicProvider<T>>,
-    value: &T,
-) -> String {
-    topic_provider
-        .and_then(|provider| provider.topic(value))
-        .unwrap_or_else(|| default_group_addr.to_string())
-}
-
-#[test]
-fn test_connector_group_address_resolution_room_based() {
-    let provider = RoomBasedGroupAddressProvider::new(1, 0);
-    let default_addr = "knx://1/0/0";
-
-    // Test 1: Living room dimmer
-    let dimmer_living = DimmerValue::new("living", 128);
-    let resolved =
-        resolve_group_address_like_connector(default_addr, Some(&provider), &dimmer_living);
-    assert_eq!(resolved, "knx://1/0/1");
-
-    // Test 2: Bedroom dimmer
-    let dimmer_bedroom = DimmerValue::new("bedroom", 255);
-    let resolved =
-        resolve_group_address_like_connector(default_addr, Some(&provider), &dimmer_bedroom);
-    assert_eq!(resolved, "knx://1/0/2");
-
-    // Test 3: Kitchen dimmer
-    let dimmer_kitchen = DimmerValue::new("kitchen", 64);
-    let resolved =
-        resolve_group_address_like_connector(default_addr, Some(&provider), &dimmer_kitchen);
-    assert_eq!(resolved, "knx://1/0/3");
-
-    // Test 4: Unknown room → sub-address 0
-    let dimmer_unknown = DimmerValue::new("garage", 100);
-    let resolved =
-        resolve_group_address_like_connector(default_addr, Some(&provider), &dimmer_unknown);
-    assert_eq!(resolved, "knx://1/0/0");
-}
-
-#[test]
-fn test_connector_group_address_resolution_hvac_zones() {
-    let provider = HvacZoneProvider;
-    let default_addr = "knx://5/0/0"; // Fallback for invalid zones
-
-    // Test valid zones 1-16
-    for zone in 1..=16u8 {
-        let setpoint = TemperatureSetpoint::new(zone, 21.0);
-        let resolved =
-            resolve_group_address_like_connector(default_addr, Some(&provider), &setpoint);
-        assert_eq!(resolved, format!("knx://5/0/{}", zone));
-    }
-
-    // Test invalid zone 0 → fallback
-    let invalid_zone_0 = TemperatureSetpoint::new(0, 21.0);
-    let resolved =
-        resolve_group_address_like_connector(default_addr, Some(&provider), &invalid_zone_0);
-    assert_eq!(resolved, "knx://5/0/0"); // Fallback
-
-    // Test invalid zone 17 → fallback
-    let invalid_zone_17 = TemperatureSetpoint::new(17, 21.0);
-    let resolved =
-        resolve_group_address_like_connector(default_addr, Some(&provider), &invalid_zone_17);
-    assert_eq!(resolved, "knx://5/0/0"); // Fallback
-}
-
-#[test]
-fn test_connector_group_address_resolution_emergency_switch() {
-    let provider = SwitchWithEmergencyProvider;
-    let default_addr = "knx://1/1/0";
-
-    // Test 1: Emergency switch → broadcast address
-    let emergency = SwitchState::new("emergency", true);
-    let resolved = resolve_group_address_like_connector(default_addr, Some(&provider), &emergency);
-    assert_eq!(resolved, "knx://0/0/0"); // Broadcast!
-
-    // Test 2: Normal zone switches
-    let zone_1 = SwitchState::new("zone-1", true);
-    let resolved = resolve_group_address_like_connector(default_addr, Some(&provider), &zone_1);
-    assert_eq!(resolved, "knx://1/1/1");
-
-    let zone_5 = SwitchState::new("zone-5", false);
-    let resolved = resolve_group_address_like_connector(default_addr, Some(&provider), &zone_5);
-    assert_eq!(resolved, "knx://1/1/5");
-
-    // Test 3: Unknown zone → fallback
-    let lobby = SwitchState::new("lobby", true);
-    let resolved = resolve_group_address_like_connector(default_addr, Some(&provider), &lobby);
-    assert_eq!(resolved, "knx://1/1/0"); // Fallback
-}
-
-#[test]
-fn test_connector_group_address_no_provider() {
-    let default_addr = "knx://1/0/99";
-
-    // No provider → always use default
-    let dimmer = DimmerValue::new("living", 128);
-    let resolved = resolve_group_address_like_connector(default_addr, None, &dimmer);
-    assert_eq!(resolved, "knx://1/0/99");
-}
-
-/// Test inbound group address resolution (TopicResolver)
-#[test]
-fn test_inbound_group_address_resolver_simulation() {
-    fn resolve_inbound_group_addr(
-        default_addr: &str,
-        resolver: Option<&dyn Fn() -> Option<String>>,
-    ) -> String {
-        resolver
-            .and_then(|r| r())
-            .unwrap_or_else(|| default_addr.to_string())
-    }
-
-    // Test 1: Resolver returns dynamic address
-    let resolver_dynamic = || Some("knx://1/2/10".to_string());
-    let addr = resolve_inbound_group_addr("knx://1/2/0", Some(&resolver_dynamic));
-    assert_eq!(addr, "knx://1/2/10");
-
-    // Test 2: Resolver returns None → fallback
-    let resolver_fallback = || None;
-    let addr = resolve_inbound_group_addr("knx://1/2/0", Some(&resolver_fallback));
-    assert_eq!(addr, "knx://1/2/0");
-
-    // Test 3: No resolver → default
-    let addr = resolve_inbound_group_addr("knx://1/2/0", None);
-    assert_eq!(addr, "knx://1/2/0");
-
-    // Test 4: Resolver from environment (simulating config file)
-    std::env::set_var("KNX_INBOUND_ADDR", "1/3/5");
-    let resolver_env = || {
-        std::env::var("KNX_INBOUND_ADDR")
-            .ok()
-            .map(|a| format!("knx://{}", a))
-    };
-    let addr = resolve_inbound_group_addr("knx://1/0/0", Some(&resolver_env));
-    assert_eq!(addr, "knx://1/3/5");
-    std::env::remove_var("KNX_INBOUND_ADDR");
-}
diff --git a/aimdb-knx-connector/tests/topic_writer_tests.rs b/aimdb-knx-connector/tests/topic_writer_tests.rs
new file mode 100644
index 00000000..42055a0b
--- /dev/null
+++ b/aimdb-knx-connector/tests/topic_writer_tests.rs
@@ -0,0 +1,405 @@
+//! Integration tests for KNX TopicWriter and TopicResolver functionality
+//!
+//! These tests verify the dynamic group address routing features:
+//! - **TopicWriter**: Outbound (AimDB → KNX) dynamic group address selection per-value
+//! - **TopicResolver**: Inbound (KNX → AimDB) late-binding group address resolution at startup
+//!
+//! A writer writes the bare group address (`1/0/1`), not the `knx://` URL. The
+//! longest group address, `31/7/255`, is 8 bytes.
+//!
+//! The tests use mock data and don't require a running KNX/IP gateway.
+
+#![cfg(feature = "std")]
+
+use aimdb_core::buffer::BufferCfg;
+use aimdb_core::connector::{TopicBuf, TopicOverflow, TopicWriter};
+use aimdb_core::{AimDbBuilder, Producer, RuntimeContext};
+use aimdb_knx_connector::KnxConnector;
+use aimdb_tokio_adapter::net::{TokioDelay, TokioNet, TokioUdpBinder};
+use aimdb_tokio_adapter::{TokioAdapter, TokioRecordRegistrarExt};
+use std::fmt::Write as _;
+use std::net::Ipv4Addr;
+use std::sync::Arc;
+
+/// Topic capacity for every KNX writer here: the longest group address.
+const GROUP_ADDRESS_CAPACITY: usize = 8;
+
+// ============================================================================
+// Test Types
+// ============================================================================
+
+/// Dimmer value with room identifier for dynamic group address routing
+#[derive(Clone, Debug)]
+struct DimmerValue {
+    room_id: String,
+    level: u8, // 0-255
+}
+
+impl DimmerValue {
+    fn new(room_id: &str, level: u8) -> Self {
+        Self {
+            room_id: room_id.into(),
+            level,
+        }
+    }
+
+    fn to_knx_bytes(&self) -> Vec<u8> {
+        vec![self.level]
+    }
+}
+
+/// Switch state with zone identifier
+#[derive(Clone, Debug)]
+struct SwitchState {
+    zone_id: String,
+    #[allow(dead_code)]
+    is_on: bool,
+}
+
+impl SwitchState {
+    fn new(zone_id: &str, is_on: bool) -> Self {
+        Self {
+            zone_id: zone_id.into(),
+            is_on,
+        }
+    }
+
+    fn from_knx_bytes(data: &[u8]) -> Result<Self, String> {
+        if data.is_empty() {
+            return Err("Empty data".into());
+        }
+        Ok(SwitchState {
+            zone_id: "unknown".into(),
+            is_on: data[0] != 0,
+        })
+    }
+}
+
+/// Temperature setpoint for HVAC control
+#[derive(Clone, Debug)]
+struct TemperatureSetpoint {
+    hvac_zone: u8, // Zone number 1-16
+    setpoint: f32, // Temperature in Celsius
+}
+
+impl TemperatureSetpoint {
+    fn new(hvac_zone: u8, setpoint: f32) -> Self {
+        Self {
+            hvac_zone,
+            setpoint,
+        }
+    }
+
+    fn to_knx_bytes(&self) -> Vec<u8> {
+        // KNX DPT 9.001 encoding (simplified for test)
+        let value = (self.setpoint * 100.0) as i16;
+        vec![(value >> 8) as u8, value as u8]
+    }
+}
+
+// ============================================================================
+// TopicWriter Implementations (KNX Group Addresses)
+// ============================================================================
+
+/// Dynamic group address writer based on room ID
+///
+/// Routes dimmer commands to different KNX group addresses based on room.
+/// In a real building, each room has its own dimmer actuator address.
+struct RoomBasedGroupAddress {
+    /// Base group address (main/middle)
+    base_main: u8,
+    base_middle: u8,
+}
+
+impl RoomBasedGroupAddress {
+    fn new(base_main: u8, base_middle: u8) -> Self {
+        Self {
+            base_main,
+            base_middle,
+        }
+    }
+
+    fn room_to_sub(&self, room_id: &str) -> u8 {
+        // Map room names to sub-addresses
+        match room_id {
+            "living" => 1,
+            "bedroom" => 2,
+            "kitchen" => 3,
+            "bathroom" => 4,
+            "office" => 5,
+            _ => 0, // Default/unknown
+        }
+    }
+}
+
+impl TopicWriter<DimmerValue> for RoomBasedGroupAddress {
+    fn write_topic(
+        &self,
+        value: &DimmerValue,
+        out: &mut TopicBuf<'_>,
+    ) -> Result<bool, TopicOverflow> {
+        let sub = self.room_to_sub(&value.room_id);
+        write!(out, "{}/{}/{}", self.base_main, self.base_middle, sub)?;
+        Ok(true)
+    }
+}
+
+/// HVAC zone-based group address writer
+///
+/// Routes temperature setpoints to zone-specific group addresses.
+struct HvacZone;
+
+impl TopicWriter<TemperatureSetpoint> for HvacZone {
+    fn write_topic(
+        &self,
+        value: &TemperatureSetpoint,
+        out: &mut TopicBuf<'_>,
+    ) -> Result<bool, TopicOverflow> {
+        // HVAC zones mapped to group addresses 5/0/1 through 5/0/16
+        if !(1..=16).contains(&value.hvac_zone) {
+            return Ok(false); // Invalid zone, use fallback
+        }
+        write!(out, "5/0/{}", value.hvac_zone)?;
+        Ok(true)
+    }
+}
+
+/// Switch writer with emergency override
+///
+/// Demonstrates conditional routing: emergency signals go to broadcast address.
+struct SwitchWithEmergency;
+
+impl TopicWriter<SwitchState> for SwitchWithEmergency {
+    fn write_topic(
+        &self,
+        value: &SwitchState,
+        out: &mut TopicBuf<'_>,
+    ) -> Result<bool, TopicOverflow> {
+        if value.zone_id == "emergency" {
+            // Emergency signals go to broadcast group
+            out.push_str("0/0/0")?;
+        } else if let Some(zone) = value.zone_id.strip_prefix("zone-") {
+            let zone_num: u8 = zone.parse().unwrap_or(0);
+            write!(out, "1/1/{zone_num}")?;
+        } else {
+            return Ok(false); // Use default from link_to()
+        }
+        Ok(true)
+    }
+}
+
+/// The group address `writer` selects for `value`, or `default` when it
+/// returns `Ok(false)` — what the outbound route resolves.
+fn resolve<T>(writer: &dyn TopicWriter<T>, value: &T, default: &str) -> String {
+    let mut storage = [0u8; GROUP_ADDRESS_CAPACITY];
+    let mut out = TopicBuf::new(&mut storage);
+    match writer.write_topic(value, &mut out) {
+        Ok(true) => out.as_str().to_string(),
+        Ok(false) => default.to_string(),
+        Err(TopicOverflow) => panic!("group address exceeds {GROUP_ADDRESS_CAPACITY} bytes"),
+    }
+}
+
+// ============================================================================
+// Unit Tests for KNX TopicWriter
+// ============================================================================
+
+#[test]
+fn test_room_based_group_address() {
+    let writer = RoomBasedGroupAddress::new(1, 0);
+    let default = "1/0/0";
+
+    for (room, expected) in [
+        ("living", "1/0/1"),
+        ("bedroom", "1/0/2"),
+        ("kitchen", "1/0/3"),
+        // Unknown room uses sub-address 0
+        ("garage", "1/0/0"),
+    ] {
+        let dimmer = DimmerValue::new(room, 128);
+        assert_eq!(resolve(&writer, &dimmer, default), expected, "{room}");
+    }
+}
+
+#[test]
+fn test_hvac_zone_routing_and_fallback() {
+    let default = "5/0/0"; // Fallback for invalid zones
+
+    for zone in 1..=16u8 {
+        let setpoint = TemperatureSetpoint::new(zone, 21.0);
+        assert_eq!(
+            resolve(&HvacZone, &setpoint, default),
+            format!("5/0/{zone}")
+        );
+    }
+
+    // Invalid zones fall back to the link's default.
+    for zone in [0, 17] {
+        let setpoint = TemperatureSetpoint::new(zone, 20.0);
+        assert_eq!(resolve(&HvacZone, &setpoint, default), default);
+    }
+}
+
+#[test]
+fn test_switch_with_emergency() {
+    let default = "1/1/0";
+
+    // Emergency goes to broadcast
+    let emergency = SwitchState::new("emergency", true);
+    assert_eq!(resolve(&SwitchWithEmergency, &emergency, default), "0/0/0");
+
+    // Normal zones
+    let zone1 = SwitchState::new("zone-1", true);
+    assert_eq!(resolve(&SwitchWithEmergency, &zone1, default), "1/1/1");
+
+    let zone5 = SwitchState::new("zone-5", false);
+    assert_eq!(resolve(&SwitchWithEmergency, &zone5, default), "1/1/5");
+
+    // Unknown zone uses fallback
+    let lobby = SwitchState::new("lobby", true);
+    assert_eq!(resolve(&SwitchWithEmergency, &lobby, default), default);
+}
+
+/// The longest group address fits the capacity; one byte less overflows.
+#[test]
+fn test_longest_group_address_fits_the_capacity() {
+    let writer = RoomBasedGroupAddress::new(31, 7);
+    let dimmer = DimmerValue::new("living", 1);
+
+    let mut storage = [0u8; GROUP_ADDRESS_CAPACITY];
+    let mut out = TopicBuf::new(&mut storage);
+    assert_eq!(writer.write_topic(&dimmer, &mut out), Ok(true));
+    assert_eq!(out.as_str(), "31/7/1");
+
+    let writer = |out: &mut TopicBuf<'_>| write!(out, "31/7/255");
+    let mut storage = [0u8; GROUP_ADDRESS_CAPACITY];
+    assert!(writer(&mut TopicBuf::new(&mut storage)).is_ok());
+    let mut storage = [0u8; GROUP_ADDRESS_CAPACITY - 1];
+    assert!(writer(&mut TopicBuf::new(&mut storage)).is_err());
+}
+
+// ============================================================================
+// Unit Tests for KNX TopicResolver
+// ============================================================================
+
+#[test]
+fn test_topic_resolver_from_config_file() {
+    // Simulate reading group address from a config file or service
+    std::env::set_var("KNX_SWITCH_GROUP", "1/2/3");
+
+    let resolver = || {
+        std::env::var("KNX_SWITCH_GROUP")
+            .ok()
+            .map(|addr| format!("knx://{}", addr))
+    };
+
+    assert_eq!(resolver(), Some("knx://1/2/3".into()));
+
+    std::env::remove_var("KNX_SWITCH_GROUP");
+}
+
+// ============================================================================
+// Integration Tests: registration with KnxConnector
+// ============================================================================
+
+fn connector() -> KnxConnector<TokioUdpBinder, TokioDelay> {
+    KnxConnector::new(
+        TokioNet::udp(Ipv4Addr::UNSPECIFIED),
+        TokioDelay,
+        "knx://192.168.1.10:3671",
+    )
+}
+
+/// `link_to` + `with_topic_writer` builds with the connector registered.
+/// Running it would need a KNX gateway.
+#[tokio::test]
+async fn test_knx_topic_writer_with_connector_registration() {
+    let runtime = Arc::new(TokioAdapter::new().unwrap());
+    let mut builder = AimDbBuilder::new()
+        .runtime(runtime)
+        .with_connector(connector());
+
+    builder.configure::<DimmerValue>("knx.dimmer.living", |reg| {
+        reg.buffer(BufferCfg::SingleLatest)
+            .source(
+                |_ctx: RuntimeContext, producer: Producer<DimmerValue>| async move {
+                    producer.produce(DimmerValue::new("living", 200));
+                },
+            )
+            .link_to("knx://1/0/0") // Fallback group address
+            .with_topic_writer(GROUP_ADDRESS_CAPACITY, RoomBasedGroupAddress::new(1, 0))
+            .with_serializer(|_ctx, dimmer: &DimmerValue| Ok(dimmer.to_knx_bytes()))
+            .finish();
+    });
+    builder.configure::<TemperatureSetpoint>("knx.hvac.setpoint", |reg| {
+        reg.buffer(BufferCfg::SingleLatest)
+            .link_to("knx://5/0/0") // Fallback for invalid zones
+            .with_topic_writer(GROUP_ADDRESS_CAPACITY, HvacZone)
+            .with_serializer(|_ctx, sp: &TemperatureSetpoint| Ok(sp.to_knx_bytes()))
+            .finish();
+    });
+
+    assert!(builder.build().await.is_ok());
+}
+
+/// The connector pulls through `OutboundRoutes`, which rejects
+/// `with_topic_provider` links at build.
+#[tokio::test]
+async fn test_knx_rejects_a_topic_provider() {
+    use aimdb_core::connector::TopicProvider;
+
+    struct Fixed;
+    impl TopicProvider<DimmerValue> for Fixed {
+        fn topic(&self, _value: &DimmerValue) -> Option<String> {
+            Some("1/0/1".into())
+        }
+    }
+
+    let runtime = Arc::new(TokioAdapter::new().unwrap());
+    let mut builder = AimDbBuilder::new()
+        .runtime(runtime)
+        .with_connector(connector());
+    builder.configure::<DimmerValue>("knx.dimmer.living", |reg| {
+        reg.buffer(BufferCfg::SingleLatest)
+            .link_to("knx://1/0/0")
+            .with_topic_provider(Fixed)
+            .with_serializer(|_ctx, dimmer: &DimmerValue| Ok(dimmer.to_knx_bytes()))
+            .finish();
+    });
+
+    let Err(err) = builder.build().await else {
+        panic!("a topic provider must be rejected");
+    };
+    assert!(
+        format!("{err}").contains("with_topic_provider"),
+        "unexpected error: {err}"
+    );
+}
+
+#[tokio::test]
+async fn test_knx_topic_resolver_with_connector_registration() {
+    let runtime = Arc::new(TokioAdapter::new().unwrap());
+
+    std::env::set_var("KNX_SWITCH_INPUT", "1/2/10");
+
+    let mut builder = AimDbBuilder::new()
+        .runtime(runtime)
+        .with_connector(connector());
+
+    // Register switch with dynamic group address resolver
+    builder.configure::<SwitchState>("knx.switch.zone1", |reg| {
+        reg.buffer(BufferCfg::SingleLatest)
+            .link_from("knx://1/2/0") // Fallback group address
+            .with_topic_resolver(|| {
+                std::env::var("KNX_SWITCH_INPUT")
+                    .ok()
+                    .map(|addr| format!("knx://{}", addr))
+            })
+            .with_deserializer(|_ctx, data: &[u8]| SwitchState::from_knx_bytes(data))
+            .finish();
+    });
+
+    assert!(builder.build().await.is_ok());
+
+    std::env::remove_var("KNX_SWITCH_INPUT");
+}
diff --git a/docs/aimdb-usage-guide.md b/docs/aimdb-usage-guide.md
index f5dee285..9cdfdb0e 100644
--- a/docs/aimdb-usage-guide.md
+++ b/docs/aimdb-usage-guide.md
@@ -315,7 +315,7 @@ knx-pico = { git = "https://github.com/aimdb-dev/knx-pico.git", branch = "master
 use aimdb_core::prelude::*;
 use aimdb_tokio_adapter::TokioAdapter;
 use aimdb_tokio_adapter::net::{TokioDelay, TokioNet};
-use aimdb_knx_connector::{Channels, KnxConnector};
+use aimdb_knx_connector::KnxConnector;
 use std::net::Ipv4Addr;
 use std::sync::Arc;
 
@@ -324,9 +324,6 @@ struct LightState {
     is_on: bool,
 }
 
-// One channel pair per connector, held for the process lifetime.
-static CHANNELS: Channels = Channels::new();
-
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let runtime = Arc::new(TokioAdapter::new()?);
@@ -339,7 +336,6 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
             TokioNet::udp(Ipv4Addr::UNSPECIFIED),
             TokioDelay,
             "knx://192.168.1.19:3671",
-            &CHANNELS,
         ))
         .configure::<LightState>("light.state", |reg| {
             reg.buffer(BufferCfg::SingleLatest)
diff --git a/examples/embassy-knx-connector-demo/src/main.rs b/examples/embassy-knx-connector-demo/src/main.rs
index 7fd463cc..52286d37 100644
--- a/examples/embassy-knx-connector-demo/src/main.rs
+++ b/examples/embassy-knx-connector-demo/src/main.rs
@@ -45,7 +45,7 @@ use aimdb_core::{AimDbBuilder, RecordKey, RuntimeContext};
 use aimdb_embassy_adapter::io::EmbassyUart;
 use aimdb_embassy_adapter::net::{EmbassyDelay, EmbassyNet};
 use aimdb_embassy_adapter::{EmbassyAdapter, EmbassyBufferType, EmbassyRecordRegistrarExtCustom};
-use aimdb_knx_connector::connector::{Channels, KnxConnector};
+use aimdb_knx_connector::connector::KnxConnector;
 use aimdb_knx_connector::dpt::{Dpt1, Dpt9, DptDecode, DptEncode};
 use aimdb_serial_connector::SerialServer;
 use defmt::*;
@@ -273,12 +273,11 @@ async fn main(spawner: Spawner) {
     let (serial_tx, serial_rx) = uart.split();
 
     // The adapter owns the UDP socket and the clock; the connector owns the
-    // tunnelling protocol. Buffers and channels are `'static`, as on any MCU.
+    // tunnelling protocol. Socket buffers are `'static`, as on any MCU.
     static KNX_RX_META: StaticCell<[PacketMetadata; 8]> = StaticCell::new();
     static KNX_RX_BUF: StaticCell<[u8; 1024]> = StaticCell::new();
     static KNX_TX_META: StaticCell<[PacketMetadata; 8]> = StaticCell::new();
     static KNX_TX_BUF: StaticCell<[u8; 1024]> = StaticCell::new();
-    static KNX_CHANNELS: Channels<32> = Channels::new();
     let knx_binder = EmbassyNet::udp(
         *stack,
         KNX_RX_META.init([PacketMetadata::EMPTY; 8]),
@@ -291,12 +290,7 @@ async fn main(spawner: Spawner) {
     // remote `record.set` is refused — peers can list/get/subscribe, not write.
     let mut builder = AimDbBuilder::new()
         .runtime(runtime.clone())
-        .with_connector(KnxConnector::new(
-            knx_binder,
-            EmbassyDelay,
-            &gateway_url,
-            &KNX_CHANNELS,
-        ))
+        .with_connector(KnxConnector::new(knx_binder, EmbassyDelay, &gateway_url))
         .with_connector(
             SerialServer::new(EmbassyUart::new(serial_rx, serial_tx))
                 .security_policy(SecurityPolicy::read_only()),
diff --git a/examples/tokio-knx-connector-demo/README.md b/examples/tokio-knx-connector-demo/README.md
index 3f5460c1..328ebc33 100644
--- a/examples/tokio-knx-connector-demo/README.md
+++ b/examples/tokio-knx-connector-demo/README.md
@@ -27,16 +27,12 @@ Demonstrates bidirectional KNX/IP integration with AimDB using the Tokio runtime
 Edit `src/main.rs` to match your KNX setup:
 
 ```rust
-// The connector's queues, held for the process lifetime.
-static KNX_CHANNELS: Channels = Channels::new();
-
 // The adapter owns the UDP socket and the clock; the connector owns the
 // tunnelling protocol. Only the gateway URL needs changing.
 .with_connector(KnxConnector::new(
     TokioNet::udp(Ipv4Addr::UNSPECIFIED),
     TokioDelay,
     "knx://YOUR_GATEWAY_IP:3671",  // Change to your gateway IP
-    &KNX_CHANNELS,
 ))
 
 // Group addresses
diff --git a/examples/tokio-knx-connector-demo/src/main.rs b/examples/tokio-knx-connector-demo/src/main.rs
index fe266d9b..8fc88ef3 100644
--- a/examples/tokio-knx-connector-demo/src/main.rs
+++ b/examples/tokio-knx-connector-demo/src/main.rs
@@ -25,7 +25,7 @@ use aimdb_core::buffer::BufferCfg;
 use aimdb_core::remote::{AimxConfig, SecurityPolicy};
 use aimdb_core::{AimDbBuilder, DbResult, Producer, RecordKey, RuntimeContext};
 use aimdb_knx_connector::dpt::{Dpt1, Dpt9, DptDecode, DptEncode};
-use aimdb_knx_connector::{Channels, KnxConnector};
+use aimdb_knx_connector::KnxConnector;
 use aimdb_tokio_adapter::net::{TokioDelay, TokioNet};
 use aimdb_tokio_adapter::{TokioAdapter, TokioRecordRegistrarExt};
 use aimdb_uds_connector::UdsServer;
@@ -103,17 +103,14 @@ async fn main() -> DbResult<()> {
         .security_policy(SecurityPolicy::read_only())
         .max_connections(10);
 
-    // The adapter owns the UDP socket and the clock; the channels are the
-    // caller's, exactly as on the MCU.
-    static KNX_CHANNELS: Channels = Channels::new();
-
+    // The adapter owns the UDP socket and the clock; the connector owns the
+    // tunnelling protocol.
     let mut builder = AimDbBuilder::new()
         .runtime(runtime)
         .with_connector(KnxConnector::new(
             TokioNet::udp(Ipv4Addr::UNSPECIFIED),
             TokioDelay,
             "knx://192.168.1.4:3671",
-            &KNX_CHANNELS,
         ))
         .with_connector(UdsServer::from_config(remote_config));