root/fusion
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(KB-159): complete Step 4 — update frontend for streaming display

Browse Source
This commit is contained in:
gsxdsm
2026-03-30 08:55:07 -07:00
parent 9085e8d867
commit c12428e996
7 changed files with 639 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

120
packages/dashboard/app/api.ts
View File

@@ -649,6 +649,14 @@ export interface PlanningSession {
summary: PlanningSummary | null;
}
/** SSE event types for planning session streaming */
export type PlanningStreamEvent =
| { type: "thinking"; data: string }
| { type: "question"; data: PlanningQuestion }
| { type: "summary"; data: PlanningSummary }
| { type: "error"; data: string }
| { type: "complete"; data: Record<string, never> };
/** Start a new planning session with an initial plan */
export function startPlanning(initialPlan: string): Promise<PlanningSession> {
return api<PlanningSession>("/planning/start", {
@@ -657,6 +665,14 @@ export function startPlanning(initialPlan: string): Promise<PlanningSession> {
});
}
/** Start a new planning session with AI streaming support */
export function startPlanningStreaming(initialPlan: string): Promise<{ sessionId: string }> {
return api<{ sessionId: string }>("/planning/start-streaming", {
method: "POST",
body: JSON.stringify({ initialPlan }),
});
}
/** Submit a response to the current planning question */
export function respondToPlanning(
sessionId: string,
@@ -683,3 +699,107 @@ export function createTaskFromPlanning(sessionId: string): Promise<Task> {
body: JSON.stringify({ sessionId }),
});
}
/** Get the SSE stream URL for a planning session */
export function getPlanningStreamUrl(sessionId: string): string {
return `/api/planning/${encodeURIComponent(sessionId)}/stream`;
}
/** Connect to planning session SSE stream and handle events
*
* Returns an object with:
* - close: function to close the connection
* - reconnect: function to reconnect after error
*/
export function connectPlanningStream(
sessionId: string,
handlers: {
onThinking?: (data: string) => void;
onQuestion?: (data: PlanningQuestion) => void;
onSummary?: (data: PlanningSummary) => void;
onError?: (data: string) => void;
onComplete?: () => void;
}
): { close: () => void; isConnected: () => boolean } {
const url = getPlanningStreamUrl(sessionId);
const eventSource = new EventSource(url);
let isClosed = false;
eventSource.onopen = () => {
isClosed = false;
};
eventSource.onmessage = (event) => {
// Handle comment events (heartbeats)
if (event.data.startsWith(":")) return;
};
// Handle specific event types
eventSource.addEventListener("thinking", (event: Event) => {
try {
const messageEvent = event as MessageEvent;
const data = JSON.parse(messageEvent.data);
handlers.onThinking?.(data);
} catch {
const messageEvent = event as MessageEvent;
handlers.onThinking?.(messageEvent.data);
}
});
eventSource.addEventListener("question", (event: Event) => {
try {
const messageEvent = event as MessageEvent;
const data = JSON.parse(messageEvent.data) as PlanningQuestion;
handlers.onQuestion?.(data);
} catch (err) {
console.error("[planning] Failed to parse question event:", err);
}
});
eventSource.addEventListener("summary", (event: Event) => {
try {
const messageEvent = event as MessageEvent;
const data = JSON.parse(messageEvent.data) as PlanningSummary;
handlers.onSummary?.(data);
} catch (err) {
console.error("[planning] Failed to parse summary event:", err);
}
});
eventSource.addEventListener("error", (event: Event) => {
try {
const messageEvent = event as MessageEvent;
const data = JSON.parse(messageEvent.data);
handlers.onError?.(data.message || data);
} catch {
const messageEvent = event as MessageEvent;
handlers.onError?.(messageEvent.data || "Stream error");
}
close();
});
eventSource.addEventListener("complete", () => {
handlers.onComplete?.();
close();
});
// Handle connection errors
eventSource.onerror = () => {
if (!isClosed) {
handlers.onError?.("Connection lost");
close();
}
};
function close() {
if (!isClosed) {
isClosed = true;
eventSource.close();
}
}
return {
close,
isConnected: () => !isClosed && eventSource.readyState === EventSource.OPEN,
};
}

