feat: add background AI sessions with persistent storage and SSE streaming

Introduces ai_sessions table (schema v9), AiSessionStore, and background
session support for mission interviews, planning, and subtask breakdown.
Adds BackgroundTasksIndicator component, SSE-based progress streaming,
and dashboard API routes for session management.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
gsxdsm
2026-04-04 13:42:02 -07:00
parent 4f2b840eb4
commit eef87a2aea
23 changed files with 1128 additions and 68 deletions

View File

@@ -69,6 +69,7 @@ describe("Database", () => {
expect(tableNames).toContain("milestones");
expect(tableNames).toContain("slices");
expect(tableNames).toContain("mission_features");
expect(tableNames).toContain("ai_sessions");
});
it("creates all expected indexes", () => {
@@ -83,10 +84,12 @@ describe("Database", () => {
expect(indexNames).toContain("idxArchivedTasksId");
expect(indexNames).toContain("idxAgentHeartbeatsAgentId");
expect(indexNames).toContain("idxAgentHeartbeatsRunId");
expect(indexNames).toContain("idxAiSessionsStatus");
expect(indexNames).toContain("idxAiSessionsType");
});
it("seeds schema version", () => {
expect(db.getSchemaVersion()).toBe(8);
expect(db.getSchemaVersion()).toBe(9);
});
it("seeds lastModified", () => {
@@ -109,7 +112,7 @@ describe("Database", () => {
it("is idempotent - calling init() twice does not fail", () => {
expect(() => db.init()).not.toThrow();
expect(db.getSchemaVersion()).toBe(8);
expect(db.getSchemaVersion()).toBe(9);
});
it("does not overwrite existing config on re-init", () => {
@@ -716,7 +719,7 @@ describe("schema migrations", () => {
db.init();
// Verify version bumped to 5 (includes v1→v2, v2→v3, v3→v4, and v4→v5 migrations)
expect(db.getSchemaVersion()).toBe(8);
expect(db.getSchemaVersion()).toBe(9);
// Verify new columns exist and existing data is intact
const cols = db.prepare("PRAGMA table_info(tasks)").all() as Array<{ name: string }>;
@@ -741,11 +744,11 @@ describe("schema migrations", () => {
const db = new Database(kbDir);
db.init();
expect(db.getSchemaVersion()).toBe(8);
expect(db.getSchemaVersion()).toBe(9);
// Re-init should not fail
db.init();
expect(db.getSchemaVersion()).toBe(8);
expect(db.getSchemaVersion()).toBe(9);
db.close();
});
@@ -840,7 +843,7 @@ describe("schema migrations", () => {
db.init();
// Verify version bumped to 5
expect(db.getSchemaVersion()).toBe(8);
expect(db.getSchemaVersion()).toBe(9);
// Verify new columns exist and existing data is intact
const cols = db.prepare("PRAGMA table_info(tasks)").all() as Array<{ name: string }>;
@@ -1050,7 +1053,7 @@ describe("createDatabase factory", () => {
const db = createDatabase(kbDir);
db.init();
expect(db.getSchemaVersion()).toBe(8);
expect(db.getSchemaVersion()).toBe(9);
expect(db.getLastModified()).toBeGreaterThan(0);
db.close();

View File

@@ -59,7 +59,7 @@ export function fromJson<T>(json: string | null | undefined): T | undefined {
// ── Schema Definition ────────────────────────────────────────────────
const SCHEMA_VERSION = 8;
const SCHEMA_VERSION = 9;
function normalizeTaskComments(
steeringComments: SteeringComment[] | undefined,
@@ -424,8 +424,32 @@ export class Database {
});
}
if (version < 9) {
this.applyMigration(9, () => {
this.db.exec(`
CREATE TABLE IF NOT EXISTS ai_sessions (
id TEXT PRIMARY KEY,
type TEXT NOT NULL,
status TEXT NOT NULL,
title TEXT NOT NULL,
inputPayload TEXT NOT NULL,
conversationHistory TEXT DEFAULT '[]',
currentQuestion TEXT,
result TEXT,
thinkingOutput TEXT DEFAULT '',
error TEXT,
projectId TEXT,
createdAt TEXT NOT NULL,
updatedAt TEXT NOT NULL
)
`);
this.db.exec(`CREATE INDEX IF NOT EXISTS idxAiSessionsStatus ON ai_sessions(status)`);
this.db.exec(`CREATE INDEX IF NOT EXISTS idxAiSessionsType ON ai_sessions(type)`);
});
}
// Future migrations go here:
// if (version < 9) { this.applyMigration(9, () => { ... }); }
// if (version < 10) { this.applyMigration(10, () => { ... }); }
}
/**

View File

@@ -2753,6 +2753,11 @@ ${stepsSection}`;
return this.tasksDir;
}
/** Expose the shared Database instance for co-located stores (e.g. AiSessionStore). */
getDatabase(): Database {
return this.db;
}
private generateSpecifiedPrompt(task: Task): string {
const deps =
task.dependencies.length > 0

View File

@@ -28,6 +28,7 @@ import { AgentListModal } from "./components/AgentListModal";
import { AgentsView } from "./components/AgentsView";
import { ScriptsModal } from "./components/ScriptsModal";
import { ExecutorStatusBar } from "./components/ExecutorStatusBar";
import { useBackgroundSessions } from "./hooks/useBackgroundSessions";
import { useTasks } from "./hooks/useTasks";
import { useProjects } from "./hooks/useProjects";
import { useCurrentProject } from "./hooks/useCurrentProject";
@@ -52,6 +53,9 @@ function AppInner() {
// Theme management
const { themeMode, colorTheme, setThemeMode, setColorTheme } = useTheme();
// Background AI sessions
const { sessions: bgSessions, generating: bgGenerating, needsInput: bgNeedsInput, dismissSession: bgDismiss } = useBackgroundSessions(currentProject?.id);
// View state
const [viewMode, setViewMode] = useState<ViewMode>(() => {
if (typeof window !== "undefined") {
@@ -90,6 +94,9 @@ function AppInner() {
const [missionsOpen, setMissionsOpen] = useState(false);
const [agentsOpen, setAgentsOpen] = useState(false);
const [scriptsOpen, setScriptsOpen] = useState(false);
const [planningResumeSessionId, setPlanningResumeSessionId] = useState<string | undefined>(undefined);
const [subtaskResumeSessionId, setSubtaskResumeSessionId] = useState<string | undefined>(undefined);
const [missionResumeSessionId, setMissionResumeSessionId] = useState<string | undefined>(undefined);
const [terminalInitialCommand, setTerminalInitialCommand] = useState<string | undefined>(undefined);
const [settingsInitialSection, setSettingsInitialSection] = useState<SectionId | undefined>(undefined);
const [setupWizardOpen, setSetupWizardOpen] = useState(false);
@@ -366,6 +373,7 @@ function AppInner() {
const handlePlanningClose = useCallback(() => {
setIsPlanningOpen(false);
setPlanningInitialPlan(null);
setPlanningResumeSessionId(undefined);
}, []);
const handlePlanningTaskCreated = useCallback((task: Task) => {
addToast(`Created ${task.id} from planning mode`, "success");
@@ -388,6 +396,7 @@ function AppInner() {
const handleSubtaskClose = useCallback(() => {
setIsSubtaskOpen(false);
setSubtaskInitialDescription(null);
setSubtaskResumeSessionId(undefined);
}, []);
const handleSubtaskTasksCreated = useCallback((createdTasks: Task[]) => {
@@ -626,7 +635,27 @@ function AppInner() {
{renderMainContent()}
</div>
{viewMode === "project" && currentProject && (
<ExecutorStatusBar tasks={tasks} projectId={currentProject.id} taskStuckTimeoutMs={taskStuckTimeoutMs} />
<ExecutorStatusBar
tasks={tasks}
projectId={currentProject.id}
taskStuckTimeoutMs={taskStuckTimeoutMs}
backgroundSessions={bgSessions}
backgroundGenerating={bgGenerating}
backgroundNeedsInput={bgNeedsInput}
onOpenBackgroundSession={(session) => {
if (session.type === "planning") {
setPlanningResumeSessionId(session.id);
setIsPlanningOpen(true);
} else if (session.type === "subtask") {
setSubtaskResumeSessionId(session.id);
setIsSubtaskOpen(true);
} else if (session.type === "mission_interview") {
setMissionResumeSessionId(session.id);
setMissionsOpen(true);
}
}}
onDismissBackgroundSession={bgDismiss}
/>
)}
{detailTask && (
<TaskDetailModal
@@ -673,6 +702,7 @@ function AppInner() {
tasks={tasks}
initialPlan={planningInitialPlan ?? undefined}
projectId={currentProject?.id}
resumeSessionId={planningResumeSessionId}
/>
<SubtaskBreakdownModal
isOpen={isSubtaskOpen}
@@ -680,6 +710,7 @@ function AppInner() {
initialDescription={subtaskInitialDescription ?? ""}
onTasksCreated={handleSubtaskTasksCreated}
projectId={currentProject?.id}
resumeSessionId={subtaskResumeSessionId}
/>
<TerminalModal
isOpen={terminalOpen}
@@ -759,9 +790,10 @@ function AppInner() {
/>
<MissionManager
isOpen={missionsOpen}
onClose={() => setMissionsOpen(false)}
onClose={() => { setMissionsOpen(false); setMissionResumeSessionId(undefined); }}
addToast={addToast}
projectId={currentProject?.id}
resumeSessionId={missionResumeSessionId}
availableTasks={tasks.map((t) => ({ id: t.id, title: t.title }))}
onSelectTask={(taskId) => {
const task = tasks.find((t) => t.id === taskId);

View File

@@ -2593,3 +2593,42 @@ export function connectMissionInterviewStream(
isConnected: () => !isClosed && eventSource.readyState === EventSource.OPEN,
};
}
// ── AI Sessions (Background Tasks) ─────────────────────────────────────────
export interface AiSessionSummary {
id: string;
type: "planning" | "subtask" | "mission_interview";
status: "generating" | "awaiting_input" | "complete" | "error";
title: string;
projectId: string | null;
updatedAt: string;
}
export interface AiSessionDetail extends AiSessionSummary {
inputPayload: string;
conversationHistory: string;
currentQuestion: string | null;
result: string | null;
thinkingOutput: string;
error: string | null;
createdAt: string;
}
export async function fetchAiSessions(projectId?: string): Promise<AiSessionSummary[]> {
const params = projectId ? `?projectId=${encodeURIComponent(projectId)}` : "";
const res = await fetch(buildApiUrl(`/ai-sessions${params}`));
if (!res.ok) return [];
const data = await res.json();
return data.sessions ?? [];
}
export async function fetchAiSession(id: string): Promise<AiSessionDetail | null> {
const res = await fetch(buildApiUrl(`/ai-sessions/${encodeURIComponent(id)}`));
if (!res.ok) return null;
return res.json();
}
export async function deleteAiSession(id: string): Promise<void> {
await fetch(buildApiUrl(`/ai-sessions/${encodeURIComponent(id)}`), { method: "DELETE" });
}

View File

@@ -0,0 +1,162 @@
import { useState, useRef, useEffect } from "react";
import { Lightbulb, Layers, Target, Loader2, HelpCircle, X } from "lucide-react";
import type { AiSessionSummary } from "../api";
interface BackgroundTasksIndicatorProps {
sessions: AiSessionSummary[];
generating: number;
needsInput: number;
onOpenSession: (session: AiSessionSummary) => void;
onDismissSession: (id: string) => void;
}
const TYPE_ICONS = {
planning: Lightbulb,
subtask: Layers,
mission_interview: Target,
} as const;
const TYPE_LABELS = {
planning: "Planning",
subtask: "Subtask Breakdown",
mission_interview: "Mission Interview",
} as const;
export function BackgroundTasksIndicator({
sessions,
generating,
needsInput,
onOpenSession,
onDismissSession,
}: BackgroundTasksIndicatorProps) {
const [popoverOpen, setPopoverOpen] = useState(false);
const containerRef = useRef<HTMLDivElement>(null);
// Close popover on outside click
useEffect(() => {
if (!popoverOpen) return;
const handler = (e: MouseEvent) => {
if (containerRef.current && !containerRef.current.contains(e.target as Node)) {
setPopoverOpen(false);
}
};
document.addEventListener("mousedown", handler);
return () => document.removeEventListener("mousedown", handler);
}, [popoverOpen]);
if (sessions.length === 0) return null;
const total = sessions.length;
const hasAttention = needsInput > 0;
return (
<div ref={containerRef} className="background-tasks-indicator" style={{ position: "relative" }}>
<button
className="background-tasks-indicator__pill"
onClick={() => setPopoverOpen((prev) => !prev)}
style={{
display: "inline-flex",
alignItems: "center",
gap: "6px",
padding: "2px 10px",
borderRadius: "12px",
border: "1px solid var(--border-color)",
background: hasAttention ? "var(--triage)" : "var(--surface-secondary)",
color: hasAttention ? "#fff" : "var(--text-primary)",
cursor: "pointer",
fontSize: "12px",
fontWeight: 500,
lineHeight: "20px",
whiteSpace: "nowrap",
}}
title={`${total} background AI task${total !== 1 ? "s" : ""}${needsInput > 0 ? ` (${needsInput} need${needsInput !== 1 ? "" : "s"} input)` : ""}`}
>
{generating > 0 && (
<Loader2 size={12} style={{ animation: "spin 1s linear infinite" }} />
)}
{needsInput > 0 && generating === 0 && <HelpCircle size={12} />}
<span>AI {total}</span>
</button>
{popoverOpen && (
<div
className="background-tasks-indicator__popover"
style={{
position: "absolute",
bottom: "calc(100% + 8px)",
left: 0,
minWidth: "280px",
maxWidth: "360px",
background: "var(--surface-primary)",
border: "1px solid var(--border-color)",
borderRadius: "8px",
boxShadow: "0 4px 12px rgba(0,0,0,0.15)",
zIndex: 1000,
overflow: "hidden",
}}
>
<div style={{ padding: "8px 12px", borderBottom: "1px solid var(--border-color)", fontSize: "12px", fontWeight: 600, color: "var(--text-secondary)" }}>
Background Tasks
</div>
<div style={{ maxHeight: "240px", overflowY: "auto" }}>
{sessions.map((session) => {
const Icon = TYPE_ICONS[session.type];
const isGenerating = session.status === "generating";
const isAwaiting = session.status === "awaiting_input";
return (
<div
key={session.id}
style={{
display: "flex",
alignItems: "center",
gap: "8px",
padding: "8px 12px",
borderBottom: "1px solid var(--border-color)",
cursor: "pointer",
}}
onClick={() => {
onOpenSession(session);
setPopoverOpen(false);
}}
>
<Icon size={14} style={{ flexShrink: 0, color: "var(--text-secondary)" }} />
<div style={{ flex: 1, minWidth: 0 }}>
<div style={{ fontSize: "13px", fontWeight: 500, overflow: "hidden", textOverflow: "ellipsis", whiteSpace: "nowrap" }}>
{session.title}
</div>
<div style={{ fontSize: "11px", color: "var(--text-secondary)" }}>
{TYPE_LABELS[session.type]}
{isGenerating && " — generating..."}
{isAwaiting && " — needs input"}
</div>
</div>
{isGenerating && <Loader2 size={14} style={{ flexShrink: 0, animation: "spin 1s linear infinite", color: "var(--color-success)" }} />}
{isAwaiting && <HelpCircle size={14} style={{ flexShrink: 0, color: "var(--triage)" }} />}
<button
onClick={(e) => {
e.stopPropagation();
onDismissSession(session.id);
}}
style={{
flexShrink: 0,
background: "none",
border: "none",
cursor: "pointer",
padding: "2px",
color: "var(--text-secondary)",
borderRadius: "4px",
}}
title="Dismiss"
>
<X size={12} />
</button>
</div>
);
})}
</div>
</div>
)}
</div>
);
}

View File

@@ -2,7 +2,8 @@ import { useMemo } from "react";
import type { Task } from "@fusion/core";
import { Activity, AlertTriangle, Clock, Pause, Play, Zap } from "lucide-react";
import { useExecutorStats } from "../hooks/useExecutorStats";
import type { ExecutorState } from "../api";
import type { ExecutorState, AiSessionSummary } from "../api";
import { BackgroundTasksIndicator } from "./BackgroundTasksIndicator";
interface ExecutorStatusBarProps {
/** Task list (shared with the board to keep counts in sync) */
@@ -11,6 +12,12 @@ interface ExecutorStatusBarProps {
projectId?: string;
/** Project-level stuck task timeout in milliseconds (undefined = disabled) */
taskStuckTimeoutMs?: number;
/** Background AI sessions */
backgroundSessions?: AiSessionSummary[];
backgroundGenerating?: number;
backgroundNeedsInput?: number;
onOpenBackgroundSession?: (session: AiSessionSummary) => void;
onDismissBackgroundSession?: (id: string) => void;
}
/**
@@ -60,7 +67,7 @@ function getStateDisplay(state: ExecutorState): { label: string; color: string;
* - Executor state badge (idle/running/paused)
* - Last activity timestamp
*/
export function ExecutorStatusBar({ tasks, projectId, taskStuckTimeoutMs }: ExecutorStatusBarProps) {
export function ExecutorStatusBar({ tasks, projectId, taskStuckTimeoutMs, backgroundSessions, backgroundGenerating, backgroundNeedsInput, onOpenBackgroundSession, onDismissBackgroundSession }: ExecutorStatusBarProps) {
const { stats, loading, error } = useExecutorStats(tasks, projectId, taskStuckTimeoutMs);
const stateDisplay = useMemo(() => getStateDisplay(stats.executorState), [stats.executorState]);
@@ -94,6 +101,20 @@ export function ExecutorStatusBar({ tasks, projectId, taskStuckTimeoutMs }: Exec
role="status"
aria-label="Executor status"
>
{/* Background AI tasks indicator */}
{backgroundSessions && backgroundSessions.length > 0 && onOpenBackgroundSession && onDismissBackgroundSession && (
<>
<BackgroundTasksIndicator
sessions={backgroundSessions}
generating={backgroundGenerating ?? 0}
needsInput={backgroundNeedsInput ?? 0}
onOpenSession={onOpenBackgroundSession}
onDismissSession={onDismissBackgroundSession}
/>
<span className="executor-status-bar__divider" aria-hidden="true" />
</>
)}
{/* Running tasks */}
<div className="executor-status-bar__segment">
<span

View File

@@ -6,6 +6,7 @@ import {
cancelMissionInterview,
createMissionFromInterview,
connectMissionInterviewStream,
fetchAiSession,
type MissionPlanSummary,
type MissionPlanMilestone,
type MissionPlanSlice,
@@ -35,6 +36,7 @@ interface MissionInterviewModalProps {
onMissionCreated: (mission: MissionWithHierarchy) => void;
projectId?: string;
initialGoal?: string;
resumeSessionId?: string;
}
interface QuestionResponse {
@@ -60,6 +62,7 @@ export function MissionInterviewModal({
onMissionCreated,
projectId,
initialGoal: initialGoalProp,
resumeSessionId,
}: MissionInterviewModalProps) {
const [missionGoal, setMissionGoal] = useState("");
const [view, setView] = useState<ViewState>({ type: "initial" });
@@ -150,6 +153,79 @@ export function MissionInterviewModal({
}
}, [isOpen]);
// Reconnect to a persisted session when resumeSessionId is provided
useEffect(() => {
if (!isOpen || !resumeSessionId || view.type !== "initial") return;
let cancelled = false;
fetchAiSession(resumeSessionId).then((session) => {
if (cancelled || !session) return;
if (session.status === "awaiting_input" && session.currentQuestion) {
try {
const question = JSON.parse(session.currentQuestion) as import("@fusion/core").PlanningQuestion;
currentSessionIdRef.current = session.id;
setHasProgress(true);
setView({ type: "question", sessionId: session.id, question });
} catch {
setError("Failed to restore session question.");
}
} else if (session.status === "complete" && session.result) {
try {
const summary = JSON.parse(session.result) as MissionPlanSummary;
currentSessionIdRef.current = session.id;
setHasProgress(true);
setEditedSummary(summary);
setView({ type: "summary", sessionId: session.id, summary });
} catch {
setError("Failed to restore session result.");
}
} else if (session.status === "generating") {
currentSessionIdRef.current = session.id;
setHasProgress(true);
if (session.thinkingOutput) {
setStreamingOutput(session.thinkingOutput);
}
setView({ type: "loading" });
const connection = connectMissionInterviewStream(session.id, projectId, {
onThinking: (data) => {
setStreamingOutput((prev) => prev + data);
},
onQuestion: (question) => {
setView({ type: "question", sessionId: session.id, question });
setStreamingOutput("");
},
onSummary: (summary) => {
setView({ type: "summary", sessionId: session.id, summary });
setEditedSummary(summary);
setStreamingOutput("");
},
onError: (message) => {
setError(message);
setView({ type: "initial" });
setStreamingOutput("");
currentSessionIdRef.current = null;
},
onComplete: () => {
currentSessionIdRef.current = null;
},
});
streamConnectionRef.current = connection;
} else if (session.status === "error") {
setError(session.error ?? "The session encountered an error.");
}
}).catch(() => {
if (!cancelled) setError("Failed to resume session.");
});
return () => {
cancelled = true;
};
}, [isOpen, resumeSessionId, view.type, projectId]);
// Cleanup stream on unmount
useEffect(() => {
return () => {

View File

@@ -60,6 +60,7 @@ interface MissionManagerProps {
projectId?: string;
onSelectTask?: (taskId: string) => void;
availableTasks?: Array<{ id: string; title?: string }>;
resumeSessionId?: string;
}
// Status badge colors — use CSS custom-property-compatible tokens
@@ -146,7 +147,7 @@ const EMPTY_FEATURE_FORM: FeatureFormData = {
status: "defined",
};
export function MissionManager({ isOpen, onClose, addToast, projectId, onSelectTask, availableTasks = [] }: MissionManagerProps) {
export function MissionManager({ isOpen, onClose, addToast, projectId, onSelectTask, availableTasks = [], resumeSessionId }: MissionManagerProps) {
const [missions, setMissions] = useState<Mission[]>([]);
const [selectedMission, setSelectedMission] = useState<MissionWithHierarchy | null>(null);
const [loading, setLoading] = useState(true);
@@ -183,6 +184,13 @@ export function MissionManager({ isOpen, onClose, addToast, projectId, onSelectT
// AI Interview modal
const [showInterviewModal, setShowInterviewModal] = useState(false);
// Auto-open interview modal when resuming a session
useEffect(() => {
if (isOpen && resumeSessionId) {
setShowInterviewModal(true);
}
}, [isOpen, resumeSessionId]);
// Delete confirmation
const [deleteConfirmId, setDeleteConfirmId] = useState<{ type: string; id: string } | null>(null);
@@ -1418,6 +1426,7 @@ export function MissionManager({ isOpen, onClose, addToast, projectId, onSelectT
addToast("Mission created from AI interview", "success");
}}
projectId={projectId}
resumeSessionId={resumeSessionId}
/>
</div>
);

View File

@@ -7,6 +7,7 @@ import {
cancelPlanning,
createTaskFromPlanning,
connectPlanningStream,
fetchAiSession,
type PlanningSession,
} from "../api";
import { Lightbulb, X, Loader2, CheckCircle, ArrowLeft, ArrowRight, Sparkles } from "lucide-react";
@@ -18,6 +19,8 @@ interface PlanningModeModalProps {
tasks: Task[];
initialPlan?: string;
projectId?: string;
/** When set, reconnect to a persisted background session instead of starting fresh */
resumeSessionId?: string;
}
interface QuestionResponse {
@@ -37,7 +40,7 @@ const EXAMPLE_PLANS = [
"Refactor the task card component for better performance",
];
export function PlanningModeModal({ isOpen, onClose, onTaskCreated, tasks, initialPlan: initialPlanProp, projectId }: PlanningModeModalProps) {
export function PlanningModeModal({ isOpen, onClose, onTaskCreated, tasks, initialPlan: initialPlanProp, projectId, resumeSessionId }: PlanningModeModalProps) {
const [initialPlan, setInitialPlan] = useState("");
const [view, setView] = useState<ViewState>({ type: "initial" });
const [error, setError] = useState<string | null>(null);
@@ -132,6 +135,59 @@ export function PlanningModeModal({ isOpen, onClose, onTaskCreated, tasks, initi
}
}, [isOpen, initialPlanProp, view.type, handleStartPlanning]);
// Resume a persisted background session
useEffect(() => {
if (!isOpen || !resumeSessionId || view.type !== "initial") return;
let cancelled = false;
(async () => {
try {
const session = await fetchAiSession(resumeSessionId);
if (cancelled || !session) return;
currentSessionIdRef.current = resumeSessionId;
if (session.status === "awaiting_input" && session.currentQuestion) {
const question = JSON.parse(session.currentQuestion);
setView({ type: "question", session: { sessionId: resumeSessionId, currentQuestion: question, summary: null } });
if (session.thinkingOutput) setStreamingOutput(session.thinkingOutput);
setHasProgress(true);
} else if (session.status === "complete" && session.result) {
const summary = JSON.parse(session.result);
setView({ type: "summary", session: { sessionId: resumeSessionId, currentQuestion: null, summary }, summary });
setEditedSummary(summary);
setHasProgress(true);
} else if (session.status === "generating") {
setView({ type: "loading" });
if (session.thinkingOutput) setStreamingOutput(session.thinkingOutput);
// Connect to live SSE stream to pick up new events
const connection = connectPlanningStream(resumeSessionId, projectId, {
onThinking: (data) => setStreamingOutput((prev) => prev + data),
onQuestion: (question) => {
setView({ type: "question", session: { sessionId: resumeSessionId, currentQuestion: question, summary: null } });
setStreamingOutput("");
setHasProgress(true);
},
onSummary: (summary) => {
setView({ type: "summary", session: { sessionId: resumeSessionId, currentQuestion: null, summary }, summary });
setEditedSummary(summary);
setStreamingOutput("");
setHasProgress(true);
},
onError: (message) => { setError(message); setView({ type: "initial" }); },
onComplete: () => { currentSessionIdRef.current = null; },
});
streamConnectionRef.current = connection;
} else if (session.status === "error") {
setError(session.error || "Session failed");
setView({ type: "initial" });
}
} catch {
setError("Failed to resume session");
}
})();
return () => { cancelled = true; };
}, [isOpen, resumeSessionId, view.type, projectId]);
// Reset hasAutoStarted when modal closes
useEffect(() => {
if (!isOpen) {

View File

@@ -5,6 +5,7 @@ import {
connectSubtaskStream,
createTasksFromBreakdown,
cancelSubtaskBreakdown,
fetchAiSession,
type SubtaskItem,
} from "../api";
import { CheckCircle, Loader2, ListTree, Plus, Trash2, X, GripVertical, ArrowUp, ArrowDown } from "lucide-react";
@@ -16,6 +17,7 @@ interface SubtaskBreakdownModalProps {
onTasksCreated: (tasks: Task[]) => void;
parentTaskId?: string;
projectId?: string;
resumeSessionId?: string;
}
type ViewState =
@@ -54,7 +56,7 @@ function hasDependencyCycle(subtasks: SubtaskItem[]): boolean {
return subtasks.some((item) => visit(item.id));
}
export function SubtaskBreakdownModal({ isOpen, onClose, initialDescription, onTasksCreated, parentTaskId, projectId }: SubtaskBreakdownModalProps) {
export function SubtaskBreakdownModal({ isOpen, onClose, initialDescription, onTasksCreated, parentTaskId, projectId, resumeSessionId }: SubtaskBreakdownModalProps) {
const [view, setView] = useState<ViewState>({ type: "initial" });
const [subtasks, setSubtasks] = useState<SubtaskItem[]>([]);
const [thinkingOutput, setThinkingOutput] = useState("");
@@ -147,6 +149,42 @@ export function SubtaskBreakdownModal({ isOpen, onClose, initialDescription, onT
}
}, [isOpen, initialDescription, beginBreakdown, resetState]);
useEffect(() => {
if (!isOpen || !resumeSessionId || view.type !== "initial") return;
void (async () => {
try {
const session = await fetchAiSession(resumeSessionId);
if (!session) return;
if (session.status === "generating" || session.status === "awaiting_input") {
setThinkingOutput(session.thinkingOutput ?? "");
setView({ type: "generating", sessionId: resumeSessionId });
streamRef.current?.close();
streamRef.current = connectSubtaskStream(resumeSessionId, projectId, {
onThinking: (data) => setThinkingOutput((prev) => prev + data),
onSubtasks: (items) => {
setSubtasks(items);
setView({ type: "editing", sessionId: resumeSessionId });
setDirty(false);
},
onError: (message) => {
setError(message);
setView({ type: "initial" });
},
});
} else if (session.status === "complete" && session.result) {
const items = JSON.parse(session.result) as SubtaskItem[];
setSubtasks(items);
setView({ type: "editing", sessionId: resumeSessionId });
} else if (session.status === "error") {
setError(session.error ?? "Session encountered an error");
}
} catch (err: any) {
setError(err.message || "Failed to resume session");
}
})();
}, [isOpen, resumeSessionId, view.type, projectId]);
useEffect(() => {
return () => {
streamRef.current?.close();

View File

@@ -0,0 +1,84 @@
import { useState, useEffect, useCallback, useRef } from "react";
import { fetchAiSessions, deleteAiSession, type AiSessionSummary } from "../api";
interface UseBackgroundSessionsResult {
sessions: AiSessionSummary[];
generating: number;
needsInput: number;
dismissSession: (id: string) => void;
refresh: () => void;
}
export function useBackgroundSessions(projectId?: string): UseBackgroundSessionsResult {
const [sessions, setSessions] = useState<AiSessionSummary[]>([]);
const eventSourceRef = useRef<EventSource | null>(null);
const refresh = useCallback(() => {
fetchAiSessions(projectId).then(setSessions).catch(() => {});
}, [projectId]);
// Initial fetch
useEffect(() => {
refresh();
}, [refresh]);
// Listen for SSE events
useEffect(() => {
const params = projectId ? `?projectId=${encodeURIComponent(projectId)}` : "";
const es = new EventSource(`/api/events${params}`);
eventSourceRef.current = es;
const handleUpdated = (e: MessageEvent) => {
try {
const updated = JSON.parse(e.data) as AiSessionSummary;
setSessions((prev) => {
const idx = prev.findIndex((s) => s.id === updated.id);
if (idx >= 0) {
const next = [...prev];
next[idx] = updated;
return next;
}
// New session — only add if active
if (updated.status === "generating" || updated.status === "awaiting_input") {
return [updated, ...prev];
}
return prev;
});
} catch { /* ignore */ }
};
const handleDeleted = (e: MessageEvent) => {
try {
const id = JSON.parse(e.data);
setSessions((prev) => prev.filter((s) => s.id !== id));
} catch { /* ignore */ }
};
es.addEventListener("ai_session:updated", handleUpdated);
es.addEventListener("ai_session:deleted", handleDeleted);
return () => {
es.removeEventListener("ai_session:updated", handleUpdated);
es.removeEventListener("ai_session:deleted", handleDeleted);
es.close();
};
}, [projectId]);
const dismissSession = useCallback((id: string) => {
deleteAiSession(id).catch(() => {});
setSessions((prev) => prev.filter((s) => s.id !== id));
}, []);
// Filter to only active sessions
const active = sessions.filter(
(s) => s.status === "generating" || s.status === "awaiting_input"
);
return {
sessions: active,
generating: active.filter((s) => s.status === "generating").length,
needsInput: active.filter((s) => s.status === "awaiting_input").length,
dismissSession,
refresh,
};
}

View File

@@ -22,7 +22,8 @@
"scripts": {
"build": "vite build && tsc",
"build:client": "vite build",
"dev": "vite build && tsc && tsc --noEmit -p tsconfig.app.json && vite dev",
"dev": "pnpm build && pnpm typecheck && pnpm dev:serve",
"dev:serve": "vite dev",
"test": "vitest run",
"typecheck": "tsc --noEmit && tsc --noEmit -p tsconfig.app.json",
"postinstall": "chmod +x node_modules/.pnpm/node-pty*/node_modules/node-pty/prebuilds/darwin-*/spawn-helper node_modules/.pnpm/node-pty*/node_modules/node-pty/prebuilds/darwin-*/*.node 2>/dev/null || true"

View File

@@ -0,0 +1,258 @@
/**
* AI Session Store
*
* Persists long-running AI session state (planning, subtask breakdown,
* mission interview) to SQLite so users can dismiss modals and return
* later — even from a different browser.
*
* The in-memory session Maps in planning.ts / subtask-breakdown.ts /
* mission-interview.ts remain the source of truth for live agent state.
* This store is the persistence shadow, updated at each state transition.
*/
import { EventEmitter } from "node:events";
import type { Database } from "@fusion/core";
// ── Types ───────────────────────────────────────────────────────────────
export type AiSessionType = "planning" | "subtask" | "mission_interview";
export type AiSessionStatus = "generating" | "awaiting_input" | "complete" | "error";
export interface AiSessionRow {
id: string;
type: AiSessionType;
status: AiSessionStatus;
title: string;
inputPayload: string; // JSON string
conversationHistory: string; // JSON string: [{question, response}]
currentQuestion: string | null; // JSON string or null
result: string | null; // JSON string or null
thinkingOutput: string;
error: string | null;
projectId: string | null;
createdAt: string;
updatedAt: string;
}
/** Summary returned by listActive (omits large fields) */
export interface AiSessionSummary {
id: string;
type: AiSessionType;
status: AiSessionStatus;
title: string;
projectId: string | null;
updatedAt: string;
}
export interface AiSessionStoreEvents {
"ai_session:updated": [AiSessionSummary];
"ai_session:deleted": [string]; // session id
}
// ── Constants ───────────────────────────────────────────────────────────
/** Max stored thinking output (50 KB). Older content trimmed from front. */
const MAX_THINKING_BYTES = 50 * 1024;
/** Debounce interval for thinking-only writes (ms). */
const THINKING_DEBOUNCE_MS = 2000;
// ── Store ───────────────────────────────────────────────────────────────
export class AiSessionStore extends EventEmitter<AiSessionStoreEvents> {
/** Pending debounce timers for thinking-only writes, keyed by session id. */
private thinkingTimers = new Map<string, ReturnType<typeof setTimeout>>();
constructor(private db: Database) {
super();
}
// ── CRUD ────────────────────────────────────────────────────────────
/**
* Insert or update an AI session row.
* Emits `ai_session:updated` after writing.
*/
upsert(session: AiSessionRow): void {
const now = new Date().toISOString();
const thinking = trimThinking(session.thinkingOutput);
this.db
.prepare(
`INSERT INTO ai_sessions (id, type, status, title, inputPayload, conversationHistory, currentQuestion, result, thinkingOutput, error, projectId, createdAt, updatedAt)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(id) DO UPDATE SET
status = excluded.status,
title = excluded.title,
conversationHistory = excluded.conversationHistory,
currentQuestion = excluded.currentQuestion,
result = excluded.result,
thinkingOutput = excluded.thinkingOutput,
error = excluded.error,
updatedAt = excluded.updatedAt`,
)
.run(
session.id,
session.type,
session.status,
session.title,
session.inputPayload,
session.conversationHistory,
session.currentQuestion ?? null,
session.result ?? null,
thinking,
session.error ?? null,
session.projectId ?? null,
session.createdAt || now,
now,
);
// Cancel any pending thinking debounce for this session
this.clearThinkingTimer(session.id);
this.emit("ai_session:updated", toSummary(session, now));
}
/**
* Update only the thinkingOutput field, debounced to reduce write frequency.
* Flushes immediately if `flush` is true (e.g. on status transition).
*/
updateThinking(sessionId: string, thinkingOutput: string, flush = false): void {
if (flush) {
this.clearThinkingTimer(sessionId);
this.writeThinking(sessionId, thinkingOutput);
return;
}
// Debounce: reset timer
this.clearThinkingTimer(sessionId);
const timer = setTimeout(() => {
this.thinkingTimers.delete(sessionId);
this.writeThinking(sessionId, thinkingOutput);
}, THINKING_DEBOUNCE_MS);
this.thinkingTimers.set(sessionId, timer);
}
/**
* Fetch a single session by ID. Returns null if not found.
*/
get(id: string): AiSessionRow | null {
const row = this.db
.prepare("SELECT * FROM ai_sessions WHERE id = ?")
.get(id) as unknown as AiSessionRow | undefined;
return row ?? null;
}
/**
* List active sessions (generating or awaiting_input).
* Optionally filtered by projectId.
*/
listActive(projectId?: string): AiSessionSummary[] {
if (projectId) {
return this.db
.prepare(
`SELECT id, type, status, title, projectId, updatedAt FROM ai_sessions
WHERE status IN ('generating', 'awaiting_input') AND projectId = ?
ORDER BY updatedAt DESC`,
)
.all(projectId) as unknown as AiSessionSummary[];
}
return this.db
.prepare(
`SELECT id, type, status, title, projectId, updatedAt FROM ai_sessions
WHERE status IN ('generating', 'awaiting_input')
ORDER BY updatedAt DESC`,
)
.all() as unknown as AiSessionSummary[];
}
/**
* Delete a session by ID. Emits `ai_session:deleted`.
*/
delete(id: string): void {
this.clearThinkingTimer(id);
this.db.prepare("DELETE FROM ai_sessions WHERE id = ?").run(id);
this.emit("ai_session:deleted", id);
}
/**
* Recover sessions after server restart.
* - `generating` sessions with a currentQuestion -> `awaiting_input`
* - `generating` sessions without -> `error`
*/
recoverStaleSessions(): number {
const now = new Date().toISOString();
let recovered = 0;
// Sessions that were generating and had a pending question — recoverable
const withQuestion = this.db
.prepare(
`UPDATE ai_sessions SET status = 'awaiting_input', updatedAt = ?
WHERE status = 'generating' AND currentQuestion IS NOT NULL`,
)
.run(now);
recovered += Number((withQuestion as any).changes ?? 0);
// Sessions that were generating with no question — unrecoverable
const withoutQuestion = this.db
.prepare(
`UPDATE ai_sessions SET status = 'error', error = 'Session interrupted — please restart', updatedAt = ?
WHERE status = 'generating' AND currentQuestion IS NULL`,
)
.run(now);
recovered += Number((withoutQuestion as any).changes ?? 0);
if (recovered > 0) {
console.log(`[ai-session-store] Recovered ${recovered} stale sessions after restart`);
}
return recovered;
}
/**
* Clean up completed/error sessions older than the given age (ms).
*/
cleanupOld(maxAgeMs: number): number {
const cutoff = new Date(Date.now() - maxAgeMs).toISOString();
const result = this.db
.prepare(
`DELETE FROM ai_sessions WHERE status IN ('complete', 'error') AND updatedAt < ?`,
)
.run(cutoff);
return Number((result as any).changes ?? 0);
}
// ── Internal ────────────────────────────────────────────────────────
private writeThinking(sessionId: string, thinkingOutput: string): void {
const now = new Date().toISOString();
this.db
.prepare("UPDATE ai_sessions SET thinkingOutput = ?, updatedAt = ? WHERE id = ?")
.run(trimThinking(thinkingOutput), now, sessionId);
}
private clearThinkingTimer(id: string): void {
const timer = this.thinkingTimers.get(id);
if (timer) {
clearTimeout(timer);
this.thinkingTimers.delete(id);
}
}
}
// ── Helpers ─────────────────────────────────────────────────────────────
function trimThinking(output: string): string {
if (output.length <= MAX_THINKING_BYTES) return output;
return output.slice(output.length - MAX_THINKING_BYTES);
}
function toSummary(session: AiSessionRow, updatedAt: string): AiSessionSummary {
return {
id: session.id,
type: session.type,
status: session.status,
title: session.title,
projectId: session.projectId,
updatedAt,
};
}

View File

@@ -17,6 +17,7 @@
import type { PlanningQuestion } from "@fusion/core";
import { randomUUID } from "node:crypto";
import { EventEmitter } from "node:events";
import type { AiSessionStore, AiSessionRow } from "./ai-session-store.js";
// Dynamic import for @fusion/engine to avoid resolution issues in test environment
// eslint-disable-next-line @typescript-eslint/consistent-type-imports, @typescript-eslint/no-explicit-any
@@ -177,6 +178,44 @@ interface RateLimitEntry {
const sessions = new Map<string, MissionInterviewSession>();
const rateLimits = new Map<string, RateLimitEntry>();
// ── AI Session Persistence ────────────────────────────────────────────────
let _aiSessionStore: AiSessionStore | undefined;
export function setAiSessionStore(store: AiSessionStore): void {
_aiSessionStore = store;
}
function persistMissionSession(session: MissionInterviewSession, status: "generating" | "awaiting_input" | "complete" | "error", error?: string): void {
if (!_aiSessionStore) return;
const row: AiSessionRow = {
id: session.id,
type: "mission_interview",
status,
title: session.missionTitle.slice(0, 120),
inputPayload: JSON.stringify({ missionTitle: session.missionTitle }),
conversationHistory: JSON.stringify(session.history),
currentQuestion: session.currentQuestion ? JSON.stringify(session.currentQuestion) : null,
result: session.summary ? JSON.stringify(session.summary) : null,
thinkingOutput: session.thinkingOutput,
error: error ?? null,
projectId: null,
createdAt: session.createdAt.toISOString(),
updatedAt: new Date().toISOString(),
};
_aiSessionStore.upsert(row);
}
function persistMissionThinking(sessionId: string, thinkingOutput: string): void {
if (!_aiSessionStore) return;
_aiSessionStore.updateThinking(sessionId, thinkingOutput);
}
function unpersistMissionSession(sessionId: string): void {
if (!_aiSessionStore) return;
_aiSessionStore.delete(sessionId);
}
// ── Cleanup Interval ────────────────────────────────────────────────────────
function cleanupExpiredSessions(): void {
@@ -474,6 +513,7 @@ async function initializeAgent(session: MissionInterviewSession, rootDir: string
tools: "readonly",
onThinking: (delta: string) => {
session.thinkingOutput += delta;
persistMissionThinking(session.id, session.thinkingOutput);
missionInterviewStreamManager.broadcast(session.id, {
type: "thinking",
data: delta,
@@ -597,6 +637,7 @@ async function continueAgentConversation(session: MissionInterviewSession, messa
if (parsed.type === "question") {
session.currentQuestion = parsed.data;
session.updatedAt = new Date();
persistMissionSession(session, "awaiting_input");
missionInterviewStreamManager.broadcast(session.id, {
type: "question",
data: parsed.data,
@@ -605,6 +646,7 @@ async function continueAgentConversation(session: MissionInterviewSession, messa
session.summary = parsed.data;
session.currentQuestion = undefined;
session.updatedAt = new Date();
persistMissionSession(session, "complete");
missionInterviewStreamManager.broadcast(session.id, {
type: "summary",
data: parsed.data,
@@ -613,6 +655,7 @@ async function continueAgentConversation(session: MissionInterviewSession, messa
}
} catch (err) {
console.error(`[mission-interview] Agent conversation error for session ${session.id}:`, err);
persistMissionSession(session, "error", err instanceof Error ? err.message : "AI processing failed");
missionInterviewStreamManager.broadcast(session.id, {
type: "error",
data: err instanceof Error ? err.message : "AI processing failed",
@@ -653,10 +696,12 @@ export async function createMissionInterviewSession(
};
sessions.set(sessionId, session);
persistMissionSession(session, "generating");
// Initialize AI agent in background
initializeAgent(session, rootDir).catch((err) => {
console.error(`[mission-interview] Failed to initialize agent for session ${sessionId}:`, err);
persistMissionSession(session, "error", err.message || "Failed to initialize AI agent");
missionInterviewStreamManager.broadcast(sessionId, {
type: "error",
data: err.message || "Failed to initialize AI agent",
@@ -688,6 +733,7 @@ export async function submitMissionInterviewResponse(
question: session.currentQuestion,
response: responses,
});
persistMissionSession(session, "generating");
// If AI agent is active, use it for next question
if (session.agent) {
@@ -729,6 +775,7 @@ export async function cancelMissionInterviewSession(sessionId: string): Promise<
missionInterviewStreamManager.cleanupSession(sessionId);
sessions.delete(sessionId);
unpersistMissionSession(sessionId);
}
export function getMissionInterviewSession(sessionId: string): MissionInterviewSession | undefined {
@@ -746,6 +793,7 @@ export function cleanupMissionInterviewSession(sessionId: string): void {
}
missionInterviewStreamManager.cleanupSession(sessionId);
sessions.delete(sessionId);
unpersistMissionSession(sessionId);
}
/**

View File

@@ -21,6 +21,7 @@ import type {
} from "@fusion/core";
import { randomUUID } from "node:crypto";
import { EventEmitter } from "node:events";
import type { AiSessionStore, AiSessionRow } from "./ai-session-store.js";
// Dynamic import for @fusion/engine to avoid resolution issues in test environment
// eslint-disable-next-line @typescript-eslint/consistent-type-imports, @typescript-eslint/no-explicit-any
@@ -144,6 +145,49 @@ const sessions = new Map<string, Session>();
/** Rate limiting state indexed by IP */
const rateLimits = new Map<string, RateLimitEntry>();
// ── AI Session Persistence ────────────────────────────────────────────────
/** Optional store for persisting session state across reloads/browsers. */
let _aiSessionStore: AiSessionStore | undefined;
/** Wire up the AI session persistence store. Called once from server.ts. */
export function setAiSessionStore(store: AiSessionStore): void {
_aiSessionStore = store;
}
/** Persist the current session state to SQLite (no-op if store not wired). */
function persistSession(session: Session, status: "generating" | "awaiting_input" | "complete" | "error", projectId?: string, error?: string): void {
if (!_aiSessionStore) return;
const row: AiSessionRow = {
id: session.id,
type: "planning",
status,
title: session.initialPlan.slice(0, 120),
inputPayload: JSON.stringify({ initialPlan: session.initialPlan }),
conversationHistory: JSON.stringify(session.history),
currentQuestion: session.currentQuestion ? JSON.stringify(session.currentQuestion) : null,
result: session.summary ? JSON.stringify(session.summary) : null,
thinkingOutput: session.thinkingOutput,
error: error ?? null,
projectId: projectId ?? null,
createdAt: session.createdAt.toISOString(),
updatedAt: new Date().toISOString(),
};
_aiSessionStore.upsert(row);
}
/** Persist only thinking output (debounced). */
function persistThinking(sessionId: string, thinkingOutput: string): void {
if (!_aiSessionStore) return;
_aiSessionStore.updateThinking(sessionId, thinkingOutput);
}
/** Remove session from persistence. */
function unpersistSession(sessionId: string): void {
if (!_aiSessionStore) return;
_aiSessionStore.delete(sessionId);
}
// ── Cleanup Interval ────────────────────────────────────────────────────────
/**
@@ -598,10 +642,12 @@ export async function createSessionWithAgent(
};
sessions.set(sessionId, session);
persistSession(session, "generating");
// Initialize AI agent in background - it will stream via planningStreamManager
initializeAgent(session, rootDir).catch((err) => {
console.error(`[planning] Failed to initialize agent for session ${sessionId}:`, err);
persistSession(session, "error", undefined, err.message || "Failed to initialize AI agent");
planningStreamManager.broadcast(sessionId, {
type: "error",
data: err.message || "Failed to initialize AI agent",
@@ -625,6 +671,7 @@ async function initializeAgent(session: Session, rootDir: string): Promise<void>
tools: "readonly",
onThinking: (delta: string) => {
session.thinkingOutput += delta;
persistThinking(session.id, session.thinkingOutput);
planningStreamManager.broadcast(session.id, {
type: "thinking",
data: delta,
@@ -765,6 +812,7 @@ async function continueAgentConversation(session: Session, message: string): Pro
if (parsed.type === "question") {
session.currentQuestion = parsed.data;
session.updatedAt = new Date();
persistSession(session, "awaiting_input");
planningStreamManager.broadcast(session.id, {
type: "question",
data: parsed.data,
@@ -773,6 +821,7 @@ async function continueAgentConversation(session: Session, message: string): Pro
session.summary = parsed.data;
session.currentQuestion = undefined;
session.updatedAt = new Date();
persistSession(session, "complete");
planningStreamManager.broadcast(session.id, {
type: "summary",
data: parsed.data,
@@ -781,6 +830,7 @@ async function continueAgentConversation(session: Session, message: string): Pro
}
} catch (err) {
console.error(`[planning] Agent conversation error for session ${session.id}:`, err);
persistSession(session, "error", undefined, err instanceof Error ? err.message : "AI processing failed");
planningStreamManager.broadcast(session.id, {
type: "error",
data: err instanceof Error ? err.message : "AI processing failed",
@@ -997,6 +1047,7 @@ export async function submitResponse(
question: session.currentQuestion,
response: responses,
});
persistSession(session, "generating");
// If AI agent is active, use it for next question
if (session.agent) {
@@ -1089,6 +1140,7 @@ export async function cancelSession(sessionId: string): Promise<void> {
planningStreamManager.cleanupSession(sessionId);
sessions.delete(sessionId);
unpersistSession(sessionId);
}
/**
@@ -1126,6 +1178,7 @@ export function cleanupSession(sessionId: string): void {
}
planningStreamManager.cleanupSession(sessionId);
sessions.delete(sessionId);
unpersistSession(sessionId);
}
/**

View File

@@ -25,6 +25,10 @@ import {
} from "./github-webhooks.js";
import { createMissionRouter } from "./mission-routes.js";
import { getOrCreateProjectStore } from "./project-store-resolver.js";
import { AiSessionStore } from "./ai-session-store.js";
import { getSession as getPlanningSession, cleanupSession as cleanupPlanningSession } from "./planning.js";
import { getSubtaskSession, cleanupSubtaskSession } from "./subtask-breakdown.js";
import { getMissionInterviewSession, cleanupMissionInterviewSession } from "./mission-interview.js";
/**
* Minimal interface matching pi-coding-agent's ModelRegistry API surface
@@ -6542,6 +6546,80 @@ Output ONLY the prompt text (no markdown, no explanations).`;
// Mount mission routes at /api/missions
router.use("/missions", createMissionRouter(store));
// ── AI Session Routes (Background Tasks) ─────────────────────────────────
const aiSessionStore = options?.aiSessionStore;
/**
* GET /api/ai-sessions
* List active background AI sessions (generating or awaiting_input).
* Query: { projectId?: string }
*/
router.get("/ai-sessions", (req, res) => {
if (!aiSessionStore) {
res.json({ sessions: [] });
return;
}
const projectId = typeof req.query.projectId === "string" ? req.query.projectId : undefined;
const sessions = aiSessionStore.listActive(projectId);
res.json({ sessions });
});
/**
* GET /api/ai-sessions/:id
* Get full session state for modal reconnection.
*/
router.get("/ai-sessions/:id", (req, res) => {
if (!aiSessionStore) {
res.status(404).json({ error: "AI sessions not available" });
return;
}
const session = aiSessionStore.get(req.params.id);
if (!session) {
res.status(404).json({ error: "Session not found" });
return;
}
res.json(session);
});
/**
* DELETE /api/ai-sessions/:id
* Dismiss/cancel a background AI session.
* Also cleans up the in-memory agent if still alive.
*/
router.delete("/ai-sessions/:id", (req, res) => {
if (!aiSessionStore) {
res.status(404).json({ error: "AI sessions not available" });
return;
}
const { id } = req.params;
const session = aiSessionStore.get(id);
if (!session) {
res.status(404).json({ error: "Session not found" });
return;
}
// Clean up the in-memory agent based on session type
try {
switch (session.type) {
case "planning":
if (getPlanningSession(id)) cleanupPlanningSession(id);
break;
case "subtask":
if (getSubtaskSession(id)) cleanupSubtaskSession(id);
break;
case "mission_interview":
if (getMissionInterviewSession(id)) cleanupMissionInterviewSession(id);
break;
}
} catch {
// Agent may already be cleaned up — that's fine
}
aiSessionStore.delete(id);
res.json({ ok: true });
});
// ── Directory Browsing ────────────────────────────────────────────────────────
/**

View File

@@ -16,6 +16,10 @@ import { getCurrentGitHubRepo, parseBadgeUrl } from "./github.js";
import { WebSocketManager, type BadgeSnapshot } from "./websocket.js";
import type { BadgePubSub } from "./badge-pubsub.js";
import { createBadgePubSub, type BadgePubSubMessage } from "./badge-pubsub.js";
import { AiSessionStore } from "./ai-session-store.js";
import { setAiSessionStore as setPlanningAiSessionStore } from "./planning.js";
import { setAiSessionStore as setSubtaskAiSessionStore } from "./subtask-breakdown.js";
import { setAiSessionStore as setMissionAiSessionStore } from "./mission-interview.js";
const __dirname = dirname(fileURLToPath(import.meta.url));
@@ -34,6 +38,8 @@ export interface ServerOptions {
badgePubSub?: BadgePubSub;
/** Optional AutomationStore for scheduled task management */
automationStore?: AutomationStore;
/** Optional AiSessionStore — if not provided, one is created from the default store's database */
aiSessionStore?: AiSessionStore;
}
type DashboardExpressApp = ReturnType<typeof express> & {
@@ -113,7 +119,7 @@ export function createServer(store: TaskStore, options?: ServerOptions): ReturnT
app.get("/api/events", rateLimit(RATE_LIMITS.sse), async (req, res) => {
const projectId = typeof req.query.projectId === "string" ? req.query.projectId : undefined;
if (!projectId) {
createSSE(store, store.getMissionStore())(req, res);
createSSE(store, store.getMissionStore(), aiSessionStore)(req, res);
return;
}
@@ -121,7 +127,7 @@ export function createServer(store: TaskStore, options?: ServerOptions): ReturnT
// Use the shared project-store resolver so SSE listeners attach to
// the same EventEmitter used by project-scoped task API routes.
const scopedStore = await getOrCreateProjectStore(projectId);
createSSE(scopedStore, scopedStore.getMissionStore())(req, res);
createSSE(scopedStore, scopedStore.getMissionStore(), aiSessionStore)(req, res);
} catch (err: any) {
res.status(500).json({ error: err.message ?? "Failed to open project event stream" });
}
@@ -274,8 +280,15 @@ export function createServer(store: TaskStore, options?: ServerOptions): ReturnT
});
}
// Create AiSessionStore for background task persistence
const aiSessionStore = options?.aiSessionStore ?? new AiSessionStore(store.getDatabase());
aiSessionStore.recoverStaleSessions();
setPlanningAiSessionStore(aiSessionStore);
setSubtaskAiSessionStore(aiSessionStore);
setMissionAiSessionStore(aiSessionStore);
// REST API
app.use("/api", createApiRoutes(store, options));
app.use("/api", createApiRoutes(store, { ...options, aiSessionStore }));
// API 404 Handler - Return JSON for unmatched API routes (instead of falling through to SPA)
app.use("/api", (_req: express.Request, res: express.Response) => {

View File

@@ -1,5 +1,6 @@
import type { Request, Response } from "express";
import type { TaskStore, MissionStore } from "@fusion/core";
import type { AiSessionStore } from "./ai-session-store.js";
let activeConnections = 0;
@@ -23,7 +24,7 @@ function safeWrite(res: Response, data: string): boolean {
}
}
export function createSSE(store: TaskStore, missionStore?: MissionStore) {
export function createSSE(store: TaskStore, missionStore?: MissionStore, aiSessionStore?: AiSessionStore) {
return (_req: Request, res: Response) => {
res.setHeader("Content-Type", "text/event-stream");
res.setHeader("Cache-Control", "no-cache");
@@ -36,41 +37,13 @@ export function createSSE(store: TaskStore, missionStore?: MissionStore) {
// Send initial heartbeat
res.write(": connected\n\n");
/** Detach all listeners and clean up. Idempotent. */
let cleaned = false;
const cleanup = () => {
if (cleaned) return;
cleaned = true;
activeConnections--;
clearInterval(heartbeat);
store.off("task:created", onCreated);
store.off("task:moved", onMoved);
store.off("task:updated", onUpdated);
store.off("task:deleted", onDeleted);
store.off("task:merged", onMerged);
if (missionStore) {
missionStore.off("mission:created", onMissionCreated);
missionStore.off("mission:updated", onMissionUpdated);
missionStore.off("mission:deleted", onMissionDeleted);
missionStore.off("milestone:created", onMilestoneCreated);
missionStore.off("milestone:updated", onMilestoneUpdated);
missionStore.off("milestone:deleted", onMilestoneDeleted);
missionStore.off("slice:created", onSliceCreated);
missionStore.off("slice:updated", onSliceUpdated);
missionStore.off("slice:deleted", onSliceDeleted);
missionStore.off("slice:activated", onSliceActivated);
missionStore.off("feature:created", onFeatureCreated);
missionStore.off("feature:updated", onFeatureUpdated);
missionStore.off("feature:deleted", onFeatureDeleted);
missionStore.off("feature:linked", onFeatureLinked);
}
};
/** Write an SSE message; clean up on failure. */
const send = (data: string) => {
if (!safeWrite(res, data)) cleanup();
};
// --- Event handler definitions ---
const onCreated = (task: any) => {
send(`event: task:created\ndata: ${JSON.stringify(task)}\n\n`);
};
@@ -87,13 +60,6 @@ export function createSSE(store: TaskStore, missionStore?: MissionStore) {
send(`event: task:merged\ndata: ${JSON.stringify(result)}\n\n`);
};
store.on("task:created", onCreated);
store.on("task:moved", onMoved);
store.on("task:updated", onUpdated);
store.on("task:deleted", onDeleted);
store.on("task:merged", onMerged);
// Mission store event listeners (only wired up when missionStore is provided)
const onMissionCreated = (data: any) => {
send(`event: mission:created\ndata: ${JSON.stringify(data)}\n\n`);
};
@@ -137,6 +103,56 @@ export function createSSE(store: TaskStore, missionStore?: MissionStore) {
send(`event: feature:linked\ndata: ${JSON.stringify(data)}\n\n`);
};
const onAiSessionUpdated = (data: any) => {
send(`event: ai_session:updated\ndata: ${JSON.stringify(data)}\n\n`);
};
const onAiSessionDeleted = (data: any) => {
send(`event: ai_session:deleted\ndata: ${JSON.stringify(data)}\n\n`);
};
// --- Cleanup (all handlers are defined above, safe to reference) ---
let cleaned = false;
const cleanup = () => {
if (cleaned) return;
cleaned = true;
activeConnections--;
clearInterval(heartbeat);
store.off("task:created", onCreated);
store.off("task:moved", onMoved);
store.off("task:updated", onUpdated);
store.off("task:deleted", onDeleted);
store.off("task:merged", onMerged);
if (missionStore) {
missionStore.off("mission:created", onMissionCreated);
missionStore.off("mission:updated", onMissionUpdated);
missionStore.off("mission:deleted", onMissionDeleted);
missionStore.off("milestone:created", onMilestoneCreated);
missionStore.off("milestone:updated", onMilestoneUpdated);
missionStore.off("milestone:deleted", onMilestoneDeleted);
missionStore.off("slice:created", onSliceCreated);
missionStore.off("slice:updated", onSliceUpdated);
missionStore.off("slice:deleted", onSliceDeleted);
missionStore.off("slice:activated", onSliceActivated);
missionStore.off("feature:created", onFeatureCreated);
missionStore.off("feature:updated", onFeatureUpdated);
missionStore.off("feature:deleted", onFeatureDeleted);
missionStore.off("feature:linked", onFeatureLinked);
}
if (aiSessionStore) {
aiSessionStore.off("ai_session:updated", onAiSessionUpdated);
aiSessionStore.off("ai_session:deleted", onAiSessionDeleted);
}
};
// --- Subscribe ---
store.on("task:created", onCreated);
store.on("task:moved", onMoved);
store.on("task:updated", onUpdated);
store.on("task:deleted", onDeleted);
store.on("task:merged", onMerged);
if (missionStore) {
missionStore.on("mission:created", onMissionCreated);
missionStore.on("mission:updated", onMissionUpdated);
@@ -154,6 +170,11 @@ export function createSSE(store: TaskStore, missionStore?: MissionStore) {
missionStore.on("feature:linked", onFeatureLinked);
}
if (aiSessionStore) {
aiSessionStore.on("ai_session:updated", onAiSessionUpdated);
aiSessionStore.on("ai_session:deleted", onAiSessionDeleted);
}
// Heartbeat every 30s to keep connection alive.
// Sent as a named event so the client's EventSource can detect it
// (SSE comments starting with ":" are silently consumed and never

View File

@@ -1,6 +1,7 @@
import type { TaskStore } from "@fusion/core";
import { randomUUID } from "node:crypto";
import { EventEmitter } from "node:events";
import type { AiSessionStore, AiSessionRow } from "./ai-session-store.js";
// eslint-disable-next-line @typescript-eslint/no-explicit-any
let createKbAgent: any;
@@ -49,6 +50,46 @@ const CLEANUP_INTERVAL_MS = 5 * 60 * 1000;
const sessions = new Map<string, SubtaskSession & { updatedAt: Date; agent?: any; thinkingOutput: string }>();
// ── AI Session Persistence ────────────────────────────────────────────────
let _aiSessionStore: AiSessionStore | undefined;
export function setAiSessionStore(store: AiSessionStore): void {
_aiSessionStore = store;
}
type SubtaskInternalSession = SubtaskSession & { updatedAt: Date; agent?: any; thinkingOutput: string };
function persistSubtaskSession(session: SubtaskInternalSession, status: "generating" | "complete" | "error", error?: string): void {
if (!_aiSessionStore) return;
const row: AiSessionRow = {
id: session.sessionId,
type: "subtask",
status,
title: session.initialDescription.slice(0, 120),
inputPayload: JSON.stringify({ initialDescription: session.initialDescription }),
conversationHistory: "[]",
currentQuestion: null,
result: session.subtasks.length > 0 ? JSON.stringify(session.subtasks) : null,
thinkingOutput: session.thinkingOutput,
error: error ?? session.error ?? null,
projectId: null,
createdAt: session.createdAt.toISOString(),
updatedAt: new Date().toISOString(),
};
_aiSessionStore.upsert(row);
}
function persistSubtaskThinking(sessionId: string, thinkingOutput: string): void {
if (!_aiSessionStore) return;
_aiSessionStore.updateThinking(sessionId, thinkingOutput);
}
function unpersistSubtaskSession(sessionId: string): void {
if (!_aiSessionStore) return;
_aiSessionStore.delete(sessionId);
}
export const SUBTASK_BREAKDOWN_PROMPT = `You are a task decomposition assistant for the kb task board system.
Analyze the user's task description and break it down into 2-5 smaller, independently executable subtasks.
@@ -147,6 +188,7 @@ export async function createSubtaskSession(initialDescription: string, _store?:
thinkingOutput: "",
};
sessions.set(sessionId, session);
persistSubtaskSession(session, "generating");
const cwd = rootDir ?? process.cwd();
generateSubtasks(sessionId, cwd).catch((err) => {
@@ -155,6 +197,7 @@ export async function createSubtaskSession(initialDescription: string, _store?:
existing.status = "error";
existing.error = err instanceof Error ? err.message : "Failed to generate subtasks";
existing.updatedAt = new Date();
persistSubtaskSession(existing, "error", existing.error);
subtaskStreamManager.broadcast(sessionId, { type: "error", data: existing.error });
});
@@ -183,6 +226,7 @@ async function generateSubtasks(sessionId: string, cwd: string): Promise<void> {
if (!current) return;
current.thinkingOutput += delta;
current.updatedAt = new Date();
persistSubtaskThinking(sessionId, current.thinkingOutput);
subtaskStreamManager.broadcast(sessionId, { type: "thinking", data: delta });
},
onText: (delta: string) => {
@@ -269,6 +313,7 @@ function completeSession(sessionId: string, subtasks: SubtaskItem[]): void {
session.status = "complete";
session.error = undefined;
session.updatedAt = new Date();
persistSubtaskSession(session, "complete");
subtaskStreamManager.broadcast(sessionId, { type: "subtasks", data: session.subtasks });
subtaskStreamManager.broadcast(sessionId, { type: "complete" });
}
@@ -298,6 +343,7 @@ export async function cancelSubtaskSession(sessionId: string): Promise<void> {
}
subtaskStreamManager.cleanupSession(sessionId);
sessions.delete(sessionId);
unpersistSubtaskSession(sessionId);
}
export function cleanupSubtaskSession(sessionId: string): void {
@@ -309,6 +355,7 @@ export function cleanupSubtaskSession(sessionId: string): void {
}
subtaskStreamManager.cleanupSession(sessionId);
sessions.delete(sessionId);
unpersistSubtaskSession(sessionId);
}
export function __resetSubtaskBreakdownState(): void {