From 575b5c21bdb4639cef72678b4e3764df3acf7320 Mon Sep 17 00:00:00 2001 From: edgarriba Date: Sat, 16 May 2026 18:34:34 +0200 Subject: [PATCH 01/13] feat(agent): grab_frame tool + ContentBlock::Image + Gemini vision support MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Enables the camera-vision demo: "what do the cameras see?" in the Chat tab calls grab_frame for each sensor node, injects JPEG frames into the next Gemini turn, and returns one coherent natural-language analysis. Architecture: - Camera node (rtsp-camera) owns encoding: kornia-rs RGBA→RGB→resize→JPEG - Agent grab_frame tool is a thin Zenoh query to bubbaloop/local/.../grab_frame - Images are injected out-of-band after tool_results via user_with_labeled_images - Gemini role-coalescing merges consecutive user messages (required for HTTP 200) New primitives: - ContentBlock::Image + ImageSource (provider-agnostic, Anthropic wire format) - Message::user_with_labeled_images (interleaved text labels + images) - GeminiPart::InlineData + role-coalescing in to_gemini_contents - Dispatcher image_stash (Mutex>) - PlatformOperations::zenoh_session() for grab_frame Zenoh access - agent/camera.rs: Zenoh query client returning (receipt_json, ImageAttachment) Co-Authored-By: Claude Opus 4.7 --- Cargo.lock | 1 + crates/bubbaloop/Cargo.toml | 3 + crates/bubbaloop/src/agent/camera.rs | 92 +++++++++++++++++++ crates/bubbaloop/src/agent/dispatch.rs | 85 ++++++++++++++++- crates/bubbaloop/src/agent/mod.rs | 17 ++++ crates/bubbaloop/src/agent/prompt.rs | 8 +- crates/bubbaloop/src/agent/provider/gemini.rs | 46 ++++++++-- crates/bubbaloop/src/agent/provider/mod.rs | 37 +++++++- crates/bubbaloop/src/mcp/daemon_platform.rs | 4 + crates/bubbaloop/src/mcp/mock_platform.rs | 4 + crates/bubbaloop/src/mcp/platform.rs | 6 ++ crates/bubbaloop/src/mcp/rbac.rs | 3 +- 12 files changed, 295 insertions(+), 11 deletions(-) create mode 100644 crates/bubbaloop/src/agent/camera.rs diff --git a/Cargo.lock b/Cargo.lock index cd423f10..cb9f4c9b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -356,6 +356,7 @@ dependencies = [ "anyhow", "argh", "axum", + "base64", "chrono", "ciborium", "cron", diff --git a/crates/bubbaloop/Cargo.toml b/crates/bubbaloop/Cargo.toml index ac3fe0b7..7aefc52b 100644 --- a/crates/bubbaloop/Cargo.toml +++ b/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"] } diff --git a/crates/bubbaloop/src/agent/camera.rs b/crates/bubbaloop/src/agent/camera.rs new file mode 100644 index 00000000..eec7fb0b --- /dev/null +++ b/crates/bubbaloop/src/agent/camera.rs @@ -0,0 +1,92 @@ +//! 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, + 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)); + } + + 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, + }, + )) +} diff --git a/crates/bubbaloop/src/agent/dispatch.rs b/crates/bubbaloop/src/agent/dispatch.rs index a9b7a9d8..f1e75959 100644 --- a/crates/bubbaloop/src/agent/dispatch.rs +++ b/crates/bubbaloop/src/agent/dispatch.rs @@ -5,6 +5,7 @@ //! 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::{ @@ -12,6 +13,7 @@ use crate::mcp::platform::{ }; 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 { job_notify: Option>, episodic_decay_half_life_days: u32, telemetry: Option>, + /// Images captured by `grab_frame` during a turn, keyed by tool_use_id. + image_stash: std::sync::Mutex>, } impl Dispatcher

{ @@ -57,6 +61,7 @@ impl Dispatcher

{ job_notify: None, episodic_decay_half_life_days: 7, telemetry: None, + image_stash: std::sync::Mutex::new(HashMap::new()), } } @@ -95,6 +100,30 @@ impl Dispatcher

{ 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 { + 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(); } } @@ -663,6 +692,29 @@ impl Dispatcher

{ "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. Check the Current Sensor Inventory for \ + available camera node names. Returns a JSON receipt with capture \ + metadata (timestamp, resolution, frame_id)." + .to_string(), + input_schema: json!({ + "type": "object", + "properties": { + "node_name": { + "type": "string", + "description": "Name of the camera node (e.g. 'tapo-terrace', 'tapo-entrance')" + }, + "machine_id": { + "type": "string", + "description": "Machine ID override (defaults to the local machine)" + } + }, + "required": ["node_name"] + }), + }, ] } @@ -723,6 +775,7 @@ impl Dispatcher

