Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 15 additions & 0 deletions crates/bubbaloop/src/agent/camera.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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())
Expand Down
10 changes: 6 additions & 4 deletions crates/bubbaloop/src/agent/dispatch.rs
Original file line number Diff line number Diff line change
Expand Up @@ -699,16 +699,18 @@ impl<P: PlatformOperations> Dispatcher<P> {
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",
Expand Down
10 changes: 9 additions & 1 deletion crates/bubbaloop/src/agent/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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(())
}
Expand Down
9 changes: 7 additions & 2 deletions crates/bubbaloop/src/agent/prompt.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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`) \
Expand Down
10 changes: 10 additions & 0 deletions crates/bubbaloop/src/agent/provider/gemini.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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,
Expand Down
36 changes: 31 additions & 5 deletions dashboard/src/components/CameraView.tsx
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down Expand Up @@ -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 ? {
Expand All @@ -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)
Expand All @@ -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.
Expand Down
15 changes: 10 additions & 5 deletions dashboard/src/components/ChatView.tsx
Original file line number Diff line number Diff line change
Expand Up @@ -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(() => {
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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();
Expand All @@ -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;
}
}

Expand All @@ -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]);

Expand Down
30 changes: 30 additions & 0 deletions dashboard/src/components/Dashboard.tsx
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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;
}
Expand Down Expand Up @@ -313,6 +323,12 @@ export function Dashboard({ cameras: initialCameras, availableTopics: staticTopi
</svg>
Network Panel
</button>
<button className="add-panel-option" onClick={() => addPanel('timeseries')}>
<svg width="16" height="16" viewBox="0 0 24 24" fill="none" stroke="currentColor" strokeWidth="2">
<polyline points="22 12 18 12 15 21 9 3 6 12 2 12" />
</svg>
Timeseries Plot
</button>
</div>
)}
</div>
Expand Down Expand Up @@ -414,6 +430,20 @@ export function Dashboard({ cameras: initialCameras, availableTopics: staticTopi
onRemove={() => removePanel(panel.id)}
/>
);
case 'timeseries':
return (
<SortableTimeseriesCard
key={panel.id}
id={panel.id}
panelName={panel.name}
topic={panel.topic}
windowSecs={(panel as { windowSecs?: number }).windowSecs}
isHidden={isHidden}
onTopicChange={(topic: string) => updatePanel(panel.id, { topic })}
onRemove={() => removePanel(panel.id)}
availableTopics={availableTopics}
/>
);
default:
return null;
}
Expand Down
53 changes: 53 additions & 0 deletions dashboard/src/components/SortableTimeseriesCard.tsx
Original file line number Diff line number Diff line change
@@ -0,0 +1,53 @@
import React from 'react';
import { useSortable } from '@dnd-kit/sortable';
import { CSS } from '@dnd-kit/utilities';
import { TimeseriesView } from './TimeseriesView';

interface SortableTimeseriesCardProps {
id: string;
panelName: string;
topic: string;
windowSecs?: number;
isHidden?: boolean;
onTopicChange: (topic: string) => void;
onRemove: () => void;
availableTopics: Array<{ display: string; raw: string }>;
}

export const SortableTimeseriesCard = React.memo(function SortableTimeseriesCard({
id,
panelName,
topic,
windowSecs,
isHidden = false,
onTopicChange,
onRemove,
availableTopics,
}: SortableTimeseriesCardProps) {
const { attributes, listeners, setNodeRef, transform, transition, isDragging } = useSortable({ id });

const style: React.CSSProperties = {
transform: CSS.Transform.toString(transform),
transition,
opacity: isDragging ? 0.5 : 1,
minWidth: 0,
overflow: 'hidden',
display: isHidden ? 'none' : undefined,
height: 220,
};

return (
<div ref={setNodeRef} style={style}>
<TimeseriesView
panelName={panelName}
topic={topic}
windowSecs={windowSecs}
isHidden={isHidden}
onTopicChange={onTopicChange}
onRemove={onRemove}
availableTopics={availableTopics}
dragHandleProps={{ ...attributes, ...listeners }}
/>
</div>
);
});
Loading
Loading