feat(FN-1148): add AI session keep-alive pings for SSE streams

- Add AiSessionStore.ping() and expose POST /api/ai-sessions/:id/ping for lightweight heartbeat updates without emitting high-frequency session events
- Add pingSession() client API and integrate 25s keep-alive timers into planning, subtask, and mission interview resilient SSE connections
- Ensure keep-alive timers stop on stream completion, fatal errors, and explicit close while treating ping failures as best-effort non-fatal behavior
- Expand API/store/routes coverage with targeted tests for ping endpoint behavior, ping store semantics, and SSE keep-alive lifecycle handling
- Stabilize MissionStore latest-error lookup ordering with a rowid tiebreaker when timestamps are equal

packages/core/src/mission-store.ts

@@ -479,7 +479,7 @@ export class MissionStore extends EventEmitter<MissionStoreEvents> {
       SELECT timestamp, description
       FROM mission_events
       WHERE missionId = ? AND eventType = 'error'
-      ORDER BY timestamp DESC
+      ORDER BY timestamp DESC, rowid DESC
       LIMIT 1
     `).get(missionId) as { timestamp: string; description: string } | undefined;
 

packages/dashboard/app/api.test.ts

@@ -3,6 +3,8 @@ import {
   fetchTaskDetail,
   updateTask,
   connectPlanningStream,
+  connectSubtaskStream,
+  connectMissionInterviewStream,
   assignTask,
   fetchAgentTasks,
   archiveTask,
@@ -2722,6 +2724,7 @@ describe("Mission mutation coverage with 204 responses", () => {
 
 describe("resilient SSE reconnect", () => {
   const OriginalEventSource = globalThis.EventSource;
+  const originalFetch = globalThis.fetch;
 
   class ControlledEventSource {
     static instances: ControlledEventSource[] = [];
@@ -2776,11 +2779,13 @@ describe("resilient SSE reconnect", () => {
     vi.useFakeTimers();
     ControlledEventSource.instances = [];
     (globalThis as any).EventSource = ControlledEventSource;
+    globalThis.fetch = vi.fn().mockReturnValue(mockFetchResponse(true, { ok: true }));
   });
 
   afterEach(() => {
     vi.useRealTimers();
     (globalThis as any).EventSource = OriginalEventSource;
+    globalThis.fetch = originalFetch;
   });
 
   it("reconnects with backoff and deduplicates replayed events", () => {
@@ -2849,4 +2854,90 @@ describe("resilient SSE reconnect", () => {
 
     expect(ControlledEventSource.instances).toHaveLength(1);
   });
+
+  it("starts planning keep-alive on open and stops on explicit close", () => {
+    const connection = connectPlanningStream("session-keepalive", undefined, {});
+    const stream = ControlledEventSource.instances[0]!;
+
+    stream.emitOpen();
+    vi.advanceTimersByTime(25_000);
+
+    expect(globalThis.fetch).toHaveBeenCalledWith(
+      "/api/ai-sessions/session-keepalive/ping",
+      expect.objectContaining({ method: "POST" }),
+    );
+
+    const pingCallsBeforeClose = (globalThis.fetch as ReturnType<typeof vi.fn>).mock.calls.length;
+    connection.close();
+
+    vi.advanceTimersByTime(50_000);
+
+    expect((globalThis.fetch as ReturnType<typeof vi.fn>).mock.calls.length).toBe(pingCallsBeforeClose);
+    expect(stream.readyState).toBe(ControlledEventSource.CLOSED);
+  });
+
+  it("stops subtask keep-alive after complete event", () => {
+    connectSubtaskStream("subtask-session", undefined, {});
+    const stream = ControlledEventSource.instances[0]!;
+
+    stream.emitOpen();
+    vi.advanceTimersByTime(25_000);
+
+    expect(globalThis.fetch).toHaveBeenCalledWith(
+      "/api/ai-sessions/subtask-session/ping",
+      expect.objectContaining({ method: "POST" }),
+    );
+
+    const pingCallsBeforeComplete = (globalThis.fetch as ReturnType<typeof vi.fn>).mock.calls.length;
+    stream.emitEvent("complete", "");
+
+    vi.advanceTimersByTime(50_000);
+
+    expect((globalThis.fetch as ReturnType<typeof vi.fn>).mock.calls.length).toBe(pingCallsBeforeComplete);
+    expect(stream.readyState).toBe(ControlledEventSource.CLOSED);
+  });
+
+  it("stops mission interview keep-alive after complete event", () => {
+    connectMissionInterviewStream("mission-session", undefined, {});
+    const stream = ControlledEventSource.instances[0]!;
+
+    stream.emitOpen();
+    vi.advanceTimersByTime(25_000);
+
+    expect(globalThis.fetch).toHaveBeenCalledWith(
+      "/api/ai-sessions/mission-session/ping",
+      expect.objectContaining({ method: "POST" }),
+    );
+
+    const pingCallsBeforeComplete = (globalThis.fetch as ReturnType<typeof vi.fn>).mock.calls.length;
+    stream.emitEvent("complete", "");
+
+    vi.advanceTimersByTime(50_000);
+
+    expect((globalThis.fetch as ReturnType<typeof vi.fn>).mock.calls.length).toBe(pingCallsBeforeComplete);
+    expect(stream.readyState).toBe(ControlledEventSource.CLOSED);
+  });
+
+  it("silently ignores keep-alive ping failures", async () => {
+    globalThis.fetch = vi.fn().mockRejectedValue(new Error("network down"));
+    const onThinking = vi.fn();
+    const onError = vi.fn();
+
+    connectPlanningStream("session-ping-failure", undefined, {
+      onThinking,
+      onError,
+    });
+
+    const stream = ControlledEventSource.instances[0]!;
+    stream.emitOpen();
+
+    vi.advanceTimersByTime(25_000);
+    await Promise.resolve();
+
+    stream.emitEvent("thinking", JSON.stringify("still-streaming"));
+
+    expect(onThinking).toHaveBeenCalledWith("still-streaming");
+    expect(onError).not.toHaveBeenCalled();
+    expect(stream.readyState).toBe(ControlledEventSource.OPEN);
+  });
 });

packages/dashboard/app/api.ts

@@ -1430,6 +1430,24 @@ function createResilientEventSource(
   };
 }
 
+function startKeepAlive(
+  sessionId: string,
+  projectId?: string,
+  intervalMs = 25_000,
+): { stop: () => void } {
+  const timer = setInterval(() => {
+    void pingSession(sessionId, projectId).catch(() => {
+      // Best-effort keepalive: ignore failures so streams remain active.
+    });
+  }, intervalMs);
+
+  return {
+    stop: () => {
+      clearInterval(timer);
+    },
+  };
+}
+
 /** Get the SSE stream URL for a planning session */
 export function getPlanningStreamUrl(sessionId: string, projectId?: string): string {
   return buildApiUrl(withProjectId(`/planning/${encodeURIComponent(sessionId)}/stream`, projectId));
@@ -1454,11 +1472,21 @@ export function connectPlanningStream(
   options?: { maxReconnectAttempts?: number },
 ): { close: () => void; isConnected: () => boolean } {
   const url = getPlanningStreamUrl(sessionId, projectId);
+  let keepAlive: { stop: () => void } | null = null;
   let connection: { close: () => void; isConnected: () => boolean } | null = null;
 
+  const stopKeepAlive = () => {
+    keepAlive?.stop();
+    keepAlive = null;
+  };
+
   const resilient = createResilientEventSource(
     url,
     {
+      onOpen: () => {
+        stopKeepAlive();
+        keepAlive = startKeepAlive(sessionId, projectId);
+      },
       onMessage: (event) => {
         if (event.data.startsWith(":")) return;
       },
@@ -1503,13 +1531,21 @@ export function connectPlanningStream(
       maxReconnectAttempts: options?.maxReconnectAttempts,
       onConnectionStateChange: handlers.onConnectionStateChange,
       onFatalError: (message) => {
+        stopKeepAlive();
         handlers.onError?.(message);
       },
     },
   );
 
-  connection = resilient;
-  return resilient;
+  connection = {
+    close: () => {
+      stopKeepAlive();
+      resilient.close();
+    },
+    isConnected: resilient.isConnected,
+  };
+
+  return connection;
 }
 
 // ── Automation / Scheduled Tasks ──────────────────────────────────
@@ -1782,11 +1818,21 @@ export function connectSubtaskStream(
   },
   options?: { maxReconnectAttempts?: number },
 ): { close: () => void; isConnected: () => boolean } {
+  let keepAlive: { stop: () => void } | null = null;
   let connection: { close: () => void; isConnected: () => boolean } | null = null;
 
+  const stopKeepAlive = () => {
+    keepAlive?.stop();
+    keepAlive = null;
+  };
+
   const resilient = createResilientEventSource(
     getSubtaskStreamUrl(sessionId, projectId),
     {
+      onOpen: () => {
+        stopKeepAlive();
+        keepAlive = startKeepAlive(sessionId, projectId);
+      },
       events: {
         thinking: (event) => {
           try {
@@ -1822,13 +1868,21 @@ export function connectSubtaskStream(
       maxReconnectAttempts: options?.maxReconnectAttempts,
       onConnectionStateChange: handlers.onConnectionStateChange,
       onFatalError: (message) => {
+        stopKeepAlive();
         handlers.onError?.(message);
       },
     },
   );
 
-  connection = resilient;
-  return resilient;
+  connection = {
+    close: () => {
+      stopKeepAlive();
+      resilient.close();
+    },
+    isConnected: resilient.isConnected,
+  };
+
+  return connection;
 }
 
 export function createTasksFromBreakdown(
@@ -3131,11 +3185,21 @@ export function connectMissionInterviewStream(
   options?: { maxReconnectAttempts?: number },
 ): { close: () => void; isConnected: () => boolean } {
   const url = buildApiUrl(withProjectId(`/missions/interview/${encodeURIComponent(sessionId)}/stream`, projectId));
+  let keepAlive: { stop: () => void } | null = null;
   let connection: { close: () => void; isConnected: () => boolean } | null = null;
 
+  const stopKeepAlive = () => {
+    keepAlive?.stop();
+    keepAlive = null;
+  };
+
   const resilient = createResilientEventSource(
     url,
     {
+      onOpen: () => {
+        stopKeepAlive();
+        keepAlive = startKeepAlive(sessionId, projectId);
+      },
       onMessage: (event) => {
         if (event.data.startsWith(":")) return;
       },
@@ -3180,13 +3244,21 @@ export function connectMissionInterviewStream(
       maxReconnectAttempts: options?.maxReconnectAttempts,
       onConnectionStateChange: handlers.onConnectionStateChange,
       onFatalError: (message) => {
+        stopKeepAlive();
         handlers.onError?.(message);
       },
     },
   );
 
-  connection = resilient;
-  return resilient;
+  connection = {
+    close: () => {
+      stopKeepAlive();
+      resilient.close();
+    },
+    isConnected: resilient.isConnected,
+  };
+
+  return connection;
 }
 
 // ── AI Sessions (Background Tasks) ─────────────────────────────────────────
@@ -3228,6 +3300,12 @@ export async function deleteAiSession(id: string): Promise<void> {
   await fetch(buildApiUrl(`/ai-sessions/${encodeURIComponent(id)}`), { method: "DELETE" });
 }
 
+export function pingSession(sessionId: string, projectId?: string): Promise<{ ok: boolean }> {
+  return api<{ ok: boolean }>(withProjectId(`/ai-sessions/${encodeURIComponent(sessionId)}/ping`, projectId), {
+    method: "POST",
+  });
+}
+
 // ── Messages API ──────────────────────────────────────────────────────────
 
 /** Response shape for GET /messages/inbox */

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

@@ -205,6 +205,30 @@ describe("AiSessionStore", () => {
     expect(projectA.every((session) => session.projectId === "project-a")).toBe(true);
   });
 
+  it("ping updates updatedAt for existing sessions without emitting updates", () => {
+    seedSession({ id: "S-ping", status: "awaiting_input" });
+
+    const staleTs = new Date(Date.now() - 60_000).toISOString();
+    db.prepare("UPDATE ai_sessions SET updatedAt = ? WHERE id = ?").run(staleTs, "S-ping");
+
+    const onUpdated = vi.fn();
+    store.on("ai_session:updated", onUpdated);
+
+    const updated = store.ping("S-ping");
+
+    expect(updated).toBe(true);
+    expect(store.get("S-ping")?.updatedAt).not.toBe(staleTs);
+    expect(onUpdated).not.toHaveBeenCalled();
+  });
+
+  it("ping returns false for nonexistent sessions", () => {
+    const onUpdated = vi.fn();
+    store.on("ai_session:updated", onUpdated);
+
+    expect(store.ping("missing-session")).toBe(false);
+    expect(onUpdated).not.toHaveBeenCalled();
+  });
+
   it("listRecoverable returns awaiting_input and generating sessions", () => {
     seedSession({ id: "S-generating", status: "generating", ageMs: 3_000 });
     seedSession({ id: "S-awaiting", status: "awaiting_input", ageMs: 1_000 });

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

@@ -145,6 +145,20 @@ export class AiSessionStore extends EventEmitter<AiSessionStoreEvents> {
     return row ?? null;
   }
 
+  /**
+   * Lightweight heartbeat for active sessions.
+   * Updates only `updatedAt` and intentionally does NOT emit
+   * `ai_session:updated` to avoid high-frequency SSE broadcasts.
+   */
+  ping(id: string): boolean {
+    const now = new Date().toISOString();
+    const result = this.db
+      .prepare("UPDATE ai_sessions SET updatedAt = ? WHERE id = ?")
+      .run(now, id) as { changes?: number };
+
+    return Number(result.changes ?? 0) > 0;
+  }
+
   /**
    * List active sessions (generating or awaiting_input).
    * Optionally filtered by projectId.

packages/dashboard/src/routes.test.ts

@@ -6366,6 +6366,46 @@ describe("Git Management endpoints", () => {
   });
 });
 
+describe("POST /api/ai-sessions/:id/ping", () => {
+  let store: TaskStore;
+
+  beforeEach(() => {
+    store = createMockStore();
+  });
+
+  it("returns 200 when the session exists", async () => {
+    const mockAiSessionStore = {
+      ping: vi.fn().mockReturnValue(true),
+    };
+
+    const app = express();
+    app.use(express.json());
+    app.use("/api", createApiRoutes(store, { aiSessionStore: mockAiSessionStore as any }));
+
+    const res = await REQUEST(app, "POST", "/api/ai-sessions/session-123/ping");
+
+    expect(res.status).toBe(200);
+    expect(res.body).toEqual({ ok: true });
+    expect(mockAiSessionStore.ping).toHaveBeenCalledWith("session-123");
+  });
+
+  it("returns 404 when the session does not exist", async () => {
+    const mockAiSessionStore = {
+      ping: vi.fn().mockReturnValue(false),
+    };
+
+    const app = express();
+    app.use(express.json());
+    app.use("/api", createApiRoutes(store, { aiSessionStore: mockAiSessionStore as any }));
+
+    const res = await REQUEST(app, "POST", "/api/ai-sessions/missing-session/ping");
+
+    expect(res.status).toBe(404);
+    expect(res.body).toEqual({ error: "Session not found" });
+    expect(mockAiSessionStore.ping).toHaveBeenCalledWith("missing-session");
+  });
+});
+
 describe("Terminal session routes", () => {
   let store: TaskStore;
 

packages/dashboard/src/routes.ts

@@ -8469,6 +8469,26 @@ Output ONLY the prompt text (no markdown, no explanations).`;
     res.json(session);
   });
 
+  /**
+   * POST /api/ai-sessions/:id/ping
+   * Lightweight keep-alive touch for active AI sessions.
+   */
+  router.post("/ai-sessions/:id/ping", (req, res) => {
+    if (!aiSessionStore) {
+      res.status(404).json({ error: "AI sessions not available" });
+      return;
+    }
+
+    const { id } = req.params;
+    const updated = aiSessionStore.ping(id);
+    if (!updated) {
+      res.status(404).json({ error: "Session not found" });
+      return;
+    }
+
+    res.json({ ok: true });
+  });
+
   /**
    * DELETE /api/ai-sessions/:id
    * Dismiss/cancel a background AI session.