feat(session): split Dispatch into shared Dispatch + per-connection Session

Phase 3 server-port prep (issue #39). The AimX wire reshape dissolved every
seam except one: record.drain needs lazy per-connection cursors, but Dispatch
is a shared Arc<D> with &self — nowhere for mutable per-connection state.

Split the dispatch role:
- Dispatch (Send + Sync, one Arc per server): authenticate + open() factory.
- Session (Send, one Box per accepted connection): call/subscribe/write on
  &mut self. run_session owns the Box<dyn Session> and threads &mut into it.
  subscribe is defaulted to NotFound (its 'static stream is side-neutral).

Additive + object-safe (mirrors the Phase-2 encode_inbound/decode_outbound
precedent); recorded in 037. Engine round-trip + AimX client tests still green;
contracts still cross-compile to thumbv7em.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: lxsaah

aimdb-client/tests/aimx_session.rs

@@ -14,7 +14,7 @@ use aimdb_client::AimxConnection;
 use aimdb_core::session::aimx::{AimxCodec, UdsConnection};
 use aimdb_core::session::{
     serve, AuthError, BoxFut, BoxStream, Connection, Dispatch, Listener, Payload, PeerInfo,
-    RpcError, SessionConfig, SessionCtx, TransportError, TransportResult,
+    RpcError, Session, SessionConfig, SessionCtx, TransportError, TransportResult,
 };
 use serde_json::json;
 use tokio::net::UnixListener;
@@ -59,9 +59,21 @@ impl Dispatch for TestDispatch {
         Box::pin(async { Ok(SessionCtx::default()) })
     }
 
+    fn open(&self, _ctx: &SessionCtx) -> Box<dyn Session> {
+        Box::new(TestSession {
+            writes: self.writes.clone(),
+        })
+    }
+}
+
+/// Per-connection session for the test dispatch.
+struct TestSession {
+    writes: WriteLog,
+}
+
+impl Session for TestSession {
     fn call<'a>(
-        &'a self,
-        _ctx: &'a SessionCtx,
+        &'a mut self,
         method: &'a str,
         params: Payload,
     ) -> BoxFut<'a, Result<Payload, RpcError>> {
@@ -91,11 +103,7 @@ impl Dispatch for TestDispatch {
         })
     }
 
-    fn subscribe(
-        &self,
-        _ctx: &SessionCtx,
-        topic: &str,
-    ) -> Result<BoxStream<'static, Payload>, RpcError> {
+    fn subscribe(&mut self, topic: &str) -> Result<BoxStream<'static, Payload>, RpcError> {
         // Three synthetic updates derived from the topic, then end.
         let items: Vec<Payload> = (1..=3)
             .map(|i| payload(json!({ "topic": topic, "n": i })))
@@ -104,8 +112,7 @@ impl Dispatch for TestDispatch {
     }
 
     fn write<'a>(
-        &'a self,
-        _ctx: &'a SessionCtx,
+        &'a mut self,
         topic: &'a str,
         payload: Payload,
     ) -> BoxFut<'a, Result<(), RpcError>> {

aimdb-core/src/session/mod.rs

