From 57b5d538f07b927a12cc195920814c3fc73703e0 Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Wed, 8 Apr 2026 09:15:25 -0700 Subject: [PATCH] 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 | 2 +- packages/dashboard/app/api.test.ts | 91 +++++++++++++++++++ packages/dashboard/app/api.ts | 90 ++++++++++++++++-- .../dashboard/src/ai-session-store.test.ts | 24 +++++ packages/dashboard/src/ai-session-store.ts | 14 +++ packages/dashboard/src/routes.test.ts | 40 ++++++++ packages/dashboard/src/routes.ts | 20 ++++ 7 files changed, 274 insertions(+), 7 deletions(-) diff --git a/packages/core/src/mission-store.ts b/packages/core/src/mission-store.ts index 909f5f496..6b5ac4091 100644 --- a/packages/core/src/mission-store.ts +++ b/packages/core/src/mission-store.ts @@ -479,7 +479,7 @@ export class MissionStore extends EventEmitter { 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; diff --git a/packages/dashboard/app/api.test.ts b/packages/dashboard/app/api.test.ts index 777f519ac..3d50f5c22 100644 --- a/packages/dashboard/app/api.test.ts +++ b/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).mock.calls.length; + connection.close(); + + vi.advanceTimersByTime(50_000); + + expect((globalThis.fetch as ReturnType).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).mock.calls.length; + stream.emitEvent("complete", ""); + + vi.advanceTimersByTime(50_000); + + expect((globalThis.fetch as ReturnType).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).mock.calls.length; + stream.emitEvent("complete", ""); + + vi.advanceTimersByTime(50_000); + + expect((globalThis.fetch as ReturnType).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); + }); }); diff --git a/packages/dashboard/app/api.ts b/packages/dashboard/app/api.ts index 3689b3fb9..88f3b4aeb 100644 --- a/packages/dashboard/app/api.ts +++ b/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 { 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 */ diff --git a/packages/dashboard/src/ai-session-store.test.ts b/packages/dashboard/src/ai-session-store.test.ts index dcba2ff5a..625b7ccaa 100644 --- a/packages/dashboard/src/ai-session-store.test.ts +++ b/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 }); diff --git a/packages/dashboard/src/ai-session-store.ts b/packages/dashboard/src/ai-session-store.ts index 04f5c7b06..4b3464db2 100644 --- a/packages/dashboard/src/ai-session-store.ts +++ b/packages/dashboard/src/ai-session-store.ts @@ -145,6 +145,20 @@ export class AiSessionStore extends EventEmitter { 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. diff --git a/packages/dashboard/src/routes.test.ts b/packages/dashboard/src/routes.test.ts index 5e4466890..0d3b2541e 100644 --- a/packages/dashboard/src/routes.test.ts +++ b/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; diff --git a/packages/dashboard/src/routes.ts b/packages/dashboard/src/routes.ts index c0ece995c..71ad157ab 100644 --- a/packages/dashboard/src/routes.ts +++ b/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.