From 86e586de0d94126b23c292b8f08534b84f94e527 Mon Sep 17 00:00:00 2001
From: "sounds.like.lx" <147444674+lxsaah@users.noreply.github.com>
Date: Sat, 24 Oct 2026 17:07:02 +0200
Subject: [PATCH 1/6] fix(session): a pump's cleanup only removes its own
 subscription; refuse a live id (#345)

---
 aimdb-core/src/session/server.rs   |  61 +++++++----
 aimdb-core/tests/session_engine.rs | 163 +++++++++++++++++++++++++++++
 2 files changed, 204 insertions(+), 20 deletions(-)

diff --git a/aimdb-core/src/session/server.rs b/aimdb-core/src/session/server.rs
index 72a619d0..f08d6b8e 100644
--- a/aimdb-core/src/session/server.rs
+++ b/aimdb-core/src/session/server.rs
@@ -70,9 +70,9 @@ enum Step {
     Closed,
     /// A subscription update to encode and forward to the peer.
     Event(SubEvent),
-    /// A subscription pump finished on its own (stream exhausted) — reap it and
-    /// prune its `cancels` entry by the carried sub id.
-    SubDrained(String),
+    /// A subscription pump finished — reap it and prune its `cancels` entry,
+    /// but only while that entry is still the pump's own (same generation).
+    SubDrained(String, u64),
 }
 
 /// Drive one accepted [`Connection`] until it closes.
@@ -116,11 +116,15 @@ pub async fn run_session<C, D>(
     // it without limit — pumps drop on overflow (events carry a monotonic `seq`).
     let (event_tx, event_rx) = async_channel::bounded::<SubEvent>(EVENT_BUFFER);
     // Per-connection subscription pumps; the engine future is their sole owner.
-    // Each resolves to its own sub id, so the loop can prune the matching
-    // `cancels` entry (a no-op if Unsubscribe already removed it).
-    let mut subs: FuturesUnordered<BoxFut<'static, String>> = FuturesUnordered::new();
-    // sub-id → cancel handle; dropping/firing the oneshot cancels the pump.
-    let mut cancels: HashMap<String, oneshot::Sender<()>> = HashMap::new();
+    // Each resolves to its own sub id and generation, so the loop can prune the
+    // matching `cancels` entry (a no-op if Unsubscribe already removed it).
+    let mut subs: FuturesUnordered<BoxFut<'static, (String, u64)>> = FuturesUnordered::new();
+    // sub-id → (generation, cancel handle); dropping/firing the oneshot cancels
+    // the pump. The generation tells a pump's own entry from a later
+    // subscription that reuses its id after an Unsubscribe: a pump that ends
+    // late must not remove that one.
+    let mut cancels: HashMap<String, (u64, oneshot::Sender<()>)> = HashMap::new();
+    let mut next_generation: u64 = 0;
     // Reused encode scratch buffer.
     let mut out = Vec::new();
 
@@ -148,17 +152,22 @@ pub async fn run_session<C, D>(
                     Err(_) => Step::Closed,
                 },
                 // ---- drain finished subscription pumps ---------------------
-                sub_id = subs.select_next_some() => Step::SubDrained(sub_id),
+                (sub_id, generation) = subs.select_next_some() => {
+                    Step::SubDrained(sub_id, generation)
+                }
             }
         };
 
         match step {
             Step::Closed => break,
-            // A pump finished on its own (stream exhausted), not via Unsubscribe;
-            // drop its cancel handle so the ended subscription neither leaks nor
-            // keeps counting against `max_subs_per_connection`.
-            Step::SubDrained(sub_id) => {
-                cancels.remove(&sub_id);
+            // A pump finished; drop its cancel handle so the ended subscription
+            // neither leaks nor keeps counting against
+            // `max_subs_per_connection` — unless the id now belongs to a newer
+            // subscription (Unsubscribe, then Subscribe with the same id).
+            Step::SubDrained(sub_id, generation) => {
+                if cancels.get(&sub_id).is_some_and(|(g, _)| *g == generation) {
+                    cancels.remove(&sub_id);
+                }
             }
 
             Step::Event(ev) => {
@@ -202,6 +211,13 @@ pub async fn run_session<C, D>(
                         // The opening request id is the subscription's routing key;
                         // events carry it back as `Outbound::Event.sub`.
                         let sub_id = id.to_string();
+                        // An id that still names a live subscription is refused:
+                        // replacing it would cancel the live one without a word.
+                        if cancels.contains_key(&sub_id) {
+                            log_warn!("subscribe refused: id {} is still live", sub_id);
+                            send_reply_err(&mut conn, codec, &mut out, id, RpcError::Denied).await;
+                            continue;
+                        }
                         if cancels.len() >= config.limits.max_subs_per_connection {
                             send_reply_err(&mut conn, codec, &mut out, id, RpcError::Denied).await;
                             continue;
@@ -275,9 +291,12 @@ pub async fn run_session<C, D>(
                                     break;
                                 }
                                 let (cancel_tx, cancel_rx) = oneshot::channel();
-                                cancels.insert(sub_id.clone(), cancel_tx);
+                                let generation = next_generation;
+                                next_generation += 1;
+                                cancels.insert(sub_id.clone(), (generation, cancel_tx));
                                 subs.push(Box::pin(pump_subscription(
                                     sub_id,
+                                    generation,
                                     stream,
                                     event_tx.clone(),
                                     cancel_rx,
@@ -348,16 +367,18 @@ async fn send_reply_err<C: EnvelopeCodec + ?Sized>(
 /// snapshot lost at the tail of the burst still shows up as a gap on the first
 /// event.
 ///
-/// Returns its `sub_id` so [`run_session`] can prune the `cancels` entry for a
-/// pump that ended on its own; the Unsubscribe path already removed it, so that
-/// later prune is a no-op.
+/// Returns its `sub_id` and `generation` so [`run_session`] can prune the
+/// `cancels` entry for a pump that ended on its own; the Unsubscribe path
+/// already removed it, and a newer subscription under the same id has another
+/// generation, so that later prune leaves both alone.
 async fn pump_subscription(
     sub_id: String,
+    generation: u64,
     mut stream: BoxStream<'static, SubUpdate>,
     tx: Sender<SubEvent>,
     cancel: oneshot::Receiver<()>,
     start_seq: u64,
-) -> String {
+) -> (String, u64) {
     // Fuse the cancel receiver: a bare `oneshot::Receiver` reports
     // `is_terminated()` once its sender drops, and `select_biased!` skips
     // terminated arms — so the cancel would never fire. `Fuse` keeps the arm
@@ -389,7 +410,7 @@ async fn pump_subscription(
             Err(_) => break,            // funnel disconnected — connection gone
         }
     }
-    sub_id
+    (sub_id, generation)
 }
 
 /// Accept connections from `listener` and serve each with [`run_session`],
diff --git a/aimdb-core/tests/session_engine.rs b/aimdb-core/tests/session_engine.rs
index 2521c9ef..852561dc 100644
--- a/aimdb-core/tests/session_engine.rs
+++ b/aimdb-core/tests/session_engine.rs
@@ -1098,3 +1098,166 @@ hello",
         "serve must keep waiting on the parked accept, not return"
     );
 }
+
+// ===========================================================================
+// Subscription ids (#328) — a raw peer against `serve`, so the test decides
+// exactly which ids go on the wire.
+// ===========================================================================
+
+/// Every subscription follows one live feed until it is cancelled.
+struct FeedDispatch {
+    feed: tokio::sync::broadcast::Sender<SubUpdate>,
+}
+
+impl Dispatch for FeedDispatch {
+    fn authenticate<'a>(
+        &'a self,
+        _peer: &'a PeerInfo,
+        _first: Option<&'a [u8]>,
+    ) -> BoxFut<'a, Result<SessionCtx, AuthError>> {
+        Box::pin(async { Ok(SessionCtx::default()) })
+    }
+
+    fn open(&self, _ctx: &SessionCtx) -> Box<dyn Session> {
+        Box::new(FeedSession {
+            feed: self.feed.clone(),
+        })
+    }
+}
+
+struct FeedSession {
+    feed: tokio::sync::broadcast::Sender<SubUpdate>,
+}
+
+impl Session for FeedSession {
+    fn call<'a>(
+        &'a mut self,
+        _method: &'a str,
+        params: Payload,
+    ) -> BoxFut<'a, Result<Payload, RpcError>> {
+        Box::pin(async move { Ok(params) })
+    }
+
+    fn subscribe<'a>(
+        &'a mut self,
+        _topic: &'a str,
+    ) -> BoxFut<'a, Result<BoxStream<'static, SubUpdate>, RpcError>> {
+        let rx = self.feed.subscribe();
+        Box::pin(async move {
+            let stream = futures::stream::unfold(rx, |mut rx| async move {
+                rx.recv().await.ok().map(|u| (u, rx))
+            });
+            Ok(Box::pin(stream) as BoxStream<'static, SubUpdate>)
+        })
+    }
+
+    fn write<'a>(
+        &'a mut self,
+        _topic: &'a str,
+        _payload: Payload,
+    ) -> BoxFut<'a, Result<(), RpcError>> {
+        Box::pin(async { Ok(()) })
+    }
+}
+
+/// A `serve` over [`FeedDispatch`] and a raw connection to it. The dialer is
+/// returned so it stays alive: dropping it closes the listener.
+async fn feed_session() -> (
+    Box<dyn Connection>,
+    tokio::sync::broadcast::Sender<SubUpdate>,
+    tokio::task::JoinHandle<()>,
+    ChannelDialer,
+) {
+    let (listener, dialer) = transport_pair();
+    let (feed, _) = tokio::sync::broadcast::channel(16);
+    let server = tokio::spawn(serve(
+        listener,
+        Arc::new(LineCodec),
+        Arc::new(FeedDispatch { feed: feed.clone() }),
+        SessionConfig {
+            limits: SessionLimits {
+                max_connections: 1,
+                max_subs_per_connection: 8,
+            },
+            reads_hello: false,
+            acks_subscribe: false,
+        },
+    ));
+    let conn = dialer.connect().await.expect("connect");
+    (conn, feed, server, dialer)
+}
+
+/// Every frame the server sends within `window`, as text.
+async fn frames_within(conn: &mut Box<dyn Connection>, window: Duration) -> Vec<String> {
+    let mut frames = Vec::new();
+    let deadline = tokio::time::Instant::now() + window;
+    while let Ok(Ok(Some(frame))) = tokio::time::timeout_at(deadline, conn.recv()).await {
+        frames.push(String::from_utf8(frame).expect("utf-8 frame"));
+    }
+    frames
+}
+
+/// Publishes `value` once the server's subscriptions are registered, and
+/// returns what arrived for it.
+async fn publish(
+    conn: &mut Box<dyn Connection>,
+    feed: &tokio::sync::broadcast::Sender<SubUpdate>,
+    value: &str,
+) -> Vec<String> {
+    tokio::time::sleep(Duration::from_millis(50)).await;
+    let _ = feed.send(SubUpdate::new(payload_from(value)));
+    frames_within(conn, Duration::from_millis(200)).await
+}
+
+/// A subscribe whose id names a live subscription is refused; the live one
+/// keeps delivering, once per update. Before, the new entry replaced the live
+/// one, which cancelled it, and the cleanup of the cancelled pump then removed
+/// the new entry too: the id went silent.
+#[tokio::test]
+async fn a_subscribe_reusing_a_live_id_is_refused() {
+    let (mut conn, feed, server, _dialer) = feed_session().await;
+
+    conn.send(b"SUB\n2\nfeed").await.unwrap();
+    let first = publish(&mut conn, &feed, "v1").await;
+    assert_eq!(first, ["EVENT\n2\n1\n-\nv1"]);
+
+    conn.send(b"SUB\n2\nfeed").await.unwrap();
+    let refused = frames_within(&mut conn, Duration::from_millis(200)).await;
+    assert_eq!(refused, ["REPLY\n2\nERR\ndenied"]);
+
+    let after = publish(&mut conn, &feed, "v2").await;
+    assert_eq!(
+        after,
+        ["EVENT\n2\n2\n-\nv2"],
+        "the live subscription keeps delivering"
+    );
+    server.abort();
+}
+
+/// An id reused after an Unsubscribe belongs to the new subscription: the old
+/// pump, ending late, must not remove the new entry.
+#[tokio::test]
+async fn an_id_reused_after_unsubscribe_keeps_delivering() {
+    let (mut conn, feed, server, _dialer) = feed_session().await;
+
+    conn.send(b"SUB\n2\nfeed").await.unwrap();
+    assert_eq!(
+        publish(&mut conn, &feed, "v1").await,
+        ["EVENT\n2\n1\n-\nv1"]
+    );
+
+    // Back to back: the old pump is cancelled but not yet reaped when the new
+    // subscription registers under the same id.
+    conn.send(b"UNSUB\n2").await.unwrap();
+    conn.send(b"SUB\n2\nfeed").await.unwrap();
+
+    for (n, value) in ["v2", "v3"].into_iter().enumerate() {
+        let got = publish(&mut conn, &feed, value).await;
+        assert_eq!(
+            got,
+            [format!("EVENT\n2\n{}\n-\n{value}", n + 1)],
+            "the reused id must keep delivering"
+        );
+    }
+    server.abort();
+}