@@ -260,12 +260,24 @@ pub trait Dialer: Send {
 
 // ===========================================================================
 // Layer 3 — dispatch (the semantics). Doc 034 § Layer 3 + § EnvelopeCodec.
-// RPC and streaming unify in ONE trait (Decision 2): three reply cardinalities
-// — `call` (one) / `subscribe` (many) / `write` (none).
+// RPC and streaming unify in ONE per-connection role (Decision 2): three reply
+// cardinalities — `call` (one) / `subscribe` (many) / `write` (none).
+//
+// The role is split across two traits so the shared, immutable half (one
+// `Arc<dyn Dispatch>` per server) and the per-connection mutable half (one
+// `Box<dyn Session>` per accepted connection) each own what they need:
+//
+// - [`Dispatch`] — `Send + Sync`, shared: `authenticate` + an `open` factory.
+// - [`Session`]  — `Send`, per-connection: `call` / `subscribe` / `write` on
+//   `&mut self`, so a connection can hold mutable state (e.g. `record.drain`'s
+//   lazy per-record cursors — the one seam the AimX wire reshape did not
+//   dissolve) without a lock. See doc 037 (the additive server-port refinement,
+//   mirroring the Phase-2 `encode_inbound`/`decode_outbound` precedent).
 // ===========================================================================
 
-/// The application dispatch: authenticate a session, then serve calls,
-/// subscriptions, and writes against a [`SessionCtx`].
+/// The shared application dispatch: authenticate a connection, then open a
+/// per-connection [`Session`]. One `Arc<dyn Dispatch>` is shared across every
+/// connection a server accepts, so it stays `Sync` and behind `&self`.
 pub trait Dispatch: Send + Sync {
     /// Resolve a [`SessionCtx`] from peer metadata and/or the first frame
     /// (WS supplies pre-resolved identity via [`PeerInfo`]; UDS reads a Hello).
@@ -275,28 +287,42 @@ pub trait Dispatch: Send + Sync {
         first: Option<&'a [u8]>,
     ) -> BoxFut<'a, Result<SessionCtx, AuthError>>;
 
+    /// Open the per-connection [`Session`] once, after [`authenticate`]. The
+    /// returned session owns the connection's mutable dispatch state (drain
+    /// cursors today, per-session auth identity in Phase 4) that the shared
+    /// `Arc<Self>` cannot hold behind `&self`; the engine threads `&mut` into it.
+    fn open(&self, ctx: &SessionCtx) -> Box<dyn Session>;
+}
+
+/// The per-connection session: serves calls, subscriptions, and writes for one
+/// accepted [`Connection`]. The engine ([`run_session`]) owns the
+/// `Box<dyn Session>` and threads `&mut self` into each method, so a session can
+/// hold per-connection mutable state without a lock — while the shared,
+/// immutable role stays on [`Dispatch`].
+pub trait Session: Send {
     /// One-shot RPC: one request → one reply.
     fn call<'a>(
-        &'a self,
-        ctx: &'a SessionCtx,
+        &'a mut self,
         method: &'a str,
         params: Payload,
     ) -> BoxFut<'a, Result<Payload, RpcError>>;
 
     /// Streaming: open a subscription that yields many payloads. The stream is
-    /// `'static` so it can hold its own buffer reader inside the engine's
-    /// `FuturesUnordered` (doc 034 risk list).
-    fn subscribe(
-        &self,
-        ctx: &SessionCtx,
-        topic: &str,
-    ) -> Result<BoxStream<'static, Payload>, RpcError>;
+    /// `'static` (it captures cloned handles), so it outlives the `&mut` borrow
+    /// and lives in the engine's `FuturesUnordered` (doc 034 risk list).
+    ///
+    /// Defaulted to [`RpcError::NotFound`] so a dispatch with no streaming
+    /// surface need not implement it (doc 037 § the server-port refinement —
+    /// the stream is side-neutral, so it is defaulted here for symmetry).
+    fn subscribe(&mut self, topic: &str) -> Result<BoxStream<'static, Payload>, RpcError> {
+        let _ = topic;
+        Err(RpcError::NotFound)
+    }
 
     /// Fire-and-forget write: no reply. Routes through the existing
     /// producer/arbiter path (single-writer-per-key stays intact).
     fn write<'a>(
-        &'a self,
-        ctx: &'a SessionCtx,
+        &'a mut self,
         topic: &'a str,
         payload: Payload,
     ) -> BoxFut<'a, Result<(), RpcError>>;
@@ -368,12 +394,13 @@ pub trait Source: Send {
 // `Box<dyn Trait>` from a mock per the acceptance criteria.
 // ===========================================================================
 
-#[allow(dead_code)]
+#[allow(dead_code, clippy::too_many_arguments)]
 fn _assert_object_safe(
     _connection: &dyn Connection,
     _listener: &dyn Listener,
     _dialer: &dyn Dialer,
     _dispatch: &dyn Dispatch,
+    _session: &dyn Session,
     _codec: &dyn EnvelopeCodec,
     _sink: &dyn Sink,
     _source: &dyn Source,
@@ -420,24 +447,25 @@ mod tests {
         ) -> BoxFut<'a, Result<SessionCtx, AuthError>> {
             unimplemented!()
         }
+        fn open(&self, _ctx: &SessionCtx) -> Box<dyn Session> {
+            unimplemented!()
+        }
+    }
+
+    struct MockSession;
+    impl Session for MockSession {
         fn call<'a>(
-            &'a self,
-            _ctx: &'a SessionCtx,
+            &'a mut self,
             _method: &'a str,
             _params: Payload,
         ) -> BoxFut<'a, Result<Payload, RpcError>> {
             unimplemented!()
         }
-        fn subscribe(
-            &self,
-            _ctx: &SessionCtx,
-            _topic: &str,
-        ) -> Result<BoxStream<'static, Payload>, RpcError> {
+        fn subscribe(&mut self, _topic: &str) -> Result<BoxStream<'static, Payload>, RpcError> {
             unimplemented!()
         }
         fn write<'a>(
-            &'a self,
-            _ctx: &'a SessionCtx,
+            &'a mut self,
             _topic: &'a str,
             _payload: Payload,
         ) -> BoxFut<'a, Result<(), RpcError>> {
@@ -487,6 +515,7 @@ mod tests {
         let _listener: Box<dyn Listener> = Box::new(MockListener);
         let _dialer: Box<dyn Dialer> = Box::new(MockDialer);
         let _dispatch: Box<dyn Dispatch> = Box::new(MockDispatch);
+        let _session: Box<dyn Session> = Box::new(MockSession);
         let _codec: Box<dyn EnvelopeCodec> = Box::new(MockCodec);
         let _sink: Box<dyn Sink> = Box::new(MockSink);
         let _source: Box<dyn Source> = Box::new(MockSource);

aimdb-core/src/session/server.rs

@@ -79,6 +79,11 @@ pub async fn run_session<C, D>(
         }
     };
 
