feat(remote-access-demo): enhance profiling integration and update README

Co-authored-by: Copilot <copilot@github.com>

Author: lxsaah

.vscode/mcp.json

@@ -3,6 +3,18 @@
         "aimdb-weather": {
             "type": "http",
             "url": "http://aimdb.dev/mcp"
+        },
+        "aimdb-local": {
+            "type": "stdio",
+            "command": "cargo",
+            "args": [
+                "run",
+                "--manifest-path",
+                "/aimdb_ws/aimdb/tools/aimdb-mcp/Cargo.toml",
+                "--",
+                "--socket",
+                "/tmp/aimdb-demo.sock"
+            ]
         }
     }
 }

CHANGELOG.md

@@ -30,6 +30,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Added
 
 - **Automatic stage profiling (Issue #58, RFC 014)**: New `profiling` feature on `aimdb-core` automatically measures wall-clock time per `.source()` / `.tap()` / `.link()` stage and exposes `call_count` / `total` / `avg` / `min` / `max` counters per stage. Name stages with `RecordRegistrar::with_name("...")`. Works on `no_std + alloc` via the runtime clock (`aimdb_executor::TimeOps`) and `portable-atomic` for 64-bit atomics on `thumbv7em`. New MCP tools `get_stage_profiling` (with bottleneck detection + recommendation) and `reset_stage_profiling`. Feature is off by default and zero-cost when disabled. ([aimdb-core](aimdb-core/CHANGELOG.md), [aimdb-executor](aimdb-executor/CHANGELOG.md), [aimdb-tokio-adapter](aimdb-tokio-adapter/CHANGELOG.md), [aimdb-embassy-adapter](aimdb-embassy-adapter/CHANGELOG.md), [aimdb-wasm-adapter](aimdb-wasm-adapter/CHANGELOG.md), [aimdb-client](aimdb-client/CHANGELOG.md), [tools/aimdb-mcp](tools/aimdb-mcp/CHANGELOG.md))
+- **`remote-access-demo` exercises stage profiling**: The `Temperature` and `SystemStatus` records now run from in-AimDB `.source()` + `.tap()` tasks (with `.with_name(...)` stage labels) and the demo enables the `profiling` feature. `SystemStatus.slow_status_processor` deliberately sleeps 100 ms per value so `get_stage_profiling` flags it as the bottleneck. README documents how to query and reset profiling via MCP / `socat`.
 - **`hello-mailbox` example (Issue #94)**: New minimal example demonstrating the Mailbox buffer (latest-wins) semantics using the synchronous API. First community contribution from [@ggmaldo](https://github.com/ggmaldo) — see [examples/hello-mailbox/](examples/hello-mailbox/).
 - **Writer-exclusivity validation (Issue #89)**: Combining `.source()`, `.transform()`, and `.link_from()` on the same record now panics at configuration time with a clear message instead of silently producing a last-writer-wins race on the buffer. Multiple `.link_from()` inbound connectors (fan-in) remain allowed. ([aimdb-core](aimdb-core/CHANGELOG.md))
 - **`no_std` Transform API (Design 027)**: `.transform()` and `.transform_join()` are now available on `no_std + alloc` targets — no longer Tokio-only. Multi-input join fan-in moved out of `aimdb-core` into the new `JoinFanInRuntime` traits in `aimdb-executor`, with implementations in the Tokio (`mpsc::channel`, capacity 64), Embassy (`embassy_sync::Channel`, capacity 8), and WASM (`futures_channel::mpsc`, capacity 64) adapters. ([aimdb-core](aimdb-core/CHANGELOG.md), [aimdb-executor](aimdb-executor/CHANGELOG.md), [aimdb-tokio-adapter](aimdb-tokio-adapter/CHANGELOG.md), [aimdb-embassy-adapter](aimdb-embassy-adapter/CHANGELOG.md), [aimdb-wasm-adapter](aimdb-wasm-adapter/CHANGELOG.md))

Cargo.lock

@@ -13,10 +13,11 @@ name = "client"
 path = "src/client.rs"
 
 [dependencies]
-aimdb-core = { path = "../../aimdb-core", features = ["std", "tracing"] }
+aimdb-core = { path = "../../aimdb-core", features = ["std", "tracing", "profiling"] }
 aimdb-tokio-adapter = { path = "../../aimdb-tokio-adapter", features = [
     "tokio-runtime",
     "tracing",
+    "profiling",
 ] }
 tokio = { version = "1.48", features = ["full"] }
 tracing = "0.1"

_external/embassy

@@ -5,10 +5,10 @@ This example demonstrates the AimX v1 remote access protocol with `record.list`
 ## What It Does
 
 **Server** (`server.rs`):
-- Creates an AimDB instance with 4 record types (Temperature, SystemStatus, UserEvent, Config)
+- Creates an AimDB instance with 5 record types (Temperature, SystemStatus, UserEvent, Config, AppSettings)
 - Enables remote access on Unix domain socket `/tmp/aimdb-demo.sock`
-- Uses ReadOnly security policy
-- Populates some initial data
+- Uses ReadWrite security policy (`server::AppSettings` is the only writable key)
+- Drives `Temperature` and `SystemStatus` from in-AimDB `.source()` tasks with named `.tap()` consumers, so the `profiling` feature can time every stage automatically (see [Stage Profiling](#stage-profiling) below)
 
 **Client** (`client.rs`):
 - Connects to the server via Unix domain socket
@@ -87,6 +87,35 @@ echo '{"version":"1.0","client":"test"}' | socat - UNIX-CONNECT:/tmp/aimdb-demo.
   - Write permissions
   - Timestamps
 
+## Stage Profiling
+
+The server is built with the `profiling` feature (see [Cargo.toml](Cargo.toml)) and
+registers both `Temperature` and `SystemStatus` via `.source()` + `.tap()` so AimDB
+owns the producer/consumer tasks and can time them automatically. Each stage is
+named via `.with_name("...")`:
+
+| Record         | Source stage         | Tap stage                                    |
+|----------------|----------------------|----------------------------------------------|
+| Temperature    | `temp_simulator`     | `temp_logger` (fast)                         |
+| SystemStatus   | `status_simulator`   | `slow_status_processor` (sleeps 100 ms each) |
+
+Once the server has been running for a few seconds, query stage profiling via the
+`aimdb-mcp` server using the `get_stage_profiling` tool with `record_key="SystemStatus"`.
+The result is a list of `{call_count, avg_time_ns, min_time_ns, max_time_ns, name, ...}`
+entries plus a `bottleneck` field — for SystemStatus the bottleneck will point at
+`slow_status_processor` with an average around 100 ms. Reset the counters between
+windows with the `reset_stage_profiling` tool.
+
+You can also call the raw RPC method directly:
+
+```bash
+# reset the counters (requires write permission, ReadWrite policy already enabled)
+echo '{"id":1,"method":"profiling.reset"}' | socat - UNIX-CONNECT:/tmp/aimdb-demo.sock
+```
+
+The per-stage snapshot is also embedded in each record's metadata via the
+`stage_profiling` field of `record.list`.
+
 ## Next Steps
 
 Future enhancements will add:

examples/remote-access-demo/Cargo.toml

@@ -13,7 +13,7 @@
 //! ```
 
 use aimdb_core::remote::{AimxConfig, SecurityPolicy};
-use aimdb_core::{buffer::BufferCfg, AimDbBuilder};
+use aimdb_core::{buffer::BufferCfg, AimDbBuilder, Consumer, Producer, RuntimeContext};
 use aimdb_tokio_adapter::{TokioAdapter, TokioRecordRegistrarExt};
 use serde::{Deserialize, Serialize};
 use std::sync::Arc;
@@ -96,15 +96,31 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         .with_remote_access(remote_config);
 
     // Configure records
-    // Use SpmcRing for Temperature and SystemStatus to support record.drain history
+    // Use SpmcRing for Temperature and SystemStatus to support record.drain history.
+    //
+    // Temperature and SystemStatus use `.source()` + `.tap()` so AimDB owns the
+    // producer/consumer task lifecycle — this also makes them eligible for
+    // automatic stage profiling (feature `profiling`, query via the MCP
+    // `get_stage_profiling` tool). `.with_name("...")` gives each stage a
+    // human-readable label that shows up in the profiling output.
     builder.configure::<Temperature>("server::Temperature", |reg| {
         reg.buffer(BufferCfg::SpmcRing { capacity: 100 })
-            .with_remote_access();
+            .with_remote_access()
+            .source(temperature_simulator)
+            .with_name("temp_simulator")
+            .tap(temperature_logger)
+            .with_name("temp_logger");
     });
 
     builder.configure::<SystemStatus>("server::SystemStatus", |reg| {
         reg.buffer(BufferCfg::SpmcRing { capacity: 50 })
-            .with_remote_access();
+            .with_remote_access()
+            .source(system_status_simulator)
+            .with_name("status_simulator")
+            // Deliberately slow consumer — surfaces as the bottleneck in
+            // `get_stage_profiling` for the SystemStatus record.
+            .tap(slow_status_processor)
+            .with_name("slow_status_processor");
     });
 
     builder.configure::<UserEvent>("server::UserEvent", |reg| {
@@ -131,31 +147,6 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
 
     info!("📝 Populating initial record data...");
 
-    // Produce some initial data
-    let temp_producer = db.producer::<Temperature>("server::Temperature");
-    let initial_duration = std::time::SystemTime::now()
-        .duration_since(std::time::UNIX_EPOCH)
-        .unwrap();
-    let initial_timestamp = initial_duration.as_secs() as f64
-        + initial_duration.subsec_nanos() as f64 / 1_000_000_000.0;
-
-    temp_producer
-        .produce(Temperature {
-            sensor_id: "sensor-01".to_string(),
-            celsius: 22.5,
-            timestamp: initial_timestamp,
-        })
-        .await?;
-
-    let status_producer = db.producer::<SystemStatus>("server::SystemStatus");
-    status_producer
-        .produce(SystemStatus {
-            uptime_seconds: 3600,
-            cpu_usage: 15.3,
-            memory_usage: 42.7,
-        })
-        .await?;
-
     let config_producer = db.producer::<Config>("server::Config");
     config_producer
         .produce(Config {
@@ -177,68 +168,8 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     )?;
 
     info!("✅ Initial data populated");
-
-    // Spawn background task to continuously update Temperature
-    info!("🌡️  Starting live temperature simulator...");
-    let temp_producer_clone = temp_producer.clone();
-    tokio::spawn(async move {
-        let mut counter = 0u64;
-        loop {
-            tokio::time::sleep(tokio::time::Duration::from_secs(2)).await;
-
-            counter += 1;
-            let temp = 20.0 + (counter as f64 * 0.5) + (counter as f64 % 10.0);
-
-            let duration = std::time::SystemTime::now()
-                .duration_since(std::time::UNIX_EPOCH)
-                .unwrap();
-            let timestamp =
-                duration.as_secs() as f64 + duration.subsec_nanos() as f64 / 1_000_000_000.0;
-
-            let reading = Temperature {
-                sensor_id: format!("sensor-{:02}", (counter % 3) + 1),
-                celsius: temp,
-                timestamp,
-            };
-
-            if let Err(e) = temp_producer_clone.produce(reading.clone()).await {
-                tracing::error!("Failed to produce temperature: {}", e);
-            } else {
-                tracing::debug!(
-                    "📊 Produced temperature: {} °C from {}",
-                    reading.celsius,
-                    reading.sensor_id
-                );
-            }
-        }
-    });
-
-    // Spawn background task to update SystemStatus
-    info!("💻 Starting system status simulator...");
-    let status_producer_clone = status_producer.clone();
-    tokio::spawn(async move {
-        let mut uptime = 3600u64;
-        loop {
-            tokio::time::sleep(tokio::time::Duration::from_secs(5)).await;
-
-            uptime += 5;
-            let status = SystemStatus {
-                uptime_seconds: uptime,
-                cpu_usage: 10.0 + (uptime as f64 % 30.0),
-                memory_usage: 40.0 + ((uptime as f64 / 10.0) % 20.0),
-            };
-
-            if let Err(e) = status_producer_clone.produce(status.clone()).await {
-                tracing::error!("Failed to produce system status: {}", e);
-            } else {
-                tracing::debug!(
-                    "📊 Produced system status: CPU {:.1}%, MEM {:.1}%",
-                    status.cpu_usage,
-                    status.memory_usage
-                );
-            }
-        }
-    });
+    info!("🌡️  Temperature simulator running via .source() (every 2s)");
+    info!("💻 SystemStatus simulator running via .source() (every 5s)");
 
     info!("");
     info!("🎯 Server ready! Connect with:");
@@ -254,6 +185,10 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     info!("   Temperature: every 2 seconds");
     info!("   SystemStatus: every 5 seconds");
     info!("");
+    info!("📈 Stage profiling is enabled. Query it via the aimdb-mcp tools:");
+    info!("   get_stage_profiling   record_key=\"Temperature\"   → per-stage timing");
+    info!("   reset_stage_profiling                              → clear counters");
+    info!("");
     info!("Press Ctrl+C to stop the server");
 
     // Keep server running
@@ -263,3 +198,104 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
 
     Ok(())
 }
+
+// ============================================================================
+// SOURCES & TAPS — owned by AimDB so the stage-profiling feature can time them
+// ============================================================================
+
+/// Periodically produces Temperature readings.
+///
+/// Because this runs inside a `.source()` callback, every `produce()` call is
+/// timed by the `profiling` feature — see `get_stage_profiling`.
+async fn temperature_simulator(
+    ctx: RuntimeContext<TokioAdapter>,
+    temperature: Producer<Temperature, TokioAdapter>,
+) {
+    let time = ctx.time();
+    let mut counter = 0u64;
+    loop {
+        let celsius = 20.0 + (counter as f64 * 0.5) + (counter as f64 % 10.0);
+        let duration = std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .unwrap();
+        let timestamp =
+            duration.as_secs() as f64 + duration.subsec_nanos() as f64 / 1_000_000_000.0;
+
+        let reading = Temperature {
+            sensor_id: format!("sensor-{:02}", (counter % 3) + 1),
+            celsius,
+            timestamp,
+        };
+
+        if let Err(e) = temperature.produce(reading).await {
+            tracing::error!("Failed to produce temperature: {}", e);
+        }
+
+        counter += 1;
+        time.sleep(time.secs(2)).await;
+    }
+}
+
+/// Fast tap on Temperature — just logs. Stage profiling will show it as
+/// substantially faster than `slow_status_processor` on SystemStatus.
+async fn temperature_logger(
+    _ctx: RuntimeContext<TokioAdapter>,
+    consumer: Consumer<Temperature, TokioAdapter>,
+) {
+    let Ok(mut reader) = consumer.subscribe() else {
+        tracing::error!("Failed to subscribe to Temperature");
+        return;
+    };
+    while let Ok(reading) = reader.recv().await {
+        tracing::debug!(
+            "🌡️  Logged temperature: {:.1} °C from {}",
+            reading.celsius,
+            reading.sensor_id
+        );
+    }
+}
+
+/// Periodically produces SystemStatus readings.
+async fn system_status_simulator(
+    ctx: RuntimeContext<TokioAdapter>,
+    status: Producer<SystemStatus, TokioAdapter>,
+) {
+    let time = ctx.time();
+    let mut uptime = 3600u64;
+    loop {
+        let snapshot = SystemStatus {
+            uptime_seconds: uptime,
+            cpu_usage: 10.0 + (uptime as f64 % 30.0),
+            memory_usage: 40.0 + ((uptime as f64 / 10.0) % 20.0),
+        };
+
+        if let Err(e) = status.produce(snapshot).await {
+            tracing::error!("Failed to produce system status: {}", e);
+        }
+
+        uptime += 5;
+        time.sleep(time.secs(5)).await;
+    }
+}
+
+/// Intentionally slow tap on SystemStatus — sleeps 100ms per value to simulate
+/// expensive per-value processing. With profiling enabled, this stage shows up
+/// as the per-record bottleneck in `get_stage_profiling`.
+async fn slow_status_processor(
+    ctx: RuntimeContext<TokioAdapter>,
+    consumer: Consumer<SystemStatus, TokioAdapter>,
+) {
+    let time = ctx.time();
+    let Ok(mut reader) = consumer.subscribe() else {
+        tracing::error!("Failed to subscribe to SystemStatus");
+        return;
+    };
+    while let Ok(snapshot) = reader.recv().await {
+        time.sleep(time.millis(100)).await;
+        tracing::debug!(
+            "💻 Processed status: CPU {:.1}%, MEM {:.1}%",
+            snapshot.cpu_usage,
+            snapshot.memory_usage
+        );
+    }
+}