{ "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 +1786,36 @@ impl Dispatcher

{ 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(), + ) + } + }; + + log::info!("[Agent] grab_frame: {}/{}", machine_id, node_name); + match crate::agent::camera::grab_frame(&session, &node_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 +1826,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() { diff --git a/crates/bubbaloop/src/agent/mod.rs b/crates/bubbaloop/src/agent/mod.rs index 07a01807..f3befa01 100644 --- a/crates/bubbaloop/src/agent/mod.rs +++ b/crates/bubbaloop/src/agent/mod.rs @@ -9,6 +9,7 @@ //! - `prompt` — System prompt builder //! - `scheduler` — Job poller integrated with heartbeat +pub mod camera; pub mod dispatch; pub(crate) mod dispatch_security; pub mod gateway; @@ -324,6 +325,7 @@ pub async fn run_agent_turn< // Browser disconnected — cancel the turn cleanly. log::info!("[Agent] Turn cancelled by client (corr={})", correlation_id); sanitize_dangling_tool_use(memory); + dispatcher.clear_image_stash(); sink.emit(AgentEvent::error( correlation_id, "turn cancelled by client", @@ -339,6 +341,7 @@ pub async fn run_agent_turn< // The Claude API returns 400 "unexpected tool_use" on the next turn if // those blocks are left dangling. Add synthetic error tool_results. sanitize_dangling_tool_use(memory); + dispatcher.clear_image_stash(); sink.emit(AgentEvent::error( correlation_id, &format!("Turn timed out after {}s", TURN_TIMEOUT_SECS), @@ -590,7 +593,21 @@ async fn run_turn_loop< } // Add tool results to short-term memory + let tool_use_ids: Vec = tool_results.iter().map(|(id, _, _)| id.clone()).collect(); memory.short_term.push(Message::tool_results(tool_results)); + + // Inject captured camera frames as a follow-up user message with labeled images. + // Images are stashed out-of-band by grab_frame (base64 never enters ToolResult). + let attachments = dispatcher.drain_images(&tool_use_ids); + if !attachments.is_empty() { + let pairs: Vec<(String, String, String)> = attachments + .into_iter() + .map(|a| (a.label, a.media_type, a.base64)) + .collect(); + memory + .short_term + .push(Message::user_with_labeled_images(pairs)); + } } // 2.5. Pre-compaction flush (design Section 5.5) diff --git a/crates/bubbaloop/src/agent/prompt.rs b/crates/bubbaloop/src/agent/prompt.rs index ca3e9fff..5cfe1cb3 100644 --- a/crates/bubbaloop/src/agent/prompt.rs +++ b/crates/bubbaloop/src/agent/prompt.rs @@ -239,14 +239,18 @@ pub fn build_system_prompt_with_soul_path( // Scope — tells the LLM what tools it has parts.push( "\n## Tools\n\n\ - You have 49 tools across categories:\n\ + You have 43 tools across categories:\n\ - **Node management:** install, build, start, stop, restart, configure, monitor, query nodes\n\ - **System:** read and write files, run shell commands\n\ - **Memory:** search episodic memory, update beliefs, read world state\n\ - **Missions:** list, pause, resume, cancel missions\n\ - - **Safety:** register and list constraints per mission\n\n\ + - **Safety:** register and list constraints per mission\n\ + - **Vision:** `grab_frame` captures a JPEG from a camera sensor node — the image is \ + delivered directly in the next turn for you to inspect\n\n\ Use the right tool for the job. For node operations, use the dedicated node tools.\n\ For everything else, use read_file, write_file, or run_command.\n\n\ + To analyze what cameras see, call `grab_frame` for each sensor — the JPEG frame\n\ + is delivered as an image content block in the next turn for you to inspect directly.\n\n\ ### Important: recurring vs one-off actions\n\n\ - **run_command** blocks until the command exits. NEVER use shell loops (`while true`, `watch`, `for i in`) \ with run_command — they block for the full 30-second timeout and produce no output until they time out.\n\ diff --git a/crates/bubbaloop/src/agent/provider/gemini.rs b/crates/bubbaloop/src/agent/provider/gemini.rs index 37091dbb..6ee97e82 100644 --- a/crates/bubbaloop/src/agent/provider/gemini.rs +++ b/crates/bubbaloop/src/agent/provider/gemini.rs @@ -56,10 +56,22 @@ struct GeminiContent { #[serde(untagged)] enum GeminiPart { Text(GeminiTextPartOwned), + InlineData(GeminiInlineDataPart), FunctionCall(GeminiFunctionCall), FunctionResponse(GeminiFunctionResponse), } +#[derive(Serialize, Deserialize, Clone, Debug)] +struct GeminiInlineDataPart { + inline_data: GeminiInlineData, +} + +#[derive(Serialize, Deserialize, Clone, Debug)] +struct GeminiInlineData { + mime_type: String, + data: String, +} + #[derive(Serialize)] struct GeminiTextPart<'a> { text: &'a str, @@ -232,8 +244,12 @@ impl GeminiProvider { // ── Message conversion ─────────────────────────────────────────────── /// Convert our canonical `Message` slice to Gemini `contents` + optional system prompt. +/// +/// Consecutive messages with the same role are coalesced into a single `GeminiContent` entry +/// by concatenating their `parts[]`. Gemini requires strict role alternation; our agent loop +/// emits two consecutive user messages (tool_results then labeled images) in the same turn. fn to_gemini_contents(messages: &[Message]) -> Vec { - let mut out = Vec::new(); + let mut out: Vec = Vec::new(); for msg in messages { let role = match msg.role.as_str() { "assistant" => "model", @@ -245,6 +261,14 @@ fn to_gemini_contents(messages: &[Message]) -> Vec { ContentBlock::Text { text } => { parts.push(GeminiPart::Text(GeminiTextPartOwned { text: text.clone() })); } + ContentBlock::Image { source } => { + parts.push(GeminiPart::InlineData(GeminiInlineDataPart { + inline_data: GeminiInlineData { + mime_type: source.media_type.clone(), + data: source.data.clone(), + }, + })); + } ContentBlock::ToolUse { name, input, .. } => { parts.push(GeminiPart::FunctionCall(GeminiFunctionCall { function_call: GeminiFn { @@ -272,12 +296,20 @@ fn to_gemini_contents(messages: &[Message]) -> Vec { } } } - if !parts.is_empty() { - out.push(GeminiContent { - role: role.to_string(), - parts, - }); + if parts.is_empty() { + continue; } + // Coalesce consecutive same-role entries (Gemini requires strict alternation). + if let Some(last) = out.last_mut() { + if last.role == role { + last.parts.extend(parts); + continue; + } + } + out.push(GeminiContent { + role: role.to_string(), + parts, + }); } out } @@ -319,6 +351,7 @@ fn parse_candidate(candidate: &GeminiCandidate) -> (Vec, Option {} + GeminiPart::InlineData(_) => {} // images in responses are not yet surfaced } } let stop_reason = candidate.finish_reason.as_deref().map(|r| match r { @@ -530,6 +563,7 @@ async fn process_gemini_sse( } } GeminiPart::FunctionResponse(_) => {} + GeminiPart::InlineData(_) => {} // images in responses not yet surfaced } } diff --git a/crates/bubbaloop/src/agent/provider/mod.rs b/crates/bubbaloop/src/agent/provider/mod.rs index ecdfdc86..5f75ff91 100644 --- a/crates/bubbaloop/src/agent/provider/mod.rs +++ b/crates/bubbaloop/src/agent/provider/mod.rs @@ -90,7 +90,7 @@ pub type Result = std::result::Result; // ── Content blocks ────────────────────────────────────────────────── -/// A single content block in a message (text, tool_use, or tool_result). +/// A single content block in a message (text, tool_use, tool_result, or image). #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] #[serde(tag = "type")] pub enum ContentBlock { @@ -111,6 +111,19 @@ pub enum ContentBlock { #[serde(skip_serializing_if = "Option::is_none")] is_error: Option, }, + + /// Base64-encoded image (Anthropic wire format; Gemini converts to inline_data). + #[serde(rename = "image")] + Image { source: ImageSource }, +} + +/// Image source for `ContentBlock::Image`. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub struct ImageSource { + #[serde(rename = "type")] + pub kind: String, // "base64" + pub media_type: String, // e.g. "image/jpeg" + pub data: String, // base64-encoded bytes } // ── Message ───────────────────────────────────────────────────────── @@ -164,6 +177,28 @@ impl Message { .join("") } + /// Build a user message with interleaved text labels and images. + /// + /// `pairs` is `Vec<(label_text, media_type, base64_data)>`. + /// Produces content = `[Text(label_1), Image(img_1), Text(label_2), Image(img_2), ...]`. + pub fn user_with_labeled_images(pairs: Vec<(String, String, String)>) -> Self { + let mut content = Vec::with_capacity(pairs.len() * 2); + for (label, media_type, data) in pairs { + content.push(ContentBlock::Text { text: label }); + content.push(ContentBlock::Image { + source: ImageSource { + kind: "base64".to_string(), + media_type, + data, + }, + }); + } + Self { + role: "user".to_string(), + content, + } + } + /// Extract all tool_use blocks as `(id, name, input)` tuples. pub fn tool_uses(&self) -> Vec<(&str, &str, &Value)> { self.content diff --git a/crates/bubbaloop/src/mcp/daemon_platform.rs b/crates/bubbaloop/src/mcp/daemon_platform.rs index 1a75bea7..189394d9 100644 --- a/crates/bubbaloop/src/mcp/daemon_platform.rs +++ b/crates/bubbaloop/src/mcp/daemon_platform.rs @@ -809,6 +809,10 @@ impl PlatformOperations for DaemonPlatform { .map_err(|e| PlatformError::Internal(format!("Zenoh put failed: {}", e))) } + fn zenoh_session(&self) -> Option> { + Some(self.session.clone()) + } + async fn clear_episodic_memory(&self, older_than_days: u32) -> PlatformResult { let base = self .agent_db_path diff --git a/crates/bubbaloop/src/mcp/mock_platform.rs b/crates/bubbaloop/src/mcp/mock_platform.rs index b764d719..d73a41c5 100644 --- a/crates/bubbaloop/src/mcp/mock_platform.rs +++ b/crates/bubbaloop/src/mcp/mock_platform.rs @@ -450,6 +450,10 @@ impl PlatformOperations for MockPlatform { log::debug!("[MockPlatform] publish_to_topic: {}", topic); Ok(()) } + + fn zenoh_session(&self) -> Option> { + self.zenoh_session.clone() + } } // ── Tests ──────────────────────────────────────────────────────────────── diff --git a/crates/bubbaloop/src/mcp/platform.rs b/crates/bubbaloop/src/mcp/platform.rs index 59de03ac..56a42af1 100644 --- a/crates/bubbaloop/src/mcp/platform.rs +++ b/crates/bubbaloop/src/mcp/platform.rs @@ -281,6 +281,12 @@ pub trait PlatformOperations: Send + Sync + 'static { topic: &str, message: &str, ) -> impl std::future::Future> + Send; + + /// Return the underlying Zenoh session, if available. + /// + /// `DaemonPlatform` returns `Some`; `MockPlatform` returns `None`. + /// Used by `grab_frame` to pull camera samples directly from SHM. + fn zenoh_session(&self) -> Option>; } /// Parameters for creating or updating a belief. diff --git a/crates/bubbaloop/src/mcp/rbac.rs b/crates/bubbaloop/src/mcp/rbac.rs index 656a31a8..7651b838 100644 --- a/crates/bubbaloop/src/mcp/rbac.rs +++ b/crates/bubbaloop/src/mcp/rbac.rs @@ -91,7 +91,8 @@ pub fn required_tier(tool_name: &str) -> Tier { | "mission_pause" | "mission_resume" | "mission_cancel" - | "belief_update" => Tier::Operator, + | "belief_update" + | "grab_frame" => Tier::Operator, // Admin tools (system modification) "node_install" From 2488229da1b4bdad6de2807af77dff2e179f216b Mon Sep 17 00:00:00 2001 From: edgarriba Date: Sat, 16 May 2026 22:32:57 +0200 Subject: [PATCH 02/13] feat(agent): image-in-chat + instance-name resolution + dangling-tool-use fix **Image in chat (dashboard)** - `AgentEvent` gains `image` + `media_type` optional fields and a new `Image` variant in `AgentEventType`. - After draining the image stash, the agent loop emits one SSE `image` event per captured frame so the dashboard can render them inline. - `ChatView` handles `image` events: renders `` tags (max 420px wide) styled to match the existing tool-call row aesthetics. - `agent.ts` extends `AgentEvent` and `AgentEventType` with the new fields. - CLI (`agent_client.rs`, `StdoutSink`) silently ignores image events. **Instance-name resolution for grab_frame** - `PlatformOperations` gains `get_node_instance_name()`: maps the registry name (e.g. `tapo-terrace`) to the Zenoh instance name the node publishes under (e.g. `tapo_terrace_camera`), which is what the queryable key uses. - `NodeManager::get_instance_name()` returns `published_instance_name` or falls back to `effective_name()`. - `DaemonPlatform` and `MockPlatform` implement the new trait method. - `dispatch.rs` resolves the name before building the grab_frame query key. **Dangling function-call sanitization** - `sanitize_dangling_tool_use()` is now called at the *start* of every `run_agent_turn()`, not only on cancel/timeout. This prevents a Gemini HTTP 400 "function response must come immediately after function call" error when an SSE connection drops after the assistant message is stored but before the tool results are pushed into short-term memory. **Dashboard agent discovery fix** - `queryWithTimeout` races the zenoh-js receiver against a hard deadline (`timeoutMs + 500 ms`). Without this, the `for await` loop hangs because zenoh-ts receivers don't always close when the server-side timeout fires. Co-Authored-By: Claude Sonnet 4.6 --- crates/bubbaloop/src/agent/dispatch.rs | 12 +++++-- crates/bubbaloop/src/agent/gateway.rs | 32 +++++++++++++++++++ crates/bubbaloop/src/agent/mod.rs | 18 +++++++++++ crates/bubbaloop/src/cli/agent_client.rs | 6 ++++ crates/bubbaloop/src/daemon/chat_route.rs | 1 + .../bubbaloop/src/daemon/node_manager/mod.rs | 14 ++++++++ crates/bubbaloop/src/mcp/daemon_platform.rs | 8 +++++ crates/bubbaloop/src/mcp/mock_platform.rs | 11 +++++++ crates/bubbaloop/src/mcp/platform.rs | 15 +++++++++ dashboard/src/components/ChatView.tsx | 30 +++++++++++++++++ dashboard/src/lib/agent.ts | 19 ++++++++++- 11 files changed, 163 insertions(+), 3 deletions(-) diff --git a/crates/bubbaloop/src/agent/dispatch.rs b/crates/bubbaloop/src/agent/dispatch.rs index f1e75959..8bbb4796 100644 --- a/crates/bubbaloop/src/agent/dispatch.rs +++ b/crates/bubbaloop/src/agent/dispatch.rs @@ -1807,8 +1807,16 @@ impl Dispatcher

{ } }; - log::info!("[Agent] grab_frame: {}/{}", machine_id, node_name); - match crate::agent::camera::grab_frame(&session, &node_name, &machine_id).await { + // 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) diff --git a/crates/bubbaloop/src/agent/gateway.rs b/crates/bubbaloop/src/agent/gateway.rs index 5b42380a..2648cc52 100644 --- a/crates/bubbaloop/src/agent/gateway.rs +++ b/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, + /// Base64-encoded image data (only present on Image events). + #[serde(skip_serializing_if = "Option::is_none")] + pub image: Option, + /// 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, } 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, } } } diff --git a/crates/bubbaloop/src/agent/mod.rs b/crates/bubbaloop/src/agent/mod.rs index f3befa01..52f9f695 100644 --- a/crates/bubbaloop/src/agent/mod.rs +++ b/crates/bubbaloop/src/agent/mod.rs @@ -69,6 +69,9 @@ impl EventSink for StdoutSink { log::warn!("Error: {}", msg); } } + AgentEventType::Image => { + // Image frames are dashboard-only; stdout mode silently ignores them + } AgentEventType::Done => { // Done is implicit in stdout mode } @@ -287,6 +290,12 @@ pub async fn run_agent_turn< &world_state, ); + // Heal any dangling function_call turns left from a prior incomplete turn + // (e.g. SSE connection dropped after the assistant message was stored but + // before tool_results were pushed). Without this, Gemini returns HTTP 400 + // "function response must come immediately after function call turn". + sanitize_dangling_tool_use(memory); + // Add user input to short-term memory if let Some(input) = user_input { memory.short_term.push(Message::user(input)); @@ -600,6 +609,15 @@ async fn run_turn_loop< // Images are stashed out-of-band by grab_frame (base64 never enters ToolResult). let attachments = dispatcher.drain_images(&tool_use_ids); if !attachments.is_empty() { + // Stream each frame to the dashboard so it appears in the chat. + for attachment in &attachments { + sink.emit(AgentEvent::image( + correlation_id, + &attachment.media_type, + &attachment.base64, + )) + .await; + } let pairs: Vec<(String, String, String)> = attachments .into_iter() .map(|a| (a.label, a.media_type, a.base64)) diff --git a/crates/bubbaloop/src/cli/agent_client.rs b/crates/bubbaloop/src/cli/agent_client.rs index f63d60ee..0e330781 100644 --- a/crates/bubbaloop/src/cli/agent_client.rs +++ b/crates/bubbaloop/src/cli/agent_client.rs @@ -149,6 +149,9 @@ async fn send_and_render( log::warn!("Error: {}", msg); } } + AgentEventType::Image => { + // Image frames are dashboard-only; CLI ignores them + } AgentEventType::Done => { log::info!(""); return Ok(()); @@ -486,6 +489,9 @@ async fn run_tui_repl( scroll_offset = 0; } } + AgentEventType::Image => { + // Image frames are dashboard-only; TUI skips them + } AgentEventType::Done => { waiting_for_agent = false; output.push(OutputLine::Separator); diff --git a/crates/bubbaloop/src/daemon/chat_route.rs b/crates/bubbaloop/src/daemon/chat_route.rs index 25403fb1..c2f37770 100644 --- a/crates/bubbaloop/src/daemon/chat_route.rs +++ b/crates/bubbaloop/src/daemon/chat_route.rs @@ -285,6 +285,7 @@ fn event_type_str(t: &AgentEventType) -> &'static str { AgentEventType::Delta => "delta", AgentEventType::Tool => "tool", AgentEventType::ToolResult => "tool_result", + AgentEventType::Image => "image", AgentEventType::Error => "error", AgentEventType::Done => "done", AgentEventType::System => "system", diff --git a/crates/bubbaloop/src/daemon/node_manager/mod.rs b/crates/bubbaloop/src/daemon/node_manager/mod.rs index 00e6be72..b8ba4c4b 100644 --- a/crates/bubbaloop/src/daemon/node_manager/mod.rs +++ b/crates/bubbaloop/src/daemon/node_manager/mod.rs @@ -524,6 +524,20 @@ impl NodeManager { .map(|n| n.to_proto(&self.machine_id, &self.machine_hostname, &self.machine_ips)) } + /// Return the published Zenoh instance name for a node (from its config YAML `name:` field). + /// Falls back to the registry name when no config_override is set. + pub async fn get_instance_name(&self, name: &str) -> Option { + let nodes = self.nodes.read().await; + nodes + .values() + .find(|n| n.effective_name() == name) + .map(|n| { + n.published_instance_name + .clone() + .unwrap_or_else(|| n.effective_name()) + }) + } + /// Execute a command pub async fn execute_command(self: &Arc, cmd: NodeCommand) -> CommandResult { let command_type = CommandType::try_from(cmd.command).unwrap_or(CommandType::Refresh); diff --git a/crates/bubbaloop/src/mcp/daemon_platform.rs b/crates/bubbaloop/src/mcp/daemon_platform.rs index 189394d9..86265795 100644 --- a/crates/bubbaloop/src/mcp/daemon_platform.rs +++ b/crates/bubbaloop/src/mcp/daemon_platform.rs @@ -95,6 +95,7 @@ impl PlatformOperations for DaemonPlatform { node_type: n.node_type.clone(), installed: n.installed, is_built: n.is_built, + instance_name: None, } }) .collect(); @@ -813,6 +814,13 @@ impl PlatformOperations for DaemonPlatform { Some(self.session.clone()) } + async fn get_node_instance_name(&self, name: &str) -> String { + self.node_manager + .get_instance_name(name) + .await + .unwrap_or_else(|| name.to_string()) + } + async fn clear_episodic_memory(&self, older_than_days: u32) -> PlatformResult { let base = self .agent_db_path diff --git a/crates/bubbaloop/src/mcp/mock_platform.rs b/crates/bubbaloop/src/mcp/mock_platform.rs index d73a41c5..c346dae9 100644 --- a/crates/bubbaloop/src/mcp/mock_platform.rs +++ b/crates/bubbaloop/src/mcp/mock_platform.rs @@ -36,6 +36,7 @@ impl MockPlatform { node_type: "rust".to_string(), installed: true, is_built: true, + instance_name: None, }]), configs: Mutex::new(HashMap::new()), missions: Mutex::new(Vec::new()), @@ -454,6 +455,10 @@ impl PlatformOperations for MockPlatform { fn zenoh_session(&self) -> Option> { self.zenoh_session.clone() } + + async fn get_node_instance_name(&self, name: &str) -> String { + name.to_string() + } } // ── Tests ──────────────────────────────────────────────────────────────── @@ -512,6 +517,7 @@ mod tests { node_type: "python".to_string(), installed: true, is_built: true, + instance_name: None, }, NodeInfo { name: "detector".to_string(), @@ -520,6 +526,7 @@ mod tests { node_type: "rust".to_string(), installed: true, is_built: false, + instance_name: None, }, NodeInfo { name: "tracker".to_string(), @@ -528,6 +535,7 @@ mod tests { node_type: "python".to_string(), installed: false, is_built: false, + instance_name: None, }, ]; let mock = mock_with_nodes(nodes); @@ -570,6 +578,7 @@ mod tests { node_type: "rust".to_string(), installed: true, is_built: true, + instance_name: None, }, NodeInfo { name: "beta".to_string(), @@ -578,6 +587,7 @@ mod tests { node_type: "python".to_string(), installed: false, is_built: false, + instance_name: None, }, ]; let mock = mock_with_nodes(nodes); @@ -1137,6 +1147,7 @@ mod tests { node_type: "rust".to_string(), installed: true, is_built: true, + instance_name: None, }; let json = serde_json::to_value(&info).unwrap(); assert_eq!(json["name"], "test"); diff --git a/crates/bubbaloop/src/mcp/platform.rs b/crates/bubbaloop/src/mcp/platform.rs index 56a42af1..369d264d 100644 --- a/crates/bubbaloop/src/mcp/platform.rs +++ b/crates/bubbaloop/src/mcp/platform.rs @@ -30,6 +30,10 @@ pub struct NodeInfo { pub node_type: String, pub installed: bool, pub is_built: bool, + /// The Zenoh instance name (from the node's config YAML `name:` field). + /// May differ from `name` when a config_override is used (e.g. multi-instance cameras). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub instance_name: Option, } /// Command to execute on a node. @@ -287,6 +291,17 @@ pub trait PlatformOperations: Send + Sync + 'static { /// `DaemonPlatform` returns `Some`; `MockPlatform` returns `None`. /// Used by `grab_frame` to pull camera samples directly from SHM. fn zenoh_session(&self) -> Option>; + + /// Resolve the Zenoh instance name for a node registered under `name`. + /// + /// Returns the `name:` field from the node's config YAML (published_instance_name) + /// when a config_override is set; otherwise returns `name` unchanged. + /// Used by `grab_frame` to build the correct Zenoh key when the registry + /// name differs from the node's published topic prefix. + fn get_node_instance_name( + &self, + name: &str, + ) -> impl std::future::Future + Send; } /// Parameters for creating or updating a belief. diff --git a/dashboard/src/components/ChatView.tsx b/dashboard/src/components/ChatView.tsx index 130e2d5d..308235d6 100644 --- a/dashboard/src/components/ChatView.tsx +++ b/dashboard/src/components/ChatView.tsx @@ -16,6 +16,7 @@ type ChatEntry = | { kind: "assistant"; text: string } | { kind: "tool"; name: string; input: string } | { kind: "tool_result"; text: string } + | { kind: "image"; src: string } | { kind: "error"; text: string } | { kind: "system"; text: string } | { kind: "connection_lost" } @@ -184,6 +185,12 @@ function EntryRow({ entry, onDismissOauth }: { entry: ChatEntry; onDismissOauth: ); } + case "image": + return ( +

+ Camera frame +
+ ); case "error": return (
@@ -381,6 +388,16 @@ export function ChatView({ session }: { session?: Session }) { setEntries((prev) => [...prev, { kind: "tool_result", text: event.text ?? "" }]); break; + case "image": + if (event.image && event.media_type) { + flushDeltaBuffer(); + setEntries((prev) => [ + ...prev, + { kind: "image", src: `data:${event.media_type};base64,${event.image}` }, + ]); + } + break; + case "error": flushDeltaBuffer(); setEntries((prev) => [...prev, { kind: "error", text: event.text ?? "Unknown error" }]); @@ -936,6 +953,19 @@ export function ChatView({ session }: { session?: Session }) { font-style: italic; } + .chat-image { + background: transparent; + border: none; + padding: 2px 14px; + } + .chat-frame { + max-width: 420px; + width: 100%; + border-radius: 8px; + border: 1px solid var(--border-color); + display: block; + } + .chat-connection-lost { justify-content: center; align-items: center; diff --git a/dashboard/src/lib/agent.ts b/dashboard/src/lib/agent.ts index f9d0d9ed..1c922456 100644 --- a/dashboard/src/lib/agent.ts +++ b/dashboard/src/lib/agent.ts @@ -12,6 +12,7 @@ export type AgentEventType = | "delta" | "tool" | "tool_result" + | "image" | "error" | "done" | "system"; @@ -24,6 +25,10 @@ export interface AgentEvent { text?: string; /** Tool input JSON snippet (only on "tool" events) */ input?: string; + /** Base64-encoded image data (only on "image" events) */ + image?: string; + /** MIME type for the image, e.g. "image/jpeg" (only on "image" events) */ + media_type?: string; } export interface AgentManifest { @@ -84,7 +89,9 @@ async function queryWithTimeout( selector: string, timeoutMs: number ): Promise { - try { + // Collect replies from the receiver into the array, stopping after timeoutMs + // regardless of whether the zenoh-js async iterator closes on its own. + const collect = async (): Promise => { const receiver = await session.get(selector, { timeout: Duration.milliseconds.of(timeoutMs), }); @@ -106,6 +113,16 @@ async function queryWithTimeout( replies.push(new TextDecoder().decode(bytes)); } return replies; + }; + + // Race against a hard deadline — zenoh-js receivers don't always close when + // the server-side timeout fires, which leaves the for-await loop hanging. + const deadline = new Promise((resolve) => + setTimeout(() => resolve([]), timeoutMs + 500) + ); + + try { + return await Promise.race([collect(), deadline]); } catch { return []; } From 5222566cfc7a93ba60d1744ac3044bd5d8f6269e Mon Sep 17 00:00:00 2001 From: edgarriba Date: Sat, 16 May 2026 23:56:23 +0200 Subject: [PATCH 03/13] fix(agent): fix Gemini 400 on multi-turn grab_frame conversations MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two bugs caused "function response must come immediately after function call turn" errors on the second grab_frame call in a conversation: 1. **functionResponse.name mismatch** — to_gemini_contents was using the internal tool_use_id UUID ("gemini-abc123") as the functionResponse.name, but Gemini requires it to match the functionCall.name ("grab_frame"). Added tool_name_map() to scan ToolUse blocks and look up real function names when building FunctionResponse parts. 2. **base64 blob accumulation** — labeled_images were pushed permanently to short_term. On subsequent grab_frame calls, the accumulated image messages caused role-sequencing inconsistencies for Gemini's validator. Fixed by making the image message transient: it's queued in pending_image_msg for the analysis provider call only, then discarded. Dashboard images still stream via SSE events as before. Adds regression test: to_gemini_contents_function_response_uses_function_name_not_id. Co-Authored-By: Claude Sonnet 4.6 --- crates/bubbaloop/src/agent/mod.rs | 26 ++++++-- crates/bubbaloop/src/agent/provider/gemini.rs | 60 ++++++++++++++++++- 2 files changed, 80 insertions(+), 6 deletions(-) diff --git a/crates/bubbaloop/src/agent/mod.rs b/crates/bubbaloop/src/agent/mod.rs index 52f9f695..29be0169 100644 --- a/crates/bubbaloop/src/agent/mod.rs +++ b/crates/bubbaloop/src/agent/mod.rs @@ -394,6 +394,11 @@ async fn run_turn_loop< let mut tool_call_counts: HashMap = HashMap::new(); let mut loop_detected = false; let max_turns = soul.capabilities.max_turns; + // Transient camera frames for the current inner iteration only. + // Grab_frame stashes base64 JPEGs here; they are appended to the provider + // call slice but NOT stored in short_term, so they don't accumulate across + // outer conversation turns (each turn starts fresh without historical blobs). + let mut pending_image_msg: Option = None; for _turn in 0..max_turns { if loop_detected { break; @@ -405,9 +410,18 @@ async fn run_turn_loop< )) .await; } + // Build effective message slice: short_term + optional transient camera frames. + // Cow-like: only clones when there's a pending image (rare), otherwise borrows. + let extended_opt: Option> = pending_image_msg.take().map(|img| { + let mut m = memory.short_term.clone(); + m.push(img); + m + }); + let provider_messages = extended_opt.as_deref().unwrap_or(&memory.short_term); + // Stream response from provider (retry handled inside ClaudeProvider) let mut rx = match provider - .generate_stream(Some(system_prompt), &memory.short_term, tools) + .generate_stream(Some(system_prompt), provider_messages, tools) .await { Ok(r) => r, @@ -605,8 +619,11 @@ async fn run_turn_loop< let tool_use_ids: Vec = tool_results.iter().map(|(id, _, _)| id.clone()).collect(); memory.short_term.push(Message::tool_results(tool_results)); - // Inject captured camera frames as a follow-up user message with labeled images. + // Inject captured camera frames as a transient follow-up for the next provider call. // Images are stashed out-of-band by grab_frame (base64 never enters ToolResult). + // We do NOT push to short_term: storing large base64 blobs permanently causes + // Gemini's function_call/response name validator to fail on multi-turn sessions + // as the accumulated image messages break the expected sequence structure. let attachments = dispatcher.drain_images(&tool_use_ids); if !attachments.is_empty() { // Stream each frame to the dashboard so it appears in the chat. @@ -622,9 +639,8 @@ async fn run_turn_loop< .into_iter() .map(|a| (a.label, a.media_type, a.base64)) .collect(); - memory - .short_term - .push(Message::user_with_labeled_images(pairs)); + // Queued for the next generate_stream call (analysis turn); dropped after use. + pending_image_msg = Some(Message::user_with_labeled_images(pairs)); } } diff --git a/crates/bubbaloop/src/agent/provider/gemini.rs b/crates/bubbaloop/src/agent/provider/gemini.rs index 6ee97e82..ccb1660b 100644 --- a/crates/bubbaloop/src/agent/provider/gemini.rs +++ b/crates/bubbaloop/src/agent/provider/gemini.rs @@ -12,6 +12,7 @@ use super::{ use futures::StreamExt; use serde::{Deserialize, Serialize}; use serde_json::Value; +use std::collections::HashMap; // ── Constants ──────────────────────────────────────────────────────── @@ -243,12 +244,30 @@ impl GeminiProvider { // ── Message conversion ─────────────────────────────────────────────── +/// Build a map from tool_use_id → function name by scanning all ToolUse blocks. +/// +/// Gemini requires `functionResponse.name` to exactly match the `functionCall.name` +/// it's responding to. Our internal `tool_use_id` is an opaque UUID; this map lets +/// the converter look up the real function name for each response. +fn tool_name_map(messages: &[Message]) -> HashMap { + let mut map = HashMap::new(); + for msg in messages { + for block in &msg.content { + if let ContentBlock::ToolUse { id, name, .. } = block { + map.insert(id.clone(), name.clone()); + } + } + } + map +} + /// Convert our canonical `Message` slice to Gemini `contents` + optional system prompt. /// /// Consecutive messages with the same role are coalesced into a single `GeminiContent` entry /// by concatenating their `parts[]`. Gemini requires strict role alternation; our agent loop /// emits two consecutive user messages (tool_results then labeled images) in the same turn. fn to_gemini_contents(messages: &[Message]) -> Vec { + let names = tool_name_map(messages); let mut out: Vec = Vec::new(); for msg in messages { let role = match msg.role.as_str() { @@ -287,9 +306,15 @@ fn to_gemini_contents(messages: &[Message]) -> Vec { } else { serde_json::json!({ "output": content }) }; + // Use the actual function name so Gemini can match call↔response. + // Falling back to tool_use_id is safe but should never happen in practice. + let fn_name = names + .get(tool_use_id) + .map(|s| s.as_str()) + .unwrap_or(tool_use_id.as_str()); parts.push(GeminiPart::FunctionResponse(GeminiFunctionResponse { function_response: GeminiFnResponse { - name: tool_use_id.clone(), + name: fn_name.to_string(), response, }, })); @@ -689,6 +714,39 @@ mod tests { } } + /// Regression test: functionResponse.name must be the function name ("grab_frame"), + /// not the internal tool_use_id UUID. Gemini matches call↔response by name. + #[test] + fn to_gemini_contents_function_response_uses_function_name_not_id() { + let uuid_id = "gemini-550e8400-e29b-41d4-a716-446655440000"; + let msgs = vec![ + Message { + role: "assistant".to_string(), + content: vec![ContentBlock::ToolUse { + id: uuid_id.to_string(), + name: "grab_frame".to_string(), + input: json!({"node_name": "tapo-terrace"}), + }], + }, + Message::tool_results(vec![( + uuid_id.to_string(), + r#"{"camera":"tapo-terrace"}"#.to_string(), + None, + )]), + ]; + let contents = to_gemini_contents(&msgs); + // contents[1] is the user turn with function_response + match &contents[1].parts[0] { + GeminiPart::FunctionResponse(fr) => { + assert_eq!( + fr.function_response.name, "grab_frame", + "functionResponse.name must be the function name, not the UUID" + ); + } + _ => panic!("expected FunctionResponse part"), + } + } + #[test] fn to_gemini_tools_empty() { let tools = to_gemini_tools(&[]); From 71db67c5e27dc2a9c1de1376fc2b668c65feb698 Mon Sep 17 00:00:00 2001 From: edgarriba Date: Sun, 17 May 2026 06:59:10 +0200 Subject: [PATCH 04/13] fix(agent): gate grab_frame to explicit user vision requests only The agent was calling grab_frame proactively (e.g. on "Hi"), sending unsolicited camera frames to the dashboard. Added explicit guidance to the system prompt: only call grab_frame when the user's message explicitly requests visual information (see, show, visualize, analyze a camera feed). Co-Authored-By: Claude Sonnet 4.6 --- crates/bubbaloop/src/agent/prompt.rs | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/crates/bubbaloop/src/agent/prompt.rs b/crates/bubbaloop/src/agent/prompt.rs index 5cfe1cb3..b3f8a2ad 100644 --- a/crates/bubbaloop/src/agent/prompt.rs +++ b/crates/bubbaloop/src/agent/prompt.rs @@ -246,11 +246,14 @@ pub fn build_system_prompt_with_soul_path( - **Missions:** list, pause, resume, cancel missions\n\ - **Safety:** register and list constraints per mission\n\ - **Vision:** `grab_frame` captures a JPEG from a camera sensor node — the image is \ - delivered directly in the next turn for you to inspect\n\n\ + delivered directly in the next turn for you to inspect. \ + **Only call grab_frame when the user explicitly asks to see, show, visualize, \ + or analyze a camera feed.** Never call it proactively or for general status checks.\n\n\ Use the right tool for the job. For node operations, use the dedicated node tools.\n\ For everything else, use read_file, write_file, or run_command.\n\n\ To analyze what cameras see, call `grab_frame` for each sensor — the JPEG frame\n\ - is delivered as an image content block in the next turn for you to inspect directly.\n\n\ + is delivered as an image content block in the next turn for you to inspect directly.\n\ + Do NOT call grab_frame unless the user's message explicitly requests visual information.\n\n\ ### Important: recurring vs one-off actions\n\n\ - **run_command** blocks until the command exits. NEVER use shell loops (`while true`, `watch`, `for i in`) \ with run_command — they block for the full 30-second timeout and produce no output until they time out.\n\ From de2898882f4b525562abad336916399d5c6e5fd2 Mon Sep 17 00:00:00 2001 From: edgarriba Date: Sun, 17 May 2026 07:12:16 +0200 Subject: [PATCH 05/13] fix(agent): no proactive tool calls on conversational messages MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Added explicit rule at top of the "Rules" section: only call tools when the user explicitly requests an action or investigation. Greetings, chitchat, and conversational messages get a plain text reply — no system_status, grab_frame, or any other tool calls. Co-Authored-By: Claude Sonnet 4.6 --- crates/bubbaloop/src/agent/prompt.rs | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/crates/bubbaloop/src/agent/prompt.rs b/crates/bubbaloop/src/agent/prompt.rs index b3f8a2ad..8aa79c0e 100644 --- a/crates/bubbaloop/src/agent/prompt.rs +++ b/crates/bubbaloop/src/agent/prompt.rs @@ -219,7 +219,11 @@ pub fn build_system_prompt_with_soul_path( 3. **Evaluate** — After each step, check: did it work? Is the goal met?\n\ 4. **Iterate** — If not done, adjust and continue. If done, report results.\n\n\ ### Rules\n\n\ - - **Act, don't ask.** Never say \"Would you like me to...?\" or \"Should I...?\" — do it.\n\ + - **Only call tools when the user explicitly requests an action or investigation.** \ + Greetings, chitchat, or conversational messages (\"Hi\", \"How are you?\", \"Thanks\") \ + get a plain text reply — no tool calls.\n\ + - **Act, don't ask.** When the user asks you to do something, do it — never say \ + \"Would you like me to...?\" or \"Should I...?\".\n\ - **Keep going.** Don't stop after one tool call if the task needs more.\n\ - **Verify.** After making a change, confirm it worked (check health, read output, etc).\n\ - **Be concise.** Report what you did and the result. Skip preamble.\n\ From 759cb89772003e2e1fa127ef30209914fe97c540 Mon Sep 17 00:00:00 2001 From: edgarriba Date: Sun, 17 May 2026 08:11:16 +0200 Subject: [PATCH 06/13] =?UTF-8?q?fix(agent):=20batch=20memory=5Fsearch=20?= =?UTF-8?q?=E2=80=94=20discourage=20multiple=20narrow=20queries?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add explicit instruction in tool description and query param hint to combine related keywords into a single broad query rather than firing one memory_search call per keyword. Reduces 3-turn memory lookups to 1 turn for typical "check memories about X" requests. Co-Authored-By: Claude Sonnet 4.6 --- crates/bubbaloop/src/agent/dispatch.rs | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/crates/bubbaloop/src/agent/dispatch.rs b/crates/bubbaloop/src/agent/dispatch.rs index 8bbb4796..8d19f9ca 100644 --- a/crates/bubbaloop/src/agent/dispatch.rs +++ b/crates/bubbaloop/src/agent/dispatch.rs @@ -529,14 +529,17 @@ impl Dispatcher

{ 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", From 30a935e2ac50179b6ccd6dc40d2760a2c26dbe12 Mon Sep 17 00:00:00 2001 From: edgarriba Date: Sun, 17 May 2026 08:11:26 +0200 Subject: [PATCH 07/13] feat(chat): compact tool calls, image gallery, scroll-to-bottom button MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Tool calls collapse to a single expandable line showing tool name, turn number, and episode count; result fills in when it arrives - System context events (memory episodes, turn N/M) are no longer shown as chat rows — metadata is captured in refs and surfaced inside tool call lines only - Consecutive image entries are grouped into a horizontal thumbnail strip (110×62px, nowrap, overflow-x scroll); single image renders at natural size up to 320px; click any thumbnail opens a fullscreen lightbox (ESC or click outside to dismiss) - Scroll-to-bottom floating button appears when user has scrolled up more than 80px; disappears when back at bottom; sending a message resets to auto-scroll Co-Authored-By: Claude Sonnet 4.6 --- dashboard/src/components/ChatView.tsx | 412 ++++++++++++++++++++------ 1 file changed, 316 insertions(+), 96 deletions(-) diff --git a/dashboard/src/components/ChatView.tsx b/dashboard/src/components/ChatView.tsx index 308235d6..0472aae7 100644 --- a/dashboard/src/components/ChatView.tsx +++ b/dashboard/src/components/ChatView.tsx @@ -14,8 +14,7 @@ import { parseSseStream } from "../lib/sse"; type ChatEntry = | { kind: "user"; text: string } | { kind: "assistant"; text: string } - | { kind: "tool"; name: string; input: string } - | { kind: "tool_result"; text: string } + | { kind: "tool_call"; name: string; input: string; result: string | null; turn: number; episodes: number } | { kind: "image"; src: string } | { kind: "error"; text: string } | { kind: "system"; text: string } @@ -136,6 +135,88 @@ function AgentSelector({ ); } +function parseEpisodes(text: string): number { + const m = text.match(/(\d+)\s+episode/); + return m ? parseInt(m[1], 10) : 0; +} + +function ToolCallRow({ entry }: { entry: Extract }) { + const [expanded, setExpanded] = useState(false); + const pending = entry.result === null; + return ( +

+ + {expanded && ( +
+ {entry.input && ( +
+ args + 
{entry.input}
+
+ )} + {entry.result !== null && ( +
+ result + 
{entry.result}
+
+ )} +
+ )} +
+ ); +} + +function ImageGallery({ srcs }: { srcs: string[] }) { + const [lightbox, setLightbox] = useState(null); + + const openLightbox = (src: string) => setLightbox(src); + const closeLightbox = () => setLightbox(null); + + useEffect(() => { + if (!lightbox) return; + const onKey = (e: KeyboardEvent) => { if (e.key === "Escape") closeLightbox(); }; + window.addEventListener("keydown", onKey); + return () => window.removeEventListener("keydown", onKey); + }, [lightbox]); + + return ( +
+
+ {srcs.map((src, i) => ( + {`Camera openLightbox(src)} + /> + ))} +
+ {lightbox && ( +
+ Camera frame full size e.stopPropagation()} + /> + +
+ )} +
+ ); +} + function EntryRow({ entry, onDismissOauth }: { entry: ChatEntry; onDismissOauth: () => void }) { switch (entry.kind) { case "user": @@ -152,45 +233,10 @@ function EntryRow({ entry, onDismissOauth }: { entry: ChatEntry; onDismissOauth: {entry.text}
); - case "tool": - return ( -
-
- - - - {entry.name} -
- {entry.input && ( -
- args - 
{entry.input}
-
- )} -
- ); - case "tool_result": { - const preview = entry.text.length > 80 ? entry.text.slice(0, 80) + "…" : entry.text; - const hasMore = entry.text.length > 80; - return ( -
- {hasMore ? ( -
- {preview} - 
{entry.text}
-
- ) : ( - {entry.text} - )} -
- ); - } + case "tool_call": + return ; case "image": - return ( -
- Camera frame -
- ); + return null; case "error": return (
@@ -266,8 +312,12 @@ export function ChatView({ session }: { session?: Session }) { const logEndRef = useRef(null); const chatBodyRef = useRef(null); const userScrolledRef = useRef(false); + const [showScrollBtn, setShowScrollBtn] = useState(false); // Accumulates assistant delta text across SSE frames before flushing to state const deltaBufferRef = useRef(""); + // Turn counter and episode count for the current agent response + const turnRef = useRef(0); + const episodesRef = useRef(0); // Abort any in-flight SSE stream when the component unmounts useEffect(() => () => { abortRef.current?.abort(); }, []); @@ -278,6 +328,21 @@ export function ChatView({ session }: { session?: Session }) { logEndRef.current?.scrollIntoView({ behavior: "smooth" }); }, [entries]); + const handleScroll = useCallback(() => { + const el = chatBodyRef.current; + if (!el) return; + const distFromBottom = el.scrollHeight - el.scrollTop - el.clientHeight; + const atBottom = distFromBottom < 80; + userScrolledRef.current = !atBottom; + setShowScrollBtn(!atBottom); + }, []); + + const scrollToBottom = useCallback(() => { + userScrolledRef.current = false; + setShowScrollBtn(false); + logEndRef.current?.scrollIntoView({ behavior: "smooth" }); + }, []); + const runDiscovery = useCallback((s: Session) => { setDiscovering(true); discoverAgents(s) @@ -315,8 +380,11 @@ export function ChatView({ session }: { session?: Session }) { setInput(""); setEntries((prev) => [...prev, { kind: "user", text: message }]); userScrolledRef.current = false; + setShowScrollBtn(false); setStreaming(true); deltaBufferRef.current = ""; + turnRef.current = 0; + episodesRef.current = 0; const ctrl = new AbortController(); abortRef.current = ctrl; @@ -377,15 +445,35 @@ export function ChatView({ session }: { session?: Session }) { case "tool": flushDeltaBuffer(); + turnRef.current += 1; setEntries((prev) => [ ...prev, - { kind: "tool", name: event.text ?? "tool", input: event.input ?? "" }, + { + kind: "tool_call", + name: event.text ?? "tool", + input: event.input ?? "", + result: null, + turn: turnRef.current, + episodes: episodesRef.current, + }, ]); break; case "tool_result": flushDeltaBuffer(); - setEntries((prev) => [...prev, { kind: "tool_result", text: event.text ?? "" }]); + setEntries((prev) => { + const idx = [...prev].reverse().findIndex( + (e) => e.kind === "tool_call" && e.result === null + ); + if (idx === -1) return prev; + const realIdx = prev.length - 1 - idx; + const updated = [...prev]; + updated[realIdx] = { + ...(updated[realIdx] as Extract), + result: event.text ?? "", + }; + return updated; + }); break; case "image": @@ -403,10 +491,13 @@ export function ChatView({ session }: { session?: Session }) { setEntries((prev) => [...prev, { kind: "error", text: event.text ?? "Unknown error" }]); break; - case "system": + case "system": { flushDeltaBuffer(); - setEntries((prev) => [...prev, { kind: "system", text: event.text ?? "" }]); + const sysText = event.text ?? ""; + episodesRef.current = parseEpisodes(sysText); + turnRef.current = 0; break; + } case "done": flushDeltaBuffer(); @@ -465,7 +556,7 @@ export function ChatView({ session }: { session?: Session }) {
{/* Scrollable area: token setup + agent bar + messages */} -
{ userScrolledRef.current = true; }}> +
{!token && (
@@ -527,14 +618,40 @@ export function ChatView({ session }: { session?: Session }) {

Try: "what nodes are running?" or "describe the current system state"

)} - {entries.map((entry, i) => ( - - ))} + {(() => { + const nodes: React.ReactNode[] = []; + let i = 0; + while (i < entries.length) { + const entry = entries[i]; + if (entry.kind === "image") { + const start = i; + const srcs: string[] = []; + while (i < entries.length && entries[i].kind === "image") { + srcs.push((entries[i] as Extract).src); + i++; + } + nodes.push(); + } else { + nodes.push(); + i++; + } + } + return nodes; + })()}
+ {/* Scroll-to-bottom button — appears when user has scrolled up */} + {showScrollBtn && ( + + )} + {/* Input bar — full width, always pinned at bottom */}
@@ -580,6 +697,36 @@ export function ChatView({ session }: { session?: Session }) { min-height: 0; overflow: hidden; width: 100%; + position: relative; + } + + .scroll-to-bottom { + position: absolute; + bottom: 74px; + right: 24px; + width: 34px; + height: 34px; + border-radius: 50%; + background: var(--bg-secondary); + border: 1px solid var(--border-color); + color: var(--text-secondary); + cursor: pointer; + display: flex; + align-items: center; + justify-content: center; + box-shadow: 0 2px 12px rgba(0,0,0,0.35); + transition: border-color 0.15s, color 0.15s, transform 0.15s; + z-index: 10; + animation: scroll-btn-in 0.15s ease; + } + .scroll-to-bottom:hover { + border-color: var(--accent-primary); + color: var(--accent-primary); + transform: translateY(-2px); + } + @keyframes scroll-btn-in { + from { opacity: 0; transform: translateY(6px); } + to { opacity: 1; transform: translateY(0); } } /* Token setup */ @@ -875,42 +1022,69 @@ export function ChatView({ session }: { session?: Session }) { font-size: 10px; } - .chat-tool { - background: transparent; - border: none; - padding: 2px 14px; - gap: 6px; - align-items: flex-start; - flex-direction: column; + .chat-tool-call { + padding: 0 14px; } - .tool-tool-row { + .tool-call-header { display: flex; align-items: center; - gap: 6px; + gap: 7px; + width: 100%; + background: none; + border: none; + cursor: pointer; + padding: 4px 0; + color: var(--text-muted); + font-family: inherit; + text-align: left; } - .tool-name { + .tool-call-header:hover .tool-call-name { color: var(--accent-secondary); } + .tool-call-name { font-size: 12px; font-weight: 600; font-family: 'JetBrains Mono', monospace; - color: var(--accent-secondary); + color: var(--text-secondary); + transition: color 0.15s; } - .tool-details { display: block; width: 100%; } - .tool-summary { + .tool-call-meta { font-size: 11px; color: var(--text-muted); - cursor: pointer; - user-select: none; - list-style: none; - padding: 1px 7px; - border: 1px solid var(--border-color); - border-radius: 10px; - transition: border-color 0.15s, color 0.15s; + font-family: 'JetBrains Mono', monospace; + opacity: 0.7; + } + .tool-call-status { + font-size: 10px; + color: var(--accent-secondary); + font-family: 'JetBrains Mono', monospace; + margin-left: auto; + } + .tool-call-status.pending { color: var(--text-muted); } + .tool-call-chevron { + flex-shrink: 0; + transition: transform 0.15s; + color: var(--text-muted); + } + .tool-call-chevron.open { transform: rotate(180deg); } + .tool-call-body { + display: flex; + flex-direction: column; + gap: 6px; + padding: 4px 0 6px 18px; + } + .tool-call-section { + display: flex; + flex-direction: column; + gap: 3px; + } + .tool-call-label { + font-size: 10px; + text-transform: uppercase; + letter-spacing: 0.06em; + color: var(--text-muted); + font-family: 'JetBrains Mono', monospace; } - .tool-summary:hover { border-color: var(--accent-secondary); color: var(--text-secondary); } - .tool-summary::-webkit-details-marker { display: none; } .tool-json { - width: 100%; - margin: 4px 0 2px; + margin: 0; padding: 8px 10px; background: var(--bg-primary); border-radius: 6px; @@ -922,22 +1096,6 @@ export function ChatView({ session }: { session?: Session }) { overflow: hidden; } - .chat-tool-result { - background: transparent; - border: none; - padding: 1px 14px 3px 30px; - } - .tool-result-summary { - font-family: 'JetBrains Mono', monospace; - } - .tool-result-text { - font-size: 12px; - color: var(--text-secondary); - font-family: 'JetBrains Mono', monospace; - white-space: pre-wrap; - word-break: break-all; - } - .chat-error { background: rgba(255, 23, 68, 0.08); border: 1px solid rgba(255, 23, 68, 0.2); @@ -953,18 +1111,80 @@ export function ChatView({ session }: { session?: Session }) { font-style: italic; } - .chat-image { - background: transparent; - border: none; - padding: 2px 14px; + /* Image gallery */ + .chat-image-gallery { + padding: 4px 14px; } - .chat-frame { - max-width: 420px; - width: 100%; - border-radius: 8px; + .image-grid { + display: flex; + flex-direction: row; + flex-wrap: nowrap; + gap: 6px; + overflow-x: auto; + padding-bottom: 2px; + } + .image-grid.single .image-thumb { + width: auto; + max-width: 320px; + height: 100px; + } + .image-thumb { + flex-shrink: 0; + width: 110px; + height: 62px; + object-fit: cover; + border-radius: 6px; border: 1px solid var(--border-color); + cursor: pointer; + transition: border-color 0.15s, transform 0.15s; display: block; } + .image-thumb:hover { + border-color: var(--accent-primary); + transform: scale(1.04); + } + + /* Lightbox */ + .lightbox-overlay { + position: fixed; + inset: 0; + z-index: 500; + background: rgba(0, 0, 0, 0.82); + display: flex; + align-items: center; + justify-content: center; + animation: lb-fade 0.15s ease; + } + @keyframes lb-fade { + from { opacity: 0; } + to { opacity: 1; } + } + .lightbox-img { + max-width: 90vw; + max-height: 85vh; + border-radius: 8px; + border: 1px solid rgba(255,255,255,0.12); + box-shadow: 0 8px 48px rgba(0,0,0,0.7); + display: block; + } + .lightbox-close { + position: absolute; + top: 18px; + right: 22px; + background: rgba(255,255,255,0.1); + border: 1px solid rgba(255,255,255,0.2); + border-radius: 50%; + color: #fff; + width: 32px; + height: 32px; + font-size: 14px; + cursor: pointer; + display: flex; + align-items: center; + justify-content: center; + transition: background 0.15s; + } + .lightbox-close:hover { background: rgba(255,255,255,0.2); } .chat-connection-lost { justify-content: center; From 5ce3567305f311b32e897c28db6c1b6767041402 Mon Sep 17 00:00:00 2001 From: edgarriba Date: Sun, 17 May 2026 08:15:30 +0200 Subject: [PATCH 08/13] refactor(chat): simplify ChatView after code review - Extract agentColor() helper and AgentTriggerContent component to deduplicate color/name/model rendering across single-agent chip, dropdown trigger, and pill - Move agent-panel-header inline style to .agent-panel-title class - Replace [...prev].reverse().findIndex() with a backward for-loop in tool_result handler (no array copy, no index arithmetic, no cast) - Lift entries render IIFE into renderEntries() top-level helper Co-Authored-By: Claude Sonnet 4.6 --- dashboard/src/components/ChatView.tsx | 111 ++++++++++++++------------ 1 file changed, 61 insertions(+), 50 deletions(-) diff --git a/dashboard/src/components/ChatView.tsx b/dashboard/src/components/ChatView.tsx index 0472aae7..ea9cb638 100644 --- a/dashboard/src/components/ChatView.tsx +++ b/dashboard/src/components/ChatView.tsx @@ -35,6 +35,20 @@ function providerFromModel(model: string): string { return "ollama"; } +function agentColor(model: string): string { + return PROVIDER_COLORS[providerFromModel(model)] ?? "#888"; +} + +function AgentTriggerContent({ agent }: { agent: ResolvedAgent }) { + return ( + <> + + {agent.agent.agent_id} + {agent.agent.model.split("-")[0]} + + ); +} + function AgentSelector({ agents, selected, @@ -62,33 +76,22 @@ function AgentSelector({ ); } if (agents.length === 1) { - const a = agents[0]; return (
- - {a.agent.agent_id} - {a.agent.model.split("-")[0]} +
); } return (
- {/* Trigger chip showing the active agent */} - {/* Dropdown panel */} {open && ( <>
setOpen(false)} />
- Agents on {agents[0]?.agent.machine_id} + Agents on {agents[0]?.agent.machine_id}
{agents.map((a) => { - const provider = providerFromModel(a.agent.model); - const color = PROVIDER_COLORS[provider] ?? "#888"; const isSelected = a.agent.agent_id === selected?.agent.agent_id; return (
)} - {(() => { - const nodes: React.ReactNode[] = []; - let i = 0; - while (i < entries.length) { - const entry = entries[i]; - if (entry.kind === "image") { - const start = i; - const srcs: string[] = []; - while (i < entries.length && entries[i].kind === "image") { - srcs.push((entries[i] as Extract).src); - i++; - } - nodes.push(); - } else { - nodes.push(); - i++; - } - } - return nodes; - })()} + {renderEntries(entries, dismissOauth)}
@@ -880,6 +884,13 @@ export function ChatView({ session }: { session?: Session }) { text-transform: uppercase; letter-spacing: 0.06em; } + .agent-panel-title { + flex: 1; + min-width: 0; + overflow: hidden; + text-overflow: ellipsis; + white-space: nowrap; + } .agent-refresh-btn { background: none; border: none; From 1cbc0818528982874053667a3bc6a941b9f9edf6 Mon Sep 17 00:00:00 2001 From: edgarriba Date: Sun, 17 May 2026 08:32:42 +0200 Subject: [PATCH 09/13] =?UTF-8?q?fix(chat):=20address=20code=20review=20?= =?UTF-8?q?=E2=80=94=20race=20conditions=20+=20robustness?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - streamingRef (not just state) guards against concurrent handleSend invocations before setStreaming(true) commits - mountedRef prevents setState after unmount; setStreaming(false) gated in finally; error/connection_lost gated in catch - programmaticScrollRef 600ms window stops smooth-scroll feedback loop from flipping userScrolledRef and showing scroll button mid-stream - handleSelectAgent aborts in-flight stream when user switches agents - runDiscovery preserves current selection on refresh; falls back to default/first only when selected agent is no longer in the list - Token trimmed before both saveToken() and setToken() to prevent trailing-whitespace Bearer header - episodesRef only updated when parseEpisodes() returns > 0 so unrelated system messages don't clobber a valid count - Network TypeError classified as connection_lost not raw error - IME composition guard on Enter key (CJK input) Co-Authored-By: Claude Sonnet 4.6 --- dashboard/src/components/ChatView.tsx | 55 ++++++++++++++++++--------- 1 file changed, 38 insertions(+), 17 deletions(-) diff --git a/dashboard/src/components/ChatView.tsx b/dashboard/src/components/ChatView.tsx index ea9cb638..1a59ff15 100644 --- a/dashboard/src/components/ChatView.tsx +++ b/dashboard/src/components/ChatView.tsx @@ -332,26 +332,28 @@ export function ChatView({ session }: { session?: Session }) { const [streaming, setStreaming] = useState(false); const abortRef = useRef(null); + const mountedRef = useRef(true); + const streamingRef = useRef(false); + const programmaticScrollRef = useRef(0); const logEndRef = useRef(null); const chatBodyRef = useRef(null); const userScrolledRef = useRef(false); const [showScrollBtn, setShowScrollBtn] = useState(false); - // Accumulates assistant delta text across SSE frames before flushing to state const deltaBufferRef = useRef(""); - // Turn counter and episode count for the current agent response const turnRef = useRef(0); const episodesRef = useRef(0); - // Abort any in-flight SSE stream when the component unmounts - useEffect(() => () => { abortRef.current?.abort(); }, []); + useEffect(() => () => { mountedRef.current = false; abortRef.current?.abort(); }, []); - // Auto-scroll to bottom when new entries arrive, unless user has scrolled up + // Auto-scroll; stamp a 600ms window so handleScroll ignores programmatic events useEffect(() => { if (userScrolledRef.current) return; + programmaticScrollRef.current = Date.now() + 600; logEndRef.current?.scrollIntoView({ behavior: "smooth" }); }, [entries]); const handleScroll = useCallback(() => { + if (Date.now() < programmaticScrollRef.current) return; const el = chatBodyRef.current; if (!el) return; const distFromBottom = el.scrollHeight - el.scrollTop - el.clientHeight; @@ -371,9 +373,14 @@ export function ChatView({ session }: { session?: Session }) { discoverAgents(s) .then((found) => { setAgents(found); - const def = found.find((a) => a.agent.is_default) ?? found[0] ?? null; - setSelectedAgent(def); + setSelectedAgent((cur) => + found.find((a) => a.agent.agent_id === cur?.agent.agent_id) + ?? found.find((a) => a.agent.is_default) + ?? found[0] + ?? null + ); }) + .catch(() => setAgents([])) .finally(() => setDiscovering(false)); }, []); @@ -399,7 +406,8 @@ export function ChatView({ session }: { session?: Session }) { }, []); const handleSend = useCallback(async () => { - if (!selectedAgent || !input.trim() || streaming) return; + if (!selectedAgent || !input.trim() || streamingRef.current) return; + streamingRef.current = true; const message = input.trim(); setInput(""); @@ -516,8 +524,8 @@ export function ChatView({ session }: { session?: Session }) { case "system": { flushDeltaBuffer(); - const sysText = event.text ?? ""; - episodesRef.current = parseEpisodes(sysText); + const ep = parseEpisodes(event.text ?? ""); + if (ep > 0) episodesRef.current = ep; turnRef.current = 0; break; } @@ -536,12 +544,19 @@ export function ChatView({ session }: { session?: Session }) { } } catch (err) { if (ctrl.signal.aborted) return; + if (!mountedRef.current) return; flushDeltaBuffer(); - const msg = err instanceof Error ? err.message : String(err); - setEntries((prev) => [...prev, { kind: "error", text: msg }]); + const isNetworkDrop = err instanceof TypeError && err.message.toLowerCase().includes("network"); + if (isNetworkDrop) { + setEntries((prev) => [...prev, { kind: "connection_lost" }]); + } else { + const msg = err instanceof Error ? err.message : String(err); + setEntries((prev) => [...prev, { kind: "error", text: msg }]); + } } finally { + streamingRef.current = false; abortRef.current = null; - setStreaming(false); + if (mountedRef.current) setStreaming(false); } }, [selectedAgent, input, streaming, token, flushDeltaBuffer]); @@ -550,8 +565,9 @@ export function ChatView({ session }: { session?: Session }) { }, []); const handleTokenSave = useCallback(() => { - saveToken(tokenDraft); - setToken(tokenDraft.trim()); + const t = tokenDraft.trim(); + saveToken(t); + setToken(t); }, [tokenDraft]); const handleRefresh = useCallback(() => { @@ -563,9 +579,14 @@ export function ChatView({ session }: { session?: Session }) { setEntries((prev) => prev.filter((e) => e.kind !== "oauth_blocked")); }, []); + const handleSelectAgent = useCallback((a: ResolvedAgent) => { + if (streamingRef.current) abortRef.current?.abort(); + setSelectedAgent(a); + }, []); + const handleKeyDown = useCallback( (e: React.KeyboardEvent) => { - if (e.key === "Enter" && !e.shiftKey) { + if (e.key === "Enter" && !e.shiftKey && !e.nativeEvent.isComposing) { e.preventDefault(); handleSend(); } @@ -609,7 +630,7 @@ export function ChatView({ session }: { session?: Session }) { From a048134f1caa4b0a96fe3b5ce1a22da50c77ec42 Mon Sep 17 00:00:00 2001 From: edgarriba Date: Sun, 17 May 2026 08:36:48 +0200 Subject: [PATCH 10/13] fix(test): add missing instance_name field to NodeInfo in integration tests Co-Authored-By: Claude Sonnet 4.6 --- crates/bubbaloop/tests/integration_mcp.rs | 2 ++ 1 file changed, 2 insertions(+) diff --git a/crates/bubbaloop/tests/integration_mcp.rs b/crates/bubbaloop/tests/integration_mcp.rs index 325f33af..41ca5132 100644 --- a/crates/bubbaloop/tests/integration_mcp.rs +++ b/crates/bubbaloop/tests/integration_mcp.rs @@ -212,6 +212,7 @@ async fn list_nodes_multiple() { node_type: "python".to_string(), installed: true, is_built: true, + instance_name: None, }, NodeInfo { name: "detector".to_string(), @@ -220,6 +221,7 @@ async fn list_nodes_multiple() { node_type: "rust".to_string(), installed: true, is_built: false, + instance_name: None, }, ]; let h = TestHarness::with_mock(mock_with_nodes(nodes)).await; From 83836794545d97ea3b55810c7a638f13cfad89ba Mon Sep 17 00:00:00 2001 From: edgarriba Date: Sun, 17 May 2026 08:49:06 +0200 Subject: [PATCH 11/13] fix(test): use exact log-entry match in prompt_limits_episodic_to_5 Previous pattern "message " also matched "user's message explicitly" in the grab_frame prompt rule, giving count 6 instead of 5. Switch to per-index contains() check so only log entries are counted. Co-Authored-By: Claude Sonnet 4.6 --- crates/bubbaloop/src/agent/prompt.rs | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/crates/bubbaloop/src/agent/prompt.rs b/crates/bubbaloop/src/agent/prompt.rs index 8aa79c0e..86bc84e1 100644 --- a/crates/bubbaloop/src/agent/prompt.rs +++ b/crates/bubbaloop/src/agent/prompt.rs @@ -438,7 +438,9 @@ mod tests { }) .collect(); let prompt = build_system_prompt(&soul, "", &[], &episodes, None, None, None); - let count = prompt.matches("message ").count(); + let count = (0..10) + .filter(|i| prompt.contains(&format!("message {}", i))) + .count(); assert_eq!(count, 5, "should limit episodic context to 5 entries"); } From 7d37400679a507be2fda0f2c3e538ce0ddb1d219 Mon Sep 17 00:00:00 2001 From: edgarriba Date: Mon, 18 May 2026 14:28:04 +0200 Subject: [PATCH 12/13] fix(chat): clear Responding state + multi-camera grab_frame - Fix React StrictMode bug: mountedRef.current was permanently false after StrictMode's double-invoke cleanup, so setStreaming(false) in the finally block was silently skipped. Reset mountedRef.current=true in effect setup; drop the guard entirely (React 18 safe no-op). - Fix SSE generator hang: parseSseStream now accepts AbortSignal and races every reader.read() against it. Calling ctrl.abort() on done/ error immediately unblocks any pending read so generator.return() can complete and the outer finally runs. - Fix labeled-break no-op: break inside switch only exits the switch, not the for-await. Changed to sseLoop: label + break sseLoop. - Fix Stop button: ctrl.abort() now reliably interrupts the in-flight reader.read() via the abort promise race. - grab_frame: remove specific camera name examples from tool schema (biased LLM toward tapo-terrace). System prompt now instructs agent to call grab_frame for ALL camera nodes when no specific camera is named by the user. Co-Authored-By: Claude Sonnet 4.6 --- crates/bubbaloop/src/agent/camera.rs | 15 ++++++ crates/bubbaloop/src/agent/dispatch.rs | 10 ++-- crates/bubbaloop/src/agent/mod.rs | 10 +++- crates/bubbaloop/src/agent/prompt.rs | 9 +++- crates/bubbaloop/src/agent/provider/gemini.rs | 10 ++++ dashboard/src/components/CameraView.tsx | 36 +++++++++++-- dashboard/src/components/ChatView.tsx | 15 ++++-- dashboard/src/components/Dashboard.tsx | 30 +++++++++++ dashboard/src/lib/sse.ts | 52 ++++++++++++++----- dashboard/src/lib/storage.ts | 10 +++- 10 files changed, 165 insertions(+), 32 deletions(-) diff --git a/crates/bubbaloop/src/agent/camera.rs b/crates/bubbaloop/src/agent/camera.rs index eec7fb0b..77cc0962 100644 --- a/crates/bubbaloop/src/agent/camera.rs +++ b/crates/bubbaloop/src/agent/camera.rs @@ -54,6 +54,21 @@ pub async fn grab_frame( 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()) diff --git a/crates/bubbaloop/src/agent/dispatch.rs b/crates/bubbaloop/src/agent/dispatch.rs index 8d19f9ca..bc6b521a 100644 --- a/crates/bubbaloop/src/agent/dispatch.rs +++ b/crates/bubbaloop/src/agent/dispatch.rs @@ -699,16 +699,18 @@ impl Dispatcher

{ 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. Check the Current Sensor Inventory for \ - available camera node names. Returns a JSON receipt with capture \ - metadata (timestamp, resolution, frame_id)." + 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": "Name of the camera node (e.g. 'tapo-terrace', 'tapo-entrance')" + "description": "Exact name of the camera node from the Current Sensor Inventory" }, "machine_id": { "type": "string", diff --git a/crates/bubbaloop/src/agent/mod.rs b/crates/bubbaloop/src/agent/mod.rs index 29be0169..62eb1493 100644 --- a/crates/bubbaloop/src/agent/mod.rs +++ b/crates/bubbaloop/src/agent/mod.rs @@ -363,11 +363,19 @@ pub async fn run_agent_turn< // 3. Signal turn complete sink.emit(AgentEvent::done(correlation_id)).await; - // 4. Trim short-term memory + // 4. Trim short-term memory. + // After draining, advance past any leading non-user messages so the slice + // always starts with a user turn (Gemini requires contents[0].role == "user"). if memory.short_term.len() > MAX_CONVERSATION_MESSAGES { let drain_count = memory.short_term.len() - MAX_CONVERSATION_MESSAGES; memory.short_term.drain(..drain_count); } + // Advance past any leading non-user messages left by trimming. + if let Some(first_user) = memory.short_term.iter().position(|m| m.role == "user") { + if first_user > 0 { + memory.short_term.drain(..first_user); + } + } Ok(()) } diff --git a/crates/bubbaloop/src/agent/prompt.rs b/crates/bubbaloop/src/agent/prompt.rs index 86bc84e1..7b720c8f 100644 --- a/crates/bubbaloop/src/agent/prompt.rs +++ b/crates/bubbaloop/src/agent/prompt.rs @@ -255,8 +255,13 @@ pub fn build_system_prompt_with_soul_path( or analyze a camera feed.** Never call it proactively or for general status checks.\n\n\ Use the right tool for the job. For node operations, use the dedicated node tools.\n\ For everything else, use read_file, write_file, or run_command.\n\n\ - To analyze what cameras see, call `grab_frame` for each sensor — the JPEG frame\n\ - is delivered as an image content block in the next turn for you to inspect directly.\n\ + To analyze what cameras see, call `grab_frame` for **each** camera sensor node \ + listed in the Current Sensor Inventory — the JPEG frame is delivered as an image \ + content block in the next turn for you to inspect directly. \ + When the user asks generically (\"grab a frame\", \"what do cameras see\", \ + \"show me the cameras\") without naming a specific camera, \ + call grab_frame for ALL camera nodes, not just one. \ + Only target a single camera if the user names it explicitly.\n\ Do NOT call grab_frame unless the user's message explicitly requests visual information.\n\n\ ### Important: recurring vs one-off actions\n\n\ - **run_command** blocks until the command exits. NEVER use shell loops (`while true`, `watch`, `for i in`) \ diff --git a/crates/bubbaloop/src/agent/provider/gemini.rs b/crates/bubbaloop/src/agent/provider/gemini.rs index ccb1660b..ebb06890 100644 --- a/crates/bubbaloop/src/agent/provider/gemini.rs +++ b/crates/bubbaloop/src/agent/provider/gemini.rs @@ -424,6 +424,11 @@ impl ModelProvider for GeminiProvider { let status = response.status(); if !status.is_success() { let msg = response.text().await.unwrap_or_default(); + if status.as_u16() == 400 { + if let Ok(debug_json) = serde_json::to_string(&body.contents) { + log::error!("[Gemini] 400 contents dump: {}", debug_json); + } + } return Err(ProviderError::Api { status: status.as_u16(), message: msg, @@ -470,6 +475,11 @@ impl ModelProvider for GeminiProvider { let status = response.status(); if !status.is_success() { let msg = response.text().await.unwrap_or_default(); + if status.as_u16() == 400 { + if let Ok(debug_json) = serde_json::to_string(&body.contents) { + log::error!("[Gemini] 400 contents dump: {}", debug_json); + } + } return Err(ProviderError::Api { status: status.as_u16(), message: msg, diff --git a/dashboard/src/components/CameraView.tsx b/dashboard/src/components/CameraView.tsx index da9e58ab..9750c60d 100644 --- a/dashboard/src/components/CameraView.tsx +++ b/dashboard/src/components/CameraView.tsx @@ -124,11 +124,28 @@ export function CameraView({ }; }, [cameraName, handleFrame]); + // Render a JPEG frame directly to canvas via createImageBitmap (no H264 decoder needed). + const renderJpeg = useCallback((data: Uint8Array) => { + const canvas = canvasRef.current; + if (!canvas) return; + const blob = new Blob([data], { type: 'image/jpeg' }); + createImageBitmap(blob).then((bitmap) => { + if (canvas.width !== bitmap.width || canvas.height !== bitmap.height) { + canvas.width = bitmap.width; + canvas.height = bitmap.height; + setDimensions({ width: bitmap.width, height: bitmap.height }); + } + const ctx = canvas.getContext('2d'); + if (ctx) { + ctx.drawImage(bitmap, 0, 0); + frameCountRef.current++; + } + bitmap.close(); + }).catch((e) => console.error('[CameraView] JPEG render error:', e)); + }, []); + // Handle incoming samples from Zenoh. Camera frames are CBOR-only. const handleSample = useCallback((sample: Sample) => { - const decoder = decoderRef.current; - if (!decoder) return; - try { const payload = getSamplePayload(sample); const encodingInfo = getEncodingInfo(sample); @@ -156,7 +173,8 @@ export function CameraView({ msg = msg.body; } - const format: string = msg.format ?? ''; + // Support both `format` (rtsp-camera/H264) and `encoding` (oak-camera/JPEG) field names. + const format: string = msg.format ?? msg.encoding ?? ''; const data: Uint8Array = msg.data instanceof Uint8Array ? msg.data : new Uint8Array(msg.data ?? []); const rawHeader = msg.header; const header = rawHeader ? { @@ -169,12 +187,20 @@ export function CameraView({ // Store latest metadata in ref (no re-render) lastMetaRef.current = { header, format, dataSize: data.length }; + if (format === 'jpeg') { + renderJpeg(data); + return; + } + // Skip non-h264 formats (but allow empty format in case field is missing) if (format && format !== 'h264') { console.warn(`[CameraView] Unexpected format: ${format}`); return; } + const decoder = decoderRef.current; + if (!decoder) return; + // Use pub_time as timestamp (convert from nanoseconds to microseconds) const timestamp = header ? Number(header.pubTime / 1000n) @@ -187,7 +213,7 @@ export function CameraView({ } catch (e) { console.error('[CameraView] Failed to process sample:', e); } - }, []); + }, [renderJpeg]); // Auto-detect correct topic from discovered topics. // Priority: 1) exact name match, 2) first camera/*/compressed topic found. diff --git a/dashboard/src/components/ChatView.tsx b/dashboard/src/components/ChatView.tsx index 1a59ff15..72790e59 100644 --- a/dashboard/src/components/ChatView.tsx +++ b/dashboard/src/components/ChatView.tsx @@ -343,7 +343,10 @@ export function ChatView({ session }: { session?: Session }) { const turnRef = useRef(0); const episodesRef = useRef(0); - useEffect(() => () => { mountedRef.current = false; abortRef.current?.abort(); }, []); + useEffect(() => { + mountedRef.current = true; // reset on (re)mount — React StrictMode double-invokes cleanup + return () => { mountedRef.current = false; abortRef.current?.abort(); }; + }, []); // Auto-scroll; stamp a 600ms window so handleScroll ignores programmatic events useEffect(() => { @@ -459,7 +462,7 @@ export function ChatView({ session }: { session?: Session }) { return; } - for await (const frame of parseSseStream(res.body)) { + sseLoop: for await (const frame of parseSseStream(res.body, ctrl.signal)) { if (ctrl.signal.aborted) break; let event: AgentEvent; @@ -520,7 +523,8 @@ export function ChatView({ session }: { session?: Session }) { case "error": flushDeltaBuffer(); setEntries((prev) => [...prev, { kind: "error", text: event.text ?? "Unknown error" }]); - break; + ctrl.abort(); // unblock any pending reader.read() inside parseSseStream + break sseLoop; case "system": { flushDeltaBuffer(); @@ -533,7 +537,8 @@ export function ChatView({ session }: { session?: Session }) { case "done": flushDeltaBuffer(); gotDone = true; - break; + ctrl.abort(); // unblock any pending reader.read() inside parseSseStream + break sseLoop; } } @@ -556,7 +561,7 @@ export function ChatView({ session }: { session?: Session }) { } finally { streamingRef.current = false; abortRef.current = null; - if (mountedRef.current) setStreaming(false); + setStreaming(false); // safe in React 18 even if unmounted; mountedRef no longer guards this } }, [selectedAgent, input, streaming, token, flushDeltaBuffer]); diff --git a/dashboard/src/components/Dashboard.tsx b/dashboard/src/components/Dashboard.tsx index 1de08b67..c92a315d 100644 --- a/dashboard/src/components/Dashboard.tsx +++ b/dashboard/src/components/Dashboard.tsx @@ -21,6 +21,7 @@ import { SortableStatsCard } from './SortableStatsCard'; import { SortableNodesCard } from './SortableNodesCard'; import { SortableSystemTelemetryCard } from './SortableSystemTelemetryCard'; import { SortableNetworkMonitorCard } from './SortableNetworkMonitorCard'; +import { SortableTimeseriesCard } from './SortableTimeseriesCard'; import { PanelConfig, PanelType, @@ -224,6 +225,15 @@ export function Dashboard({ cameras: initialCameras, availableTopics: staticTopi type: 'network', }; break; + case 'timeseries': + newPanel = { + id: newId, + name: `Plot ${count}`, + topic: '', + type: 'timeseries', + windowSecs: 10, + }; + break; default: return; } @@ -313,6 +323,12 @@ export function Dashboard({ cameras: initialCameras, availableTopics: staticTopi Network Panel +

)}
@@ -414,6 +430,20 @@ export function Dashboard({ cameras: initialCameras, availableTopics: staticTopi onRemove={() => removePanel(panel.id)} /> ); + case 'timeseries': + return ( + updatePanel(panel.id, { topic })} + onRemove={() => removePanel(panel.id)} + availableTopics={availableTopics} + /> + ); default: return null; } diff --git a/dashboard/src/lib/sse.ts b/dashboard/src/lib/sse.ts index bf959fd2..6f1dcdb0 100644 --- a/dashboard/src/lib/sse.ts +++ b/dashboard/src/lib/sse.ts @@ -1,9 +1,3 @@ -/** - * Minimal SSE frame parser for ReadableStream bodies. - * - * Handles the server-sent events format: lines prefixed with `data:` or `event:`. - * Each complete event block (terminated by blank line) is yielded. - */ export interface SseFrame { event: string; data: string; @@ -12,26 +6,56 @@ export interface SseFrame { /** * Parse an SSE ReadableStream body into an async iterable of SseFrame objects. * + * Pass an AbortSignal to make every reader.read() race against it so the + * generator can be interrupted immediately when the signal fires — without + * this, a pending read() blocks generator.return() indefinitely when the + * server keeps the TCP connection alive (e.g. Vite dev proxy). + * * Usage: - * const res = await fetch(url, { method: 'POST', ... }); - * for await (const frame of parseSseStream(res.body!)) { ... } + * const ctrl = new AbortController(); + * const res = await fetch(url, { signal: ctrl.signal, ... }); + * for await (const frame of parseSseStream(res.body!, ctrl.signal)) { ... } */ export async function* parseSseStream( - body: ReadableStream + body: ReadableStream, + signal?: AbortSignal ): AsyncGenerator { const reader = body.getReader(); const decoder = new TextDecoder(); let buffer = ""; let currentEvent = ""; + // A promise that rejects the moment the abort signal fires. + // Kept outside the loop so we only allocate it once. + // Suppressed in finally to avoid unhandled-rejection warnings after the + // generator exits (the signal may fire after we've already cleaned up). + const abortPromise: Promise | null = signal + ? new Promise((_, reject) => { + if (signal.aborted) { + reject(signal.reason ?? new DOMException("Aborted", "AbortError")); + } else { + signal.addEventListener( + "abort", + () => reject(signal.reason ?? new DOMException("Aborted", "AbortError")), + { once: true } + ); + } + }) + : null; + try { while (true) { - const { done, value } = await reader.read(); + // Race the read against the abort signal so that ctrl.abort() (Stop + // button) or an explicit abort on "done"/"error" immediately unblocks + // a pending reader.read() — otherwise generator.return() can't run + // and the outer finally { setStreaming(false) } never executes. + const { done, value } = abortPromise + ? await Promise.race([reader.read(), abortPromise]) + : await reader.read(); if (done) break; buffer += decoder.decode(value, { stream: true }); - // Process complete lines let nl: number; while ((nl = buffer.indexOf("\n")) !== -1) { const rawLine = buffer.slice(0, nl); @@ -39,7 +63,6 @@ export async function* parseSseStream( const line = rawLine.replace(/\r$/, ""); if (line === "") { - // Blank line = end of event block; reset event type currentEvent = ""; continue; } @@ -59,6 +82,9 @@ export async function* parseSseStream( } } } finally { - reader.releaseLock(); + // Suppress any future rejection from abortPromise so it doesn't become + // an unhandled rejection after the generator has exited. + abortPromise?.catch(() => {}); + reader.cancel().catch(() => {}); } } diff --git a/dashboard/src/lib/storage.ts b/dashboard/src/lib/storage.ts index 89b34f81..01d85a64 100644 --- a/dashboard/src/lib/storage.ts +++ b/dashboard/src/lib/storage.ts @@ -14,7 +14,7 @@ const STORAGE_KEYS = { CAMERA_ORDER: 'bubbaloop-camera-order', } as const; -export type PanelType = 'camera' | 'json' | 'rawdata' | 'weather' | 'stats' | 'nodes' | 'telemetry' | 'network'; +export type PanelType = 'camera' | 'json' | 'rawdata' | 'weather' | 'stats' | 'nodes' | 'telemetry' | 'network' | 'timeseries'; export interface BasePanelConfig { id: string; @@ -55,7 +55,12 @@ export interface NetworkPanelConfig extends BasePanelConfig { type: 'network'; } -export type PanelConfig = CameraPanelConfig | JsonPanelConfig | RawDataPanelConfig | WeatherPanelConfig | StatsPanelConfig | NodesPanelConfig | TelemetryPanelConfig | NetworkPanelConfig; +export interface TimeseriesPanelConfig extends BasePanelConfig { + type: 'timeseries'; + windowSecs?: number; +} + +export type PanelConfig = CameraPanelConfig | JsonPanelConfig | RawDataPanelConfig | WeatherPanelConfig | StatsPanelConfig | NodesPanelConfig | TelemetryPanelConfig | NetworkPanelConfig | TimeseriesPanelConfig; // Legacy type for migration export interface LegacyCameraConfig { @@ -181,6 +186,7 @@ export function generatePanelId(type: PanelType): string { nodes: 'nodes', telemetry: 'telemetry', network: 'network', + timeseries: 'ts', }; const prefix = prefixes[type]; return `${prefix}-${Date.now()}-${Math.random().toString(36).slice(2, 7)}`; From 0679bd75e6d7248d9be84e1572464e09148ff99a Mon Sep 17 00:00:00 2001 From: edgarriba Date: Mon, 18 May 2026 14:31:40 +0200 Subject: [PATCH 13/13] refactor(sse): trim verbose comments for clarity Co-Authored-By: Claude Sonnet 4.6 --- dashboard/src/lib/sse.ts | 26 ++++++-------------------- 1 file changed, 6 insertions(+), 20 deletions(-) diff --git a/dashboard/src/lib/sse.ts b/dashboard/src/lib/sse.ts index 6f1dcdb0..b6c836ff 100644 --- a/dashboard/src/lib/sse.ts +++ b/dashboard/src/lib/sse.ts @@ -6,15 +6,9 @@ export interface SseFrame { /** * Parse an SSE ReadableStream body into an async iterable of SseFrame objects. * - * Pass an AbortSignal to make every reader.read() race against it so the - * generator can be interrupted immediately when the signal fires — without - * this, a pending read() blocks generator.return() indefinitely when the - * server keeps the TCP connection alive (e.g. Vite dev proxy). - * - * Usage: - * const ctrl = new AbortController(); - * const res = await fetch(url, { signal: ctrl.signal, ... }); - * for await (const frame of parseSseStream(res.body!, ctrl.signal)) { ... } + * Pass an AbortSignal so every reader.read() races against it — without this, + * a pending read() blocks generator.return() indefinitely when the server keeps + * the TCP connection alive (e.g. Vite dev proxy), preventing setStreaming(false). */ export async function* parseSseStream( body: ReadableStream, @@ -25,10 +19,8 @@ export async function* parseSseStream( let buffer = ""; let currentEvent = ""; - // A promise that rejects the moment the abort signal fires. - // Kept outside the loop so we only allocate it once. - // Suppressed in finally to avoid unhandled-rejection warnings after the - // generator exits (the signal may fire after we've already cleaned up). + // Rejects when the signal fires; raced against reader.read() each iteration. + // Suppressed in finally so a late-firing signal doesn't become an unhandled rejection. const abortPromise: Promise | null = signal ? new Promise((_, reject) => { if (signal.aborted) { @@ -45,10 +37,6 @@ export async function* parseSseStream( try { while (true) { - // Race the read against the abort signal so that ctrl.abort() (Stop - // button) or an explicit abort on "done"/"error" immediately unblocks - // a pending reader.read() — otherwise generator.return() can't run - // and the outer finally { setStreaming(false) } never executes. const { done, value } = abortPromise ? await Promise.race([reader.read(), abortPromise]) : await reader.read(); @@ -82,9 +70,7 @@ export async function* parseSseStream( } } } finally { - // Suppress any future rejection from abortPromise so it doesn't become - // an unhandled rejection after the generator has exited. - abortPromise?.catch(() => {}); + abortPromise?.catch(() => {}); // suppress late-firing signal rejection reader.cancel().catch(() => {}); } }