+    // Open the per-connection session once. It owns the connection's mutable
+    // dispatch state (e.g. `record.drain` cursors); the loop below threads
+    // `&mut` into its `call` / `subscribe` / `write`.
+    let mut session = dispatch.open(&ctx);
+
     // Event funnel: every per-subscription pump sends its updates here; the main
     // loop is the sole writer to the connection.
     let (event_tx, mut event_rx) = mpsc::unbounded_channel::<SubEvent>();
@@ -107,7 +112,7 @@ pub async fn run_session<C, D>(
                 };
                 match msg {
                     Inbound::Request { id, method, params } => {
-                        let result = dispatch.call(&ctx, &method, params).await;
+                        let result = session.call(&method, params).await;
                         out.clear();
                         if codec.encode(Outbound::Reply { id, result }, &mut out).is_err() {
                             continue;
@@ -124,7 +129,7 @@ pub async fn run_session<C, D>(
                             send_reply_err(&mut conn, codec, &mut out, id, RpcError::Denied).await;
                             continue;
                         }
-                        match dispatch.subscribe(&ctx, &topic) {
+                        match session.subscribe(&topic) {
                             Ok(stream) => {
                                 let (cancel_tx, cancel_rx) = oneshot::channel();
                                 cancels.insert(sub_id.clone(), cancel_tx);
@@ -145,8 +150,8 @@ pub async fn run_session<C, D>(
                         cancels.remove(&sub);
                     }
                     Inbound::Write { topic, payload } => {
-                        // Fire-and-forget; routes through Dispatch (single-writer-per-key intact).
-                        let _ = dispatch.write(&ctx, &topic, payload).await;
+                        // Fire-and-forget; routes through the session (single-writer-per-key intact).
+                        let _ = session.write(&topic, payload).await;
                     }
                     Inbound::Ping => {
                         out.clear();
@@ -208,7 +213,7 @@ async fn send_reply_err<C: EnvelopeCodec + ?Sized>(
     }
 }
 
-/// Pump one `Dispatch::subscribe` stream into the connection's event funnel,
+/// Pump one `Session::subscribe` stream into the connection's event funnel,
 /// tagging each update with a monotonic `seq`. Ends when the stream finishes or
 /// the cancel handle is dropped/fired (Unsubscribe or connection teardown).
 async fn pump_subscription(

aimdb-core/tests/session_engine.rs

@@ -18,7 +18,7 @@ use futures::StreamExt;
 
 use aimdb_core::session::{
     run_client, serve, AuthError, BoxFut, BoxStream, ClientConfig, CodecError, Connection, Dialer,
-    Dispatch, EnvelopeCodec, Inbound, Listener, Outbound, Payload, PeerInfo, RpcError,
+    Dispatch, EnvelopeCodec, Inbound, Listener, Outbound, Payload, PeerInfo, RpcError, Session,
     SessionConfig, SessionCtx, TransportError, TransportResult,
 };
 
@@ -257,21 +257,30 @@ impl Dispatch for EchoDispatch {
         Box::pin(async { Ok(SessionCtx::default()) })
     }
 
+    fn open(&self, _ctx: &SessionCtx) -> Box<dyn Session> {
+        Box::new(EchoSession {
+            writes: self.writes.clone(),
+        })
+    }
+}
+
+/// Per-connection echo session — the shared `EchoDispatch` clones its write log
+/// into each one at `open` time.
+struct EchoSession {
+    writes: WriteLog,
+}
+
+impl Session for EchoSession {
     fn call<'a>(
-        &'a self,
-        _ctx: &'a SessionCtx,
+        &'a mut self,
         _method: &'a str,
         params: Payload,
     ) -> BoxFut<'a, Result<Payload, RpcError>> {
         // Echo the params straight back.
         Box::pin(async move { Ok(params) })
     }
 
-    fn subscribe(
-        &self,
-        _ctx: &SessionCtx,
-        topic: &str,
-    ) -> Result<BoxStream<'static, Payload>, RpcError> {
+    fn subscribe(&mut self, topic: &str) -> Result<BoxStream<'static, Payload>, RpcError> {
         // Three synthetic updates derived from the topic, then end.
         let items: Vec<Payload> = (1..=3)
             .map(|i| payload_from(&format!("{topic}#{i}")))
@@ -280,8 +289,7 @@ impl Dispatch for EchoDispatch {
     }
 
     fn write<'a>(
-        &'a self,
-        _ctx: &'a SessionCtx,
+        &'a mut self,
         topic: &'a str,
         payload: Payload,
     ) -> BoxFut<'a, Result<(), RpcError>> {