admin/OpenJarvis
1
0
Fork 0
mirror of https://github.com/open-jarvis/OpenJarvis.git synced 2026-07-31 03:12:16 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: add real-time streaming and tool progress to InteractTab

Browse Source 
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) <noreply@anthropic.com>
This commit is contained in:
Jon Saad-Falcon
2026-03-27 18:17:10 -07:00
co-authored by Claude Opus 4.6
parent bd22eeae1e
commit de91fe6ecf
3 changed files with 244 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files
frontend/src/lib/api.ts
+26 -5
View File
@@ -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<AgentMessage> {
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<AgentMessage> {
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,
frontend/src/pages/AgentsPage.tsx
+72 -4
View File
@@ -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<HTMLDivElement>(null);
const timerRef = useRef<ReturnType<typeof setInterval> | 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
</div>
</div>
))}
{/* Progress indicator */}
{(isAgentWorking || hasPending || waitingForResponse || sending) && (
{/* Progress indicator — shown when waiting but no streamed content yet, or alongside streaming */}
{(isAgentWorking || hasPending || waitingForResponse || sending) && !streamingContent && (
<div className="flex justify-start">
<div
className="px-3 py-2 rounded-lg text-sm"
@@ -989,12 +1033,36 @@ function InteractTab({ agentId, agentStatus }: { agentId: string; agentStatus: s
{sending
? 'Sending message...'
: waitingForResponse
? 'Researching your data — searching, analyzing, writing report...'
? progressLabel || 'Initializing agent...'
: currentActivity || 'Agent is thinking...'}
</div>
</div>
</div>
)}
{/* Streaming content bubble — real-time response as it arrives */}
{waitingForResponse && streamingContent && (
<div className="flex justify-start">
<div
className="max-w-[75%] px-3 py-2 rounded-lg text-sm"
style={{
background: 'var(--color-bg-secondary)',
border: '1px solid var(--color-border)',
color: 'var(--color-text)',
}}
>
{progressLabel && (
<div className="flex items-center gap-2 mb-2 text-xs" style={{ color: 'var(--color-text-secondary)' }}>
<span className="inline-block w-2 h-2 rounded-full animate-pulse" style={{ background: 'var(--color-accent)' }} />
{progressLabel}
</div>
)}
<p className="whitespace-pre-wrap">{streamingContent}</p>
<p className="text-xs mt-1 opacity-70">
{streamElapsedMs > 0 && `${(streamElapsedMs / 1000).toFixed(1)}s elapsed`}
</p>
</div>
</div>
)}
<div ref={bottomRef} />
</div>
{/* Input area */}
src/openjarvis/server/agent_manager_routes.py
+146 -39
View File
@@ -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(),