feat(session): subscribe-ack + pump_client record mirroring (AimxClientConnector)

Phase 3 server port (issue #39), client refinements:

- Subscribe-ack (run_client): a server subscribe-failure Reply carries an id the
  client never registered as a pending call. Recognize it and drop the matching
  event sink so a rejected subscribe ends the stream (None) instead of hanging
  forever. Contract: success is acknowledged implicitly by events flowing; the
  server replies only on failure. Covered by a new session_engine test.

- pump_client(db, scheme, handle): mirrors records both directions over a running
  run_client engine — outbound routes stream local updates via ClientHandle::write;
  inbound routes subscribe and produce into local records through the Router
  (arbiter path; single-writer-per-key intact). Uses the type-erased
  consumer/serializer + producer/deserializer machinery (db.runtime_any() supplies
  the ctx for context-aware (de)serializers).

- AimxClientConnector: a ConnectorBuilder for the `aimx://` scheme so records can
  .link_to/.link_from an AimX peer; on build it dials via run_client and returns
  the pump futures + engine future for the runner (spawn-free). The registerable
  wrapper around pump_client, mirroring build_aimx_server on the server side.

New aimdb-client test mirrors a record client->server and server->client through
the real connector path.

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

Author: lxsaah

aimdb-client/tests/pump_client.rs

@@ -0,0 +1,130 @@
+//! Phase 3 server-port exit criterion (§4): `pump_client` mirrors a record
+//! **both directions** between a local AimDb and a remote AimDb over the shared
+//! session engine.
+//!
+//! Topology: a server `AimDb` (served by `build_aimx_server`) and a client
+//! `AimDb` whose records carry `aimx://` connector links. `run_client` opens the
+//! connection; `pump_client` wires the client's outbound/inbound routes to the
+//! `ClientHandle`:
+//! - **client → server**: producing the client's `cfg` record streams it to the
+//!   server via `ClientHandle::write` → the server's `record.set` path.
+//! - **server → client**: updating the server's `tele` record streams it back
+//!   through a subscription → the client's inbound producer (arbiter path).
+
+use std::sync::Arc;
+use std::time::Duration;
+
+use aimdb_core::buffer::BufferCfg;
+use aimdb_core::remote::{AimxConfig, SecurityPolicy};
+use aimdb_core::session::aimx::{build_aimx_server, AimxClientConnector};
+use aimdb_core::session::ClientConfig;
+use aimdb_core::{AimDb, AimDbBuilder};
+use aimdb_tokio_adapter::{TokioAdapter, TokioRecordRegistrarExt};
+use serde::{Deserialize, Serialize};
+use serde_json::json;
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
+struct Msg {
+    v: u64,
+}
+
+/// Re-assert `db.<key>` reaches `want`, re-driving `push` each tick so the test
+/// is robust against subscription-registration timing (a fresh subscriber may
+/// only see values produced after it attaches).
+async fn mirror_reaches(
+    db: &Arc<AimDb<TokioAdapter>>,
+    key: &str,
+    want: &serde_json::Value,
+    mut push: impl FnMut(),
+) -> bool {
+    for _ in 0..100 {
+        push();
+        tokio::time::sleep(Duration::from_millis(20)).await;
+        if db.try_latest_as_json(key).as_ref() == Some(want) {
+            return true;
+        }
+    }
+    false
+}
+
+#[tokio::test]
+async fn pump_client_mirrors_record_both_directions() {
+    let dir = tempfile::tempdir().unwrap();
+    let sock = dir.path().join("aimdb.sock");
+
+    // --- server: cfg (writable target) + tele (streamed source) ------------
+    let mut sb = AimDbBuilder::new().runtime(Arc::new(TokioAdapter));
+    sb.configure::<Msg>("cfg", |reg| {
+        reg.buffer(BufferCfg::SingleLatest).with_remote_access();
+    });
+    sb.configure::<Msg>("tele", |reg| {
+        reg.buffer(BufferCfg::SingleLatest).with_remote_access();
+    });
+    let (server_db, server_runner) = sb.build().await.expect("build server db");
+    let server_db = Arc::new(server_db);
+    tokio::spawn(server_runner.run());
+
+    let mut policy = SecurityPolicy::read_write();
+    policy.allow_write_key("cfg");
+    let config = AimxConfig::uds_default()
+        .socket_path(&sock)
+        .security_policy(policy);
+    let server = tokio::spawn(build_aimx_server(server_db.clone(), config).expect("bind server"));
+
+    // --- client: cfg links *to* the server, tele links *from* it -----------
+    // The AimxClientConnector registers the `aimx://` scheme (so the links
+    // validate) and, on build, dials the server + drives the mirroring pumps.
+    let mut cb = AimDbBuilder::new()
+        .runtime(Arc::new(TokioAdapter))
+        .with_connector(AimxClientConnector::new(&sock).with_config(ClientConfig {
+            reconnect: true,
+            reconnect_delay: Duration::from_millis(50),
+            sends_hello: false,
+        }));
+    cb.configure::<Msg>("cfg", |reg| {
+        reg.buffer(BufferCfg::SingleLatest)
+            .with_remote_access()
+            .link_to("aimx://cfg")
+            .with_serializer_raw(|m: &Msg| Ok(serde_json::to_vec(m).expect("serialize")))
+            .finish();
+    });
+    cb.configure::<Msg>("tele", |reg| {
+        reg.buffer(BufferCfg::SingleLatest)
+            .with_remote_access()
+            .link_from("aimx://tele")
+            .with_deserializer_raw(|d: &[u8]| {
+                serde_json::from_slice::<Msg>(d).map_err(|e| e.to_string())
+            })
+            .finish();
+    });
+    // build() collects the connector's engine + pump futures; the runner drives
+    // them (spawn-free engine, driven on a task here).
+    let (client_db, client_runner) = cb.build().await.expect("build client db");
+    let client_db = Arc::new(client_db);
+    tokio::spawn(client_runner.run());
+
+    // client → server: producing client `cfg` mirrors to server `cfg`.
+    let want_cfg = json!({ "v": 7 });
+    let mirrored_out = mirror_reaches(&server_db, "cfg", &want_cfg, || {
+        client_db
+            .set_record_from_json("cfg", json!({ "v": 7 }))
+            .expect("set client cfg");
+    })
+    .await;
+    assert!(
+        mirrored_out,
+        "client→server mirror did not reach the server"
+    );
+
+    // server → client: updating server `tele` mirrors to client `tele`.
+    let want_tele = json!({ "v": 9 });
+    let mirrored_in = mirror_reaches(&client_db, "tele", &want_tele, || {
+        server_db
+            .set_record_from_json("tele", json!({ "v": 9 }))
+            .expect("set server tele");
+    })
+    .await;
+    assert!(mirrored_in, "server→client mirror did not reach the client");
+
+    server.abort();
+}

aimdb-core/src/session/aimx/client_connector.rs

@@ -0,0 +1,74 @@
+//! AimX **client** connector (Phase 3 server port, std-only) — registers the
+//! `aimx://` scheme so records can `.link_to`/`.link_from` an AimX peer, and on
+//! build dials that peer and drives the mirroring pumps.
+//!
+//! This is the registerable wrapper around [`pump_client`](crate::session::pump_client):
+//! `build` opens the connection with [`run_client`](crate::session::run_client)
+//! and returns one spawn-free future per route (plus the engine future) for the
+//! runner to drive — collapsing the AimX *client* onto the shared session engine
+//! the same way [`build_aimx_server`](super::build_aimx_server) does the server.
+
+use std::future::Future;
+use std::path::PathBuf;
+use std::pin::Pin;
+
+use crate::builder::AimDb;
+use crate::connector::ConnectorBuilder;
+use crate::session::{pump_client, run_client, ClientConfig};
+use crate::{DbResult, RuntimeAdapter};
+
+use super::{AimxCodec, UdsDialer};
+
+type BoxFuture = Pin<Box<dyn Future<Output = ()> + Send + 'static>>;
+
+/// A connector that mirrors records to/from an AimX peer over a Unix-domain
+/// socket. Register it with [`AimDbBuilder::with_connector`](crate::AimDbBuilder::with_connector)
+/// so `aimx://<record>` links validate; its `build` wires every collected route
+/// to the connection.
+pub struct AimxClientConnector {
+    socket_path: PathBuf,
+    config: ClientConfig,
+}
+
+impl AimxClientConnector {
+    /// Mirror records over the AimX peer listening at `socket_path`.
+    pub fn new(socket_path: impl Into<PathBuf>) -> Self {
+        Self {
+            socket_path: socket_path.into(),
+            config: ClientConfig::default(),
+        }
+    }
+
+    /// Override the client engine config (reconnect policy, etc.).
+    pub fn with_config(mut self, config: ClientConfig) -> Self {
+        self.config = config;
+        self
+    }
+}
+
+impl<R> ConnectorBuilder<R> for AimxClientConnector
+where
+    R: RuntimeAdapter + 'static,
+{
+    fn build<'a>(
+        &'a self,
+        db: &'a AimDb<R>,
+    ) -> Pin<Box<dyn Future<Output = DbResult<Vec<BoxFuture>>> + Send + 'a>> {
+        Box::pin(async move {
+            let (handle, engine_fut) = run_client(
+                UdsDialer::new(self.socket_path.clone()),
+                AimxCodec,
+                self.config.clone(),
+            );
+            // One pump future per route; they hold `ClientHandle` clones, so the
+            // engine stays alive as long as any mirror runs. `handle` drops here.
+            let mut futures = pump_client(db, "aimx", &handle);
+            futures.push(engine_fut);
+            Ok(futures)
+        })
+    }
+
+    fn scheme(&self) -> &str {
+        "aimx"
+    }
+}

