fix: record.get answers at once and no longer drains; MCP get/drain share one session (#444)

CHANGELOG.md

@@ -31,6 +31,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Fixed — `record.get` answers at once and no longer drains (#444)
+
+- **`record.get` is a single, non-destructive read.** `aimdb-mcp`'s
+  `get_record` drops its 3 s retry, the server drops the get-to-drain fallback,
+  and a ring answers `unsupported` instead of consuming events. See
+  [aimdb-core](aimdb-core/CHANGELOG.md),
+  [aimdb-client](aimdb-client/CHANGELOG.md) and
+  [aimdb-mcp](tools/aimdb-mcp/CHANGELOG.md).
+
 ### Removed — `examples/weather-mesh-demo`
 
 - **The in-repo weather mesh demo is gone.** The fleet behind aimdb.dev lives in

Makefile

@@ -243,6 +243,8 @@ test: test-operational-events test-build-info-link
 	cargo test --package aimdb-cli --features yaml
 	@printf "$(YELLOW)  → Testing MCP server$(NC)\n"
 	cargo test --package aimdb-mcp
+	@printf "$(YELLOW)  → Testing MCP server (serial: one session for get and drain)$(NC)\n"
+	cargo test --package aimdb-mcp --features "transport-serial" --test serial_session
 	@printf "$(YELLOW)  → Testing persistence backend$(NC)\n"
 	cargo test --package aimdb-persistence
 	@printf "$(YELLOW)  → Testing persistence SQLite backend$(NC)\n"

aimdb-client/CHANGELOG.md

@@ -7,6 +7,19 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Changed
+
+- **`get_record` on a ring returns `unsupported` (#444).** A ring has no
+  current value; use `drain_record` or `subscribe`. `get_record` no longer moves
+  the connection's drain cursor, so `drain_record` returns every unread event.
+  `watch::arm` treats `unsupported` like `not_found`: no baseline.
+
+### Added
+
+- **`AimxConnection::is_closed()` (#450):** whether the connection's engine has
+  stopped. A caller keeping a connection open uses it to decide when to
+  reconnect; an `internal` error reply does not mean the connection is gone.
+
 ### Removed (breaking)
 
 - **Obsolete AimX message types (#419).** `protocol::{Request, Response,

aimdb-client/src/engine.rs

@@ -231,6 +231,16 @@ impl AimxConnection {
         &self.server_info
     }
 
+    /// Whether this connection's engine has stopped: the transport closed or
+    /// failed. The engine does not reconnect, so every later call fails; a
+    /// caller that keeps a connection open replaces it.
+    ///
+    /// This is the reliable signal: an `internal` error can also be an
+    /// ordinary server reply on a healthy connection.
+    pub fn is_closed(&self) -> bool {
+        self.engine.is_finished()
+    }
+
     /// List all registered records.
     pub async fn list_records(&self) -> ClientResult<Vec<RecordMetadata>> {
         let reply = self.call("record.list", null_payload()).await?;
@@ -239,12 +249,14 @@ impl AimxConnection {
 
     /// Get a record's current value.
     ///
-    /// For a `SingleLatest`/state record this is a non-destructive read. A ring
-    /// (`SpmcRing`) has no canonical latest, so the server returns the most recent
-    /// value from this connection's drain cursor — which **advances that cursor**
-    /// (interleaving with [`drain_record`](Self::drain_record)) and is empty until
-    /// the ring produces a value after the connection first reads it. Prefer
-    /// [`drain_record`](Self::drain_record) for ring/history records.
+    /// A single, non-destructive read of the record's snapshot (`SingleLatest`,
+    /// `Mailbox`). It returns at once and never moves this connection's drain
+    /// cursor. Server errors:
+    ///
+    /// - `not_found`: no such record, or no value yet.
+    /// - `unsupported`: the record is a ring (`SpmcRing`), which has no current
+    ///   value. Use [`drain_record`](Self::drain_record) for unread events or
+    ///   [`subscribe`](Self::subscribe) for new ones.
     ///
     /// The value comes with the stamp of its write; a reply without `ts` is a
     /// [`ClientError::Protocol`].
@@ -571,6 +583,12 @@ fn rpc_err(e: RpcError) -> ClientError {
             ClientError::server_error("not_found", "method or record not found", None)
         }
         RpcError::Denied => ClientError::server_error("denied", "permission denied", None),
+        RpcError::Unsupported => ClientError::server_error(
+            "unsupported",
+            "not supported for this record (a ring has no current value: \
+             use drain_record for unread events or subscribe for new ones)",
+            None,
+        ),
         // `Internal` today, plus any future non-exhaustive variant.
         _ => ClientError::server_error("internal", "engine/transport failure", None),
     }

aimdb-client/src/watch.rs

@@ -84,8 +84,13 @@ pub async fn arm(conn: &AimxConnection, record: &str) -> Result<Armed, ClientErr
     let baseline = match conn.get_record(record).await {
         Ok(current) => Some(current.stamp()),
         // No value yet (or an unknown record, whose subscribe refusal is
-        // picked up below).
-        Err(ClientError::ServerError { code, .. }) if code == "not_found" => None,
+        // picked up below), or a ring: it has no current value, and its
+        // subscription starts after the last write, so nothing old can count.
+        Err(ClientError::ServerError { code, .. })
+            if code == "not_found" || code == "unsupported" =>
+        {
+            None
+        }
         Err(e) => return Err(e),
     };
     let stream = match stream.next().now_or_never() {

aimdb-client/tests/aimx_session.rs

@@ -326,10 +326,11 @@ async fn exact_subscribe_sends_the_current_value_once() {
     }
 }
 
-/// `record.get` on a ring (`SpmcRing`) record has no canonical latest, so it
-/// falls back to draining the connection's cursor for the most-recent value.
+/// `record.get` is one non-destructive read (#444): it answers at once for an
+/// empty record, an unknown name and a ring, and never moves the connection's
+/// drain cursor, so `record.drain` still returns every unread event.
 #[tokio::test]
-async fn record_get_on_ring_falls_back_to_drain() {
+async fn record_get_answers_at_once_and_leaves_drain_intact() {
     let dir = tempfile::tempdir().unwrap();
     let sock = dir.path().join("aimdb.sock");
 
@@ -341,35 +342,71 @@ async fn record_get_on_ring_falls_back_to_drain() {
         reg.buffer(BufferCfg::SpmcRing { capacity: 16 })
             .with_remote_access();
     });
+    builder.configure::<Reading>("state", |reg| {
+        reg.buffer(BufferCfg::SingleLatest).with_remote_access();
+    });
     let (db, runner) = builder.build().await.expect("build db");
     let db = Arc::new(db);
     tokio::spawn(runner.run());
 
     let conn = AimxConnection::connect(sock.to_str().unwrap())
         .await
         .expect("connect");
-    let producer = db.producer::<Reading>("stream").expect("producer");
 
-    // First get opens the cursor; a fresh broadcast reader starts at the tail, so
-    // it sees nothing until a value is produced afterwards.
-    assert!(conn.get_record("stream").await.is_err());
+    // Each miss answers at once (the old MCP retry took 3 s per read).
+    let code = |r: Result<_, aimdb_client::ClientError>| match r {
+        Err(aimdb_client::ClientError::ServerError { code, .. }) => code,
+        other => panic!("expected a server error, got {other:?}"),
+    };
+    let started = std::time::Instant::now();
+    assert_eq!(code(conn.get_record("state").await), "not_found", "empty");
+    assert_eq!(
+        code(conn.get_record("missing").await),
+        "not_found",
+        "unknown"
+    );
+    assert_eq!(code(conn.get_record("stream").await), "unsupported", "ring");
+    assert!(
+        started.elapsed() < Duration::from_secs(1),
+        "three misses took {:?}",
+        started.elapsed()
+    );
 
-    // Produce after the cursor is open; get now returns the most-recent value.
-    let mut got = None;
-    for n in 1..=50u64 {
-        producer.produce(Reading { n });
-        tokio::time::sleep(Duration::from_millis(10)).await;
-        if let Ok(v) = conn.get_record("stream").await {
-            got = Some(v);
+    // Open the drain cursors, then write.
+    assert_eq!(conn.drain_record("stream").await.expect("drain").count, 0);
+    assert_eq!(conn.drain_record("state").await.expect("drain").count, 0);
+    let ring = db.producer::<Reading>("stream").expect("producer");
+    let state = db.producer::<Reading>("state").expect("producer");
+    for n in 1..=3 {
+        ring.produce(Reading { n });
+    }
+    state.produce(Reading { n: 7 });
+
+    // Reads in between: the ring still refuses, the state answers, twice.
+    assert_eq!(code(conn.get_record("stream").await), "unsupported");
+    for _ in 0..2 {
+        let got = conn.get_record("state").await.expect("state get");
+        assert_eq!(got.value, serde_json::json!({ "n": 7 }));
+    }
+
+    // Drain still sees every event the gets did not consume.
+    let mut drained = Vec::new();
+    for _ in 0..50 {
+        drained.extend(conn.drain_record("stream").await.expect("drain").values);
+        if drained.len() >= 3 {
             break;
         }
+        tokio::time::sleep(Duration::from_millis(10)).await;
     }
-    let got = got.expect("ring get returns a value after producing");
-    assert!(
-        got.value.get("n").is_some(),
-        "ring get yields a Reading: {}",
-        got.value
-    );
+    let ns: Vec<_> = drained.iter().map(|v| v.value["n"].clone()).collect();
+    assert_eq!(ns, [1, 2, 3]);
+    let state_drain = conn.drain_record("state").await.expect("drain");
+    let ns: Vec<_> = state_drain
+        .values
+        .iter()
+        .map(|v| v.value["n"].clone())
+        .collect();
+    assert_eq!(ns, [7]);
 
     drop(conn);
 }

aimdb-core/CHANGELOG.md

@@ -7,6 +7,26 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Changed (wire)
+
+- **`record.get` is a single, non-destructive read (#444).** It returns the
+  buffer's own snapshot (`SingleLatest`, `Mailbox`) and never touches the
+  connection's drain cursor, so only `record.drain` advances it. The fallback
+  that drained a ring to answer `record.get` is gone: it made a point-in-time
+  read wait for a future write and consumed events a later `record.drain`
+  should have returned. A ring (`SpmcRing`) now answers with the new error
+  code `unsupported`; read it with `record.drain` (unread events) or
+  `record.subscribe` (new ones). An empty `SingleLatest`/`Mailbox` record and
+  an unknown name answer `not_found` at once. A ring registered with
+  `buffer_raw()` and no `buffer_cfg()` answers `not_found`, since core cannot
+  tell it is a ring.
+
+### Added
+
+- **`RpcError::Unsupported` (#444),** wire code `"unsupported"`: the method
+  exists but the target cannot serve it. Older clients decode an unknown code
+  as `internal`.
+
 ### Removed (breaking)
 
 - **Obsolete AimX message types (#419).** `remote::{Request, Response,

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

@@ -147,6 +147,7 @@ fn payload_of(raw: Option<&RawValue>) -> Payload {
 fn err_code(e: &RpcError) -> &'static str {
     match e {
         RpcError::NotFound => "not_found",
+        RpcError::Unsupported => "unsupported",
         RpcError::Denied => "denied",
         RpcError::Internal => "internal",
         RpcError::VersionMismatch => "version_mismatch",
@@ -158,6 +159,7 @@ fn err_code(e: &RpcError) -> &'static str {
 fn code_err(s: &str) -> RpcError {
     match s {
         "not_found" => RpcError::NotFound,
+        "unsupported" => RpcError::Unsupported,
         "denied" => RpcError::Denied,
         "version_mismatch" => RpcError::VersionMismatch,
         _ => RpcError::Internal,
@@ -509,6 +511,7 @@ mod tests {
     fn reply_err_codes_roundtrip() {
         for err in [
             RpcError::NotFound,
+            RpcError::Unsupported,
             RpcError::Denied,
             RpcError::Internal,
             RpcError::VersionMismatch,

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

@@ -319,25 +319,37 @@ impl AimxSession {
     /// `record.get`: the record's current value and its write stamp,
     /// `{"value": <value>, "ts": <ns>}`.
     ///
-    /// A `SingleLatest`/state record exposes a non-destructive canonical latest
-    /// ([`try_latest_as_json_bytes`](crate::AimDb::try_latest_as_json_bytes)).
-    /// A ring
-    /// ([`SpmcRing`](crate::buffer::BufferCfg::SpmcRing)) has none, so we fall back
-    /// to the connection's drain cursor and return the **most recent** available
-    /// value. Two consequences of that fallback: it *advances the shared drain
-    /// cursor* (so `record.get` and `record.drain` interleave on one connection),
-    /// and it yields `NotFound` until the ring produces a value *after* the cursor
-    /// is first opened (a fresh broadcast reader starts at the tail). Use
-    /// `record.drain` for a ring's full backlog.
-    fn record_get(&mut self, name: &str) -> Result<Payload, RpcError> {
+    /// A single, non-destructive read of the buffer's own snapshot
+    /// ([`try_latest_as_json_bytes`](crate::AimDb::try_latest_as_json_bytes)):
+    /// it returns at once and never touches this connection's drain cursor.
+    ///
+    /// - `NotFound`: no such record, or a record with a snapshot (`SingleLatest`,
+    ///   `Mailbox`) that holds no value yet.
+    /// - `Unsupported`: an [`SpmcRing`](crate::buffer::BufferCfg::SpmcRing). A
+    ///   ring is a stream with no snapshot; read it with `record.drain`
+    ///   (unread events) or `record.subscribe` (future events). An application
+    ///   that needs an inspectable current state exposes a `SingleLatest`
+    ///   record next to the stream.
+    ///
+    /// The ring check reads the declared [`BufferCfg`](crate::buffer::BufferCfg):
+    /// a ring registered with `buffer_raw()` and no `buffer_cfg()` reports
+    /// `NotFound`, because core cannot tell it is a ring.
+    fn record_get(&self, name: &str) -> Result<Payload, RpcError> {
         if let Some(sample) = self.db.try_latest_as_json_bytes(name) {
             return Ok(super::value_reply(&sample.value, sample.stamp()));
         }
-        // Ring fallback: drain to the newest currently-available value (or NotFound).
-        self.drain_values(name, usize::MAX)?
-            .pop()
-            .map(|sample| to_payload(&stamped_value_json(sample)))
-            .ok_or(RpcError::NotFound)
+        let inner = self.db.inner();
+        let record = inner
+            .resolve_str(name)
+            .and_then(|id| inner.storage(id))
+            .ok_or(RpcError::NotFound)?;
+        // Only the miss path pays for this lookup.
+        let ring = crate::buffer::BufferCfg::SpmcRing { capacity: 0 }.name();
+        if record.buffer_info().0 == ring {
+            Err(RpcError::Unsupported)
+        } else {
+            Err(RpcError::NotFound)
+        }
     }
 
     /// `record.drain`: return everything accumulated since the previous drain
@@ -359,9 +371,7 @@ impl AimxSession {
     }
 
     /// Lazily open (on first call) the per-record drain cursor and read up to
-    /// `limit` values accumulated since the previous read (oldest-first). Shared
-    /// by [`record.drain`](Self::record_drain) and [`record.get`](Self::record_get)'s
-    /// ring fallback, so both read from the same per-connection cursor.
+    /// `limit` values accumulated since the previous read (oldest-first).
     fn drain_values(&mut self, name: &str, limit: usize) -> Result<Vec<Stamped<Value>>, RpcError> {
         if !self.drain_readers.contains_key(name) {
             let id = self

aimdb-core/src/session/mod.rs

@@ -270,6 +270,9 @@ pub enum CodecError {
 pub enum RpcError {
     /// No such method or topic.
     NotFound,
+    /// The method exists but the target cannot serve it, e.g. `record.get` on
+    /// a ring buffer, which has no current value to read.
+    Unsupported,
     /// The caller lacks permission for this operation.
     Denied,
     /// The handler failed.

aimdb-wasm-adapter/src/ws_bridge.rs

@@ -840,6 +840,7 @@ impl WsBridge {
 fn rpc_err_str(e: &RpcError) -> &'static str {
     match e {
         RpcError::NotFound => "not_found",
+        RpcError::Unsupported => "unsupported (a ring has no current value)",
         RpcError::Denied => "denied",
         RpcError::VersionMismatch => {
             "version_mismatch (server rejected the client protocol version)"