From fdb5d89f8dfadaa93371717549b00f57688ce70a Mon Sep 17 00:00:00 2001
From: "sounds.like.lx" <147444674+lxsaah@users.noreply.github.com>
Date: Sat, 24 Oct 2026 17:09:37 +0200
Subject: [PATCH 2/6] fix(session): a second handshake on a live connection
 starts a new session (#346)

---
 aimdb-core/src/session/aimx/dispatch.rs   |   6 +
 aimdb-core/src/session/mod.rs             |  14 ++
 aimdb-core/src/session/server.rs          |  24 +++-
 aimdb-core/tests/session_engine.rs        |  59 +++++++++
 aimdb-serial-connector/tests/reconnect.rs | 152 ++++++++++++++++++++++
 5 files changed, 254 insertions(+), 1 deletion(-)
 create mode 100644 aimdb-serial-connector/tests/reconnect.rs

diff --git a/aimdb-core/src/session/aimx/dispatch.rs b/aimdb-core/src/session/aimx/dispatch.rs
index a3e84f7e..4fcd08d1 100644
--- a/aimdb-core/src/session/aimx/dispatch.rs
+++ b/aimdb-core/src/session/aimx/dispatch.rs
@@ -102,6 +102,12 @@ impl DispatchReply {
 }
 
 impl Session for AimxSession {
+    /// `hello` opens every AimX session; a second one on a connection comes
+    /// from a new client (a serial line never closes between clients).
+    fn starts_session(&self, method: &str) -> bool {
+        method == "hello"
+    }
+
     fn call<'a>(
         &'a mut self,
         method: &'a str,
diff --git a/aimdb-core/src/session/mod.rs b/aimdb-core/src/session/mod.rs
index 98b9cba4..6adb6cf7 100644
--- a/aimdb-core/src/session/mod.rs
+++ b/aimdb-core/src/session/mod.rs
@@ -475,6 +475,20 @@ pub trait Session: Send {
         Vec::new()
     }
 
+    /// Whether a request for `method` is a client's handshake, which starts a
+    /// new session. Defaulted to `false`.
+    ///
+    /// When a handshake arrives on a connection that already had one,
+    /// [`run_session`](server::run_session) treats it as a new client: it
+    /// cancels every subscription, discards their queued events and opens a
+    /// fresh session before answering. A transport that carries one peer for
+    /// its whole life (a serial line) has no other way to tell that the client
+    /// behind it changed.
+    fn starts_session(&self, method: &str) -> bool {
+        let _ = method;
+        false
+    }
+
     /// Fire-and-forget write: no reply. Routes through the producer/arbiter path,
     /// so single-writer-per-key stays intact.
     fn write<'a>(
diff --git a/aimdb-core/src/session/server.rs b/aimdb-core/src/session/server.rs
index f08d6b8e..0d815114 100644
--- a/aimdb-core/src/session/server.rs
+++ b/aimdb-core/src/session/server.rs
@@ -114,7 +114,7 @@ pub async fn run_session<C, D>(
     // Event funnel: every per-subscription pump sends updates here; the main loop
     // is the sole writer to the connection. Bounded, so a slow client cannot grow
     // it without limit — pumps drop on overflow (events carry a monotonic `seq`).
-    let (event_tx, event_rx) = async_channel::bounded::<SubEvent>(EVENT_BUFFER);
+    let (mut event_tx, mut event_rx) = async_channel::bounded::<SubEvent>(EVENT_BUFFER);
     // Per-connection subscription pumps; the engine future is their sole owner.
     // Each resolves to its own sub id and generation, so the loop can prune the
     // matching `cancels` entry (a no-op if Unsubscribe already removed it).
@@ -125,6 +125,9 @@ pub async fn run_session<C, D>(
     // late must not remove that one.
     let mut cancels: HashMap<String, (u64, oneshot::Sender<()>)> = HashMap::new();
     let mut next_generation: u64 = 0;
+    // Whether this connection has seen a handshake; a second one starts a new
+    // session (see `Session::starts_session`).
+    let mut greeted = false;
     // Reused encode scratch buffer.
     let mut out = Vec::new();
 
@@ -195,6 +198,25 @@ pub async fn run_session<C, D>(
                 };
                 match msg {
                     Inbound::Request { id, method, params } => {
+                        if session.starts_session(&method) {
+                            if greeted {
+                                // A new client on a connection that never closed
+                                // (a serial line): nothing of the old session
+                                // may reach it or outlive it. Dropping the pumps
+                                // releases their buffer subscribers; a fresh
+                                // funnel discards their queued events.
+                                log_info!(
+                                    "new handshake on a live connection: ending {} subscription(s) of the previous client",
+                                    cancels.len()
+                                );
+                                subs = FuturesUnordered::new();
+                                cancels.clear();
+                                (event_tx, event_rx) =
+                                    async_channel::bounded::<SubEvent>(EVENT_BUFFER);
+                                session = dispatch.open(&ctx);
+                            }
+                            greeted = true;
+                        }
                         let result = session.call(&method, params).await;
                         out.clear();
                         if codec
diff --git a/aimdb-core/tests/session_engine.rs b/aimdb-core/tests/session_engine.rs
index 852561dc..1bd51b1b 100644
--- a/aimdb-core/tests/session_engine.rs
+++ b/aimdb-core/tests/session_engine.rs
@@ -1130,6 +1130,10 @@ struct FeedSession {
 }
 
 impl Session for FeedSession {
+    fn starts_session(&self, method: &str) -> bool {
+        method == "hello"
+    }
+
     fn call<'a>(
         &'a mut self,
         _method: &'a str,
@@ -1261,3 +1265,58 @@ async fn an_id_reused_after_unsubscribe_keeps_delivering() {
     }
     server.abort();
 }
+
+/// A second handshake on a connection that never closed (a serial line) is a
+/// new client: the old client's subscriptions end and release their source,
+/// and the new client's subscription under the same id delivers once per
+/// update. Before, the old subscriptions lived on, kept their sources, and the
+/// new subscribe collided with them.
+#[tokio::test]
+async fn a_second_handshake_starts_a_new_session() {
+    let (mut conn, feed, server, _dialer) = feed_session().await;
+
+    // Client A: handshake, two subscriptions.
+    conn.send(b"REQ\n1\nhello\nA").await.unwrap();
+    conn.send(b"SUB\n2\nfeed").await.unwrap();
+    conn.send(b"SUB\n3\nfeed").await.unwrap();
+    let a = publish(&mut conn, &feed, "v1").await;
+    assert_eq!(a.len(), 3, "A's reply and two events: {a:?}");
+    assert_eq!(feed.receiver_count(), 2);
+
+    // Client B on the same connection: its handshake ends A's session.
+    conn.send(b"REQ\n1\nhello\nB").await.unwrap();
+    let reply = frames_within(&mut conn, Duration::from_millis(200)).await;
+    assert_eq!(reply, ["REPLY\n1\nOK\nB"]);
+    assert_eq!(
+        feed.receiver_count(),
+        0,
+        "A's subscriptions released their source"
+    );
+
+    // No event of A's subscriptions reaches B.
+    let quiet = publish(&mut conn, &feed, "v2").await;
+    assert!(
+        quiet.is_empty(),
+        "nothing for B before it subscribes: {quiet:?}"
+    );
+
+    // B subscribes under A's id: delivered once per update.
+    conn.send(b"SUB\n2\nfeed").await.unwrap();
+    let b = publish(&mut conn, &feed, "v3").await;
+    assert_eq!(b, ["EVENT\n2\n1\n-\nv3"]);
+    assert_eq!(feed.receiver_count(), 1);
+    server.abort();
+}
+
+/// The first handshake on a connection changes nothing: subscriptions opened
+/// before it stay.
+#[tokio::test]
+async fn the_first_handshake_keeps_earlier_subscriptions() {
+    let (mut conn, feed, server, _dialer) = feed_session().await;
+
+    conn.send(b"SUB\n2\nfeed").await.unwrap();
+    conn.send(b"REQ\n1\nhello\nA").await.unwrap();
+    let got = publish(&mut conn, &feed, "v1").await;
+    assert_eq!(got, ["REPLY\n1\nOK\nA", "EVENT\n2\n1\n-\nv1"]);
+    server.abort();
+}
diff --git a/aimdb-serial-connector/tests/reconnect.rs b/aimdb-serial-connector/tests/reconnect.rs
new file mode 100644
index 00000000..5130e8b7
--- /dev/null
+++ b/aimdb-serial-connector/tests/reconnect.rs
@@ -0,0 +1,152 @@
+//! Clients one after another on one serial line (#328).
+//!
+//! A UART never closes, so the server cannot see a client leave. A new
+//! client's `hello` is the only sign that the peer changed: it must end the
+//! previous client's subscriptions, or they keep streaming and the new
+//! client's subscribe collides with them (every other reconnect got nothing).
+//!
+//! The pipe below stays open for the whole test, as a UART does; each client
+//! speaks raw AimX frames, sending what `aimdb-client` sends: `hello` as id 1,
+//! then a subscribe as id 2.
+
+#![cfg(feature = "std")]
+
+use std::sync::Arc;
+use std::time::Duration;
+
+use aimdb_core::buffer::BufferCfg;
+use aimdb_core::remote::AimxConfig;
+use aimdb_core::session::aimx::{AimxCodec, AimxDispatch};
+use aimdb_core::session::OneShotListener;
+use aimdb_core::session::{serve, Dispatch, SessionConfig, SessionLimits};
+use aimdb_core::AimDbBuilder;
+use aimdb_serial_connector::connector::framed;
+use aimdb_serial_connector::framing::{encode_frame, FrameAccumulator};
+use aimdb_tokio_adapter::net::TokioByteStream;
+use aimdb_tokio_adapter::{TokioAdapter, TokioRecordRegistrarExt};
+use serde::{Deserialize, Serialize};
+use serde_json::{json, Value};
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio::sync::mpsc;
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+struct Temp {
+    c: u64,
+}
+
+/// One AimX frame, COBS-framed for the line.
+fn frame(v: Value) -> Vec<u8> {
+    let mut wire = Vec::new();
+    encode_frame(&serde_json::to_vec(&v).unwrap(), &mut wire);
+    wire
+}
+
+/// Every frame that arrives within `window`.
+async fn frames_within(rx: &mut mpsc::UnboundedReceiver<Value>, window: Duration) -> Vec<Value> {
+    let mut frames = Vec::new();
+    let deadline = tokio::time::Instant::now() + window;
+    while let Ok(Some(f)) = tokio::time::timeout_at(deadline, rx.recv()).await {
+        frames.push(f);
+    }
+    frames
+}
+
+#[tokio::test]
+async fn every_client_on_one_serial_line_gets_its_events() {
+    let (server_end, client_end) = tokio::io::duplex(64 * 1024);
+
+    let mut builder = AimDbBuilder::new().runtime(Arc::new(TokioAdapter));
+    builder.configure::<Temp>("temp", |reg| {
+        reg.buffer(BufferCfg::SpmcRing { capacity: 16 })
+            .with_remote_access();
+    });
+    let (db, runner) = builder.build().await.expect("build db");
+    let db = Arc::new(db);
+    tokio::spawn(runner.run());
+
+    // The serial connector's session config: one peer for the line's life.
+    let dispatch: Arc<dyn Dispatch> =
+        Arc::new(AimxDispatch::new(db.clone(), AimxConfig::uds_default()));
+    tokio::spawn(serve(
+        OneShotListener::new(framed(TokioByteStream(server_end))),
+        Arc::new(AimxCodec),
+        dispatch,
+        SessionConfig {
+            limits: SessionLimits {
+                max_connections: 1,
+                max_subs_per_connection: 8,
+            },
+            reads_hello: false,
+            acks_subscribe: false,
+        },
+    ));
+
+    // The host end of the line: frames keep arriving whether or not a client
+    // reads them.
+    let (mut rd, mut wr) = tokio::io::split(client_end);
+    let (tx, mut rx) = mpsc::unbounded_channel::<Value>();
+    tokio::spawn(async move {
+        let mut acc = FrameAccumulator::new();
+        let mut buf = [0u8; 1024];
+        while let Ok(n) = rd.read(&mut buf).await {
+            if n == 0 {
+                break;
+            }
+            acc.push_bytes(&buf[..n]);
+            while let Some(Ok(f)) = acc.next_frame() {
+                if let Ok(v) = serde_json::from_slice::<Value>(&f) {
+                    let _ = tx.send(v);
+                }
+            }
+        }
+    });
+
+    let mut value = 0u64;
+    for client in ["A", "B", "C"] {
+        // Values produced while the previous client is gone, then whatever
+        // its subscription sent: lost with it.
+        for _ in 0..3 {
+            value += 1;
+            db.produce::<Temp>("temp", Temp { c: value }).unwrap();
+        }
+        frames_within(&mut rx, Duration::from_millis(100)).await;
+
+        wr.write_all(&frame(json!({"t":"req","id":1,"method":"hello",
+            "params":{"client":client,"version":"3.0"}})))
+            .await
+            .unwrap();
+        let welcome = frames_within(&mut rx, Duration::from_millis(100)).await;
+        assert!(
+            welcome.len() == 1 && welcome[0]["t"] == "reply" && welcome[0]["id"] == 1,
+            "client {client}: only its Welcome after hello, got {welcome:?}"
+        );
+
+        wr.write_all(&frame(json!({"t":"sub","id":2,"topic":"temp"})))
+            .await
+            .unwrap();
+        tokio::time::sleep(Duration::from_millis(50)).await;
+        let first = value + 1;
+        for _ in 0..5 {
+            value += 1;
+            db.produce::<Temp>("temp", Temp { c: value }).unwrap();
+        }
+        let frames = frames_within(&mut rx, Duration::from_millis(300)).await;
+        // A fresh subscription numbers its events from 1; the previous
+        // client's would continue its own count under the same id.
+        let events: Vec<(u64, u64)> = frames
+            .iter()
+            .filter(|f| f["t"] == "event" && f["sub"] == "2")
+            .map(|f| (f["seq"].as_u64().unwrap(), f["data"]["c"].as_u64().unwrap()))
+            .collect();
+        assert_eq!(
+            events,
+            (1..=5).zip(first..=value).collect::<Vec<_>>(),
+            "client {client}: each value once, from its own subscription; got {frames:?}"
+        );
+        assert!(
+            frames.iter().all(|f| f["t"] == "event"),
+            "client {client}: no error reply to its subscribe; got {frames:?}"
+        );
+        // The client goes away without a word, as a closed serial port does.
+    }
+}

From 0e130c07c27e61ff1d291b2c339db4e37f15fcad Mon Sep 17 00:00:00 2001
From: "sounds.like.lx" <147444674+lxsaah@users.noreply.github.com>
Date: Sat, 24 Oct 2026 17:24:58 +0200
Subject: [PATCH 3/6] fix(session): report a subscription whose stream ends on
 its own (#347)

---
 aimdb-core/src/session/server.rs   | 106 ++++++++++++++++++++++-------
 aimdb-core/tests/session_engine.rs |  70 +++++++++++++++++++
 2 files changed, 151 insertions(+), 25 deletions(-)

diff --git a/aimdb-core/src/session/server.rs b/aimdb-core/src/session/server.rs
index 0d815114..69c42ef8 100644
--- a/aimdb-core/src/session/server.rs
+++ b/aimdb-core/src/session/server.rs
@@ -72,7 +72,16 @@ enum Step {
     Event(SubEvent),
     /// A subscription pump finished — reap it and prune its `cancels` entry,
     /// but only while that entry is still the pump's own (same generation).
-    SubDrained(String, u64),
+    SubDrained(PumpEnd),
+}
+
+/// How a subscription pump ended, returned to [`run_session`].
+struct PumpEnd {
+    sub_id: String,
+    /// Which `cancels` entry was the pump's own.
+    generation: u64,
+    /// The stream ran out on its own, not through a cancel or a closed funnel.
+    exhausted: bool,
 }
 
 /// Drive one accepted [`Connection`] until it closes.
@@ -118,7 +127,7 @@ pub async fn run_session<C, D>(
     // Per-connection subscription pumps; the engine future is their sole owner.
     // Each resolves to its own sub id and generation, so the loop can prune the
     // matching `cancels` entry (a no-op if Unsubscribe already removed it).
-    let mut subs: FuturesUnordered<BoxFut<'static, (String, u64)>> = FuturesUnordered::new();
+    let mut subs: FuturesUnordered<BoxFut<'static, PumpEnd>> = FuturesUnordered::new();
     // sub-id → (generation, cancel handle); dropping/firing the oneshot cancels
     // the pump. The generation tells a pump's own entry from a later
     // subscription that reuses its id after an Unsubscribe: a pump that ends
@@ -155,9 +164,7 @@ pub async fn run_session<C, D>(
                     Err(_) => Step::Closed,
                 },
                 // ---- drain finished subscription pumps ---------------------
-                (sub_id, generation) = subs.select_next_some() => {
-                    Step::SubDrained(sub_id, generation)
-                }
+                end = subs.select_next_some() => Step::SubDrained(end),
             }
         };
 
@@ -167,26 +174,43 @@ pub async fn run_session<C, D>(
             // neither leaks nor keeps counting against
             // `max_subs_per_connection` — unless the id now belongs to a newer
             // subscription (Unsubscribe, then Subscribe with the same id).
-            Step::SubDrained(sub_id, generation) => {
-                if cancels.get(&sub_id).is_some_and(|(g, _)| *g == generation) {
-                    cancels.remove(&sub_id);
+            Step::SubDrained(end) => {
+                let own = cancels
+                    .get(&end.sub_id)
+                    .is_some_and(|(g, _)| *g == end.generation);
+                if !own {
+                    continue;
+                }
+                cancels.remove(&end.sub_id);
+                // A subscription the client still holds ended on its own: its
+                // source closed, or never opened (a buffer with no subscriber
+                // slot left). Tell the client, which turns the error reply to
+                // its subscribe id into a terminal error, instead of leaving an
+                // acknowledged subscription that never delivers.
+                if end.exhausted {
+                    // Its last events may still sit in the funnel: the pump
+                    // runs inside the reap arm, so a stream that is ready at
+                    // once fills the funnel and ends in one poll. Send them
+                    // first, so the error is the subscription's last word.
+                    let mut flushed = true;
+                    while let Ok(ev) = event_rx.try_recv() {
+                        if !send_event(&mut conn, codec, &mut out, ev).await {
+                            flushed = false;
+                            break;
+                        }
+                    }
+                    if !flushed {
+                        break;
+                    }
+                    log_warn!("subscription {} ended: its source closed", end.sub_id);
+                    if let Ok(id) = end.sub_id.parse::<u64>() {
+                        send_reply_err(&mut conn, codec, &mut out, id, RpcError::Internal).await;
+                    }
                 }
             }
 
             Step::Event(ev) => {
-                out.clear();
-                let encoded = codec
-                    .encode(
-                        Outbound::Event {
-                            sub: &ev.sub,
-                            seq: ev.seq,
-                            topic: ev.topic.as_deref(),
-                            data: ev.data,
-                        },
-                        &mut out,
-                    )
-                    .is_ok();
-                if encoded && conn.send(&out).await.is_err() {
+                if !send_event(&mut conn, codec, &mut out, ev).await {
                     break;
                 }
             }
@@ -355,6 +379,29 @@ pub async fn run_session<C, D>(
     drop(subs);
 }
 
+/// Encode + send one subscription event. `false` only when the connection is
+/// gone; an event that does not encode is skipped.
+async fn send_event<C: EnvelopeCodec + ?Sized>(
+    conn: &mut Box<dyn Connection>,
+    codec: &C,
+    out: &mut Vec<u8>,
+    ev: SubEvent,
+) -> bool {
+    out.clear();
+    let encoded = codec
+        .encode(
+            Outbound::Event {
+                sub: &ev.sub,
+                seq: ev.seq,
+                topic: ev.topic.as_deref(),
+                data: ev.data,
+            },
+            out,
+        )
+        .is_ok();
+    !encoded || conn.send(out).await.is_ok()
+}
+
 /// Encode + send a `Reply` carrying an [`RpcError`]; best-effort (a send/encode
 /// failure just ends this attempt — the caller's loop handles a dead connection
 /// on its next `send`).
@@ -392,7 +439,8 @@ async fn send_reply_err<C: EnvelopeCodec + ?Sized>(
 /// Returns its `sub_id` and `generation` so [`run_session`] can prune the
 /// `cancels` entry for a pump that ended on its own; the Unsubscribe path
 /// already removed it, and a newer subscription under the same id has another
-/// generation, so that later prune leaves both alone.
+/// generation, so that later prune leaves both alone. `exhausted` tells a
+/// stream that ran out from a cancel or a closed funnel.
 async fn pump_subscription(
     sub_id: String,
     generation: u64,
@@ -400,13 +448,14 @@ async fn pump_subscription(
     tx: Sender<SubEvent>,
     cancel: oneshot::Receiver<()>,
     start_seq: u64,
-) -> (String, u64) {
+) -> PumpEnd {
     // Fuse the cancel receiver: a bare `oneshot::Receiver` reports
     // `is_terminated()` once its sender drops, and `select_biased!` skips
     // terminated arms — so the cancel would never fire. `Fuse` keeps the arm
     // polled until it actually resolves.
     let mut cancel = cancel.fuse();
     let mut seq: u64 = start_seq;
+    let mut exhausted = false;
     loop {
         // Independent arms, so a direct `select_biased!` is fine here.
         let update = select_biased! {
@@ -415,7 +464,10 @@ async fn pump_subscription(
             // `BoxStream` is not `FusedStream`, so fuse the per-iteration `next`.
             next = stream.next().fuse() => match next {
                 Some(update) => update,
-                None => break, // stream exhausted
+                None => {
+                    exhausted = true;
+                    break;
+                }
             },
         };
         seq += update.skipped + 1;
@@ -432,7 +484,11 @@ async fn pump_subscription(
             Err(_) => break,            // funnel disconnected — connection gone
         }
     }
-    (sub_id, generation)
+    PumpEnd {
+        sub_id,
+        generation,
+        exhausted,
+    }
 }
 
 /// Accept connections from `listener` and serve each with [`run_session`],
diff --git a/aimdb-core/tests/session_engine.rs b/aimdb-core/tests/session_engine.rs
index 1bd51b1b..e58f180d 100644
--- a/aimdb-core/tests/session_engine.rs
+++ b/aimdb-core/tests/session_engine.rs
@@ -1320,3 +1320,73 @@ async fn the_first_handshake_keeps_earlier_subscriptions() {
     assert_eq!(got, ["REPLY\n1\nOK\nA", "EVENT\n2\n1\n-\nv1"]);
     server.abort();
 }
+
+/// A subscription whose source ends on its own (here after three updates; on
+/// a device, a buffer that had no subscriber slot left) is reported: the
+/// client gets its events, then a terminal `Internal` error, then the end.
+/// Before, the stream just stopped, so an acknowledged subscription that
+/// never delivered looked like a quiet record.
+#[tokio::test]
+async fn a_subscription_that_ends_on_its_own_is_reported() {
+    let (listener, dialer) = transport_pair();
+    let server = tokio::spawn(serve(
+        listener,
+        Arc::new(LineCodec),
+        Arc::new(EchoDispatch {
+            writes: Arc::new(Mutex::new(Vec::new())),
+        }),
+        SessionConfig::default(),
+    ));
+    let (handle, client_fut) = run_client(
+        dialer,
+        LineCodec,
+        ClientConfig {
+            reconnect: false,
+            reconnect_delay: 10,
+            max_reconnect_delay: 10,
+            max_reconnect_attempts: 0,
+            keepalive_interval: None,
+            max_offline_queue: 1024,
+            sends_hello: false,
+        },
+        Arc::new(TestClock),
+    );
+    let client = tokio::spawn(client_fut);
+
+    let mut stream = handle.subscribe("temp").unwrap();
+    let mut items = Vec::new();
+    while let Ok(Some(item)) = tokio::time::timeout(Duration::from_secs(2), stream.next()).await {
+        items.push(item.map(|u| String::from_utf8(u.data.to_vec()).unwrap()));
+    }
+    assert_eq!(
+        items,
+        [
+            Ok("temp#1".to_string()),
+            Ok("temp#2".to_string()),
+            Ok("temp#3".to_string()),
+            Err(RpcError::Internal),
+        ]
+    );
+
+    drop(stream);
+    drop(handle);
+    let _ = client.await;
+    server.abort();
+}
+
+/// Only a subscription the client still holds is reported when it ends: an
+/// unsubscribed one, and the previous client's after a new handshake, end in
+/// silence.
+#[tokio::test]
+async fn an_ended_subscription_the_client_dropped_is_not_reported() {
+    let (mut conn, feed, server, _dialer) = feed_session().await;
+
+    conn.send(b"SUB\n2\nfeed").await.unwrap();
+    conn.send(b"UNSUB\n2").await.unwrap();
+    conn.send(b"REQ\n1\nhello\nA").await.unwrap();
+    conn.send(b"SUB\n3\nfeed").await.unwrap();
+    conn.send(b"REQ\n4\nhello\nB").await.unwrap();
+    let frames = publish(&mut conn, &feed, "v1").await;
+    assert_eq!(frames, ["REPLY\n1\nOK\nA", "REPLY\n4\nOK\nB"]);
+    server.abort();
+}

From 8a2e14ce53ead407e54a47acc03d608716197aaf Mon Sep 17 00:00:00 2001
From: "sounds.like.lx" <147444674+lxsaah@users.noreply.github.com>
Date: Sat, 24 Oct 2026 17:26:41 +0200
Subject: [PATCH 4/6] fix(examples): give the embassy MQTT demo's sensors a
 remote-access slot (#348)

---
 examples/embassy-mqtt-connector-demo/src/main.rs | 9 ++++++---
 1 file changed, 6 insertions(+), 3 deletions(-)

diff --git a/examples/embassy-mqtt-connector-demo/src/main.rs b/examples/embassy-mqtt-connector-demo/src/main.rs
index 2181f690..78c52bdb 100644
--- a/examples/embassy-mqtt-connector-demo/src/main.rs
+++ b/examples/embassy-mqtt-connector-demo/src/main.rs
@@ -441,8 +441,11 @@ async fn main(spawner: Spawner) {
     // Using compile-time safe SensorKey enum - typos caught at compile time!
     // ========================================================================
 
+    // Three subscriber slots per sensor: the logger tap, the MQTT link and one
+    // remote-access subscriber (`aimdb watch` over the serial port). With two,
+    // a remote subscribe finds no slot left.
     builder.configure::<Temperature>(SensorKey::TempIndoor, |reg| {
-        reg.buffer_sized::<16, 2>(EmbassyBufferType::SpmcRing)
+        reg.buffer_sized::<16, 3>(EmbassyBufferType::SpmcRing)
             .with_remote_access()
             .source(indoor_temp_producer)
             .tap(temperature_logger)
@@ -452,7 +455,7 @@ async fn main(spawner: Spawner) {
     });
 
     builder.configure::<Temperature>(SensorKey::TempOutdoor, |reg| {
-        reg.buffer_sized::<16, 2>(EmbassyBufferType::SpmcRing)
+        reg.buffer_sized::<16, 3>(EmbassyBufferType::SpmcRing)
             .with_remote_access()
             .source(outdoor_temp_producer)
             .tap(temperature_logger)
@@ -462,7 +465,7 @@ async fn main(spawner: Spawner) {
     });
 
     builder.configure::<Temperature>(SensorKey::TempServerRoom, |reg| {
-        reg.buffer_sized::<16, 2>(EmbassyBufferType::SpmcRing)
+        reg.buffer_sized::<16, 3>(EmbassyBufferType::SpmcRing)
             .with_remote_access()
             .source(server_room_temp_producer)
             .tap(temperature_logger)

From 0361e374e7e392e2f18d5f511605a65bc00f502b Mon Sep 17 00:00:00 2001
From: "sounds.like.lx" <147444674+lxsaah@users.noreply.github.com>
Date: Sat, 24 Oct 2026 17:38:47 +0200
Subject: [PATCH 5/6] fix(client): a subscription the server ended no longer
 reads as a transport failure (#349)

---
 aimdb-client/src/engine.rs | 58 +++++++++++++++++++++++++++++++++++++-
 1 file changed, 57 insertions(+), 1 deletion(-)

diff --git a/aimdb-client/src/engine.rs b/aimdb-client/src/engine.rs
index ed3cf29f..bbed6aa8 100644
--- a/aimdb-client/src/engine.rs
+++ b/aimdb-client/src/engine.rs
@@ -399,7 +399,7 @@ fn decode_updates(
             }),
             Err(e) => {
                 *terminated = true;
-                Err(rpc_err(e))
+                Err(sub_err(e))
             }
         };
         futures::future::ready(Some(out))
@@ -434,6 +434,24 @@ fn rpc_err(e: RpcError) -> ClientError {
     }
 }
 
+/// Map the terminal error of a subscription stream onto a [`ClientError`].
+///
+/// Unlike a call's error it always comes from the server: a dropped
+/// connection ends the stream without one. So `Internal` here is not a
+/// transport failure but the server ending the subscription, because its
+/// source closed or could not be opened (a record buffer with no subscriber
+/// slot left, for example).
+fn sub_err(e: RpcError) -> ClientError {
+    match e {
+        RpcError::NotFound | RpcError::Denied => rpc_err(e),
+        _ => ClientError::server_error(
+            "internal",
+            "the server ended the subscription (its source closed or could not be opened)",
+            None,
+        ),
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -542,4 +560,42 @@ mod tests {
             "unexpected terminal error: {err}"
         );
     }
+
+    /// A subscription the server ended (`Internal` as its terminal item) says
+    /// so, rather than blaming the transport: a dropped connection would have
+    /// ended the stream without an error.
+    #[tokio::test]
+    async fn a_subscription_the_server_ended_says_so() {
+        let raw = raw_stream(vec![ok_update("1", 0), Err(RpcError::Internal)]);
+        let out: Vec<_> = decode_updates(raw).collect().await;
+
+        assert_eq!(out.len(), 2);
+        let err = out[1].as_ref().expect_err("the end reaches the caller");
+        assert!(
+            matches!(
+                err,
+                ClientError::ServerError { code, message, .. }
+                    if code == "internal" && message.contains("ended the subscription")
+            ),
+            "unexpected terminal error: {err}"
+        );
+    }
+
+    /// A call's `Internal` keeps naming the engine or transport: for calls the
+    /// client engine raises it itself when the connection drops.
+    #[test]
+    fn a_calls_internal_error_still_names_the_transport() {
+        let err = rpc_err(RpcError::Internal);
+        assert!(
+            matches!(
+                &err,
+                ClientError::ServerError { message, .. } if message == "engine/transport failure"
+            ),
+            "{err}"
+        );
+        assert!(matches!(
+            sub_err(RpcError::Denied),
+            ClientError::ServerError { code, .. } if code == "denied"
+        ));
+    }
 }

From 51c749248bf73cd88841ba9f395697b3c1bb2cf1 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Alexander=20Schn=C3=B6rch?=
 <alexander.schnoerch@outlook.com>
Date: Sat, 24 Oct 2026 18:10:30 +0200
Subject: [PATCH 6/6] docs(#328): changelogs and serial session docs for the
 session reset

- Changelogs: `aimdb-core` (session engine fixes, `Session::starts_session`),
  `aimdb-serial-connector`, `aimdb-client`, and the root changelog.
- `aimdb-serial-connector`: "Sessions on a line that never closes" in the
  crate docs: a client's `hello` starts a new session; a client that leaves
  without another connecting keeps its subscriptions until the next `hello`;
  size subscriber slots for remote access on top of a record's taps and links.
- `aimdb-core`: drop a redundant explicit doc link target on
  `Session::starts_session`, which failed `make doc` (`-D warnings`).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01YPDjQ947yKjLLAq3Xzgu3e
---
 CHANGELOG.md                            |  8 +++++++
 aimdb-client/CHANGELOG.md               | 10 +++++++++
 aimdb-core/CHANGELOG.md                 | 30 +++++++++++++++++++++++++
 aimdb-core/src/session/mod.rs           |  2 +-
 aimdb-serial-connector/CHANGELOG.md     | 10 +++++++++
 aimdb-serial-connector/src/connector.rs |  4 ++++
 aimdb-serial-connector/src/lib.rs       | 15 +++++++++++++
 7 files changed, 78 insertions(+), 1 deletion(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 8ee531d3..7b3cf4f6 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -76,6 +76,14 @@ is unchanged. ([aimdb-core](aimdb-core/CHANGELOG.md))
 
 ### Fixed
 
+- **AimX over serial: every reconnect gets its events.** A client's `hello`
+  on a live connection starts a new session, an ended subscription no longer
+  cancels a newer one with the same id, and a subscription whose source ended
+  is reported instead of staying silent (#328).
+  ([aimdb-core](aimdb-core/CHANGELOG.md),
+  [aimdb-serial-connector](aimdb-serial-connector/CHANGELOG.md),
+  [aimdb-client](aimdb-client/CHANGELOG.md))
+
 - **The embedded MQTT backend advertises the largest packet it receives**
   (`Maximum Packet Size` 3,328 in CONNECT), so an oversized retained message is
   withheld by the broker instead of reconnecting the client forever.
diff --git a/aimdb-client/CHANGELOG.md b/aimdb-client/CHANGELOG.md
index 502a829b..2de04294 100644
--- a/aimdb-client/CHANGELOG.md
+++ b/aimdb-client/CHANGELOG.md
@@ -7,6 +7,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Fixed
+
+- **A subscription the server ended no longer reads as a transport failure
+  (#328).** A subscription stream's terminal `Internal` error now says "the
+  server ended the subscription (its source closed or could not be opened)";
+  it used to say "engine/transport failure", although a dropped connection
+  ends a subscription stream without an error. Calls keep the old text, which
+  fits them: for a call, the client engine raises `Internal` itself when the
+  connection drops.
+
 ## [0.7.0] - 2026-09-18
 
 ### Changed
diff --git a/aimdb-core/CHANGELOG.md b/aimdb-core/CHANGELOG.md
index cf4558b4..d203fd98 100644
--- a/aimdb-core/CHANGELOG.md
+++ b/aimdb-core/CHANGELOG.md
@@ -129,6 +129,36 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   never got a value, a join ran without that input, and with default
   features nothing was logged.
 
+### Fixed
+
+- **Every client on a serial line gets its events, not every other one (#328).**
+  A UART never closes, so the AimX session engine saw every client behind it
+  as one session: a departed client's subscriptions kept streaming and held
+  their buffer subscriber slots, and the next client's subscribe collided with
+  them by id. Three changes in `run_session`:
+  - **A second handshake starts a new session.** `Session::starts_session`
+    (new, default `false`; `AimxSession`: `hello`) names a client's handshake.
+    When one arrives on a connection that already had one, the engine drops
+    every subscription pump (releasing its buffer subscriber), discards their
+    queued events and opens a fresh session before answering. This applies
+    to every transport; elsewhere a second `hello` on one connection does not
+    happen with a well-behaved client.
+  - **An ended pump only removes its own subscription.** Subscriptions carry
+    a generation, so a pump ending late (after `unsub`, then `sub` with the
+    same id) no longer cancels the newer subscription. A `sub` whose id is
+    still live is refused with `Denied` instead of silently replacing it.
+  - **A subscription whose stream ends on its own is reported.** When its
+    source closes, or never opened (an Embassy SPMC ring with no subscriber
+    slot left), the client gets the remaining events and then an `Internal`
+    error under the subscribe id, which it turns into a terminal error,
+    instead of an acknowledged subscription that never delivers. An
+    unsubscribed subscription, or the previous client's after a new
+    handshake, still ends in silence.
+
+  A client that leaves without another one connecting is still invisible:
+  its subscriptions run until the next handshake, since the engine has no
+  timer to detect a silent peer.
+
 ## [2.0.0] - 2026-09-18
 
 ### Added
diff --git a/aimdb-core/src/session/mod.rs b/aimdb-core/src/session/mod.rs
index 6adb6cf7..79a8caac 100644
--- a/aimdb-core/src/session/mod.rs
+++ b/aimdb-core/src/session/mod.rs
@@ -479,7 +479,7 @@ pub trait Session: Send {
     /// new session. Defaulted to `false`.
     ///
     /// When a handshake arrives on a connection that already had one,
-    /// [`run_session`](server::run_session) treats it as a new client: it
+    /// [`run_session`] treats it as a new client: it
     /// cancels every subscription, discards their queued events and opens a
     /// fresh session before answering. A transport that carries one peer for
     /// its whole life (a serial line) has no other way to tell that the client
diff --git a/aimdb-serial-connector/CHANGELOG.md b/aimdb-serial-connector/CHANGELOG.md
index f19db501..56e79f93 100644
--- a/aimdb-serial-connector/CHANGELOG.md
+++ b/aimdb-serial-connector/CHANGELOG.md
@@ -7,6 +7,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Fixed
+
+- **Reconnecting over serial works every time (#328).** Every other client
+  that connected to a `SerialServer` without a board reset got no events: the
+  previous client's subscriptions lived on in the one session a UART carries.
+  A client's `hello` now starts a new session, ending the previous client's
+  subscriptions (see `aimdb-core`). A client that leaves without another one
+  connecting keeps its subscriptions streaming into the UART until the next
+  `hello`.
+
 ## [0.1.0] - 2026-09-18
 
 ### Changed
diff --git a/aimdb-serial-connector/src/connector.rs b/aimdb-serial-connector/src/connector.rs
index 56b80776..39190604 100644
--- a/aimdb-serial-connector/src/connector.rs
+++ b/aimdb-serial-connector/src/connector.rs
@@ -134,6 +134,10 @@ impl SerialClient {
 }
 
 /// Serves AimX over a moved-in serial stream.
+///
+/// One session for the line's lifetime; a client's `hello` starts a new one,
+/// ending the previous client's subscriptions (see the crate docs, "Sessions
+/// on a line that never closes").
 pub struct SerialServer<S> {
     stream: OneShot<S>,
     config: AimxConfig,
diff --git a/aimdb-serial-connector/src/lib.rs b/aimdb-serial-connector/src/lib.rs
index 89860027..26e4050a 100644
--- a/aimdb-serial-connector/src/lib.rs
+++ b/aimdb-serial-connector/src/lib.rs
@@ -22,6 +22,21 @@
 //! host) and this crate contributes only the COBS framer. A UART is
 //! point-to-point, so the stream is moved in and served once.
 //!
+//! # Sessions on a line that never closes
+//!
+//! A UART has no connect or disconnect: the server serves one connection for
+//! the firmware's lifetime and cannot see a client leave. A client's `hello`
+//! is what marks a new one: the session engine then ends the previous
+//! client's session (its subscriptions stop and release their buffer slots)
+//! and answers the new client from a fresh one. So every client that connects
+//! gets its events, however often clients come and go.
+//!
+//! Until the next `hello`, a client that left keeps its subscriptions: they
+//! go on streaming into the UART and hold their buffer slots. Size the
+//! subscriber slots of a record with remote access for the subscribers a
+//! client opens (one per record it watches) on top of the record's own taps
+//! and links.
+//!
 //! Both speak the `serial://` scheme by default ([`DEFAULT_SCHEME`]).
 
 #![cfg_attr(not(feature = "std"), no_std)]