aimdb-core/src/session/aimx/mod.rs

@@ -6,10 +6,12 @@
 //! and the accepting transport ([`UdsListener`]) + server [`AimxDispatch`] that
 //! [`build_aimx_server`] drives via `serve`.
 
+mod client_connector;
 mod codec;
 mod dispatch;
 mod transport;
 
+pub use client_connector::AimxClientConnector;
 pub use codec::AimxCodec;
 pub use dispatch::{build_aimx_server, AimxDispatch};
 pub use transport::{UdsConnection, UdsDialer, UdsListener};

aimdb-core/src/session/client.rs

@@ -16,13 +16,17 @@
 //! it never spawns.
 
 use std::collections::HashMap;
+use std::sync::Arc;
 use std::time::Duration;
 
 use tokio::sync::{mpsc, oneshot};
 
 use super::{
     BoxFut, BoxStream, Connection, Dialer, EnvelopeCodec, Inbound, Outbound, Payload, RpcError,
 };
+use crate::connector::SerializerKind;
+use crate::router::RouterBuilder;
+use crate::{AimDb, RuntimeAdapter};
 
 /// Client engine knobs.
 #[derive(Debug, Clone)]
@@ -239,6 +243,15 @@ where
                     Ok(Outbound::Reply { id, result }) => {
                         if let Some(tx) = pending.remove(&id) {
                             let _ = tx.send(result);
+                        } else if result.is_err() {
+                            // Subscribe-ack contract: a successful subscribe is
+                            // acknowledged implicitly by its events flowing; the
+                            // server replies only on *failure* (unknown record,
+                            // sub cap). Such a Reply carries the subscribe `id`,
+                            // which was never registered as a pending call — so
+                            // drop the matching event sink to end the stream
+                            // (`None`) instead of leaving it hanging forever.
+                            subs.remove(&id.to_string());
                         }
                     }
                     Ok(Outbound::Event { sub, seq: _, data }) => {
@@ -311,3 +324,87 @@ where
         }
     }
 }
