fix(chat): clear Responding state + multi-camera grab_frame

crates/bubbaloop/Cargo.toml

@@ -84,6 +84,9 @@ chrono.workspace = true
 notify = { version = "7", default-features = false, features = ["macos_kqueue"] }
 toml = "0.8"
 
+# Camera vision: base64 for encoding grab_frame JPEG responses
+base64 = "0.22"
+
 # TUI for agent chat REPL
 ratatui = { version = "0.29", default-features = false, features = ["crossterm"] }
 crossterm = { version = "0.28", features = ["event-stream"] }

crates/bubbaloop/src/agent/camera.rs

@@ -0,0 +1,107 @@
+//! Camera frame capture for the agent's `grab_frame` tool.
+//!
+//! Sends a Zenoh query to the camera node's `grab_frame` queryable and returns
+//! the JPEG frame + metadata. The actual image encoding (RGBA→JPEG via kornia-rs)
+//! lives in the camera node itself — the daemon receives only the base64 result.
+
+use std::sync::Arc;
+use std::time::Duration;
+
+/// An image captured from a camera, ready for injection into the next agent turn.
+pub struct ImageAttachment {
+    pub media_type: String,
+    pub base64: String,
+    /// Text label prepended before the image in the labeled-pair message.
+    pub label: String,
+}
+
+/// Query the camera node's `grab_frame` queryable and return
+/// `(receipt_json, ImageAttachment)` on success.
+pub async fn grab_frame(
+    session: &Arc<zenoh::Session>,
+    node_name: &str,
+    machine_id: &str,
+) -> anyhow::Result<(String, ImageAttachment)> {
+    let key = format!("bubbaloop/local/{}/{}/grab_frame", machine_id, node_name);
+
+    let replies = session
+        .get(&key)
+        .timeout(Duration::from_secs(5))
+        .await
+        .map_err(|e| anyhow::anyhow!("Zenoh get failed: {}", e))?;
+
+    let reply = tokio::time::timeout(Duration::from_secs(5), replies.recv_async())
+        .await
+        .map_err(|_| {
+            anyhow::anyhow!(
+                "camera '{}' did not respond within 5s on {}",
+                node_name,
+                key
+            )
+        })?
+        .map_err(|_| anyhow::anyhow!("grab_frame channel closed for '{}'", node_name))?;
+
+    let sample = reply
+        .result()
+        .map_err(|e| anyhow::anyhow!("grab_frame reply error for '{}': {:?}", node_name, e))?;
+
+    let payload = sample.payload().to_bytes();
+    let v: serde_json::Value = serde_json::from_slice(payload.as_ref())
+        .map_err(|e| anyhow::anyhow!("grab_frame JSON parse failed: {}", e))?;
+
+    // Surface error from camera node
+    if let Some(err) = v.get("error").and_then(|e| e.as_str()) {
+        return Err(anyhow::anyhow!("camera '{}' error: {}", node_name, err));
+    }
+
+    // Reject stale frames — camera is likely stopped or not streaming.
+    // 10 s is generous for live cameras; anything older means the cached
+    // frame is from a previous run and would mislead vision analysis.
+    const MAX_AGE_MS: u64 = 10_000;
+    if let Some(age_ms) = v.get("age_ms").and_then(|a| a.as_u64()) {
+        if age_ms > MAX_AGE_MS {
+            return Err(anyhow::anyhow!(
+                "camera '{}' frame is stale ({}ms old, max {}ms) — node may not be streaming",
+                node_name,
+                age_ms,
+                MAX_AGE_MS
+            ));
+        }
+    }
+
+    let jpeg_b64 = v
+        .get("jpeg_b64")
+        .and_then(|s| s.as_str())
+        .ok_or_else(|| anyhow::anyhow!("grab_frame response missing jpeg_b64"))?
+        .to_string();
+
+    let media_type = v
+        .get("media_type")
+        .and_then(|s| s.as_str())
+        .unwrap_or("image/jpeg")
+        .to_string();
+
+    let label = v
+        .get("label")
+        .and_then(|s| s.as_str())
+        .unwrap_or(node_name)
+        .to_string();
+
+    // Build receipt JSON from metadata fields (strip the large jpeg_b64 field)
+    let receipt = {
+        let mut r = v.clone();
+        if let serde_json::Value::Object(ref mut m) = r {
+            m.remove("jpeg_b64");
+        }
+        r.to_string()
+    };
+
+    Ok((
+        receipt,
+        ImageAttachment {
+            media_type,
+            base64: jpeg_b64,
+            label,
+        },
+    ))
+}

crates/bubbaloop/src/agent/dispatch.rs

@@ -5,13 +5,15 @@
 //! exposes 47 tools to external clients; the agent here uses the subset it
 //! needs for its own reasoning loop.
 
