71 record history api (#72)

* add design document

* feat: add record draining functionality to AimDB client and server

- Implemented `drain_record` and `drain_record_with_limit` methods in AimxClient for draining values from records.
- Added `DrainResponse` struct to encapsulate the response from drain calls.
- Updated connection handler to manage drain readers for records.
- Enhanced buffer traits with non-blocking `try_recv` method for various buffer types.
- Modified existing record configuration methods to use `.with_remote_access()` instead of `.with_serialization()`.
- Introduced `drain_record` tool in MCP server for remote access to drained values.
- Updated documentation and examples to reflect new draining capabilities.

* feat: add drain integration tests for AimX protocol and update dependencies

* style: format assertions for clarity in buffer and drain integration tests

* chore: update subproject commit for embassy dependency

* fix: update stm32-metapac source URL and adjust button handler type for async mode

* Refactor MCP server and remove subscription management

- Removed the SubscriptionManager and related subscription handling from the MCP server.
- Updated the MCP server initialization to eliminate notification directory and subscription manager dependencies.
- Cleaned up the Notification struct by removing unused notification creation methods.
- Removed subscription-related tools and tests, including subscribe_record, unsubscribe_record, and list_subscriptions.
- Adjusted server capabilities to disable resource subscriptions.
- Updated relevant documentation and comments to reflect the removal of subscription features.

* feat: implement record drain functionality and enhance connection management

* implement transform api

* chore: update documentation for database builder and transform modules

* feat(graph): introduce dependency graph module and introspection tools

- Added a new `graph` module to the `aimdb-core` crate, implementing the `DependencyGraph` structure and related types for managing and introspecting the database topology.
- Implemented methods for building and validating the dependency graph, including cycle detection and topological sorting.
- Created new tools in the `aimdb-mcp` crate for graph introspection: `graph_nodes`, `graph_edges`, and `graph_topo_order`, allowing users to retrieve metadata about records and their relationships.
- Extended the AimX protocol with new methods for graph introspection, including `graph.describe`, `graph.lineage`, and `graph.dot`, providing remote clients with visibility into data flow and dependencies.
- Updated documentation to reflect the new graph features and their usage in the AimDB ecosystem.

* feat(graph): add commands for dependency graph introspection and output formatting

* feat: add record drain and graph introspection APIs, enhance changelog documentation

* chore: update subproject commit for embassy

* chore: update stm32-metapac source URL to latest tag

* Apply suggestion from @Copilot

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

* Apply suggestion from @Copilot

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

* feat: enhance error handling in buffer reader and improve limit parameter conversion

---------

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

Author: lxsaah

.vscode/mcp.json

@@ -2,7 +2,7 @@
     "servers": {
         "aimdb": {
             "type": "stdio",
-            "command": "/aimdb/target/debug/aimdb-mcp",
+            "command": "/aimdb_ws/aimdb/target/release/aimdb-mcp",
             "args": [],
             "env": {
                 "RUST_LOG": "info"

Cargo.lock

@@ -7,7 +7,21 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
-No changes yet.
+### Added
+
+- **Record Drain API**: New methods for batch history access
+  - `drain_record(name)`: Drain all pending values since last drain call
+  - `drain_record_with_limit(name, limit)`: Drain with maximum count limit
+  - `DrainResponse` struct with `record_name`, `values` (JSON array), and `count`
+  - Cold-start semantics: first drain creates reader and returns empty
+- **Graph Introspection API**: New methods for dependency graph exploration
+  - `graph_nodes()`: Get all nodes with origin, buffer type, tap count, outbound status
+  - `graph_edges()`: Get all directed edges showing data flow
+  - `graph_topo_order()`: Get record keys in topological (spawn) order
+
+### Changed
+
+- **Re-export**: `DrainResponse` now re-exported from crate root for convenience
 
 ## [0.4.0] - 2025-12-25
 

_external/embassy

@@ -7,6 +7,7 @@ use crate::protocol::{
     cli_hello, parse_message, serialize_message, Event, EventMessage, RecordMetadata, Request,
     RequestExt, Response, ResponseExt, WelcomeMessage,
 };
+use serde::{Deserialize, Serialize};
 use serde_json::json;
 use std::path::{Path, PathBuf};
 use std::time::Duration;
@@ -174,6 +175,71 @@ impl AimxClient {
         Ok(event_msg.event)
     }
 
+    /// Drain all pending values from a record's drain reader.
+    ///
+    /// Returns all values accumulated since the last drain call,
+    /// in chronological order. This is a destructive read — drained
+    /// values will not be returned again.
+    ///
+    /// The first call for a given record creates the drain reader and
+    /// returns empty (cold start). Subsequent calls return accumulated values.
+    pub async fn drain_record(&mut self, name: &str) -> ClientResult<DrainResponse> {
+        let params = json!({ "name": name });
+        let result = self.send_request("record.drain", Some(params)).await?;
+        let response: DrainResponse = serde_json::from_value(result)?;
+        Ok(response)
+    }
+
+    /// Drain with a limit on the number of values returned.
+    pub async fn drain_record_with_limit(
+        &mut self,
+        name: &str,
+        limit: u32,
+    ) -> ClientResult<DrainResponse> {
+        let params = json!({
+            "name": name,
+            "limit": limit,
+        });
+        let result = self.send_request("record.drain", Some(params)).await?;
+        let response: DrainResponse = serde_json::from_value(result)?;
+        Ok(response)
+    }
+
+    // ========================================================================
+    // Graph Introspection Methods
+    // ========================================================================
+
+    /// Get all nodes in the dependency graph.
+    ///
+    /// Returns a list of GraphNode objects representing all records
+    /// and their connections in the database.
+    pub async fn graph_nodes(&mut self) -> ClientResult<Vec<serde_json::Value>> {
+        let result = self.send_request("graph.nodes", None).await?;
+        let nodes: Vec<serde_json::Value> = serde_json::from_value(result)?;
+        Ok(nodes)
+    }
+
+    /// Get all edges in the dependency graph.
+    ///
+    /// Returns a list of GraphEdge objects representing data flow
+    /// connections between records.
+    pub async fn graph_edges(&mut self) -> ClientResult<Vec<serde_json::Value>> {
+        let result = self.send_request("graph.edges", None).await?;
+        let edges: Vec<serde_json::Value> = serde_json::from_value(result)?;
+        Ok(edges)
+    }
+
+    /// Get the topological ordering of records.
+    ///
+    /// Returns the record keys in topological order, ensuring all
+    /// dependencies are listed before dependents. Useful for understanding
+    /// data flow and initialization order.
+    pub async fn graph_topo_order(&mut self) -> ClientResult<Vec<String>> {
+        let result = self.send_request("graph.topo_order", None).await?;
+        let order: Vec<String> = serde_json::from_value(result)?;
+        Ok(order)
+    }
+
     /// Write a message to the stream
     async fn write_message<T: serde::Serialize>(&mut self, msg: &T) -> ClientResult<()> {
         let data = serialize_message(msg)?;
@@ -197,3 +263,14 @@ impl AimxClient {
         parse_message(&line).map_err(|e| e.into())
     }
 }
+
+/// Response from a record.drain call
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct DrainResponse {
+    /// Echo of the queried record name
+    pub record_name: String,
+    /// Chronologically ordered values (raw JSON, as written by the producer)
+    pub values: Vec<serde_json::Value>,
+    /// Number of values returned
+    pub count: usize,
+}

aimdb-client/CHANGELOG.md

@@ -39,7 +39,7 @@ pub mod error;
 pub mod protocol;
 
 // Re-export main types for convenience
-pub use connection::AimxClient;
+pub use connection::{AimxClient, DrainResponse};
 pub use discovery::{discover_instances, find_instance, InstanceInfo};
 pub use error::{ClientError, ClientResult};
 pub use protocol::{

aimdb-client/src/connection.rs

@@ -9,6 +9,39 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- **Transform API (Design 020)**: Reactive data transformations between records
+  - **Single-Input Transforms**: `transform_raw()` method on `RecordRegistrar` for creating reactive derivations
+  - **Multi-Input Joins**: `transform_join_raw()` for combining multiple input records with stateful handlers
+  - **`TransformBuilder`**: Fluent API with `.with_state()` and `.map()` for transform configuration
+  - **`JoinBuilder`**: Multi-input builder with `.input::<T>(key)` and `.with_state().on_trigger()` pattern
+  - **`TransformDescriptor`**: Type-erased descriptor for storing transform configuration
+  - **`JoinTrigger`**: Event type for multi-input join handlers with index and type-erased value
+  - Transforms are spawned as tasks during `AimDb::build()` and subscribe to input buffers
+  - Mutual exclusion enforced: a record cannot have both `.source()` and `.transform()`
+  - Full tracing integration for transform lifecycle events
+- **Graph Introspection API (Design 021)**: Dependency graph visualization and introspection
+  - **`RecordOrigin` enum**: Classifies record data sources (`Source`, `Link`, `Transform`, `TransformJoin`, `Passive`)
+  - **`GraphNode` struct**: Node metadata including origin, buffer config, tap count, outbound status
+  - **`GraphEdge` struct**: Directed edge with `from`, `to`, and edge type classification
+  - **`DependencyGraph` struct**: Full graph with nodes, edges, and topological ordering
+  - New `AnyRecord` trait methods: `has_transform()`, `record_origin()`, `buffer_info()`, `transform_input_keys()`
+  - `RecordId::new()` now accepts `RecordOrigin` parameter for accurate metadata
+  - Graph methods in `AimDbInner`: `build_dependency_graph()`, `graph_nodes()`, `graph_edges()`, `graph_topo_order()`
+- **Record Drain API (Design 019)**: Non-blocking batch history access
+  - **`try_recv()` on BufferReader**: Non-blocking receive returning `Ok(T)`, `Err(BufferEmpty)`, or `Err(BufferLagged)`
+  - **`try_recv_json()` on JsonBufferReader**: JSON-serialized non-blocking receive for remote access
+  - **`record.drain` AimX protocol method**: Drain accumulated values since last call with optional limit
+  - Cold-start semantics: first drain creates reader and returns empty
+  - Supports `SpmcRing` (full history), `SingleLatest` (at most 1), `Mailbox` (at most 1)
+  - Handler maintains per-connection drain readers via `ConnectionState`
+- **Extension Macro**: `impl_record_registrar_ext!` macro in `ext_macros.rs` for generating runtime adapter extension traits
+
+### Changed
+
+- **Renamed `.with_serialization()` to `.with_remote_access()`**: Clearer naming for JSON serialization configuration
+- **`RecordId` constructor**: Now requires `RecordOrigin` parameter for dependency graph support
+- **`set_from_json` protection**: Now also rejects writes on records with active transforms (in addition to sources)
+
 - **Dynamic Topic/Destination Routing (Design 018)**: Complete support for dynamic topic resolution
   - **Outbound (`TopicProvider` trait)**: Dynamically determine MQTT topics or KNX group addresses based on data being published
     - New `TopicProvider<T>` trait for type-safe topic determination

aimdb-client/src/lib.rs

@@ -103,6 +103,16 @@ pub trait BufferReader<T: Clone + Send>: Send {
     /// - **SingleLatest**: Waits for value change, returns most recent
     /// - **Mailbox**: Waits for slot value, takes and clears it
     fn recv(&mut self) -> Pin<Box<dyn Future<Output = Result<T, DbError>> + Send + '_>>;
+
+    /// Non-blocking receive — returns immediately.
+    ///
+    /// Returns `Err(DbError::BufferEmpty)` if no pending values.
+    ///
+    /// # Behavior by Buffer Type
+    /// - **SPMC Ring**: Returns next buffered value, or `BufferEmpty` if caught up
+    /// - **SingleLatest**: Returns value if changed since last read, or `BufferEmpty`
+    /// - **Mailbox**: Takes and returns slot value, or `BufferEmpty` if empty
+    fn try_recv(&mut self) -> Result<T, DbError>;
 }
 
 /// Reader trait for consuming JSON-serialized values from a buffer (std only)
@@ -114,7 +124,7 @@ pub trait BufferReader<T: Clone + Send>: Send {
 /// at compile time, by serializing values to JSON on each `recv_json()` call.
 ///
 /// # Requirements
-/// - Record must be configured with `.with_serialization()`
+/// - Record must be configured with `.with_remote_access()`
 /// - Only available with `std` feature (requires serde_json)
 ///
 /// # Example
@@ -139,6 +149,11 @@ pub trait JsonBufferReader: Send {
     fn recv_json(
         &mut self,
     ) -> Pin<Box<dyn Future<Output = Result<serde_json::Value, DbError>> + Send + '_>>;
+
+    /// Non-blocking receive as JSON — returns immediately.
+    ///
+    /// Returns `Err(DbError::BufferEmpty)` if no pending values.
+    fn try_recv_json(&mut self) -> Result<serde_json::Value, DbError>;
 }
 
 /// Snapshot of buffer metrics at a point in time
@@ -269,6 +284,10 @@ mod tests {
                 })
             })
         }
+
+        fn try_recv(&mut self) -> Result<T, DbError> {
+            Err(DbError::BufferEmpty)
+        }
     }
 
     #[test]