+
+/// Mirror records between a local [`AimDb`] and a remote peer over a running
+/// [`run_client`] engine — the connector-link half of the client capability.
+///
+/// For the given connector `scheme` (e.g. `"aimx"`):
+/// - **outbound** routes (`db.collect_outbound_routes`) stream local record
+///   updates to the remote via [`ClientHandle::write`];
+/// - **inbound** routes (`db.collect_inbound_routes`) subscribe to the remote and
+///   produce each update into the local record through the producer/arbiter path
+///   — single-writer-per-key stays intact (a mirrored-in record is produced
+///   through its inbound producer, never a direct co-writer).
+///
+/// Returns one spawn-free pump future per route for the runner to drive
+/// (mirroring the `ConnectorBuilder::build -> Vec<BoxFuture>` spine); it drives
+/// the **same** engine as [`run_client`], never a second one.
+pub fn pump_client<R>(
+    db: &AimDb<R>,
+    scheme: &str,
+    handle: &ClientHandle,
+) -> Vec<BoxFut<'static, ()>>
+where
+    R: RuntimeAdapter + 'static,
+{
+    use futures_util::StreamExt;
+
+    // The type-erased runtime context for context-aware (de)serializers.
+    let ctx = db.runtime_any();
+    let mut pumps: Vec<BoxFut<'static, ()>> = Vec::new();
+
+    // --- outbound: local record updates -> remote `write` ------------------
+    for (destination, consumer, serializer, _config, topic_provider) in
+        db.collect_outbound_routes(scheme)
+    {
+        let handle = handle.clone();
+        let ctx = ctx.clone();
+        pumps.push(Box::pin(async move {
+            let mut reader = match consumer.subscribe_any().await {
+                Ok(r) => r,
+                Err(_e) => return,
+            };
+            while let Ok(value) = reader.recv_any().await {
+                // Dynamic destination (topic provider) or the static link target.
+                let dest = topic_provider
+                    .as_ref()
+                    .and_then(|p| p.topic_any(&*value))
+                    .unwrap_or_else(|| destination.clone());
+                let bytes = match &serializer {
+                    SerializerKind::Raw(ser) => match ser(&*value) {
+                        Ok(b) => b,
+                        Err(_e) => continue,
+                    },
+                    SerializerKind::Context(ser) => match ser(ctx.clone(), &*value) {
+                        Ok(b) => b,
+                        Err(_e) => continue,
+                    },
+                };
+                if handle.write(dest, Payload::from(bytes.as_slice())).is_err() {
+                    break; // engine stopped — all handles dropped
+                }
+            }
+        }));
+    }
+
+    // --- inbound: remote events -> local producer (via the Router) ---------
+    // The Router applies each route's deserializer and produces the value; one
+    // subscription per unique remote topic feeds it.
+    let router = Arc::new(RouterBuilder::from_routes(db.collect_inbound_routes(scheme)).build());
+    for id in router.resource_ids() {
+        let handle = handle.clone();
+        let router = router.clone();
+        let ctx = ctx.clone();
+        pumps.push(Box::pin(async move {
+            let mut stream = match handle.subscribe(id.as_ref()) {
+                Ok(s) => s,
+                Err(_e) => return,
+            };
+            while let Some(payload) = stream.next().await {
+                let _ = router.route(id.as_ref(), &payload, Some(&ctx)).await;
+            }
+        }));
+    }
+
+    pumps
+}

aimdb-core/src/session/mod.rs

@@ -46,7 +46,7 @@ mod server;
 pub mod aimx;
 
 #[cfg(feature = "std")]
-pub use client::{run_client, ClientConfig, ClientHandle};
+pub use client::{pump_client, run_client, ClientConfig, ClientHandle};
 #[cfg(feature = "std")]
 pub use server::{run_session, serve, SessionConfig};