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..ce30a633 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/components/SortableTimeseriesCard.tsx b/dashboard/src/components/SortableTimeseriesCard.tsx new file mode 100644 index 00000000..d1b9a390 --- /dev/null +++ b/dashboard/src/components/SortableTimeseriesCard.tsx @@ -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 ( +

+ +
+ ); +}); diff --git a/dashboard/src/components/TimeseriesView.tsx b/dashboard/src/components/TimeseriesView.tsx new file mode 100644 index 00000000..3c295be2 --- /dev/null +++ b/dashboard/src/components/TimeseriesView.tsx @@ -0,0 +1,361 @@ +import { useRef, useEffect, useCallback, useState } from 'react'; +import { Sample } from '@eclipse-zenoh/zenoh-ts'; +import { getSamplePayload, getEncodingInfo, hasExplicitEncoding, EncodingPredefined } from '../lib/zenoh'; +import { decode as cborDecode } from 'cbor-x'; +import { useZenohSubscription } from '../hooks/useZenohSubscription'; + +const COLORS = [ + '#3d5afe', '#00e5ff', '#00c853', + '#ffd600', '#ff6d00', '#e040fb', + '#ff1744', '#00bfa5', '#76ff03', + '#f50057', '#40c4ff', '#ffab40', +]; + +// Fields excluded from channel discovery — they are timestamps / sequence numbers, +// not sensor signals, and their large magnitudes would corrupt the Y scale. +const EXCLUDE_PATHS = new Set(['timestamp_us', 'timestamp_ns', 'timestamp', 'sequence', 'seq']); +const EXCLUDE_SUFFIXES = ['_us', '_ns', '_ms', 'time', 'Time']; + +function isExcluded(path: string): boolean { + const leaf = path.split('.').pop()!; + if (EXCLUDE_PATHS.has(leaf)) return true; + return EXCLUDE_SUFFIXES.some(s => leaf.endsWith(s) || leaf.startsWith('timestamp')); +} + +interface Point { t: number; v: number } +interface Channel { path: string; points: Point[]; color: string } + +function flattenNumeric(obj: unknown, prefix = ''): { path: string; value: number }[] { + if (typeof obj === 'number') return prefix && !isExcluded(prefix) ? [{ path: prefix, value: obj }] : []; + if (typeof obj === 'bigint') return prefix && !isExcluded(prefix) ? [{ path: prefix, value: Number(obj) }] : []; + if (obj && typeof obj === 'object' && !ArrayBuffer.isView(obj) && !Array.isArray(obj)) { + return Object.entries(obj as Record).flatMap(([k, v]) => + flattenNumeric(v, prefix ? `${prefix}.${k}` : k) + ); + } + return []; +} + +interface DragHandleProps { [key: string]: unknown } + +interface TimeseriesViewProps { + panelName: string; + topic: string; + windowSecs?: number; + isHidden?: boolean; + onTopicChange?: (topic: string) => void; + onRemove?: () => void; + availableTopics?: Array<{ display: string; raw: string }>; + dragHandleProps?: DragHandleProps; +} + +export function TimeseriesView({ + panelName, + topic, + windowSecs = 10, + isHidden = false, + onTopicChange, + onRemove, + availableTopics = [], + dragHandleProps, +}: TimeseriesViewProps) { + const canvasRef = useRef(null); + const channelsRef = useRef>(new Map()); + const rafRef = useRef(null); + const [channelNames, setChannelNames] = useState([]); + const [disabled, setDisabled] = useState>(new Set()); + const disabledRef = useRef>(new Set()); + const [showTopicEditor, setShowTopicEditor] = useState(!topic); + const lastValuesRef = useRef>(new Map()); + + const windowMs = windowSecs * 1000; + + // Keep disabledRef in sync with state so the draw loop sees current value + useEffect(() => { disabledRef.current = disabled; }, [disabled]); + + const toggleChannel = useCallback((path: string) => { + setDisabled(prev => { + const next = new Set(prev); + next.has(path) ? next.delete(path) : next.add(path); + return next; + }); + }, []); + + const draw = useCallback(() => { + const canvas = canvasRef.current; + if (!canvas || canvas.width === 0 || canvas.height === 0) { + rafRef.current = requestAnimationFrame(draw); + return; + } + const ctx = canvas.getContext('2d'); + if (!ctx) { rafRef.current = requestAnimationFrame(draw); return; } + + const W = canvas.width; + const H = canvas.height; + const now = Date.now(); + const tMin = now - windowMs; + const dis = disabledRef.current; + + ctx.clearRect(0, 0, W, H); + ctx.fillStyle = '#15151f'; + ctx.fillRect(0, 0, W, H); + + const channels = Array.from(channelsRef.current.values()); + if (channels.length === 0) { rafRef.current = requestAnimationFrame(draw); return; } + + // Y-axis label width — reserve left margin for tick labels + ctx.font = '9px "JetBrains Mono", monospace'; + const Y_MARGIN = 42; + + const toX = (t: number) => Y_MARGIN + ((t - tMin) / windowMs) * (W - Y_MARGIN); + + // Collect visible points per channel, compute global [lo, hi] across all enabled channels. + type VisibleData = { visible: Point[] }; + const visMap = new Map(); + let globalLo = Infinity, globalHi = -Infinity; + + for (const ch of channels) { + const cutoff = tMin - 200; + while (ch.points.length > 1 && ch.points[0].t < cutoff) ch.points.shift(); + const visible = ch.points.filter(p => p.t >= tMin); + visMap.set(ch.path, { visible }); + if (dis.has(ch.path) || visible.length < 2) continue; + for (const p of visible) { + if (p.v < globalLo) globalLo = p.v; + if (p.v > globalHi) globalHi = p.v; + } + } + + // Shared absolute Y axis with 10% padding + const span = (globalHi - globalLo) || 1; + const pad = span * 0.1; + const yLo = globalLo - pad; + const yHi = globalHi + pad; + const MARGIN_V = 10; + const toY = (v: number) => MARGIN_V + (H - 2 * MARGIN_V) - ((v - yLo) / (yHi - yLo)) * (H - 2 * MARGIN_V); + + // Y-axis grid lines + labels (5 ticks) + const TICK_COUNT = 5; + for (let i = 0; i <= TICK_COUNT; i++) { + const v = yLo + (yHi - yLo) * (i / TICK_COUNT); + const y = Math.round(toY(v)) + 0.5; + // Grid line + ctx.strokeStyle = '#1e1e2e'; + ctx.lineWidth = 1; + ctx.beginPath(); ctx.moveTo(Y_MARGIN, y); ctx.lineTo(W, y); ctx.stroke(); + // Tick label — right-aligned in the margin + const label = Math.abs(v) >= 100 ? v.toFixed(0) : Math.abs(v) >= 1 ? v.toFixed(1) : v.toFixed(3); + ctx.fillStyle = '#606070'; + ctx.textAlign = 'right'; + ctx.fillText(label, Y_MARGIN - 4, y + 3); + } + ctx.textAlign = 'left'; + + // Zero line (distinct from grid) + if (yLo < 0 && yHi > 0) { + const zy = toY(0); + ctx.strokeStyle = '#2a2a3a'; + ctx.lineWidth = 1; + ctx.beginPath(); ctx.moveTo(Y_MARGIN, zy + 0.5); ctx.lineTo(W, zy + 0.5); ctx.stroke(); + } + + // Y-axis border line + ctx.strokeStyle = '#2a2a3a'; + ctx.lineWidth = 1; + ctx.beginPath(); ctx.moveTo(Y_MARGIN + 0.5, 0); ctx.lineTo(Y_MARGIN + 0.5, H); ctx.stroke(); + + const newLastValues = new Map(); + for (const ch of channels) { + const { visible } = visMap.get(ch.path)!; + const last = visible[visible.length - 1]; + if (last) newLastValues.set(ch.path, last.v); + if (dis.has(ch.path) || visible.length < 2) continue; + + ctx.strokeStyle = ch.color; + ctx.lineWidth = 1.5; + ctx.lineJoin = 'round'; + ctx.beginPath(); + ctx.moveTo(toX(visible[0].t), toY(visible[0].v)); + for (let j = 1; j < visible.length; j++) { + ctx.lineTo(toX(visible[j].t), toY(visible[j].v)); + } + ctx.stroke(); + } + + lastValuesRef.current = newLastValues; + + rafRef.current = requestAnimationFrame(draw); + }, [windowMs]); + + useEffect(() => { + if (isHidden) { if (rafRef.current) cancelAnimationFrame(rafRef.current); return; } + rafRef.current = requestAnimationFrame(draw); + return () => { if (rafRef.current) cancelAnimationFrame(rafRef.current); }; + }, [draw, isHidden]); + + useEffect(() => { + const canvas = canvasRef.current; + if (!canvas) return; + const ro = new ResizeObserver(() => { + canvas.width = canvas.offsetWidth; + canvas.height = canvas.offsetHeight; + }); + ro.observe(canvas); + return () => ro.disconnect(); + }, [topic]); + + const handleSample = useCallback((sample: Sample) => { + try { + const payload = getSamplePayload(sample); + const encodingInfo = getEncodingInfo(sample); + if (hasExplicitEncoding(encodingInfo)) { + const id = encodingInfo.id; + if (id !== EncodingPredefined.APPLICATION_CBOR && + id !== EncodingPredefined.APPLICATION_OCTET_STREAM && + id !== EncodingPredefined.APPLICATION_JSON && + id !== EncodingPredefined.TEXT_PLAIN) return; + } + + // eslint-disable-next-line @typescript-eslint/no-explicit-any + let msg: any; + try { msg = cborDecode(payload); } + catch { try { msg = JSON.parse(new TextDecoder().decode(payload)); } catch { return; } } + + if (msg && typeof msg === 'object' && 'body' in msg) msg = msg.body; + + const leaves = flattenNumeric(msg); + if (leaves.length === 0) return; + + const now = Date.now(); + const channels = channelsRef.current; + let added = false; + for (const { path, value } of leaves) { + if (!channels.has(path)) { + channels.set(path, { path, points: [], color: COLORS[channels.size % COLORS.length] }); + added = true; + } + channels.get(path)!.points.push({ t: now, v: value }); + } + if (added) setChannelNames(Array.from(channels.keys())); + } catch { /* ignore */ } + }, []); + + useZenohSubscription(topic, handleSample); + + const applyTopic = useCallback((raw: string) => { + if (!raw) return; + // Pass exactly as-is — subscription manager handles wildcard internally + onTopicChange?.(raw); + channelsRef.current.clear(); + setChannelNames([]); + setShowTopicEditor(false); + }, [onTopicChange]); + + const nonMeta = availableTopics.filter(t => + !t.display.endsWith('/health') && + !t.display.endsWith('/schema') && + !t.display.endsWith('/manifest') && + !t.display.endsWith('/compressed') + ); + + return ( +
+ {/* Header — same structure as CameraView */} +
+ + + + + {panelName} + + {channelNames.length > 0 && ( + + {channelNames.length}ch · {windowSecs}s + + )} + + {onRemove && ( + + )} +
+ + {/* Topic picker */} + {showTopicEditor && ( +
+ + {nonMeta.length === 0 && ( + + No topics discovered yet. Topics appear after nodes start publishing. + + )} +
+ )} + + {/* Empty state */} + {!topic && !showTopicEditor && ( +
+ Select a topic to begin plotting +
+ )} + + {/* Canvas */} + {topic && ( + + )} + + {/* Legend — HTML row so buttons are clickable */} + {channelNames.length > 0 && ( +
+ {Array.from(channelsRef.current.values()).map(ch => { + const off = disabled.has(ch.path); + 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)}`;