+use crate::agent::camera::ImageAttachment;
 use crate::agent::dispatch_security;
 use crate::agent::provider::{ContentBlock, ToolDefinition};
 use crate::mcp::platform::{
     ConfigureContextParams, NodeCommand, PlatformOperations, RegisterAlertParams,
 };
 use crate::validation;
 use serde_json::{json, Value};
+use std::collections::HashMap;
 use std::sync::Arc;
 
 /// Result of a tool handler: text output and whether it was an error.
@@ -44,6 +46,8 @@ pub struct Dispatcher<P: PlatformOperations> {
     job_notify: Option<Arc<tokio::sync::Notify>>,
     episodic_decay_half_life_days: u32,
     telemetry: Option<Arc<crate::daemon::telemetry::TelemetryService>>,
+    /// Images captured by `grab_frame` during a turn, keyed by tool_use_id.
+    image_stash: std::sync::Mutex<HashMap<String, ImageAttachment>>,
 }
 
 impl<P: PlatformOperations> Dispatcher<P> {
@@ -57,6 +61,7 @@ impl<P: PlatformOperations> Dispatcher<P> {
             job_notify: None,
             episodic_decay_half_life_days: 7,
             telemetry: None,
+            image_stash: std::sync::Mutex::new(HashMap::new()),
         }
     }
 
@@ -95,6 +100,30 @@ impl<P: PlatformOperations> Dispatcher<P> {
             job_notify,
             episodic_decay_half_life_days,
             telemetry: None,
+            image_stash: std::sync::Mutex::new(HashMap::new()),
+        }
+    }
+
+    /// Stash an image attachment captured by `grab_frame` for the current turn.
+    pub fn stash_image(&self, tool_use_id: String, attachment: ImageAttachment) {
+        if let Ok(mut stash) = self.image_stash.lock() {
+            stash.insert(tool_use_id, attachment);
+        }
+    }
+
+    /// Drain all stashed images for the given tool_use_ids (in order).
+    pub fn drain_images(&self, tool_use_ids: &[String]) -> Vec<ImageAttachment> {
+        let mut stash = self.image_stash.lock().unwrap_or_else(|e| e.into_inner());
+        tool_use_ids
+            .iter()
+            .filter_map(|id| stash.remove(id))
+            .collect()
+    }
+
+    /// Clear the image stash (called at the end of each turn).
+    pub fn clear_image_stash(&self) {
+        if let Ok(mut stash) = self.image_stash.lock() {
+            stash.clear();
         }
     }
 
