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

feat(FN-1146): add configurable AI session cleanup lifecycle

Browse Source 
- Add aiSessionTtlMs and aiSessionCleanupIntervalMs project settings with defaults and bounds for cleanup scheduling
- Extend AiSessionStore cleanup to expire stale in-progress sessions, emit deletion events, and support start/stop scheduled cleanup loops
- Wire server startup/shutdown to load cleanup settings and manage the scheduled ai_sessions sweep lifecycle
- Align planning, subtask breakdown, and mission interview in-memory session retention to a 7-day TTL with shared deletion-driven cleanup
- Update schema/tests/docs for migration v15 ai_sessions indexing and end-to-end cleanup/TTL behavior stability
This commit is contained in:
gsxdsm
2026-04-08 06:44:09 -07:00
parent 220c269b0d
commit 00d5f36240
12 changed files with 506 additions and 144 deletions
Download Patch File Download Diff File Expand all files Collapse all files

165
packages/dashboard/src/ai-session-store.test.ts
View File

@@ -1,4 +1,4 @@
import { describe, it, expect, beforeEach, afterEach } from "vitest";
import { describe, it, expect, beforeEach, afterEach, vi } from "vitest";
import { mkdtempSync } from "node:fs";
import { rm } from "node:fs/promises";
import { tmpdir } from "node:os";
@@ -6,7 +6,7 @@ import { join } from "node:path";
import { Database } from "@fusion/core";
import { AiSessionStore, type AiSessionRow, type AiSessionStatus } from "./ai-session-store.js";
describe("AiSessionStore.listActive", () => {
describe("AiSessionStore", () => {
let tmpRoot: string;
let db: Database;
let store: AiSessionStore;
@@ -19,6 +19,8 @@ describe("AiSessionStore.listActive", () => {
});
afterEach(async () => {
store.stopScheduledCleanup();
vi.useRealTimers();
try {
db.close();
} catch {
@@ -27,7 +29,7 @@ describe("AiSessionStore.listActive", () => {
await rm(tmpRoot, { recursive: true, force: true });
});
function createSession(id: string, status: AiSessionStatus, projectId: string | null = null): AiSessionRow {
function makeRow(id: string, status: AiSessionStatus, projectId: string | null = null): AiSessionRow {
const now = new Date().toISOString();
return {
id,
@@ -46,32 +48,155 @@ describe("AiSessionStore.listActive", () => {
};
}
it("returns generating, awaiting_input, and complete sessions", () => {
store.upsert(createSession("S-1", "generating"));
store.upsert(createSession("S-2", "awaiting_input"));
store.upsert(createSession("S-3", "complete"));
store.upsert(createSession("S-4", "error"));
function seedSession(params: {
id: string;
status: AiSessionStatus;
ageMs?: number;
projectId?: string | null;
currentQuestion?: object | null;
error?: string | null;
}): void {
const { id, status, ageMs = 0, projectId = null, currentQuestion = null, error } = params;
const row = makeRow(id, status, projectId);
row.currentQuestion = currentQuestion ? JSON.stringify(currentQuestion) : null;
row.error = error ?? row.error;
store.upsert(row);
const active = store.listActive();
const statuses = active.map((session) => session.status).sort();
if (ageMs > 0) {
const staleTs = new Date(Date.now() - ageMs).toISOString();
db.prepare("UPDATE ai_sessions SET updatedAt = ? WHERE id = ?").run(staleTs, id);
}
}
expect(statuses).toEqual(["awaiting_input", "complete", "generating"]);
expect(active.map((session) => session.id)).toEqual(expect.arrayContaining(["S-1", "S-2", "S-3"]));
it("cleanupOld removes stale sessions across all statuses and emits deleted events", () => {
const deletedIds: string[] = [];
store.on("ai_session:deleted", (id) => deletedIds.push(id));
seedSession({ id: "S-complete", status: "complete", ageMs: 2 * 60 * 60 * 1000 });
seedSession({ id: "S-error", status: "error", ageMs: 2 * 60 * 60 * 1000 });
seedSession({ id: "S-generating", status: "generating", ageMs: 2 * 60 * 60 * 1000 });
seedSession({ id: "S-awaiting", status: "awaiting_input", ageMs: 2 * 60 * 60 * 1000 });
seedSession({ id: "S-fresh", status: "generating", ageMs: 5 * 60 * 1000 });
const removed = store.cleanupOld(60 * 60 * 1000);
expect(removed).toBe(4);
expect(store.get("S-complete")).toBeNull();
expect(store.get("S-error")).toBeNull();
expect(store.get("S-generating")).toBeNull();
expect(store.get("S-awaiting")).toBeNull();
expect(store.get("S-fresh")).not.toBeNull();
expect(deletedIds.sort()).toEqual(["S-awaiting", "S-complete", "S-error", "S-generating"]);
});
it("excludes sessions with error status", () => {
store.upsert(createSession("S-err", "error"));
it("cleanupOld marks stale generating/awaiting_input sessions as error before delete", () => {
db.exec(`
CREATE TABLE IF NOT EXISTS ai_session_status_audit (
id TEXT NOT NULL,
oldStatus TEXT NOT NULL,
newStatus TEXT NOT NULL,
error TEXT
);
`);
db.exec(`
CREATE TRIGGER IF NOT EXISTS trg_ai_sessions_mark_expired
AFTER UPDATE OF status ON ai_sessions
WHEN NEW.status = 'error' AND OLD.status IN ('generating', 'awaiting_input')
BEGIN
INSERT INTO ai_session_status_audit (id, oldStatus, newStatus, error)
VALUES (NEW.id, OLD.status, NEW.status, NEW.error);
END;
`);
seedSession({ id: "S-generating", status: "generating", ageMs: 2 * 60 * 60 * 1000, error: null });
seedSession({ id: "S-awaiting", status: "awaiting_input", ageMs: 2 * 60 * 60 * 1000, error: null });
store.cleanupOld(60 * 60 * 1000);
const auditRows = db
.prepare("SELECT id, oldStatus, newStatus, error FROM ai_session_status_audit ORDER BY id")
.all() as Array<{ id: string; oldStatus: string; newStatus: string; error: string }>;
expect(auditRows).toEqual([
{
id: "S-awaiting",
oldStatus: "awaiting_input",
newStatus: "error",
error: "Session expired",
},
{
id: "S-generating",
oldStatus: "generating",
newStatus: "error",
error: "Session expired",
},
]);
});
it("startScheduledCleanup and stopScheduledCleanup control cleanup interval", () => {
vi.useFakeTimers();
seedSession({ id: "S-old", status: "complete", ageMs: 2 * 60 * 1000 });
store.startScheduledCleanup(1_000, 60_000);
vi.advanceTimersByTime(1_000);
expect(store.get("S-old")).toBeNull();
seedSession({ id: "S-old-2", status: "complete", ageMs: 2 * 60 * 1000 });
store.stopScheduledCleanup();
vi.advanceTimersByTime(5_000);
expect(store.get("S-old-2")).not.toBeNull();
});
it("supports configurable TTL values", () => {
seedSession({ id: "S-older", status: "complete", ageMs: 2 * 60 * 60 * 1000 });
seedSession({ id: "S-recent", status: "complete", ageMs: 30 * 60 * 1000 });
const removedWithShortTtl = store.cleanupOld(60 * 60 * 1000);
expect(removedWithShortTtl).toBe(1);
expect(store.get("S-older")).toBeNull();
expect(store.get("S-recent")).not.toBeNull();
const removedWithLongTtl = store.cleanupOld(3 * 60 * 60 * 1000);
expect(removedWithLongTtl).toBe(0);
});
it("recoverStaleSessions keeps recoverable sessions and marks unrecoverable ones as error", () => {
seedSession({
id: "S-recoverable",
status: "generating",
currentQuestion: { id: "q-1", type: "text", question: "Continue?" },
});
seedSession({ id: "S-broken", status: "generating", currentQuestion: null });
const recovered = store.recoverStaleSessions();
expect(recovered).toBe(2);
expect(store.get("S-recoverable")?.status).toBe("awaiting_input");
expect(store.get("S-broken")?.status).toBe("error");
expect(store.get("S-broken")?.error).toBe("Session interrupted — please restart");
});
it("listActive only returns generating/awaiting_input sessions", () => {
seedSession({ id: "S-generating", status: "generating" });
seedSession({ id: "S-awaiting", status: "awaiting_input" });
seedSession({ id: "S-complete", status: "complete" });
seedSession({ id: "S-error", status: "error" });
const active = store.listActive();
expect(active).toEqual([]);
expect(active.map((session) => session.status).sort()).toEqual(["awaiting_input", "generating"]);
expect(active.map((session) => session.id).sort()).toEqual(["S-awaiting", "S-generating"]);
});
it("filters active sessions by projectId", () => {
store.upsert(createSession("S-a1", "generating", "project-a"));
store.upsert(createSession("S-a2", "complete", "project-a"));
store.upsert(createSession("S-b1", "awaiting_input", "project-b"));
store.upsert(createSession("S-none", "complete", null));
it("listActive filters by projectId", () => {
seedSession({ id: "S-a1", status: "generating", projectId: "project-a" });
seedSession({ id: "S-a2", status: "awaiting_input", projectId: "project-a" });
seedSession({ id: "S-b1", status: "awaiting_input", projectId: "project-b" });
seedSession({ id: "S-a-done", status: "complete", projectId: "project-a" });
const projectA = store.listActive("project-a");

90
packages/dashboard/src/ai-session-store.ts
View File

@@ -62,6 +62,8 @@ const THINKING_DEBOUNCE_MS = 2000;
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>>();
/** Interval used for periodic stale-session cleanup. */
private cleanupTimer: ReturnType<typeof setInterval> | undefined;
constructor(private db: Database) {
super();
@@ -144,7 +146,7 @@ export class AiSessionStore extends EventEmitter<AiSessionStoreEvents> {
}
/**
* List active sessions (generating, awaiting_input, or complete).
* List active sessions (generating or awaiting_input).
* Optionally filtered by projectId.
*/
listActive(projectId?: string): AiSessionSummary[] {
@@ -152,7 +154,7 @@ export class AiSessionStore extends EventEmitter<AiSessionStoreEvents> {
return this.db
.prepare(
`SELECT id, type, status, title, projectId, updatedAt FROM ai_sessions
WHERE status IN ('generating', 'awaiting_input', 'complete') AND projectId = ?
WHERE status IN ('generating', 'awaiting_input') AND projectId = ?
ORDER BY updatedAt DESC`,
)
.all(projectId) as unknown as AiSessionSummary[];
@@ -160,7 +162,7 @@ export class AiSessionStore extends EventEmitter<AiSessionStoreEvents> {
return this.db
.prepare(
`SELECT id, type, status, title, projectId, updatedAt FROM ai_sessions
WHERE status IN ('generating', 'awaiting_input', 'complete')
WHERE status IN ('generating', 'awaiting_input')
ORDER BY updatedAt DESC`,
)
.all() as unknown as AiSessionSummary[];
@@ -209,16 +211,88 @@ export class AiSessionStore extends EventEmitter<AiSessionStoreEvents> {
}
/**
* Clean up completed/error sessions older than the given age (ms).
* Clean up stale sessions older than the given age (ms).
*
* For stale in-progress sessions (`generating`, `awaiting_input`), status is first
* transitioned to `error` with a "Session expired" marker before deletion.
* Returns the number of deleted sessions.
*/
cleanupOld(maxAgeMs: number): number {
const cutoff = new Date(Date.now() - maxAgeMs).toISOString();
const result = this.db
const stale = this.db
.prepare(
`DELETE FROM ai_sessions WHERE status IN ('complete', 'error') AND updatedAt < ?`,
`SELECT id FROM ai_sessions
WHERE updatedAt < ?
AND status IN ('complete', 'error', 'generating', 'awaiting_input')`,
)
.run(cutoff);
return Number((result as any).changes ?? 0);
.all(cutoff) as Array<{ id: string }>;
if (stale.length === 0) {
return 0;
}
this.db.transaction(() => {
this.db
.prepare(
`UPDATE ai_sessions
SET status = 'error',
error = CASE
WHEN error IS NULL OR error = '' THEN 'Session expired'
ELSE error
END
WHERE updatedAt < ?
AND status IN ('generating', 'awaiting_input')`,
)
.run(cutoff);
this.db
.prepare(
`DELETE FROM ai_sessions
WHERE updatedAt < ?
AND status IN ('complete', 'error', 'generating', 'awaiting_input')`,
)
.run(cutoff);
});
for (const { id } of stale) {
this.clearThinkingTimer(id);
this.emit("ai_session:deleted", id);
}
return stale.length;
}
/**
* Start periodic stale-session cleanup using the provided schedule and TTL.
*/
startScheduledCleanup(cleanupIntervalMs: number, ttlMs: number): void {
this.stopScheduledCleanup();
const runCleanup = () => {
try {
const deleted = this.cleanupOld(ttlMs);
if (deleted > 0) {
console.log(`[ai-session-store] Cleaned up ${deleted} stale sessions`);
}
} catch (err) {
console.error("[ai-session-store] Scheduled cleanup failed:", err);
}
};
this.cleanupTimer = setInterval(runCleanup, cleanupIntervalMs);
this.cleanupTimer.unref?.();
}
/**
* Stop periodic stale-session cleanup if currently running.
*/
stopScheduledCleanup(): void {
if (!this.cleanupTimer) {
return;
}
clearInterval(this.cleanupTimer);
this.cleanupTimer = undefined;
}
// ── Internal ────────────────────────────────────────────────────────

65
packages/dashboard/src/mission-interview.ts
View File

@@ -43,8 +43,8 @@ const engineReady = initEngine();
// ── Constants ───────────────────────────────────────────────────────────────
/** Session TTL in milliseconds (30 minutes) */
const SESSION_TTL_MS = 30 * 60 * 1000;
/** Session TTL in milliseconds (7 days) */
const SESSION_TTL_MS = 7 * 24 * 60 * 60 * 1000;
/** Cleanup interval in milliseconds (5 minutes) */
const CLEANUP_INTERVAL_MS = 5 * 60 * 1000;
@@ -182,9 +182,34 @@ const rateLimits = new Map<string, RateLimitEntry>();
// ── AI Session Persistence ────────────────────────────────────────────────
let _aiSessionStore: AiSessionStore | undefined;
let _aiSessionDeletedListener: ((sessionId: string) => void) | undefined;
export function setAiSessionStore(store: AiSessionStore): void {
if (_aiSessionStore && _aiSessionDeletedListener) {
_aiSessionStore.off("ai_session:deleted", _aiSessionDeletedListener);
}
_aiSessionStore = store;
_aiSessionDeletedListener = (sessionId: string) => {
cleanupInMemoryMissionSession(sessionId);
};
_aiSessionStore.on("ai_session:deleted", _aiSessionDeletedListener);
}
function cleanupInMemoryMissionSession(sessionId: string): boolean {
const session = sessions.get(sessionId);
if (!session) {
return false;
}
if (session.agent) {
try { session.agent.session.dispose?.(); } catch { /* ignore */ }
session.agent = undefined;
}
missionInterviewStreamManager.cleanupSession(sessionId);
sessions.delete(sessionId);
return true;
}
function persistMissionSession(session: MissionInterviewSession, status: "generating" | "awaiting_input" | "complete" | "error", error?: string): void {
@@ -223,11 +248,7 @@ function cleanupExpiredSessions(): void {
const now = Date.now();
for (const [id, session] of sessions) {
if (now - session.updatedAt.getTime() > SESSION_TTL_MS) {
if (session.agent) {
try { session.agent.session.dispose?.(); } catch { /* ignore */ }
}
missionInterviewStreamManager.cleanupSession(id);
sessions.delete(id);
cleanupInMemoryMissionSession(id);
}
}
for (const [ip, entry] of rateLimits) {
@@ -798,18 +819,11 @@ export async function submitMissionInterviewResponse(
}
export async function cancelMissionInterviewSession(sessionId: string): Promise<void> {
const session = sessions.get(sessionId);
if (!session) {
const removed = cleanupInMemoryMissionSession(sessionId);
if (!removed) {
throw new SessionNotFoundError(`Mission interview session ${sessionId} not found or expired`);
}
if (session.agent) {
try { session.agent.session.dispose?.(); } catch { /* ignore */ }
session.agent = undefined;
}
missionInterviewStreamManager.cleanupSession(sessionId);
sessions.delete(sessionId);
unpersistMissionSession(sessionId);
}
@@ -822,12 +836,7 @@ export function getMissionInterviewSummary(sessionId: string): MissionPlanSummar
}
export function cleanupMissionInterviewSession(sessionId: string): void {
const session = sessions.get(sessionId);
if (session?.agent) {
try { session.agent.session.dispose?.(); } catch { /* ignore */ }
}
missionInterviewStreamManager.cleanupSession(sessionId);
sessions.delete(sessionId);
cleanupInMemoryMissionSession(sessionId);
unpersistMissionSession(sessionId);
}
@@ -835,14 +844,18 @@ export function cleanupMissionInterviewSession(sessionId: string): void {
* Reset all mission interview state. Used for testing only.
*/
export function __resetMissionInterviewState(): void {
for (const [, session] of sessions) {
if (session.agent) {
try { session.agent.session.dispose?.(); } catch { /* ignore */ }
}
for (const [id] of sessions) {
cleanupInMemoryMissionSession(id);
}
sessions.clear();
rateLimits.clear();
missionInterviewStreamManager.reset();
if (_aiSessionStore && _aiSessionDeletedListener) {
_aiSessionStore.off("ai_session:deleted", _aiSessionDeletedListener);
}
_aiSessionDeletedListener = undefined;
_aiSessionStore = undefined;
}
// ── Custom Errors ───────────────────────────────────────────────────────────

18
packages/dashboard/src/planning.test.ts
View File

@@ -19,6 +19,7 @@ import {
parseAgentResponse,
generateSubtasksFromPlanning,
formatInterviewQA,
SESSION_TTL_MS,
} from "./planning.js";
import type { PlanningQuestion, PlanningSummary } from "@fusion/core";
@@ -483,23 +484,20 @@ describe("planning module", () => {
});
describe("session TTL", () => {
it("sessions expire after TTL", async () => {
it("uses a 7-day TTL constant", () => {
expect(SESSION_TTL_MS).toBe(7 * 24 * 60 * 60 * 1000);
});
it("does not expire sessions within the old 30-minute window", async () => {
vi.useFakeTimers({ shouldAdvanceTime: true });
try {
const mockIp = getUniqueIp();
const { sessionId } = await createSession(mockIp, initialPlan, undefined, TEST_ROOT_DIR);
// Verify session exists
expect(getSession(sessionId)).toBeDefined();
// Advance time by 31 minutes
// Advance beyond the old 30-minute TTL used prior to FN-1146.
vi.advanceTimersByTime(31 * 60 * 1000);
// Trigger cleanup by creating a new session
await createSession(getUniqueIp(), "Another plan", undefined, TEST_ROOT_DIR);
// Note: Session should be expired after cleanup runs
// We can't directly verify as cleanup is async
expect(getSession(sessionId)).toBeDefined();
} finally {
vi.useRealTimers();
}

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

@@ -95,8 +95,8 @@ For questions:
For completion:
{\n "type": "complete",\n "data": {\n "title": "Task title",\n "description": "Detailed description",\n "suggestedSize": "S|M|L",\n "suggestedDependencies": [],\n "keyDeliverables": ["Item 1", "Item 2"]\n }\n}`;
/** Session TTL in milliseconds (30 minutes) */
const SESSION_TTL_MS = 30 * 60 * 1000;
/** Session TTL in milliseconds (7 days) */
export const SESSION_TTL_MS = 7 * 24 * 60 * 60 * 1000;
/** Cleanup interval in milliseconds (5 minutes) */
const CLEANUP_INTERVAL_MS = 5 * 60 * 1000;
@@ -154,10 +154,39 @@ const rateLimits = new Map<string, RateLimitEntry>();
/** Optional store for persisting session state across reloads/browsers. */
let _aiSessionStore: AiSessionStore | undefined;
let _aiSessionDeletedListener: ((sessionId: string) => void) | undefined;
/** Wire up the AI session persistence store. Called once from server.ts. */
export function setAiSessionStore(store: AiSessionStore): void {
if (_aiSessionStore && _aiSessionDeletedListener) {
_aiSessionStore.off("ai_session:deleted", _aiSessionDeletedListener);
}
_aiSessionStore = store;
_aiSessionDeletedListener = (sessionId: string) => {
cleanupInMemorySession(sessionId);
};
_aiSessionStore.on("ai_session:deleted", _aiSessionDeletedListener);
}
function cleanupInMemorySession(sessionId: string): boolean {
const session = sessions.get(sessionId);
if (!session) {
return false;
}
if (session.agent) {
try {
session.agent.session.dispose?.();
} catch (err) {
console.error(`[planning] Error disposing agent for session ${sessionId}:`, err);
}
session.agent = undefined;
}
planningStreamManager.cleanupSession(sessionId);
sessions.delete(sessionId);
return true;
}
/** Persist the current session state to SQLite (no-op if store not wired). */
@@ -207,8 +236,9 @@ function cleanupExpiredSessions(): void {
// Clean up expired sessions
for (const [id, session] of sessions) {
if (now - session.updatedAt.getTime() > SESSION_TTL_MS) {
sessions.delete(id);
cleanedSessions++;
if (cleanupInMemorySession(id)) {
cleanedSessions++;
}
}
}
@@ -1144,25 +1174,11 @@ export function formatInterviewQA(
* Cancel and cleanup a planning session.
*/
export async function cancelSession(sessionId: string): Promise<void> {
const session = sessions.get(sessionId);
if (!session) {
const removed = cleanupInMemorySession(sessionId);
if (!removed) {
throw new SessionNotFoundError(`Planning session ${sessionId} not found or expired`);
}
// Cleanup AI agent if present
if (session.agent) {
try {
session.agent.session.dispose?.();
} catch (err) {
console.error(`[planning] Error disposing agent for session ${sessionId}:`, err);
}
session.agent = undefined;
}
// Cleanup SSE subscriptions
planningStreamManager.cleanupSession(sessionId);
sessions.delete(sessionId);
unpersistSession(sessionId);
}
@@ -1250,16 +1266,7 @@ export function generateSubtasksFromPlanning(sessionId: string): SubtaskItem[] {
* Cleanup a session (used after task creation).
*/
export function cleanupSession(sessionId: string): void {
const session = sessions.get(sessionId);
if (session?.agent) {
try {
session.agent.session.dispose?.();
} catch {
// Ignore errors during cleanup
}
}
planningStreamManager.cleanupSession(sessionId);
sessions.delete(sessionId);
cleanupInMemorySession(sessionId);
unpersistSession(sessionId);
}
@@ -1268,18 +1275,18 @@ export function cleanupSession(sessionId: string): void {
*/
export function __resetPlanningState(): void {
// Cleanup all agent sessions
for (const [id, session] of sessions) {
if (session.agent) {
try {
session.agent.session.dispose?.();
} catch {
// Ignore errors during cleanup
}
}
for (const [id] of sessions) {
cleanupInMemorySession(id);
}
sessions.clear();
rateLimits.clear();
planningStreamManager.reset();
if (_aiSessionStore && _aiSessionDeletedListener) {
_aiSessionStore.off("ai_session:deleted", _aiSessionDeletedListener);
}
_aiSessionDeletedListener = undefined;
_aiSessionStore = undefined;
}
/**

2
packages/dashboard/src/routes.test.ts
View File

@@ -2111,7 +2111,7 @@ describe("PATCH /tasks/:id/assign and GET /agents/:id/tasks", () => {
expect(res.status).toBe(404);
expect(res.body.error).toBe("Agent not found");
expect(store.listTasks).not.toHaveBeenCalled();
});
}, 30_000);
});
describe("Attachment routes", () => {

57
packages/dashboard/src/server.ts
View File

@@ -23,6 +23,14 @@ import { setAiSessionStore as setMissionAiSessionStore } from "./mission-intervi
const __dirname = dirname(fileURLToPath(import.meta.url));
const DEFAULT_AI_SESSION_TTL_MS = 7 * 24 * 60 * 60 * 1000;
const MIN_AI_SESSION_TTL_MS = 10 * 60 * 1000;
const MAX_AI_SESSION_TTL_MS = 30 * 24 * 60 * 60 * 1000;
const DEFAULT_AI_SESSION_CLEANUP_INTERVAL_MS = 60 * 60 * 1000;
const MIN_AI_SESSION_CLEANUP_INTERVAL_MS = 60 * 1000;
const MAX_AI_SESSION_CLEANUP_INTERVAL_MS = 24 * 60 * 60 * 1000;
export interface ServerOptions {
/** Custom merge handler — when provided, used instead of store.mergeTask */
onMerge?: (taskId: string) => Promise<MergeResult>;
@@ -97,6 +105,18 @@ function normalizeListenArgsForTests(args: unknown[]): unknown[] {
return args;
}
function resolveBoundedMs(
value: number | undefined,
fallback: number,
min: number,
max: number,
): number {
if (typeof value !== "number" || !Number.isFinite(value)) {
return fallback;
}
return Math.min(max, Math.max(min, value));
}
export function createServer(store: TaskStore, options?: ServerOptions): ReturnType<typeof express> {
const app = express();
const mutationRateLimit = rateLimit(RATE_LIMITS.mutation);
@@ -302,6 +322,39 @@ export function createServer(store: TaskStore, options?: ServerOptions): ReturnT
setSubtaskAiSessionStore(aiSessionStore);
setMissionAiSessionStore(aiSessionStore);
const loadSettings = (store as { getSettings?: () => Promise<{ aiSessionTtlMs?: number; aiSessionCleanupIntervalMs?: number }> }).getSettings;
if (typeof loadSettings === "function") {
void loadSettings
.call(store)
.then((settings) => {
const ttlMs = resolveBoundedMs(
settings.aiSessionTtlMs,
DEFAULT_AI_SESSION_TTL_MS,
MIN_AI_SESSION_TTL_MS,
MAX_AI_SESSION_TTL_MS,
);
const cleanupIntervalMs = resolveBoundedMs(
settings.aiSessionCleanupIntervalMs,
DEFAULT_AI_SESSION_CLEANUP_INTERVAL_MS,
MIN_AI_SESSION_CLEANUP_INTERVAL_MS,
MAX_AI_SESSION_CLEANUP_INTERVAL_MS,
);
aiSessionStore.startScheduledCleanup(cleanupIntervalMs, ttlMs);
})
.catch((err) => {
console.warn("[server] Failed to load settings for AI session cleanup; using defaults", err);
aiSessionStore.startScheduledCleanup(
DEFAULT_AI_SESSION_CLEANUP_INTERVAL_MS,
DEFAULT_AI_SESSION_TTL_MS,
);
});
} else {
aiSessionStore.startScheduledCleanup(
DEFAULT_AI_SESSION_CLEANUP_INTERVAL_MS,
DEFAULT_AI_SESSION_TTL_MS,
);
}
// REST API
app.use("/api", createApiRoutes(store, { ...options, aiSessionStore }));
@@ -340,6 +393,10 @@ export function createServer(store: TaskStore, options?: ServerOptions): ReturnT
const normalizedArgs = normalizeListenArgsForTests(args) as Parameters<typeof originalListen>;
const server = originalListen(...normalizedArgs);
server.once("close", () => {
aiSessionStore.stopScheduledCleanup();
});
if (!dashboardApp.__kbWebSocketsAttached) {
dashboardApp.__kbWebSocketsAttached = true;
setupTerminalWebSocket(dashboardApp, server);

70
packages/dashboard/src/subtask-breakdown.ts
View File

@@ -46,7 +46,7 @@ export type SubtaskStreamEvent =
export type SubtaskStreamCallback = (event: SubtaskStreamEvent, eventId?: number) => void;
const SESSION_TTL_MS = 30 * 60 * 1000;
const SESSION_TTL_MS = 7 * 24 * 60 * 60 * 1000;
const CLEANUP_INTERVAL_MS = 5 * 60 * 1000;
const sessions = new Map<string, SubtaskSession & { updatedAt: Date; agent?: any; thinkingOutput: string }>();
@@ -54,13 +54,39 @@ const sessions = new Map<string, SubtaskSession & { updatedAt: Date; agent?: any
// ── AI Session Persistence ────────────────────────────────────────────────
let _aiSessionStore: AiSessionStore | undefined;
let _aiSessionDeletedListener: ((sessionId: string) => void) | undefined;
export function setAiSessionStore(store: AiSessionStore): void {
if (_aiSessionStore && _aiSessionDeletedListener) {
_aiSessionStore.off("ai_session:deleted", _aiSessionDeletedListener);
}
_aiSessionStore = store;
_aiSessionDeletedListener = (sessionId: string) => {
cleanupInMemorySubtaskSession(sessionId);
};
_aiSessionStore.on("ai_session:deleted", _aiSessionDeletedListener);
}
type SubtaskInternalSession = SubtaskSession & { updatedAt: Date; agent?: any; thinkingOutput: string };
function cleanupInMemorySubtaskSession(sessionId: string): boolean {
const session = sessions.get(sessionId);
if (!session) {
return false;
}
try {
session.agent?.session?.dispose?.();
} catch {
// ignore cleanup errors
}
subtaskStreamManager.cleanupSession(sessionId);
sessions.delete(sessionId);
return true;
}
function persistSubtaskSession(session: SubtaskInternalSession, status: "generating" | "complete" | "error", error?: string): void {
if (!_aiSessionStore) return;
const row: AiSessionRow = {
@@ -125,13 +151,7 @@ function cleanupExpiredSessions(): void {
const now = Date.now();
for (const [id, session] of sessions) {
if (now - session.updatedAt.getTime() > SESSION_TTL_MS) {
try {
session.agent?.session?.dispose?.();
} catch {
// ignore cleanup failures
}
sessions.delete(id);
subtaskStreamManager.cleanupSession(id);
cleanupInMemorySubtaskSession(id);
}
}
}
@@ -367,42 +387,30 @@ export function getSubtaskSession(sessionId: string): SubtaskSession | undefined
}
export async function cancelSubtaskSession(sessionId: string): Promise<void> {
const session = sessions.get(sessionId);
if (!session) {
const removed = cleanupInMemorySubtaskSession(sessionId);
if (!removed) {
throw new SessionNotFoundError(`Subtask session ${sessionId} not found or expired`);
}
try {
session.agent?.session?.dispose?.();
} catch {
// ignore dispose errors
}
subtaskStreamManager.cleanupSession(sessionId);
sessions.delete(sessionId);
unpersistSubtaskSession(sessionId);
}
export function cleanupSubtaskSession(sessionId: string): void {
const session = sessions.get(sessionId);
try {
session?.agent?.session?.dispose?.();
} catch {
// ignore cleanup errors
}
subtaskStreamManager.cleanupSession(sessionId);
sessions.delete(sessionId);
cleanupInMemorySubtaskSession(sessionId);
unpersistSubtaskSession(sessionId);
}
export function __resetSubtaskBreakdownState(): void {
for (const [, session] of sessions) {
try {
session.agent?.session?.dispose?.();
} catch {
// ignore cleanup errors
}
for (const [id] of sessions) {
cleanupInMemorySubtaskSession(id);
}
sessions.clear();
subtaskStreamManager.reset();
if (_aiSessionStore && _aiSessionDeletedListener) {
_aiSessionStore.off("ai_session:deleted", _aiSessionDeletedListener);
}
_aiSessionDeletedListener = undefined;
_aiSessionStore = undefined;
}
export class SessionNotFoundError extends Error {