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..77cc0962 --- /dev/null +++ b/crates/bubbaloop/src/agent/camera.rs @@ -0,0 +1,107 @@ +//! Camera frame capture for the agent's `grab_frame` tool. +//! +//! Sends a Zenoh query to the camera node's `grab_frame` queryable and returns +//! the JPEG frame + metadata. The actual image encoding (RGBA→JPEG via kornia-rs) +//! lives in the camera node itself — the daemon receives only the base64 result. + +use std::sync::Arc; +use std::time::Duration; + +/// An image captured from a camera, ready for injection into the next agent turn. +pub struct ImageAttachment { + pub media_type: String, + pub base64: String, + /// Text label prepended before the image in the labeled-pair message. + pub label: String, +} + +/// Query the camera node's `grab_frame` queryable and return +/// `(receipt_json, ImageAttachment)` on success. +pub async fn grab_frame( + session: &Arc, + node_name: &str, + machine_id: &str, +) -> anyhow::Result<(String, ImageAttachment)> { + let key = format!("bubbaloop/local/{}/{}/grab_frame", machine_id, node_name); + + let replies = session + .get(&key) + .timeout(Duration::from_secs(5)) + .await + .map_err(|e| anyhow::anyhow!("Zenoh get failed: {}", e))?; + + let reply = tokio::time::timeout(Duration::from_secs(5), replies.recv_async()) + .await + .map_err(|_| { + anyhow::anyhow!( + "camera '{}' did not respond within 5s on {}", + node_name, + key + ) + })? + .map_err(|_| anyhow::anyhow!("grab_frame channel closed for '{}'", node_name))?; + + let sample = reply + .result() + .map_err(|e| anyhow::anyhow!("grab_frame reply error for '{}': {:?}", node_name, e))?; + + let payload = sample.payload().to_bytes(); + let v: serde_json::Value = serde_json::from_slice(payload.as_ref()) + .map_err(|e| anyhow::anyhow!("grab_frame JSON parse failed: {}", e))?; + + // Surface error from camera node + if let Some(err) = v.get("error").and_then(|e| e.as_str()) { + return Err(anyhow::anyhow!("camera '{}' error: {}", node_name, err)); + } + + // Reject stale frames — camera is likely stopped or not streaming. + // 10 s is generous for live cameras; anything older means the cached + // frame is from a previous run and would mislead vision analysis. + const MAX_AGE_MS: u64 = 10_000; + if let Some(age_ms) = v.get("age_ms").and_then(|a| a.as_u64()) { + if age_ms > MAX_AGE_MS { + return Err(anyhow::anyhow!( + "camera '{}' frame is stale ({}ms old, max {}ms) — node may not be streaming", + node_name, + age_ms, + MAX_AGE_MS + )); + } + } + + let jpeg_b64 = v + .get("jpeg_b64") + .and_then(|s| s.as_str()) + .ok_or_else(|| anyhow::anyhow!("grab_frame response missing jpeg_b64"))? + .to_string(); + + let media_type = v + .get("media_type") + .and_then(|s| s.as_str()) + .unwrap_or("image/jpeg") + .to_string(); + + let label = v + .get("label") + .and_then(|s| s.as_str()) + .unwrap_or(node_name) + .to_string(); + + // Build receipt JSON from metadata fields (strip the large jpeg_b64 field) + let receipt = { + let mut r = v.clone(); + if let serde_json::Value::Object(ref mut m) = r { + m.remove("jpeg_b64"); + } + r.to_string() + }; + + Ok(( + receipt, + ImageAttachment { + media_type, + base64: jpeg_b64, + label, + }, + )) +} diff --git a/crates/bubbaloop/src/agent/dispatch.rs b/crates/bubbaloop/src/agent/dispatch.rs index a9b7a9d8..bc6b521a 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(); } } @@ -500,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", @@ -663,6 +695,31 @@ 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. \ + Use the exact node name from the Current Sensor Inventory. \ + When the user asks about cameras without naming one, call grab_frame \ + for EVERY camera node in the inventory — do not pick just one. \ + Returns a JSON receipt with capture metadata (timestamp, resolution, frame_id)." + .to_string(), + input_schema: json!({ + "type": "object", + "properties": { + "node_name": { + "type": "string", + "description": "Exact name of the camera node from the Current Sensor Inventory" + }, + "machine_id": { + "type": "string", + "description": "Machine ID override (defaults to the local machine)" + } + }, + "required": ["node_name"] + }), + }, ] } @@ -723,6 +780,7 @@ impl 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 +1791,44 @@ 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(), + ) + } + }; + + // Registry name may differ from the Zenoh instance name (e.g. "tapo-terrace" vs + // "tapo_terrace_camera" when a config_override sets a different `name:` field). + let instance_name = self.platform.get_node_instance_name(&node_name).await; + log::info!( + "[Agent] grab_frame: {}/{} (instance: {})", + machine_id, + node_name, + instance_name + ); + match crate::agent::camera::grab_frame(&session, &instance_name, &machine_id).await { + Ok((receipt, attachment)) => { + self.stash_image(tool_use_id.to_string(), attachment); + ToolResult::success(receipt) + } + Err(e) => ToolResult::error(format!("grab_frame failed for '{}': {}", node_name, e)), + } + } } #[cfg(test)] @@ -1743,7 +1839,7 @@ mod tests { use zenoh::pubsub::Subscriber; use zenoh::sample::Sample; - const TOOL_COUNT: usize = 42; + const TOOL_COUNT: usize = 43; #[test] fn tool_definitions_count() { 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 07a01807..62eb1493 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; @@ -68,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 } @@ -286,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)); @@ -324,6 +334,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 +350,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), @@ -351,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(()) } @@ -382,6 +402,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; @@ -393,9 +418,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, @@ -590,7 +624,32 @@ 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 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. + 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)) + .collect(); + // Queued for the next generate_stream call (analysis turn); dropped after use. + pending_image_msg = Some(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..7b720c8f 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\ @@ -239,14 +243,26 @@ 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. \ + **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** 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`) \ with run_command — they block for the full 30-second timeout and produce no output until they time out.\n\ @@ -427,7 +443,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"); } diff --git a/crates/bubbaloop/src/agent/provider/gemini.rs b/crates/bubbaloop/src/agent/provider/gemini.rs index 37091dbb..ebb06890 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 ──────────────────────────────────────────────────────── @@ -56,10 +57,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, @@ -231,9 +244,31 @@ 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 mut out = Vec::new(); + let names = tool_name_map(messages); + let mut out: Vec = Vec::new(); for msg in messages { let role = match msg.role.as_str() { "assistant" => "model", @@ -245,6 +280,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 { @@ -263,21 +306,35 @@ 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, }, })); } } } - 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 +376,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 { @@ -366,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, @@ -412,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, @@ -530,6 +598,7 @@ async fn process_gemini_sse( } } GeminiPart::FunctionResponse(_) => {} + GeminiPart::InlineData(_) => {} // images in responses not yet surfaced } } @@ -655,6 +724,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(&[]); 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/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 1a75bea7..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(); @@ -809,6 +810,17 @@ 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 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 b764d719..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()), @@ -450,6 +451,14 @@ impl PlatformOperations for MockPlatform { log::debug!("[MockPlatform] publish_to_topic: {}", topic); Ok(()) } + + fn zenoh_session(&self) -> Option> { + self.zenoh_session.clone() + } + + async fn get_node_instance_name(&self, name: &str) -> String { + name.to_string() + } } // ── Tests ──────────────────────────────────────────────────────────────── @@ -508,6 +517,7 @@ mod tests { node_type: "python".to_string(), installed: true, is_built: true, + instance_name: None, }, NodeInfo { name: "detector".to_string(), @@ -516,6 +526,7 @@ mod tests { node_type: "rust".to_string(), installed: true, is_built: false, + instance_name: None, }, NodeInfo { name: "tracker".to_string(), @@ -524,6 +535,7 @@ mod tests { node_type: "python".to_string(), installed: false, is_built: false, + instance_name: None, }, ]; let mock = mock_with_nodes(nodes); @@ -566,6 +578,7 @@ mod tests { node_type: "rust".to_string(), installed: true, is_built: true, + instance_name: None, }, NodeInfo { name: "beta".to_string(), @@ -574,6 +587,7 @@ mod tests { node_type: "python".to_string(), installed: false, is_built: false, + instance_name: None, }, ]; let mock = mock_with_nodes(nodes); @@ -1133,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 59de03ac..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. @@ -281,6 +285,23 @@ 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>; + + /// 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/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" 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; 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 130e2d5d..72790e59 100644 --- a/dashboard/src/components/ChatView.tsx +++ b/dashboard/src/components/ChatView.tsx @@ -14,8 +14,8 @@ 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 } | { kind: "connection_lost" } @@ -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 ( + {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": @@ -151,39 +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 null; case "error": return (
@@ -243,6 +296,29 @@ function EntryRow({ entry, onDismissOauth }: { entry: ChatEntry; onDismissOauth: } } +function renderEntries(entries: ChatEntry[], onDismissOauth: () => void): React.ReactNode[] { + 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) { + const next = entries[i]; + if (next.kind !== "image") break; + srcs.push(next.src); + i++; + } + nodes.push(); + } else { + nodes.push(); + i++; + } + } + return nodes; +} + // ── Main component ──────────────────────────────────────────────────── export function ChatView({ session }: { session?: Session }) { @@ -256,29 +332,58 @@ 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); - // Accumulates assistant delta text across SSE frames before flushing to state + const [showScrollBtn, setShowScrollBtn] = useState(false); const deltaBufferRef = useRef(""); + 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 = true; // reset on (re)mount — React StrictMode double-invokes cleanup + return () => { 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; + 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) .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)); }, []); @@ -295,21 +400,27 @@ export function ChatView({ session }: { session?: Session }) { setEntries((prev) => { const last = prev[prev.length - 1]; if (last?.kind === "assistant") { - return [...prev.slice(0, -1), { kind: "assistant", text: last.text + text }]; + const updated = [...prev]; + updated[updated.length - 1] = { kind: "assistant", text: last.text + text }; + return updated; } return [...prev, { kind: "assistant", text }]; }); }, []); 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(""); 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; @@ -351,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; @@ -370,31 +481,64 @@ 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) => { + for (let j = prev.length - 1; j >= 0; j--) { + const e = prev[j]; + if (e.kind === "tool_call" && e.result === null) { + const updated = [...prev]; + updated[j] = { ...e, result: event.text ?? "" }; + return updated; + } + } + return prev; + }); + 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" }]); - break; + ctrl.abort(); // unblock any pending reader.read() inside parseSseStream + break sseLoop; - case "system": + case "system": { flushDeltaBuffer(); - setEntries((prev) => [...prev, { kind: "system", text: event.text ?? "" }]); + const ep = parseEpisodes(event.text ?? ""); + if (ep > 0) episodesRef.current = ep; + turnRef.current = 0; break; + } case "done": flushDeltaBuffer(); gotDone = true; - break; + ctrl.abort(); // unblock any pending reader.read() inside parseSseStream + break sseLoop; } } @@ -405,12 +549,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); + setStreaming(false); // safe in React 18 even if unmounted; mountedRef no longer guards this } }, [selectedAgent, input, streaming, token, flushDeltaBuffer]); @@ -419,8 +570,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(() => { @@ -432,9 +584,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(); } @@ -448,7 +605,7 @@ export function ChatView({ session }: { session?: Session }) {
{/* Scrollable area: token setup + agent bar + messages */} -
{ userScrolledRef.current = true; }}> +
{!token && (
@@ -478,7 +635,7 @@ export function ChatView({ session }: { session?: Session }) { @@ -510,14 +667,21 @@ export function ChatView({ session }: { session?: Session }) {

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

)} - {entries.map((entry, i) => ( - - ))} + {renderEntries(entries, dismissOauth)}
+ {/* Scroll-to-bottom button — appears when user has scrolled up */} + {showScrollBtn && ( + + )} + {/* Input bar — full width, always pinned at bottom */}
@@ -563,6 +727,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 */ @@ -716,6 +910,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; @@ -858,42 +1059,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; @@ -905,22 +1133,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); @@ -936,6 +1148,81 @@ export function ChatView({ session }: { session?: Session }) { font-style: italic; } + /* Image gallery */ + .chat-image-gallery { + padding: 4px 14px; + } + .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; align-items: center; 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/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 []; } diff --git a/dashboard/src/lib/sse.ts b/dashboard/src/lib/sse.ts index bf959fd2..b6c836ff 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,44 @@ export interface SseFrame { /** * Parse an SSE ReadableStream body into an async iterable of SseFrame objects. * - * Usage: - * const res = await fetch(url, { method: 'POST', ... }); - * for await (const frame of parseSseStream(res.body!)) { ... } + * 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 + body: ReadableStream, + signal?: AbortSignal ): AsyncGenerator { const reader = body.getReader(); const decoder = new TextDecoder(); let buffer = ""; let currentEvent = ""; + // 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) { + 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(); + 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 +51,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 +70,7 @@ export async function* parseSseStream( } } } finally { - reader.releaseLock(); + abortPromise?.catch(() => {}); // suppress late-firing signal rejection + 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)}`;