@@ -500,14 +529,17 @@ impl<P: PlatformOperations> Dispatcher<P> {
                 name: "memory_search".to_string(),
                 description: "Search episodic memory for past conversations, tool results, and \
                     agent observations. Uses BM25 full-text search with temporal decay. Returns \
-                    the most relevant entries."
+                    the most relevant entries. \
+                    IMPORTANT: combine all related keywords into ONE query (e.g. \"camera grab_frame terrace\") \
+                    rather than calling this tool multiple times with separate queries. \
+                    One broad query is faster and covers more results than several narrow ones."
                     .to_string(),
                 input_schema: json!({
                     "type": "object",
                     "properties": {
                         "query": {
                             "type": "string",
-                            "description": "Search query (keywords or phrases)"
+                            "description": "Search query — combine multiple keywords with spaces for a single broad search"
                         },
                         "limit": {
                             "type": "integer",
@@ -663,6 +695,31 @@ impl<P: PlatformOperations> Dispatcher<P> {
                     "required": ["topic", "message"]
                 }),
             },
+            ToolDefinition {
+                name: "grab_frame".to_string(),
+                description: "Capture the latest JPEG frame from a camera sensor node. \
+                    The frame is delivered as an image content block in the next turn so \
+                    you can inspect it directly. \
+                    Use the exact node name from the Current Sensor Inventory. \
+                    When the user asks about cameras without naming one, call grab_frame \
+                    for EVERY camera node in the inventory — do not pick just one. \
+                    Returns a JSON receipt with capture metadata (timestamp, resolution, frame_id)."
+                    .to_string(),
+                input_schema: json!({
+                    "type": "object",
+                    "properties": {
+                        "node_name": {
+                            "type": "string",
+                            "description": "Exact name of the camera node from the Current Sensor Inventory"
+                        },
+                        "machine_id": {
+                            "type": "string",
+                            "description": "Machine ID override (defaults to the local machine)"
+                        }
+                    },
+                    "required": ["node_name"]
+                }),
+            },
         ]
     }
 
@@ -723,6 +780,7 @@ impl<P: PlatformOperations> Dispatcher<P> {
             "telemetry_history" => self.handle_get_telemetry_history(input).await,
             "telemetry_config_update" => self.handle_update_telemetry_config(input).await,
             "publish_to_topic" => self.handle_publish_to_topic(input).await,
+            "grab_frame" => self.handle_grab_frame(tool_use_id, input).await,
             _ => ToolResult::error(format!(
                 "Unknown tool: {}. Use your available tools: node management \
                  (node_list, node_start, etc.), system (read_file, write_file, \
@@ -1733,6 +1791,44 @@ impl<P: PlatformOperations> Dispatcher<P> {
             Err(e) => ToolResult::error(format!("Error: {}", e)),
         }
     }
+
+    async fn handle_grab_frame(&self, tool_use_id: &str, input: &Value) -> ToolResult {
+        let node_name = match input.get("node_name").and_then(|v| v.as_str()) {
+            Some(n) => n.to_string(),
+            None => return ToolResult::error("Missing required parameter: node_name".to_string()),
+        };
+        let machine_id = input
+            .get("machine_id")
+            .and_then(|v| v.as_str())
+            .unwrap_or(&self.machine_id)
+            .to_string();
+
+        let session = match self.platform.zenoh_session() {
+            Some(s) => s,
+            None => {
+                return ToolResult::error(
+                    "grab_frame is not available in this context (no Zenoh session)".to_string(),
+                )
+            }
+        };
+
+        // Registry name may differ from the Zenoh instance name (e.g. "tapo-terrace" vs
+        // "tapo_terrace_camera" when a config_override sets a different `name:` field).
+        let instance_name = self.platform.get_node_instance_name(&node_name).await;
+        log::info!(
+            "[Agent] grab_frame: {}/{} (instance: {})",
+            machine_id,
+            node_name,
+            instance_name
+        );
+        match crate::agent::camera::grab_frame(&session, &instance_name, &machine_id).await {
+            Ok((receipt, attachment)) => {
+                self.stash_image(tool_use_id.to_string(), attachment);
+                ToolResult::success(receipt)
+            }
+            Err(e) => ToolResult::error(format!("grab_frame failed for '{}': {}", node_name, e)),
+        }
+    }
 }
 
 #[cfg(test)]
@@ -1743,7 +1839,7 @@ mod tests {
     use zenoh::pubsub::Subscriber;
     use zenoh::sample::Sample;
 
-    const TOOL_COUNT: usize = 42;
+    const TOOL_COUNT: usize = 43;
 
     #[test]
     fn tool_definitions_count() {

crates/bubbaloop/src/agent/gateway.rs

@@ -34,6 +34,8 @@ pub enum AgentEventType {
     Tool,
     /// Tool call result.
     ToolResult,
+    /// Camera frame captured by grab_frame (base64-encoded JPEG in `image` field).
+    Image,
     /// Error message.
     Error,
     /// Turn complete — no more events for this correlation ID.
@@ -59,6 +61,12 @@ pub struct AgentEvent {
     /// Truncated tool input JSON (only present on Tool events).
     #[serde(skip_serializing_if = "Option::is_none")]
     pub input: Option<String>,
+    /// Base64-encoded image data (only present on Image events).
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub image: Option<String>,
+    /// MIME type for the image (e.g. "image/jpeg"). Only present on Image events.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub media_type: Option<String>,
 }
 
 impl AgentEvent {
@@ -69,6 +77,20 @@ impl AgentEvent {
             event_type: AgentEventType::Delta,
             text: Some(text.to_string()),
             input: None,
+            image: None,
+            media_type: None,
+        }
+    }
+
+    /// Create an Image event (camera frame captured by grab_frame).
+    pub fn image(id: &str, media_type: &str, base64: &str) -> Self {
+        Self {
+            id: id.to_string(),
+            event_type: AgentEventType::Image,
+            text: None,
+            input: None,
+            image: Some(base64.to_string()),
+            media_type: Some(media_type.to_string()),
         }
     }
 
@@ -81,6 +103,8 @@ impl AgentEvent {
             event_type: AgentEventType::Tool,
             text: Some(tool_name.to_string()),
             input: input.map(|s| s.to_string()),
+            image: None,
+            media_type: None,
         }
     }
 
@@ -91,6 +115,8 @@ impl AgentEvent {
             event_type: AgentEventType::ToolResult,
             text: Some(result.to_string()),
             input: None,
+            image: None,
+            media_type: None,
         }
     }
 
@@ -101,6 +127,8 @@ impl AgentEvent {
             event_type: AgentEventType::Error,
             text: Some(message.to_string()),
             input: None,
+            image: None,
+            media_type: None,
         }
     }
 
@@ -111,6 +139,8 @@ impl AgentEvent {
             event_type: AgentEventType::Done,
             text: None,
             input: None,
+            image: None,
+            media_type: None,
         }
     }
 
@@ -124,6 +154,8 @@ impl AgentEvent {
             event_type: AgentEventType::System,
             text: Some(message.to_string()),
             input: None,
+            image: None,
+            media_type: None,
         }
     }
 }