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>

packages/dashboard/src/ai-session-store.ts

@@ -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,
+  };
+}

packages/dashboard/src/mission-interview.ts

@@ -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);
 }
 
 /**

packages/dashboard/src/planning.ts

@@ -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);
 }
 
 /**

packages/dashboard/src/routes.ts

@@ -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 ────────────────────────────────────────────────────────
 
   /**

packages/dashboard/src/server.ts

@@ -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) => {

packages/dashboard/src/sse.ts

@@ -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

packages/dashboard/src/subtask-breakdown.ts

@@ -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 {