176
packages/dashboard/app/components/PlanningModeModal.tsx
View File

@@ -2,9 +2,11 @@ import { useState, useCallback, useEffect, useRef } from "react";
import type { Task, PlanningQuestion, PlanningSummary } from "@kb/core";
import {
startPlanning,
startPlanningStreaming,
respondToPlanning,
cancelPlanning,
createTaskFromPlanning,
connectPlanningStream,
type PlanningSession,
} from "../api";
import { Lightbulb, X, Loader2, CheckCircle, ArrowLeft, ArrowRight, Sparkles } from "lucide-react";
@@ -41,7 +43,11 @@ export function PlanningModeModal({ isOpen, onClose, onTaskCreated, tasks, initi
const [responseHistory, setResponseHistory] = useState<QuestionResponse[]>([]);
const [editedSummary, setEditedSummary] = useState<PlanningSummary | null>(null);
const [hasAutoStarted, setHasAutoStarted] = useState(false);
const [streamingOutput, setStreamingOutput] = useState<string>("");
const [showThinking, setShowThinking] = useState(true);
const textareaRef = useRef<HTMLTextAreaElement>(null);
const streamConnectionRef = useRef<{ close: () => void; isConnected: () => boolean } | null>(null);
const currentSessionIdRef = useRef<string | null>(null);
// Focus textarea when opening
useEffect(() => {
@@ -70,6 +76,14 @@ export function PlanningModeModal({ isOpen, onClose, onTaskCreated, tasks, initi
}
}, [isOpen]);
// Cleanup stream connection on unmount
useEffect(() => {
return () => {
streamConnectionRef.current?.close();
streamConnectionRef.current = null;
};
}, []);
// Handle browser unload during active session
useEffect(() => {
if (!isOpen) return;
@@ -79,6 +93,8 @@ export function PlanningModeModal({ isOpen, onClose, onTaskCreated, tasks, initi
e.preventDefault();
e.returnValue = "";
}
// Close stream connection
streamConnectionRef.current?.close();
};
window.addEventListener("beforeunload", handleBeforeUnload);
@@ -109,20 +125,52 @@ export function PlanningModeModal({ isOpen, onClose, onTaskCreated, tasks, initi
if (!initialPlan.trim()) return;
setError(null);
setStreamingOutput("");
setView({ type: "loading" });
try {
const session = await startPlanning(initialPlan.trim());
if (session.currentQuestion) {
setView({ type: "question", session });
} else if (session.summary) {
setView({ type: "summary", session, summary: session.summary });
setEditedSummary(session.summary);
}
// Use streaming mode for real-time AI thinking display
const { sessionId } = await startPlanningStreaming(initialPlan.trim());
currentSessionIdRef.current = sessionId;
// Connect to SSE stream
const connection = connectPlanningStream(sessionId, {
onThinking: (data) => {
setStreamingOutput((prev) => prev + data);
},
onQuestion: (question) => {
setView({
type: "question",
session: { sessionId, currentQuestion: question, summary: null },
});
setStreamingOutput("");
},
onSummary: (summary) => {
setView({
type: "summary",
session: { sessionId, currentQuestion: null, summary },
summary,
});
setEditedSummary(summary);
setStreamingOutput("");
},
onError: (message) => {
setError(message);
setView({ type: "initial" });
setStreamingOutput("");
currentSessionIdRef.current = null;
},
onComplete: () => {
currentSessionIdRef.current = null;
},
});
streamConnectionRef.current = connection;
setResponseHistory([]);
} catch (err: any) {
setError(err.message || "Failed to start planning session");
setView({ type: "initial" });
currentSessionIdRef.current = null;
}
}, [initialPlan]);
@@ -131,20 +179,50 @@ export function PlanningModeModal({ isOpen, onClose, onTaskCreated, tasks, initi
if (!plan.trim()) return;
setError(null);
setStreamingOutput("");
setView({ type: "loading" });
try {
const session = await startPlanning(plan.trim());
if (session.currentQuestion) {
setView({ type: "question", session });
} else if (session.summary) {
setView({ type: "summary", session, summary: session.summary });
setEditedSummary(session.summary);
}
const { sessionId } = await startPlanningStreaming(plan.trim());
currentSessionIdRef.current = sessionId;
const connection = connectPlanningStream(sessionId, {
onThinking: (data) => {
setStreamingOutput((prev) => prev + data);
},
onQuestion: (question) => {
setView({
type: "question",
session: { sessionId, currentQuestion: question, summary: null },
});
setStreamingOutput("");
},
onSummary: (summary) => {
setView({
type: "summary",
session: { sessionId, currentQuestion: null, summary },
summary,
});
setEditedSummary(summary);
setStreamingOutput("");
},
onError: (message) => {
setError(message);
setView({ type: "initial" });
setStreamingOutput("");
currentSessionIdRef.current = null;
},
onComplete: () => {
currentSessionIdRef.current = null;
},
});
streamConnectionRef.current = connection;
setResponseHistory([]);
} catch (err: any) {
setError(err.message || "Failed to start planning session");
setView({ type: "initial" });
currentSessionIdRef.current = null;
}
}, []);
@@ -153,19 +231,59 @@ export function PlanningModeModal({ isOpen, onClose, onTaskCreated, tasks, initi
if (view.type !== "question") return;
const { session } = view;
const sessionId = session.sessionId;
setError(null);
setStreamingOutput("");
setView({ type: "loading" });
try {
const updatedSession = await respondToPlanning(session.sessionId, responses);
// Close previous connection if any
streamConnectionRef.current?.close();
// Submit response - this will trigger the AI to process and stream
const updatedSession = await respondToPlanning(sessionId, responses);
setResponseHistory((prev) => [...prev, responses]);
// If we got an immediate response (non-streaming mode), use it
if (updatedSession.summary) {
setView({ type: "summary", session: updatedSession, summary: updatedSession.summary });
setEditedSummary(updatedSession.summary);
} else if (updatedSession.currentQuestion) {
setView({ type: "question", session: updatedSession });
return;
}
if (updatedSession.currentQuestion) {
setView({ type: "question", session: updatedSession });
return;
}
// Otherwise, set up streaming for the next question
const connection = connectPlanningStream(sessionId, {
onThinking: (data) => {
setStreamingOutput((prev) => prev + data);
},
onQuestion: (question) => {
setView({
type: "question",
session: { sessionId, currentQuestion: question, summary: null },
});
setStreamingOutput("");
},
onSummary: (summary) => {
setView({
type: "summary",
session: { sessionId, currentQuestion: null, summary },
summary,
});
setEditedSummary(summary);
setStreamingOutput("");
},
onError: (message) => {
setError(message);
setView({ type: "question", session });
setStreamingOutput("");
},
});
streamConnectionRef.current = connection;
} catch (err: any) {
setError(err.message || "Failed to submit response");
setView({ type: "question", session });
@@ -175,6 +293,10 @@ export function PlanningModeModal({ isOpen, onClose, onTaskCreated, tasks, initi
);
const handleCancel = useCallback(async () => {
// Always close the stream connection
streamConnectionRef.current?.close();
streamConnectionRef.current = null;
if (view.type === "question" || view.type === "summary") {
try {
await cancelPlanning(view.session.sessionId);
@@ -187,6 +309,8 @@ export function PlanningModeModal({ isOpen, onClose, onTaskCreated, tasks, initi
setError(null);
setResponseHistory([]);
setEditedSummary(null);
setStreamingOutput("");
currentSessionIdRef.current = null;
onClose();
}, [view, onClose]);
@@ -309,7 +433,23 @@ export function PlanningModeModal({ isOpen, onClose, onTaskCreated, tasks, initi
{view.type === "loading" && (
<div className="planning-loading">
<Loader2 size={40} className="spin" style={{ color: "var(--todo)" }} />
<p>Thinking...</p>
<p>{streamingOutput ? "AI is thinking..." : "Connecting..."}</p>
{streamingOutput && (
<div className="planning-thinking-container">
<button
className="planning-thinking-toggle"
onClick={() => setShowThinking(!showThinking)}
type="button"
>
{showThinking ? "Hide thinking" : "Show thinking"}
</button>
{showThinking && (
<div className="planning-thinking-output">
<pre>{streamingOutput}</pre>
</div>
)}
</div>
)}
</div>
)}

27
packages/dashboard/src/planning.ts
View File

@@ -634,17 +634,32 @@ async function continueAgentConversation(session: Session, message: string): Pro
try {
// Clear thinking output for this turn
const previousThinking = session.thinkingOutput;
session.thinkingOutput = "";
// Send message to agent - it will stream thinking via onThinking callback
const response = await session.agent.session.send(message);
// Send message to agent using .prompt() - it will stream thinking via onThinking callback
await session.agent.session.prompt(message);
// Combine any thinking output with the response text
const fullResponse = session.thinkingOutput + (response?.text || "");
// Get the response text from the agent's state
const lastMessage = session.agent.session.state.messages
.filter(m => m.role === "assistant")
.pop();
let responseText = session.thinkingOutput;
if (lastMessage?.content) {
// Handle both string and array content types
if (typeof lastMessage.content === "string") {
responseText = lastMessage.content;
} else if (Array.isArray(lastMessage.content)) {
// Extract text from content blocks
responseText = lastMessage.content
.filter((c): c is { type: "text"; text: string } => c.type === "text")
.map(c => c.text)
.join("");
}
}
// Parse the JSON response
const parsed = parseAgentResponse(fullResponse);
const parsed = parseAgentResponse(responseText);
if (parsed.type === "question") {
session.currentQuestion = parsed.data;

41
packages/dashboard/src/routes.ts
View File

@@ -2169,6 +2169,44 @@ export function createApiRoutes(store: TaskStore, options?: ServerOptions): Rout
}
});
/**
* POST /api/planning/start-streaming
* Start a new planning session with AI agent streaming.
* Body: { initialPlan: string }
* Returns: { sessionId: string }
*
* After receiving sessionId, connect to GET /api/planning/:sessionId/stream
* for real-time thinking output and questions.
*/
router.post("/planning/start-streaming", async (req, res) => {
try {
const { initialPlan } = req.body;
if (!initialPlan || typeof initialPlan !== "string") {
res.status(400).json({ error: "initialPlan is required and must be a string" });
return;
}
if (initialPlan.length > 500) {
res.status(400).json({ error: "initialPlan must be 500 characters or less" });
return;
}
const ip = req.ip || req.socket.remoteAddress || "unknown";
const rootDir = store.getRootDir();
const { createSessionWithAgent, RateLimitError } = await import("./planning.js");
const sessionId = await createSessionWithAgent(ip, initialPlan, rootDir);
res.status(201).json({ sessionId });
} catch (err: any) {
if (err.name === "RateLimitError") {
res.status(429).json({ error: err.message });
} else {
res.status(500).json({ error: err.message || "Failed to start planning session" });
}
}
});
/**
* POST /api/planning/respond
* Submit a response to the current planning question.
@@ -2322,7 +2360,8 @@ export function createApiRoutes(store: TaskStore, options?: ServerOptions): Rout
// Subscribe to session events
const unsubscribe = planningStreamManager.subscribe(sessionId, (event) => {
try {
res.write(`event: ${event.type}\ndata: ${JSON.stringify(event.data ?? {})}\n\n`);
const data = (event as { data?: unknown }).data;
res.write(`event: ${event.type}\ndata: ${JSON.stringify(data ?? {})}\n\n`);
// End stream on complete or error
if (event.type === "complete" || event.type === "error") {