From de91fe6ecf74b39433cbe6e95b1f494aae15708a Mon Sep 17 00:00:00 2001 From: Jon Saad-Falcon <41205309+jonsaadfalcon@users.noreply.github.com> Date: Fri, 27 Mar 2026 18:17:10 -0700 Subject: [PATCH] feat: add real-time streaming and tool progress to InteractTab Refactor sendAgentMessage to accept onProgress/onContentDelta/onDone callbacks so the InteractTab can show content as it arrives word by word, display tool-call progress labels (e.g. "Querying data with SQL"), and render an elapsed-time footer. The backend now streams tool_progress SSE events from DeepResearchAgent before the final content, matching the Chat tab's polished streaming UX. Co-Authored-By: Claude Opus 4.6 (1M context) --- frontend/src/lib/api.ts | 31 ++- frontend/src/pages/AgentsPage.tsx | 76 ++++++- src/openjarvis/server/agent_manager_routes.py | 185 ++++++++++++++---- 3 files changed, 244 insertions(+), 48 deletions(-) diff --git a/frontend/src/lib/api.ts b/frontend/src/lib/api.ts index 1feecf87..f6a16f45 100644 --- a/frontend/src/lib/api.ts +++ b/frontend/src/lib/api.ts @@ -502,7 +502,16 @@ export async function fetchAgentState(agentId: string): Promise<{ return res.json(); } -export async function sendAgentMessage(agentId: string, content: string, mode: 'immediate' | 'queued' = 'queued'): Promise { +export async function sendAgentMessage( + agentId: string, + content: string, + mode: 'immediate' | 'queued' = 'queued', + callbacks?: { + onProgress?: (label: string) => void; + onContentDelta?: (delta: string, fullContent: string) => void; + onDone?: (fullContent: string) => void; + }, +): Promise { const res = await fetch(`${getBase()}/v1/managed-agents/${agentId}/messages`, { method: 'POST', headers: { 'Content-Type': 'application/json' }, @@ -516,23 +525,35 @@ export async function sendAgentMessage(agentId: string, content: string, mode: ' const reader = res.body.getReader(); const decoder = new TextDecoder(); let fullContent = ''; + let buffer = ''; try { while (true) { const { done, value } = await reader.read(); if (done) break; - const text = decoder.decode(value, { stream: true }); - for (const line of text.split('\n')) { + buffer += decoder.decode(value, { stream: true }); + const lines = buffer.split('\n'); + buffer = lines.pop() || ''; + for (const line of lines) { if (!line.startsWith('data: ') || line === 'data: [DONE]') continue; try { const chunk = JSON.parse(line.slice(6)); + // Check for tool progress events + const toolProgress = chunk.choices?.[0]?.tool_progress; + if (toolProgress) { + callbacks?.onProgress?.(toolProgress); + } const delta = chunk.choices?.[0]?.delta?.content || ''; - fullContent += delta; + if (delta) { + fullContent += delta; + callbacks?.onContentDelta?.(delta, fullContent); + } } catch { /* skip malformed chunks */ } } } } catch { /* stream ended */ } - // Return a synthetic message with the full content + callbacks?.onDone?.(fullContent); + return { id: '', agent_id: agentId, diff --git a/frontend/src/pages/AgentsPage.tsx b/frontend/src/pages/AgentsPage.tsx index 3c8599dd..8d251855 100644 --- a/frontend/src/pages/AgentsPage.tsx +++ b/frontend/src/pages/AgentsPage.tsx @@ -859,9 +859,13 @@ function InteractTab({ agentId, agentStatus }: { agentId: string; agentStatus: s const [input, setInput] = useState(''); const [sending, setSending] = useState(false); const [waitingForResponse, setWaitingForResponse] = useState(false); + const [progressLabel, setProgressLabel] = useState(''); + const [streamingContent, setStreamingContent] = useState(''); const [currentActivity, setCurrentActivity] = useState(''); const [liveStatus, setLiveStatus] = useState(agentStatus); + const [streamElapsedMs, setStreamElapsedMs] = useState(0); const bottomRef = useRef(null); + const timerRef = useRef | null>(null); const loadData = useCallback(async () => { try { @@ -885,6 +889,16 @@ function InteractTab({ agentId, agentStatus }: { agentId: string; agentStatus: s useEffect(() => { setLiveStatus(agentStatus); }, [agentStatus]); + // Clean up elapsed-time timer on unmount + useEffect(() => { + return () => { + if (timerRef.current) { + clearInterval(timerRef.current); + timerRef.current = null; + } + }; + }, []); + // Scroll to bottom only on initial load, not on every poll update. const hasScrolled = useRef(false); useEffect(() => { @@ -894,6 +908,13 @@ function InteractTab({ agentId, agentStatus }: { agentId: string; agentStatus: s } }, [messages]); + // Scroll to bottom when streaming content updates + useEffect(() => { + if (streamingContent) { + bottomRef.current?.scrollIntoView({ behavior: 'smooth' }); + } + }, [streamingContent]); + async function handleSend(mode: 'immediate' | 'queued') { if (!input.trim()) return; const text = input.trim(); @@ -913,9 +934,25 @@ function InteractTab({ agentId, agentStatus }: { agentId: string; agentStatus: s setMessages((prev) => [localMsg, ...prev]); setSending(false); setWaitingForResponse(true); + setProgressLabel('Initializing agent...'); + setStreamingContent(''); + + // Start elapsed-time timer + const startTime = Date.now(); + setStreamElapsedMs(0); + timerRef.current = setInterval(() => { + setStreamElapsedMs(Date.now() - startTime); + }, 100); try { - const response = await sendAgentMessage(agentId, text, mode); + const response = await sendAgentMessage(agentId, text, mode, { + onProgress: (label) => setProgressLabel(label), + onContentDelta: (_delta, full) => setStreamingContent(full), + onDone: () => { + // Clear streaming state — final content comes from the response object + setStreamingContent(''); + }, + }); // Add the agent's response as a local bubble immediately if (response && response.content) { setMessages((prev) => [ @@ -933,6 +970,13 @@ function InteractTab({ agentId, agentStatus }: { agentId: string; agentStatus: s // ignore } finally { setWaitingForResponse(false); + setStreamingContent(''); + setProgressLabel(''); + if (timerRef.current) { + clearInterval(timerRef.current); + timerRef.current = null; + } + setStreamElapsedMs(0); } } @@ -973,8 +1017,8 @@ function InteractTab({ agentId, agentStatus }: { agentId: string; agentStatus: s ))} - {/* Progress indicator */} - {(isAgentWorking || hasPending || waitingForResponse || sending) && ( + {/* Progress indicator — shown when waiting but no streamed content yet, or alongside streaming */} + {(isAgentWorking || hasPending || waitingForResponse || sending) && !streamingContent && (
)} + {/* Streaming content bubble — real-time response as it arrives */} + {waitingForResponse && streamingContent && ( +
+
+ {progressLabel && ( +
+ + {progressLabel} +
+ )} +

{streamingContent}

+

+ {streamElapsedMs > 0 && `${(streamElapsedMs / 1000).toFixed(1)}s elapsed`} +

+
+
+ )}
{/* Input area */} diff --git a/src/openjarvis/server/agent_manager_routes.py b/src/openjarvis/server/agent_manager_routes.py index 9a5c8984..93718d40 100644 --- a/src/openjarvis/server/agent_manager_routes.py +++ b/src/openjarvis/server/agent_manager_routes.py @@ -466,6 +466,48 @@ def _get_mcp_tools(app_state: Any) -> Tuple[List[Dict[str, Any]], Dict[str, Any] return openai_tools, adapters_by_name +def _sse_chunk(chunk_id: str, model: str, content: str) -> str: + """Build a single SSE content chunk.""" + import json as _json + + data = { + "id": chunk_id, + "object": "chat.completion.chunk", + "model": model, + "choices": [ + { + "index": 0, + "delta": {"content": content}, + "finish_reason": None, + } + ], + } + return f"data: {_json.dumps(data)}\n\n" + + +def _tool_progress_label(tool_name: str, args: str) -> str: + """Human-readable label for a tool call in progress.""" + labels = { + "knowledge_search": "Searching your knowledge base", + "knowledge_sql": "Querying data with SQL", + "scan_chunks": "Scanning documents for semantic matches", + "think": "Planning next step", + } + label = labels.get(tool_name, f"Using {tool_name}") + if args and tool_name != "think": + # Try to extract the query/question from args JSON + try: + import json as _json + + parsed = _json.loads(args) + q = parsed.get("query") or parsed.get("question") or "" + if q: + label += f' — "{q[:50]}"' + except Exception: + pass + return label + + async def _stream_managed_agent( *, manager: AgentManager, @@ -534,11 +576,16 @@ async def _stream_managed_agent( if dr_tools: async def generate_deep_research(): - """Run DeepResearchAgent in a thread, stream result as SSE.""" + """Run DeepResearchAgent in thread, stream progress + result.""" import asyncio + import queue + import threading from openjarvis.agents.deep_research import DeepResearchAgent + progress_q: queue.Queue = queue.Queue() + + # Patch the agent's tool executor to emit progress dr_agent = DeepResearchAgent( engine=engine, model=model, @@ -547,48 +594,108 @@ async def _stream_managed_agent( temperature=float(config.get("temperature", 0.3)), ) - # Run agent in background thread - result = await asyncio.to_thread(dr_agent.run, user_content) - content = result.content or "No results found." + # Wrap the executor to capture tool calls + original_execute = dr_agent._executor.execute - # Stream the result word-by-word for SSE - words = content.split(" ") - for i, word in enumerate(words): - token = word if i == 0 else " " + word - chunk_data = { - "id": chunk_id, - "object": "chat.completion.chunk", - "model": model, - "choices": [ - { - "index": 0, - "delta": {"content": token}, - "finish_reason": None, - } - ], - } - yield f"data: {json.dumps(chunk_data)}\n\n" + def _tracked_execute(tc): + tool_name = tc.name + args_str = ( + tc.arguments[:80] if tc.arguments else "" + ) + progress_q.put({ + "type": "tool_start", + "tool": tool_name, + "args": args_str, + }) + result = original_execute(tc) + progress_q.put({ + "type": "tool_end", + "tool": tool_name, + "success": result.success, + }) + return result - # Final chunk - finish_data = { - "id": chunk_id, - "object": "chat.completion.chunk", - "model": model, - "choices": [ - { - "index": 0, - "delta": {}, - "finish_reason": "stop", + dr_agent._executor.execute = _tracked_execute + + def _run_agent(): + try: + result = dr_agent.run(user_content) + content = ( + result.content or "No results found." + ) + progress_q.put({ + "type": "done", + "content": content, + }) + except Exception as exc: + progress_q.put({ + "type": "error", + "content": f"Error: {exc}", + }) + + thread = threading.Thread(target=_run_agent, daemon=True) + thread.start() + + # Stream progress events and final content + while True: + try: + event = await asyncio.to_thread(progress_q.get, timeout=120) + except Exception: + # Timeout + yield _sse_chunk(chunk_id, model, "Agent timed out.") + break + + if event["type"] == "tool_start": + tool = event["tool"] + args = event.get("args", "") + label = _tool_progress_label(tool, args) + progress_data = { + "id": chunk_id, + "object": "chat.completion.chunk", + "model": model, + "choices": [ + { + "index": 0, + "delta": {}, + "finish_reason": None, + "tool_progress": label, + } + ], } - ], - } - yield f"data: {json.dumps(finish_data)}\n\n" - yield "data: [DONE]\n\n" + yield f"data: {json.dumps(progress_data)}\n\n" - # Persist the response - manager.store_agent_response( - agent_id, message_id, content, - ) + elif event["type"] == "tool_end": + pass # Could emit completion signal + + elif event["type"] in ("done", "error"): + content = event["content"] + # Stream content word-by-word + words = content.split(" ") + for i, word in enumerate(words): + token = word if i == 0 else " " + word + yield _sse_chunk(chunk_id, model, token) + + # Final chunk + finish_data = { + "id": chunk_id, + "object": "chat.completion.chunk", + "model": model, + "choices": [ + { + "index": 0, + "delta": {}, + "finish_reason": "stop", + } + ], + } + yield f"data: {json.dumps(finish_data)}\n\n" + yield "data: [DONE]\n\n" + + # Persist + manager.store_agent_response( + agent_id, message_id, content, + ) + break return StreamingResponse( generate_deep_research(),