test(FN-1156): add AI session lifecycle and reconnect coverage
- Add ai-session-store unit tests for upsert/get round-trips, active filtering, stale recovery, cleanup, thinking debounce, and update/delete events - Add persistence and resume-history integration tests for planning, subtask, and mission interview sessions across SQLite reload and cancellation flows - Add reconnect and cross-tab lock tests covering SSE Last-Event-ID replay, keep-alive ping touches, optimistic lock conflicts, stale lock expiry, and lock release on tab close - Expand useBackgroundSessions hook tests for SSE-driven updates/deletes, project-scoped fetch/stream behavior, counts, dismiss, and refresh state handling
This commit is contained in:
gsxdsm
@@ -1,12 +1,17 @@
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
/**
|
||||
* Covers background AI session hook behavior: fetch lifecycle, SSE updates,
|
||||
* dismissal, counters/filters, refresh, and project scoping.
|
||||
*/
|
||||
|
||||
import { act, renderHook, waitFor } from "@testing-library/react";
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { useBackgroundSessions } from "../useBackgroundSessions";
|
||||
import {
|
||||
__destroyAiSessionSyncStoreForTests,
|
||||
__resetAiSessionSyncStoreForTests,
|
||||
useAiSessionSync,
|
||||
} from "../useAiSessionSync";
|
||||
import * as apiModule from "../../api";
|
||||
import { MockEventSource } from "../../../vitest.setup";
|
||||
|
||||
vi.mock("../../api", () => ({
|
||||
fetchAiSessions: vi.fn(),
|
||||
@@ -16,51 +21,23 @@ vi.mock("../../api", () => ({
|
||||
const mockFetchAiSessions = vi.mocked(apiModule.fetchAiSessions);
|
||||
const mockDeleteAiSession = vi.mocked(apiModule.deleteAiSession);
|
||||
|
||||
class MockEventSource {
|
||||
static instances: MockEventSource[] = [];
|
||||
|
||||
readonly url: string;
|
||||
private listeners = new Map<string, Set<(event: MessageEvent) => void>>();
|
||||
|
||||
constructor(url: string) {
|
||||
this.url = url;
|
||||
MockEventSource.instances.push(this);
|
||||
}
|
||||
|
||||
addEventListener(type: string, listener: (event: MessageEvent) => void): void {
|
||||
const set = this.listeners.get(type) ?? new Set<(event: MessageEvent) => void>();
|
||||
set.add(listener);
|
||||
this.listeners.set(type, set);
|
||||
}
|
||||
|
||||
removeEventListener(type: string, listener: (event: MessageEvent) => void): void {
|
||||
this.listeners.get(type)?.delete(listener);
|
||||
}
|
||||
|
||||
close(): void {
|
||||
this.listeners.clear();
|
||||
}
|
||||
|
||||
emit(type: string, payload: unknown): void {
|
||||
const event = { data: JSON.stringify(payload) } as MessageEvent;
|
||||
for (const listener of this.listeners.get(type) ?? []) {
|
||||
listener(event);
|
||||
}
|
||||
}
|
||||
function makeSession(overrides: Partial<apiModule.AiSessionSummary> & Pick<apiModule.AiSessionSummary, "id">): apiModule.AiSessionSummary {
|
||||
return {
|
||||
id: overrides.id,
|
||||
type: overrides.type ?? "planning",
|
||||
status: overrides.status ?? "generating",
|
||||
title: overrides.title ?? overrides.id,
|
||||
projectId: overrides.projectId ?? null,
|
||||
lockedByTab: overrides.lockedByTab ?? null,
|
||||
updatedAt: overrides.updatedAt ?? "2026-04-08T00:00:00.000Z",
|
||||
};
|
||||
}
|
||||
|
||||
describe("useBackgroundSessions", () => {
|
||||
const originalEventSource = globalThis.EventSource;
|
||||
|
||||
beforeEach(() => {
|
||||
vi.clearAllMocks();
|
||||
__resetAiSessionSyncStoreForTests();
|
||||
__destroyAiSessionSyncStoreForTests();
|
||||
|
||||
MockEventSource.instances = [];
|
||||
(globalThis as unknown as { EventSource: typeof EventSource }).EventSource =
|
||||
MockEventSource as unknown as typeof EventSource;
|
||||
|
||||
mockFetchAiSessions.mockResolvedValue([]);
|
||||
mockDeleteAiSession.mockResolvedValue(undefined);
|
||||
});
|
||||
@@ -68,71 +45,176 @@ describe("useBackgroundSessions", () => {
|
||||
afterEach(() => {
|
||||
__resetAiSessionSyncStoreForTests();
|
||||
__destroyAiSessionSyncStoreForTests();
|
||||
(globalThis as unknown as { EventSource: typeof EventSource }).EventSource = originalEventSource;
|
||||
});
|
||||
|
||||
it("merges cross-tab session updates into the local list", async () => {
|
||||
const background = renderHook(() => useBackgroundSessions("proj-1"));
|
||||
const sync = renderHook(() => useAiSessionSync());
|
||||
it("fetches and filters initial sessions", async () => {
|
||||
mockFetchAiSessions.mockResolvedValueOnce([
|
||||
makeSession({ id: "s-generating", status: "generating" }),
|
||||
makeSession({ id: "s-awaiting", status: "awaiting_input" }),
|
||||
makeSession({ id: "s-complete", status: "complete" }),
|
||||
makeSession({ id: "s-error", status: "error" }),
|
||||
makeSession({ id: "s-ignored", status: "paused" as any }),
|
||||
]);
|
||||
|
||||
const { result } = renderHook(() => useBackgroundSessions());
|
||||
|
||||
await waitFor(() => {
|
||||
expect(mockFetchAiSessions).toHaveBeenCalledWith("proj-1");
|
||||
});
|
||||
|
||||
act(() => {
|
||||
sync.result.current.broadcastUpdate({
|
||||
sessionId: "sess-cross-tab",
|
||||
status: "awaiting_input",
|
||||
needsInput: true,
|
||||
type: "planning",
|
||||
title: "Cross-tab planning",
|
||||
projectId: "proj-1",
|
||||
owningTabId: "tab-other",
|
||||
timestamp: 500,
|
||||
});
|
||||
expect(mockFetchAiSessions).toHaveBeenCalledWith(undefined);
|
||||
});
|
||||
|
||||
await waitFor(() => {
|
||||
expect(background.result.current.sessions).toHaveLength(1);
|
||||
expect(background.result.current.sessions[0]).toMatchObject({
|
||||
id: "sess-cross-tab",
|
||||
status: "awaiting_input",
|
||||
type: "planning",
|
||||
});
|
||||
expect(result.current.sessions.map((session) => session.id).sort()).toEqual([
|
||||
"s-awaiting",
|
||||
"s-complete",
|
||||
"s-error",
|
||||
"s-generating",
|
||||
]);
|
||||
});
|
||||
});
|
||||
|
||||
it("broadcasts SSE updates through the sync store", async () => {
|
||||
const background = renderHook(() => useBackgroundSessions("proj-1"));
|
||||
const sync = renderHook(() => useAiSessionSync());
|
||||
it("applies SSE-driven session updates reactively", async () => {
|
||||
const { result } = renderHook(() => useBackgroundSessions());
|
||||
|
||||
await waitFor(() => {
|
||||
expect(MockEventSource.instances.length).toBeGreaterThan(0);
|
||||
});
|
||||
|
||||
const eventSource = MockEventSource.instances[0];
|
||||
const eventSource = MockEventSource.instances[0]!;
|
||||
|
||||
act(() => {
|
||||
eventSource.emit("ai_session:updated", {
|
||||
id: "sess-sse",
|
||||
type: "subtask",
|
||||
status: "generating",
|
||||
title: "SSE session",
|
||||
projectId: "proj-1",
|
||||
lockedByTab: "tab-remote",
|
||||
updatedAt: "2026-04-08T00:00:00.000Z",
|
||||
});
|
||||
eventSource._emit(
|
||||
"ai_session:updated",
|
||||
makeSession({
|
||||
id: "sse-session",
|
||||
type: "mission_interview",
|
||||
status: "generating",
|
||||
title: "Mission stream",
|
||||
updatedAt: "2026-04-08T00:00:01.000Z",
|
||||
}),
|
||||
);
|
||||
});
|
||||
|
||||
await waitFor(() => {
|
||||
expect(background.result.current.sessions[0]?.id).toBe("sess-sse");
|
||||
expect(result.current.sessions.find((session) => session.id === "sse-session")?.status).toBe(
|
||||
"generating",
|
||||
);
|
||||
});
|
||||
|
||||
act(() => {
|
||||
eventSource._emit(
|
||||
"ai_session:updated",
|
||||
makeSession({
|
||||
id: "sse-session",
|
||||
type: "mission_interview",
|
||||
status: "awaiting_input",
|
||||
title: "Mission stream",
|
||||
updatedAt: "2026-04-08T00:00:02.000Z",
|
||||
}),
|
||||
);
|
||||
});
|
||||
|
||||
await waitFor(() => {
|
||||
const synced = sync.result.current.sessions.get("sess-sse");
|
||||
expect(synced?.status).toBe("generating");
|
||||
expect(synced?.type).toBe("subtask");
|
||||
expect(synced?.owningTabId).toBe("tab-remote");
|
||||
expect(result.current.sessions.find((session) => session.id === "sse-session")?.status).toBe(
|
||||
"awaiting_input",
|
||||
);
|
||||
});
|
||||
});
|
||||
|
||||
it("removes sessions when ai_session:deleted SSE event arrives", async () => {
|
||||
mockFetchAiSessions.mockResolvedValueOnce([
|
||||
makeSession({ id: "delete-me", status: "awaiting_input" }),
|
||||
]);
|
||||
|
||||
const { result } = renderHook(() => useBackgroundSessions());
|
||||
|
||||
await waitFor(() => {
|
||||
expect(result.current.sessions).toHaveLength(1);
|
||||
});
|
||||
|
||||
const eventSource = MockEventSource.instances[0]!;
|
||||
act(() => {
|
||||
eventSource._emit("ai_session:deleted", "delete-me");
|
||||
});
|
||||
|
||||
await waitFor(() => {
|
||||
expect(result.current.sessions).toEqual([]);
|
||||
});
|
||||
});
|
||||
|
||||
it("dismissSession calls API and updates local state", async () => {
|
||||
mockFetchAiSessions.mockResolvedValueOnce([
|
||||
makeSession({ id: "dismiss-me", status: "awaiting_input" }),
|
||||
]);
|
||||
|
||||
const { result } = renderHook(() => useBackgroundSessions());
|
||||
|
||||
await waitFor(() => {
|
||||
expect(result.current.sessions.map((session) => session.id)).toEqual(["dismiss-me"]);
|
||||
});
|
||||
|
||||
act(() => {
|
||||
result.current.dismissSession("dismiss-me");
|
||||
});
|
||||
|
||||
expect(mockDeleteAiSession).toHaveBeenCalledWith("dismiss-me");
|
||||
await waitFor(() => {
|
||||
expect(result.current.sessions).toEqual([]);
|
||||
});
|
||||
});
|
||||
|
||||
it("returns accurate generating/needsInput counts and planningSessions filter", async () => {
|
||||
mockFetchAiSessions.mockResolvedValueOnce([
|
||||
makeSession({ id: "count-generating", status: "generating", type: "planning" }),
|
||||
makeSession({ id: "count-awaiting", status: "awaiting_input", type: "subtask" }),
|
||||
makeSession({ id: "count-error-plan", status: "error", type: "planning" }),
|
||||
makeSession({ id: "count-complete", status: "complete", type: "mission_interview" }),
|
||||
]);
|
||||
|
||||
const { result } = renderHook(() => useBackgroundSessions());
|
||||
|
||||
await waitFor(() => {
|
||||
expect(result.current.generating).toBe(1);
|
||||
expect(result.current.needsInput).toBe(1);
|
||||
expect(result.current.planningSessions.map((session) => session.id).sort()).toEqual([
|
||||
"count-error-plan",
|
||||
"count-generating",
|
||||
]);
|
||||
});
|
||||
});
|
||||
|
||||
it("refresh triggers a new fetch and updates state", async () => {
|
||||
mockFetchAiSessions
|
||||
.mockResolvedValueOnce([makeSession({ id: "first-session", status: "generating" })])
|
||||
.mockResolvedValueOnce([makeSession({ id: "second-session", status: "awaiting_input" })]);
|
||||
|
||||
const { result } = renderHook(() => useBackgroundSessions());
|
||||
|
||||
await waitFor(() => {
|
||||
expect(result.current.sessions.map((session) => session.id)).toEqual(["first-session"]);
|
||||
});
|
||||
|
||||
act(() => {
|
||||
result.current.refresh();
|
||||
});
|
||||
|
||||
await waitFor(() => {
|
||||
expect(mockFetchAiSessions).toHaveBeenCalledTimes(2);
|
||||
expect(result.current.sessions.map((session) => session.id)).toEqual(["second-session"]);
|
||||
});
|
||||
});
|
||||
|
||||
it("passes projectId to API fetch and uses project-scoped SSE URL", async () => {
|
||||
const projectId = "proj-123";
|
||||
renderHook(() => useBackgroundSessions(projectId));
|
||||
|
||||
await waitFor(() => {
|
||||
expect(mockFetchAiSessions).toHaveBeenCalledWith(projectId);
|
||||
});
|
||||
|
||||
await waitFor(() => {
|
||||
expect(MockEventSource.instances.length).toBeGreaterThan(0);
|
||||
});
|
||||
|
||||
expect(MockEventSource.instances[0]?.url).toContain(`/api/events?projectId=${encodeURIComponent(projectId)}`);
|
||||
});
|
||||
});
|
||||
|
||||
309
packages/dashboard/src/__tests__/ai-session-store.test.ts
Normal file
309
packages/dashboard/src/__tests__/ai-session-store.test.ts
Normal file
@@ -0,0 +1,309 @@
|
||||
/**
|
||||
* Covers AI session persistence store round-trips, lifecycle transitions,
|
||||
* cleanup/recovery behavior, and debounce/emit semantics.
|
||||
*/
|
||||
|
||||
import { describe, it, expect, beforeEach, afterEach, vi } from "vitest";
|
||||
import { Database } from "@fusion/core";
|
||||
import { mkdtempSync } from "node:fs";
|
||||
import { rm } from "node:fs/promises";
|
||||
import { join } from "node:path";
|
||||
import { tmpdir } from "node:os";
|
||||
import {
|
||||
AiSessionStore,
|
||||
type AiSessionRow,
|
||||
type AiSessionSummary,
|
||||
} from "../ai-session-store.js";
|
||||
|
||||
function makeTmpDir(): string {
|
||||
return mkdtempSync(join(tmpdir(), "kb-ai-session-store-tests-"));
|
||||
}
|
||||
|
||||
function makeRow(
|
||||
id: string,
|
||||
overrides: Partial<AiSessionRow> = {},
|
||||
): AiSessionRow {
|
||||
const now = new Date().toISOString();
|
||||
return {
|
||||
id,
|
||||
type: "planning",
|
||||
status: "generating",
|
||||
title: `Session ${id}`,
|
||||
inputPayload: JSON.stringify({ initialPlan: `Plan ${id}`, ip: "127.0.0.1" }),
|
||||
conversationHistory: JSON.stringify([]),
|
||||
currentQuestion: null,
|
||||
result: null,
|
||||
thinkingOutput: "",
|
||||
error: null,
|
||||
projectId: null,
|
||||
createdAt: now,
|
||||
updatedAt: now,
|
||||
lockedByTab: null,
|
||||
lockedAt: null,
|
||||
...overrides,
|
||||
};
|
||||
}
|
||||
|
||||
describe("AiSessionStore (__tests__)", () => {
|
||||
let tmpDir: string;
|
||||
let kbDir: string;
|
||||
let db: Database;
|
||||
let store: AiSessionStore;
|
||||
|
||||
beforeEach(() => {
|
||||
tmpDir = makeTmpDir();
|
||||
kbDir = join(tmpDir, ".fusion");
|
||||
db = new Database(kbDir);
|
||||
db.init();
|
||||
store = new AiSessionStore(db);
|
||||
});
|
||||
|
||||
afterEach(async () => {
|
||||
store.stopScheduledCleanup();
|
||||
store.removeAllListeners();
|
||||
vi.useRealTimers();
|
||||
try {
|
||||
db.close();
|
||||
} catch {
|
||||
// no-op
|
||||
}
|
||||
await rm(tmpDir, { recursive: true, force: true });
|
||||
});
|
||||
|
||||
it("round-trips full session payload via upsert/get", () => {
|
||||
const history = [
|
||||
{
|
||||
question: { id: "q-1", type: "text", question: "What should we build?" },
|
||||
response: { "q-1": "A planner" },
|
||||
thinkingOutput: "first-think",
|
||||
},
|
||||
{
|
||||
question: { id: "q-2", type: "confirm", question: "Need tests?" },
|
||||
response: { "q-2": true },
|
||||
thinkingOutput: "second-think",
|
||||
},
|
||||
];
|
||||
const currentQuestion = {
|
||||
id: "q-3",
|
||||
type: "single_select",
|
||||
question: "Target size?",
|
||||
options: [{ id: "m", label: "Medium" }],
|
||||
};
|
||||
const result = {
|
||||
title: "Planner task",
|
||||
description: "A complete planning summary",
|
||||
suggestedSize: "M",
|
||||
suggestedDependencies: ["FN-100"],
|
||||
keyDeliverables: ["API", "UI", "Tests"],
|
||||
};
|
||||
|
||||
const row = makeRow("sess-roundtrip", {
|
||||
status: "awaiting_input",
|
||||
title: "Roundtrip Session",
|
||||
inputPayload: JSON.stringify({ initialPlan: "Build planning", ip: "10.0.0.1" }),
|
||||
conversationHistory: JSON.stringify(history),
|
||||
currentQuestion: JSON.stringify(currentQuestion),
|
||||
result: JSON.stringify(result),
|
||||
thinkingOutput: "Thought stream",
|
||||
error: null,
|
||||
projectId: "proj-a",
|
||||
});
|
||||
|
||||
store.upsert(row);
|
||||
|
||||
const persisted = store.get(row.id);
|
||||
expect(persisted).not.toBeNull();
|
||||
expect(persisted).toMatchObject({
|
||||
id: row.id,
|
||||
type: "planning",
|
||||
status: "awaiting_input",
|
||||
title: "Roundtrip Session",
|
||||
projectId: "proj-a",
|
||||
thinkingOutput: "Thought stream",
|
||||
error: null,
|
||||
});
|
||||
expect(JSON.parse(persisted!.inputPayload)).toEqual(JSON.parse(row.inputPayload));
|
||||
expect(JSON.parse(persisted!.conversationHistory)).toEqual(history);
|
||||
expect(JSON.parse(persisted!.currentQuestion ?? "null")).toEqual(currentQuestion);
|
||||
expect(JSON.parse(persisted!.result ?? "null")).toEqual(result);
|
||||
});
|
||||
|
||||
it("upsert updates an existing row on id conflict", () => {
|
||||
const id = "sess-conflict";
|
||||
store.upsert(
|
||||
makeRow(id, {
|
||||
status: "generating",
|
||||
conversationHistory: JSON.stringify([{ question: { id: "q-1" }, response: { "q-1": "initial" } }]),
|
||||
}),
|
||||
);
|
||||
|
||||
store.upsert(
|
||||
makeRow(id, {
|
||||
status: "error",
|
||||
conversationHistory: JSON.stringify([{ question: { id: "q-1" }, response: { "q-1": "updated" } }]),
|
||||
error: "Failed to parse AI response",
|
||||
}),
|
||||
);
|
||||
|
||||
const rowCount = db.prepare("SELECT COUNT(*) as count FROM ai_sessions WHERE id = ?").get(id) as {
|
||||
count: number;
|
||||
};
|
||||
expect(rowCount.count).toBe(1);
|
||||
|
||||
const updated = store.get(id);
|
||||
expect(updated?.status).toBe("error");
|
||||
expect(updated?.error).toBe("Failed to parse AI response");
|
||||
expect(JSON.parse(updated?.conversationHistory ?? "[]")).toEqual([
|
||||
{ question: { id: "q-1" }, response: { "q-1": "updated" } },
|
||||
]);
|
||||
});
|
||||
|
||||
it("listActive returns only generating/awaiting_input/error ordered by updatedAt desc", () => {
|
||||
store.upsert(makeRow("active-generating", { status: "generating" }));
|
||||
store.upsert(makeRow("active-awaiting", { status: "awaiting_input" }));
|
||||
store.upsert(makeRow("inactive-complete", { status: "complete" }));
|
||||
store.upsert(makeRow("active-error", { status: "error" }));
|
||||
|
||||
db.prepare("UPDATE ai_sessions SET updatedAt = ? WHERE id = ?").run("2026-01-01T00:00:01.000Z", "active-generating");
|
||||
db.prepare("UPDATE ai_sessions SET updatedAt = ? WHERE id = ?").run("2026-01-01T00:00:03.000Z", "active-awaiting");
|
||||
db.prepare("UPDATE ai_sessions SET updatedAt = ? WHERE id = ?").run("2026-01-01T00:00:02.000Z", "active-error");
|
||||
db.prepare("UPDATE ai_sessions SET updatedAt = ? WHERE id = ?").run("2026-01-01T00:00:04.000Z", "inactive-complete");
|
||||
|
||||
const active = store.listActive();
|
||||
|
||||
expect(active.map((item) => item.id)).toEqual([
|
||||
"active-awaiting",
|
||||
"active-error",
|
||||
"active-generating",
|
||||
]);
|
||||
expect(active.every((item) => ["generating", "awaiting_input", "error"].includes(item.status))).toBe(true);
|
||||
});
|
||||
|
||||
it("listActive filters by projectId", () => {
|
||||
store.upsert(makeRow("a-1", { status: "generating", projectId: "proj-a" }));
|
||||
store.upsert(makeRow("a-2", { status: "awaiting_input", projectId: "proj-a" }));
|
||||
store.upsert(makeRow("a-3", { status: "error", projectId: "proj-a" }));
|
||||
store.upsert(makeRow("b-1", { status: "generating", projectId: "proj-b" }));
|
||||
store.upsert(makeRow("a-complete", { status: "complete", projectId: "proj-a" }));
|
||||
|
||||
const filtered = store.listActive("proj-a");
|
||||
|
||||
expect(filtered.map((row) => row.id).sort()).toEqual(["a-1", "a-2", "a-3"]);
|
||||
expect(filtered.every((row) => row.projectId === "proj-a")).toBe(true);
|
||||
});
|
||||
|
||||
it("delete removes row and emits ai_session:deleted", () => {
|
||||
const onDeleted = vi.fn();
|
||||
store.on("ai_session:deleted", onDeleted);
|
||||
|
||||
store.upsert(makeRow("sess-delete", { status: "awaiting_input" }));
|
||||
expect(store.get("sess-delete")).not.toBeNull();
|
||||
|
||||
store.delete("sess-delete");
|
||||
|
||||
expect(store.get("sess-delete")).toBeNull();
|
||||
expect(onDeleted).toHaveBeenCalledWith("sess-delete");
|
||||
});
|
||||
|
||||
it("recoverStaleSessions promotes recoverable rows and errors unrecoverable ones", () => {
|
||||
store.upsert(
|
||||
makeRow("recoverable", {
|
||||
status: "generating",
|
||||
currentQuestion: JSON.stringify({ id: "q-1", type: "text", question: "Continue?" }),
|
||||
}),
|
||||
);
|
||||
store.upsert(makeRow("unrecoverable", { status: "generating", currentQuestion: null }));
|
||||
|
||||
const changed = store.recoverStaleSessions();
|
||||
|
||||
expect(changed).toBe(2);
|
||||
expect(store.get("recoverable")?.status).toBe("awaiting_input");
|
||||
expect(store.get("unrecoverable")?.status).toBe("error");
|
||||
expect(store.get("unrecoverable")?.error).toContain("Session interrupted");
|
||||
});
|
||||
|
||||
it("cleanupOld removes only old terminal rows", () => {
|
||||
store.upsert(makeRow("old-complete", { status: "complete" }));
|
||||
store.upsert(makeRow("old-error", { status: "error" }));
|
||||
store.upsert(makeRow("old-generating", { status: "generating" }));
|
||||
store.upsert(makeRow("fresh-complete", { status: "complete" }));
|
||||
|
||||
const staleTs = new Date(Date.now() - 4 * 60 * 60 * 1000).toISOString();
|
||||
const freshTs = new Date(Date.now() - 20 * 60 * 1000).toISOString();
|
||||
db.prepare("UPDATE ai_sessions SET updatedAt = ? WHERE id IN (?, ?, ?)").run(
|
||||
staleTs,
|
||||
"old-complete",
|
||||
"old-error",
|
||||
"old-generating",
|
||||
);
|
||||
db.prepare("UPDATE ai_sessions SET updatedAt = ? WHERE id = ?").run(freshTs, "fresh-complete");
|
||||
|
||||
const removed = store.cleanupOld(60 * 60 * 1000);
|
||||
|
||||
expect(removed).toBe(2);
|
||||
expect(store.get("old-complete")).toBeNull();
|
||||
expect(store.get("old-error")).toBeNull();
|
||||
expect(store.get("old-generating")).not.toBeNull();
|
||||
expect(store.get("fresh-complete")).not.toBeNull();
|
||||
});
|
||||
|
||||
it("trims thinkingOutput to the last 50KB on upsert", () => {
|
||||
const maxBytes = 50 * 1024;
|
||||
const oversized = `${"x".repeat(1024)}${"y".repeat(maxBytes + 2000)}`;
|
||||
|
||||
store.upsert(makeRow("sess-thinking-trim", { thinkingOutput: oversized }));
|
||||
|
||||
const persisted = store.get("sess-thinking-trim");
|
||||
expect(persisted).not.toBeNull();
|
||||
expect(persisted!.thinkingOutput.length).toBe(maxBytes);
|
||||
expect(persisted!.thinkingOutput).toBe(oversized.slice(oversized.length - maxBytes));
|
||||
});
|
||||
|
||||
it("updateThinking debounces writes unless flush=true", () => {
|
||||
vi.useFakeTimers();
|
||||
store.upsert(makeRow("sess-thinking-debounce", { thinkingOutput: "initial" }));
|
||||
|
||||
store.updateThinking("sess-thinking-debounce", "deferred-write");
|
||||
expect(store.get("sess-thinking-debounce")?.thinkingOutput).toBe("initial");
|
||||
|
||||
vi.advanceTimersByTime(1999);
|
||||
expect(store.get("sess-thinking-debounce")?.thinkingOutput).toBe("initial");
|
||||
|
||||
vi.advanceTimersByTime(1);
|
||||
expect(store.get("sess-thinking-debounce")?.thinkingOutput).toBe("deferred-write");
|
||||
|
||||
store.updateThinking("sess-thinking-debounce", "queued-write");
|
||||
store.updateThinking("sess-thinking-debounce", "flushed-write", true);
|
||||
|
||||
expect(store.get("sess-thinking-debounce")?.thinkingOutput).toBe("flushed-write");
|
||||
|
||||
vi.advanceTimersByTime(5000);
|
||||
expect(store.get("sess-thinking-debounce")?.thinkingOutput).toBe("flushed-write");
|
||||
});
|
||||
|
||||
it("emits ai_session:updated summary on upsert", () => {
|
||||
const onUpdated = vi.fn<[AiSessionSummary]>();
|
||||
store.on("ai_session:updated", onUpdated);
|
||||
|
||||
store.upsert(
|
||||
makeRow("sess-event", {
|
||||
status: "awaiting_input",
|
||||
title: "Session Event",
|
||||
projectId: "proj-events",
|
||||
}),
|
||||
);
|
||||
|
||||
expect(onUpdated).toHaveBeenCalledTimes(1);
|
||||
expect(onUpdated).toHaveBeenCalledWith(
|
||||
expect.objectContaining({
|
||||
id: "sess-event",
|
||||
type: "planning",
|
||||
status: "awaiting_input",
|
||||
title: "Session Event",
|
||||
projectId: "proj-events",
|
||||
lockedByTab: null,
|
||||
updatedAt: expect.any(String),
|
||||
}),
|
||||
);
|
||||
});
|
||||
});
|
||||
171
packages/dashboard/src/__tests__/session-cross-tab.test.ts
Normal file
171
packages/dashboard/src/__tests__/session-cross-tab.test.ts
Normal file
@@ -0,0 +1,171 @@
|
||||
/**
|
||||
* Covers optimistic locking and cross-tab continuity primitives:
|
||||
* lock conflicts, beacon release, stale lock expiry, SSE summaries, and stale cleanup.
|
||||
*/
|
||||
|
||||
// @vitest-environment node
|
||||
|
||||
import express from "express";
|
||||
import { beforeEach, afterEach, describe, expect, it } from "vitest";
|
||||
import { mkdtempSync } from "node:fs";
|
||||
import { rm } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import { TaskStore } from "@fusion/core";
|
||||
import { AiSessionStore, type AiSessionRow } from "../ai-session-store.js";
|
||||
import { createApiRoutes } from "../routes.js";
|
||||
import { request } from "../test-request.js";
|
||||
|
||||
function makeRow(id: string, overrides: Partial<AiSessionRow> = {}): AiSessionRow {
|
||||
const now = new Date().toISOString();
|
||||
return {
|
||||
id,
|
||||
type: "planning",
|
||||
status: "awaiting_input",
|
||||
title: `Session ${id}`,
|
||||
inputPayload: JSON.stringify({ initialPlan: "Cross-tab lock test" }),
|
||||
conversationHistory: "[]",
|
||||
currentQuestion: JSON.stringify({ id: "q-1", type: "text", question: "Q" }),
|
||||
result: null,
|
||||
thinkingOutput: "",
|
||||
error: null,
|
||||
projectId: "proj-locks",
|
||||
createdAt: now,
|
||||
updatedAt: now,
|
||||
lockedByTab: null,
|
||||
lockedAt: null,
|
||||
...overrides,
|
||||
};
|
||||
}
|
||||
|
||||
describe("cross-tab session locking", () => {
|
||||
let tmpRoot: string;
|
||||
let taskStore: TaskStore;
|
||||
let aiSessionStore: AiSessionStore;
|
||||
let app: express.Express;
|
||||
|
||||
beforeEach(async () => {
|
||||
tmpRoot = mkdtempSync(join(tmpdir(), "kb-session-cross-tab-"));
|
||||
taskStore = new TaskStore(tmpRoot);
|
||||
await taskStore.init();
|
||||
aiSessionStore = new AiSessionStore(taskStore.getDatabase());
|
||||
|
||||
app = express();
|
||||
app.use(express.json());
|
||||
app.use("/api", createApiRoutes(taskStore, { aiSessionStore }));
|
||||
});
|
||||
|
||||
afterEach(async () => {
|
||||
try {
|
||||
taskStore.close();
|
||||
} catch {
|
||||
// no-op
|
||||
}
|
||||
await rm(tmpRoot, { recursive: true, force: true });
|
||||
});
|
||||
|
||||
it("enforces optimistic lock conflicts and reports current holder", () => {
|
||||
aiSessionStore.upsert(makeRow("lock-conflict"));
|
||||
|
||||
const first = aiSessionStore.acquireLock("lock-conflict", "tab-a");
|
||||
const second = aiSessionStore.acquireLock("lock-conflict", "tab-b");
|
||||
|
||||
expect(first).toEqual({ acquired: true, currentHolder: null });
|
||||
expect(second).toEqual({ acquired: false, currentHolder: "tab-a" });
|
||||
expect(aiSessionStore.getLockHolder("lock-conflict").tabId).toBe("tab-a");
|
||||
});
|
||||
|
||||
it("supports concurrent lock attempts where only one tab acquires", async () => {
|
||||
aiSessionStore.upsert(makeRow("lock-race"));
|
||||
|
||||
const [resultA, resultB] = await Promise.all([
|
||||
Promise.resolve(aiSessionStore.acquireLock("lock-race", "tab-a")),
|
||||
Promise.resolve(aiSessionStore.acquireLock("lock-race", "tab-b")),
|
||||
]);
|
||||
|
||||
const acquiredCount = [resultA, resultB].filter((result) => result.acquired).length;
|
||||
const denied = [resultA, resultB].find((result) => !result.acquired);
|
||||
|
||||
expect(acquiredCount).toBe(1);
|
||||
expect(denied?.currentHolder).toMatch(/^tab-[ab]$/);
|
||||
});
|
||||
|
||||
it("releases lock on tab close beacon endpoint", async () => {
|
||||
aiSessionStore.upsert(makeRow("lock-beacon"));
|
||||
aiSessionStore.acquireLock("lock-beacon", "tab-a");
|
||||
|
||||
const response = await request(
|
||||
app,
|
||||
"DELETE",
|
||||
"/api/ai-sessions/lock-beacon/lock/beacon?tabId=tab-a",
|
||||
);
|
||||
|
||||
expect(response.status).toBe(200);
|
||||
expect(aiSessionStore.getLockHolder("lock-beacon")).toEqual({ tabId: null, lockedAt: null });
|
||||
});
|
||||
|
||||
it("expires stale locks and clears ownership", () => {
|
||||
aiSessionStore.upsert(makeRow("lock-expiry"));
|
||||
aiSessionStore.acquireLock("lock-expiry", "tab-expired");
|
||||
|
||||
const staleTimestamp = new Date(Date.now() - 31 * 60 * 1000).toISOString();
|
||||
taskStore
|
||||
.getDatabase()
|
||||
.prepare("UPDATE ai_sessions SET lockedAt = ? WHERE id = ?")
|
||||
.run(staleTimestamp, "lock-expiry");
|
||||
|
||||
const released = aiSessionStore.releaseStaleLocks(30 * 60 * 1000);
|
||||
|
||||
expect(released).toBe(1);
|
||||
expect(aiSessionStore.getLockHolder("lock-expiry")).toEqual({ tabId: null, lockedAt: null });
|
||||
});
|
||||
|
||||
it("emits ai_session:updated summaries on lock acquisition/release transitions", () => {
|
||||
aiSessionStore.upsert(makeRow("lock-sse"));
|
||||
const summaries: Array<{ id: string; lockedByTab: string | null }> = [];
|
||||
|
||||
aiSessionStore.on("ai_session:updated", (summary) => {
|
||||
summaries.push({ id: summary.id, lockedByTab: summary.lockedByTab });
|
||||
});
|
||||
|
||||
aiSessionStore.acquireLock("lock-sse", "tab-a");
|
||||
aiSessionStore.releaseLock("lock-sse", "tab-a");
|
||||
aiSessionStore.forceAcquireLock("lock-sse", "tab-b");
|
||||
|
||||
expect(summaries).toEqual(
|
||||
expect.arrayContaining([
|
||||
{ id: "lock-sse", lockedByTab: "tab-a" },
|
||||
{ id: "lock-sse", lockedByTab: null },
|
||||
{ id: "lock-sse", lockedByTab: "tab-b" },
|
||||
]),
|
||||
);
|
||||
|
||||
const listActive = aiSessionStore.listActive("proj-locks");
|
||||
const latest = listActive.find((session) => session.id === "lock-sse");
|
||||
expect(latest?.lockedByTab).toBe("tab-b");
|
||||
});
|
||||
|
||||
it("cleans stale active sessions and leaves fresh sessions intact", () => {
|
||||
aiSessionStore.upsert(makeRow("stale-generating", { status: "generating" }));
|
||||
aiSessionStore.upsert(makeRow("stale-awaiting", { status: "awaiting_input" }));
|
||||
aiSessionStore.upsert(makeRow("fresh-generating", { status: "generating" }));
|
||||
|
||||
const stale = new Date(Date.now() - 8 * 24 * 60 * 60 * 1000).toISOString();
|
||||
const fresh = new Date(Date.now() - 60 * 1000).toISOString();
|
||||
taskStore
|
||||
.getDatabase()
|
||||
.prepare("UPDATE ai_sessions SET updatedAt = ? WHERE id IN (?, ?)")
|
||||
.run(stale, "stale-generating", "stale-awaiting");
|
||||
taskStore
|
||||
.getDatabase()
|
||||
.prepare("UPDATE ai_sessions SET updatedAt = ? WHERE id = ?")
|
||||
.run(fresh, "fresh-generating");
|
||||
|
||||
const summary = aiSessionStore.cleanupStaleSessions(7 * 24 * 60 * 60 * 1000);
|
||||
|
||||
expect(summary.orphanedDeleted).toBe(2);
|
||||
expect(aiSessionStore.get("stale-generating")).toBeNull();
|
||||
expect(aiSessionStore.get("stale-awaiting")).toBeNull();
|
||||
expect(aiSessionStore.get("fresh-generating")).not.toBeNull();
|
||||
});
|
||||
});
|
||||
275
packages/dashboard/src/__tests__/session-error-recovery.test.ts
Normal file
275
packages/dashboard/src/__tests__/session-error-recovery.test.ts
Normal file
@@ -0,0 +1,275 @@
|
||||
/**
|
||||
* Covers error-state persistence, SSE error broadcasts, and retry recovery flows
|
||||
* for planning, subtask breakdown, and mission interview sessions.
|
||||
*/
|
||||
|
||||
// @vitest-environment node
|
||||
|
||||
import { beforeEach, afterEach, describe, expect, it, vi } from "vitest";
|
||||
import { mkdtempSync } from "node:fs";
|
||||
import { rm } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import { Database } from "@fusion/core";
|
||||
import { AiSessionStore } from "../ai-session-store.js";
|
||||
import {
|
||||
__resetPlanningState,
|
||||
__setCreateKbAgent,
|
||||
createSession,
|
||||
getSession,
|
||||
planningStreamManager,
|
||||
retrySession,
|
||||
setAiSessionStore as setPlanningAiSessionStore,
|
||||
submitResponse,
|
||||
} from "../planning.js";
|
||||
import {
|
||||
__resetSubtaskBreakdownState,
|
||||
createSubtaskSession,
|
||||
getSubtaskSession,
|
||||
retrySubtaskSession,
|
||||
setAiSessionStore as setSubtaskAiSessionStore,
|
||||
subtaskStreamManager,
|
||||
} from "../subtask-breakdown.js";
|
||||
import {
|
||||
__resetMissionInterviewState,
|
||||
createMissionInterviewSession,
|
||||
getMissionInterviewSession,
|
||||
missionInterviewStreamManager,
|
||||
retryMissionInterviewSession,
|
||||
setAiSessionStore as setMissionAiSessionStore,
|
||||
submitMissionInterviewResponse,
|
||||
} from "../mission-interview.js";
|
||||
|
||||
const { mockCreateKbAgent } = vi.hoisted(() => ({
|
||||
mockCreateKbAgent: vi.fn(),
|
||||
}));
|
||||
|
||||
vi.mock("@fusion/engine", () => ({
|
||||
createKbAgent: mockCreateKbAgent,
|
||||
}));
|
||||
|
||||
function makeTmpDir(): string {
|
||||
return mkdtempSync(join(tmpdir(), "kb-session-error-recovery-"));
|
||||
}
|
||||
|
||||
function createMockAgent(responses: string[]) {
|
||||
const queue = [...responses];
|
||||
const messages: Array<{ role: string; content: string }> = [];
|
||||
|
||||
return {
|
||||
session: {
|
||||
state: { messages },
|
||||
prompt: vi.fn(async (_input: string) => {
|
||||
const response = queue.shift() ?? queue[queue.length - 1] ?? "{}";
|
||||
messages.push({ role: "assistant", content: response });
|
||||
}),
|
||||
dispose: vi.fn(),
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
async function waitFor(check: () => boolean, timeoutMs = 2000): Promise<void> {
|
||||
const start = Date.now();
|
||||
while (!check()) {
|
||||
if (Date.now() - start > timeoutMs) {
|
||||
throw new Error("Timed out waiting for condition");
|
||||
}
|
||||
await new Promise((resolve) => setTimeout(resolve, 10));
|
||||
}
|
||||
}
|
||||
|
||||
describe("session error recovery", () => {
|
||||
let tmpDir: string;
|
||||
let db: Database;
|
||||
let aiSessionStore: AiSessionStore;
|
||||
|
||||
beforeEach(() => {
|
||||
vi.clearAllMocks();
|
||||
__resetPlanningState();
|
||||
__resetSubtaskBreakdownState();
|
||||
__resetMissionInterviewState();
|
||||
|
||||
tmpDir = makeTmpDir();
|
||||
db = new Database(join(tmpDir, ".fusion"));
|
||||
db.init();
|
||||
aiSessionStore = new AiSessionStore(db);
|
||||
|
||||
setPlanningAiSessionStore(aiSessionStore);
|
||||
setSubtaskAiSessionStore(aiSessionStore);
|
||||
setMissionAiSessionStore(aiSessionStore);
|
||||
});
|
||||
|
||||
afterEach(async () => {
|
||||
__setCreateKbAgent(undefined as any);
|
||||
__resetPlanningState();
|
||||
__resetSubtaskBreakdownState();
|
||||
__resetMissionInterviewState();
|
||||
|
||||
try {
|
||||
db.close();
|
||||
} catch {
|
||||
// no-op
|
||||
}
|
||||
|
||||
await rm(tmpDir, { recursive: true, force: true });
|
||||
});
|
||||
|
||||
it("captures planning parse failures as error state, preserves history, and allows retry", async () => {
|
||||
const errorEvents: string[] = [];
|
||||
const unsubscribe = planningStreamManager.subscribe("pending", () => {
|
||||
// placeholder; replaced below once session id exists
|
||||
});
|
||||
unsubscribe();
|
||||
|
||||
__setCreateKbAgent(
|
||||
async () =>
|
||||
createMockAgent([
|
||||
JSON.stringify({
|
||||
type: "question",
|
||||
data: { id: "q-1", type: "text", question: "First question" },
|
||||
}),
|
||||
"not-json",
|
||||
"still-not-json",
|
||||
]),
|
||||
);
|
||||
|
||||
const { sessionId } = await createSession("127.0.0.101", "Planning error flow", undefined, "/tmp/project");
|
||||
|
||||
const unsubscribeError = planningStreamManager.subscribe(sessionId, (event) => {
|
||||
if (event.type === "error") {
|
||||
errorEvents.push(String(event.data));
|
||||
}
|
||||
});
|
||||
|
||||
const responseAfterFailure = await submitResponse(
|
||||
sessionId,
|
||||
{ "q-1": "trigger error" },
|
||||
"/tmp/project",
|
||||
);
|
||||
expect(responseAfterFailure.type).toBe("question");
|
||||
if (responseAfterFailure.type === "question") {
|
||||
// currentQuestion remains the same when parsing fails
|
||||
expect(responseAfterFailure.data.id).toBe("q-1");
|
||||
}
|
||||
|
||||
const persistedError = aiSessionStore.get(sessionId);
|
||||
expect(persistedError?.status).toBe("error");
|
||||
expect(persistedError?.error).toContain("AI returned no valid JSON");
|
||||
expect(JSON.parse(persistedError?.conversationHistory ?? "[]")).toHaveLength(1);
|
||||
expect(errorEvents.length).toBeGreaterThan(0);
|
||||
|
||||
__setCreateKbAgent(
|
||||
async () =>
|
||||
createMockAgent([
|
||||
JSON.stringify({
|
||||
type: "question",
|
||||
data: { id: "q-retry", type: "text", question: "Recovered question" },
|
||||
}),
|
||||
]),
|
||||
);
|
||||
|
||||
await retrySession(sessionId, "/tmp/project");
|
||||
|
||||
const persistedRecovered = aiSessionStore.get(sessionId);
|
||||
expect(persistedRecovered?.status).toBe("awaiting_input");
|
||||
expect(persistedRecovered?.error).toBeNull();
|
||||
expect(getSession(sessionId)?.currentQuestion?.id).toBe("q-retry");
|
||||
|
||||
unsubscribeError();
|
||||
});
|
||||
|
||||
it("captures subtask generation errors, broadcasts SSE error, and retries to completion", async () => {
|
||||
const subtaskErrors: string[] = [];
|
||||
|
||||
mockCreateKbAgent.mockImplementationOnce(async () => createMockAgent(["{not-json"]));
|
||||
|
||||
const session = await createSubtaskSession("Subtask error flow", undefined, "/tmp/project");
|
||||
|
||||
const unsubscribe = subtaskStreamManager.subscribe(session.sessionId, (event) => {
|
||||
if (event.type === "error") {
|
||||
subtaskErrors.push(String(event.data));
|
||||
}
|
||||
});
|
||||
|
||||
await waitFor(() => aiSessionStore.get(session.sessionId)?.status === "error");
|
||||
|
||||
const persistedError = aiSessionStore.get(session.sessionId);
|
||||
expect(persistedError?.status).toBe("error");
|
||||
expect(String(persistedError?.error).length).toBeGreaterThan(0);
|
||||
expect(subtaskErrors.length).toBeGreaterThan(0);
|
||||
|
||||
mockCreateKbAgent.mockImplementationOnce(async () =>
|
||||
createMockAgent([
|
||||
JSON.stringify({
|
||||
subtasks: [
|
||||
{
|
||||
id: "subtask-1",
|
||||
title: "Recovered",
|
||||
description: "Recovered after retry",
|
||||
suggestedSize: "S",
|
||||
dependsOn: [],
|
||||
},
|
||||
],
|
||||
}),
|
||||
]),
|
||||
);
|
||||
|
||||
await retrySubtaskSession(session.sessionId, "/tmp/project");
|
||||
|
||||
await waitFor(() => aiSessionStore.get(session.sessionId)?.status === "complete");
|
||||
expect(getSubtaskSession(session.sessionId)?.status).toBe("complete");
|
||||
expect(aiSessionStore.get(session.sessionId)?.error).toBeNull();
|
||||
|
||||
unsubscribe();
|
||||
});
|
||||
|
||||
it("captures mission parse failure with history preserved and recovers via retry", async () => {
|
||||
const missionErrors: string[] = [];
|
||||
|
||||
mockCreateKbAgent.mockImplementation(async () =>
|
||||
createMockAgent([
|
||||
JSON.stringify({
|
||||
type: "question",
|
||||
data: { id: "q-m-1", type: "text", question: "Mission question" },
|
||||
}),
|
||||
"invalid-json",
|
||||
"invalid-json-again",
|
||||
]),
|
||||
);
|
||||
|
||||
const sessionId = await createMissionInterviewSession("127.0.0.111", "Mission error flow", "/tmp/project");
|
||||
await waitFor(() => Boolean(getMissionInterviewSession(sessionId)?.currentQuestion));
|
||||
|
||||
const unsubscribe = missionInterviewStreamManager.subscribe(sessionId, (event) => {
|
||||
if (event.type === "error") {
|
||||
missionErrors.push(String(event.data));
|
||||
}
|
||||
});
|
||||
|
||||
await submitMissionInterviewResponse(sessionId, { "q-m-1": "trigger mission error" }, "/tmp/project");
|
||||
|
||||
const persistedError = aiSessionStore.get(sessionId);
|
||||
expect(persistedError?.status).toBe("error");
|
||||
expect(String(persistedError?.error)).toContain("AI returned no valid JSON");
|
||||
expect(JSON.parse(persistedError?.conversationHistory ?? "[]")).toHaveLength(1);
|
||||
expect(missionErrors.length).toBeGreaterThan(0);
|
||||
|
||||
mockCreateKbAgent.mockImplementation(async () =>
|
||||
createMockAgent([
|
||||
JSON.stringify({
|
||||
type: "question",
|
||||
data: { id: "q-m-retry", type: "text", question: "Recovered mission question" },
|
||||
}),
|
||||
]),
|
||||
);
|
||||
|
||||
await retryMissionInterviewSession(sessionId, "/tmp/project");
|
||||
|
||||
const persistedRecovered = aiSessionStore.get(sessionId);
|
||||
expect(persistedRecovered?.status).toBe("awaiting_input");
|
||||
expect(persistedRecovered?.error).toBeNull();
|
||||
expect(getMissionInterviewSession(sessionId)?.currentQuestion?.id).toBe("q-m-retry");
|
||||
|
||||
unsubscribe();
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,325 @@
|
||||
/**
|
||||
* Covers SQLite persistence round-trips for planning, subtask, and mission sessions,
|
||||
* including transition snapshots, recovery, and delete semantics.
|
||||
*/
|
||||
|
||||
// @vitest-environment node
|
||||
|
||||
import { beforeEach, afterEach, describe, expect, it, vi } from "vitest";
|
||||
import { mkdtempSync } from "node:fs";
|
||||
import { rm } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import { Database } from "@fusion/core";
|
||||
import { AiSessionStore, type AiSessionRow } from "../ai-session-store.js";
|
||||
import {
|
||||
__resetPlanningState,
|
||||
__setCreateKbAgent,
|
||||
cancelSession,
|
||||
createSession,
|
||||
setAiSessionStore as setPlanningAiSessionStore,
|
||||
submitResponse,
|
||||
} from "../planning.js";
|
||||
import {
|
||||
__resetSubtaskBreakdownState,
|
||||
createSubtaskSession,
|
||||
setAiSessionStore as setSubtaskAiSessionStore,
|
||||
} from "../subtask-breakdown.js";
|
||||
import {
|
||||
__resetMissionInterviewState,
|
||||
createMissionInterviewSession,
|
||||
getMissionInterviewSession,
|
||||
setAiSessionStore as setMissionAiSessionStore,
|
||||
submitMissionInterviewResponse,
|
||||
} from "../mission-interview.js";
|
||||
|
||||
const { mockCreateKbAgent } = vi.hoisted(() => ({
|
||||
mockCreateKbAgent: vi.fn(),
|
||||
}));
|
||||
|
||||
vi.mock("@fusion/engine", () => ({
|
||||
createKbAgent: mockCreateKbAgent,
|
||||
}));
|
||||
|
||||
function makeTmpDir(): string {
|
||||
return mkdtempSync(join(tmpdir(), "kb-session-roundtrip-"));
|
||||
}
|
||||
|
||||
function createMockAgent(responses: string[]) {
|
||||
const messages: Array<{ role: string; content: string }> = [];
|
||||
let index = 0;
|
||||
|
||||
return {
|
||||
session: {
|
||||
state: { messages },
|
||||
prompt: vi.fn(async (_message: string) => {
|
||||
const response = responses[index++] ?? responses[responses.length - 1] ?? responses[0] ?? "{}";
|
||||
messages.push({ role: "assistant", content: response });
|
||||
}),
|
||||
dispose: vi.fn(),
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
async function waitFor(check: () => boolean, timeoutMs = 2000): Promise<void> {
|
||||
const start = Date.now();
|
||||
while (!check()) {
|
||||
if (Date.now() - start > timeoutMs) {
|
||||
throw new Error("Timed out waiting for condition");
|
||||
}
|
||||
await new Promise((resolve) => setTimeout(resolve, 10));
|
||||
}
|
||||
}
|
||||
|
||||
describe("session persistence round-trip", () => {
|
||||
let tmpDir: string;
|
||||
let db: Database;
|
||||
let aiSessionStore: AiSessionStore;
|
||||
|
||||
beforeEach(() => {
|
||||
vi.clearAllMocks();
|
||||
__resetPlanningState();
|
||||
__resetSubtaskBreakdownState();
|
||||
__resetMissionInterviewState();
|
||||
|
||||
tmpDir = makeTmpDir();
|
||||
db = new Database(join(tmpDir, ".fusion"));
|
||||
db.init();
|
||||
aiSessionStore = new AiSessionStore(db);
|
||||
|
||||
setPlanningAiSessionStore(aiSessionStore);
|
||||
setSubtaskAiSessionStore(aiSessionStore);
|
||||
setMissionAiSessionStore(aiSessionStore);
|
||||
});
|
||||
|
||||
afterEach(async () => {
|
||||
__setCreateKbAgent(undefined as any);
|
||||
__resetPlanningState();
|
||||
__resetSubtaskBreakdownState();
|
||||
__resetMissionInterviewState();
|
||||
|
||||
try {
|
||||
db.close();
|
||||
} catch {
|
||||
// no-op
|
||||
}
|
||||
|
||||
await rm(tmpDir, { recursive: true, force: true });
|
||||
});
|
||||
|
||||
it("persists planning session transitions across generating → awaiting_input → complete", async () => {
|
||||
__setCreateKbAgent(
|
||||
async () =>
|
||||
createMockAgent([
|
||||
JSON.stringify({
|
||||
type: "question",
|
||||
data: { id: "q-1", type: "text", question: "First question" },
|
||||
}),
|
||||
JSON.stringify({
|
||||
type: "question",
|
||||
data: { id: "q-2", type: "text", question: "Second question" },
|
||||
}),
|
||||
JSON.stringify({
|
||||
type: "complete",
|
||||
data: {
|
||||
title: "Planned",
|
||||
description: "Plan summary",
|
||||
suggestedSize: "M",
|
||||
suggestedDependencies: [],
|
||||
keyDeliverables: ["One", "Two"],
|
||||
},
|
||||
}),
|
||||
]),
|
||||
);
|
||||
|
||||
const { sessionId } = await createSession("127.0.0.31", "Plan persistence", undefined, "/tmp/project");
|
||||
|
||||
const afterCreate = aiSessionStore.get(sessionId);
|
||||
expect(afterCreate?.status).toBe("awaiting_input");
|
||||
expect(JSON.parse(afterCreate?.conversationHistory ?? "[]")).toEqual([]);
|
||||
expect(JSON.parse(afterCreate?.currentQuestion ?? "null")?.id).toBe("q-1");
|
||||
|
||||
await submitResponse(sessionId, { "q-1": "Answer one" }, "/tmp/project");
|
||||
const afterFirstResponse = aiSessionStore.get(sessionId);
|
||||
expect(afterFirstResponse?.status).toBe("awaiting_input");
|
||||
expect(JSON.parse(afterFirstResponse?.currentQuestion ?? "null")?.id).toBe("q-2");
|
||||
expect(JSON.parse(afterFirstResponse?.conversationHistory ?? "[]")).toHaveLength(1);
|
||||
|
||||
await submitResponse(sessionId, { "q-2": "Answer two" }, "/tmp/project");
|
||||
const afterComplete = aiSessionStore.get(sessionId);
|
||||
expect(afterComplete?.status).toBe("complete");
|
||||
expect(afterComplete?.currentQuestion).toBeNull();
|
||||
expect(JSON.parse(afterComplete?.conversationHistory ?? "[]")).toHaveLength(2);
|
||||
expect(JSON.parse(afterComplete?.result ?? "null")?.title).toBe("Planned");
|
||||
});
|
||||
|
||||
it("recovers generating sessions from SQLite while preserving history and currentQuestion", () => {
|
||||
const now = new Date().toISOString();
|
||||
|
||||
const rows: AiSessionRow[] = [
|
||||
{
|
||||
id: "planning-recover",
|
||||
type: "planning",
|
||||
status: "generating",
|
||||
title: "Planning recover",
|
||||
inputPayload: JSON.stringify({ initialPlan: "Plan" }),
|
||||
conversationHistory: JSON.stringify([{ question: { id: "q-1" }, response: { "q-1": "a" } }]),
|
||||
currentQuestion: JSON.stringify({ id: "q-2", type: "text", question: "Next?" }),
|
||||
result: null,
|
||||
thinkingOutput: "thinking",
|
||||
error: null,
|
||||
projectId: null,
|
||||
createdAt: now,
|
||||
updatedAt: now,
|
||||
lockedByTab: null,
|
||||
lockedAt: null,
|
||||
},
|
||||
{
|
||||
id: "subtask-recover",
|
||||
type: "subtask",
|
||||
status: "generating",
|
||||
title: "Subtask recover",
|
||||
inputPayload: JSON.stringify({ initialDescription: "Breakdown" }),
|
||||
conversationHistory: JSON.stringify([{ note: "history" }]),
|
||||
currentQuestion: JSON.stringify({ id: "q-sub", type: "text", question: "placeholder" }),
|
||||
result: null,
|
||||
thinkingOutput: "thinking",
|
||||
error: null,
|
||||
projectId: null,
|
||||
createdAt: now,
|
||||
updatedAt: now,
|
||||
lockedByTab: null,
|
||||
lockedAt: null,
|
||||
},
|
||||
{
|
||||
id: "mission-recover",
|
||||
type: "mission_interview",
|
||||
status: "generating",
|
||||
title: "Mission recover",
|
||||
inputPayload: JSON.stringify({ missionTitle: "Mission" }),
|
||||
conversationHistory: JSON.stringify([{ question: { id: "q-m" }, response: { "q-m": "a" } }]),
|
||||
currentQuestion: JSON.stringify({ id: "q-m-2", type: "text", question: "Next mission question" }),
|
||||
result: null,
|
||||
thinkingOutput: "thinking",
|
||||
error: null,
|
||||
projectId: null,
|
||||
createdAt: now,
|
||||
updatedAt: now,
|
||||
lockedByTab: null,
|
||||
lockedAt: null,
|
||||
},
|
||||
];
|
||||
|
||||
for (const row of rows) {
|
||||
aiSessionStore.upsert(row);
|
||||
}
|
||||
|
||||
const recoveredCount = aiSessionStore.recoverStaleSessions();
|
||||
expect(recoveredCount).toBe(3);
|
||||
|
||||
for (const row of rows) {
|
||||
const recovered = aiSessionStore.get(row.id);
|
||||
expect(recovered?.status).toBe("awaiting_input");
|
||||
expect(JSON.parse(recovered?.conversationHistory ?? "[]")).toEqual(JSON.parse(row.conversationHistory));
|
||||
expect(JSON.parse(recovered?.currentQuestion ?? "null")).toEqual(JSON.parse(row.currentQuestion ?? "null"));
|
||||
}
|
||||
});
|
||||
|
||||
it("persists completed subtask results into AiSessionStore result JSON", async () => {
|
||||
mockCreateKbAgent.mockImplementation(async () =>
|
||||
createMockAgent([
|
||||
JSON.stringify({
|
||||
subtasks: [
|
||||
{
|
||||
id: "subtask-1",
|
||||
title: "Prepare",
|
||||
description: "Prepare setup",
|
||||
suggestedSize: "S",
|
||||
dependsOn: [],
|
||||
},
|
||||
{
|
||||
id: "subtask-2",
|
||||
title: "Implement",
|
||||
description: "Implement feature",
|
||||
suggestedSize: "M",
|
||||
dependsOn: ["subtask-1"],
|
||||
},
|
||||
],
|
||||
}),
|
||||
]),
|
||||
);
|
||||
|
||||
const session = await createSubtaskSession("Generate subtasks", undefined, "/tmp/project");
|
||||
|
||||
await waitFor(() => aiSessionStore.get(session.sessionId)?.status === "complete");
|
||||
|
||||
const persisted = aiSessionStore.get(session.sessionId);
|
||||
expect(persisted?.status).toBe("complete");
|
||||
|
||||
const result = JSON.parse(persisted?.result ?? "[]") as Array<{ title: string }>;
|
||||
expect(result.map((subtask) => subtask.title)).toEqual(["Prepare", "Implement"]);
|
||||
});
|
||||
|
||||
it("persists mission interview history and result round-trip", async () => {
|
||||
mockCreateKbAgent.mockImplementation(async () =>
|
||||
createMockAgent([
|
||||
JSON.stringify({
|
||||
type: "question",
|
||||
data: { id: "q-m-1", type: "text", question: "What are we building?" },
|
||||
}),
|
||||
JSON.stringify({
|
||||
type: "complete",
|
||||
data: {
|
||||
missionTitle: "Mission Plan",
|
||||
missionDescription: "Plan details",
|
||||
milestones: [
|
||||
{
|
||||
title: "Milestone A",
|
||||
slices: [
|
||||
{
|
||||
title: "Slice A",
|
||||
features: [{ title: "Feature A", acceptanceCriteria: "Pass" }],
|
||||
},
|
||||
],
|
||||
},
|
||||
],
|
||||
},
|
||||
}),
|
||||
]),
|
||||
);
|
||||
|
||||
const sessionId = await createMissionInterviewSession("127.0.0.44", "Mission persistence", "/tmp/project");
|
||||
await waitFor(() => Boolean(getMissionInterviewSession(sessionId)?.currentQuestion));
|
||||
|
||||
await submitMissionInterviewResponse(sessionId, { "q-m-1": "A mission" }, "/tmp/project");
|
||||
|
||||
const persisted = aiSessionStore.get(sessionId);
|
||||
expect(persisted?.status).toBe("complete");
|
||||
|
||||
const history = JSON.parse(persisted?.conversationHistory ?? "[]") as Array<{ question: { id: string } }>;
|
||||
expect(history).toHaveLength(1);
|
||||
expect(history[0]?.question.id).toBe("q-m-1");
|
||||
|
||||
const result = JSON.parse(persisted?.result ?? "null") as { missionTitle?: string };
|
||||
expect(result?.missionTitle).toBe("Mission Plan");
|
||||
});
|
||||
|
||||
it("removes persisted rows when a planning session is cancelled", async () => {
|
||||
__setCreateKbAgent(
|
||||
async () =>
|
||||
createMockAgent([
|
||||
JSON.stringify({
|
||||
type: "question",
|
||||
data: { id: "q-cancel", type: "text", question: "Cancel me?" },
|
||||
}),
|
||||
]),
|
||||
);
|
||||
|
||||
const { sessionId } = await createSession("127.0.0.55", "Cancel persistence", undefined, "/tmp/project");
|
||||
expect(aiSessionStore.get(sessionId)).not.toBeNull();
|
||||
|
||||
await cancelSession(sessionId);
|
||||
|
||||
expect(aiSessionStore.get(sessionId)).toBeNull();
|
||||
});
|
||||
});
|
||||
288
packages/dashboard/src/__tests__/session-reconnect.test.ts
Normal file
288
packages/dashboard/src/__tests__/session-reconnect.test.ts
Normal file
@@ -0,0 +1,288 @@
|
||||
/**
|
||||
* Covers SSE reconnect behavior: Last-Event-ID replay, reconnect catch-up,
|
||||
* and keep-alive ping handling for persisted AI sessions.
|
||||
*/
|
||||
|
||||
// @vitest-environment node
|
||||
|
||||
import express from "express";
|
||||
import { mkdtempSync } from "node:fs";
|
||||
import { rm } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import { beforeEach, afterEach, describe, expect, it, vi } from "vitest";
|
||||
import { TaskStore } from "@fusion/core";
|
||||
import { createApiRoutes } from "../routes.js";
|
||||
import { request, get } from "../test-request.js";
|
||||
import { AiSessionStore, type AiSessionRow } from "../ai-session-store.js";
|
||||
import {
|
||||
__resetPlanningState,
|
||||
__setCreateKbAgent,
|
||||
createSession,
|
||||
submitResponse,
|
||||
setAiSessionStore as setPlanningAiSessionStore,
|
||||
} from "../planning.js";
|
||||
import {
|
||||
__resetSubtaskBreakdownState,
|
||||
createSubtaskSession,
|
||||
getSubtaskSession,
|
||||
setAiSessionStore as setSubtaskAiSessionStore,
|
||||
} from "../subtask-breakdown.js";
|
||||
import {
|
||||
__resetMissionInterviewState,
|
||||
createMissionInterviewSession,
|
||||
getMissionInterviewSession,
|
||||
submitMissionInterviewResponse,
|
||||
setAiSessionStore as setMissionAiSessionStore,
|
||||
} from "../mission-interview.js";
|
||||
|
||||
const { mockCreateKbAgent } = vi.hoisted(() => ({
|
||||
mockCreateKbAgent: vi.fn(),
|
||||
}));
|
||||
|
||||
vi.mock("@fusion/engine", () => ({
|
||||
createKbAgent: mockCreateKbAgent,
|
||||
}));
|
||||
|
||||
function makePlanningAgent(responses: string[]) {
|
||||
const messages: Array<{ role: string; content: string }> = [];
|
||||
let index = 0;
|
||||
return {
|
||||
session: {
|
||||
state: { messages },
|
||||
prompt: vi.fn(async (_input: string) => {
|
||||
const response = responses[index++] ?? responses[responses.length - 1] ?? responses[0] ?? "{}";
|
||||
messages.push({ role: "assistant", content: response });
|
||||
}),
|
||||
dispose: vi.fn(),
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
async function waitForCondition(check: () => boolean, timeoutMs = 2_000): Promise<void> {
|
||||
const start = Date.now();
|
||||
while (!check()) {
|
||||
if (Date.now() - start > timeoutMs) {
|
||||
throw new Error("Timed out waiting for condition");
|
||||
}
|
||||
await new Promise((resolve) => setTimeout(resolve, 10));
|
||||
}
|
||||
}
|
||||
|
||||
function extractEventId(body: string, eventName: string): number {
|
||||
const pattern = new RegExp(`id: (\\d+)\\nevent: ${eventName}`, "m");
|
||||
const match = body.match(pattern);
|
||||
if (!match) {
|
||||
throw new Error(`Missing event ${eventName} in body: ${body}`);
|
||||
}
|
||||
return Number.parseInt(match[1]!, 10);
|
||||
}
|
||||
|
||||
describe("session reconnect + replay", () => {
|
||||
let tmpRoot: string;
|
||||
let store: TaskStore;
|
||||
let aiSessionStore: AiSessionStore;
|
||||
let app: express.Express;
|
||||
|
||||
beforeEach(async () => {
|
||||
vi.clearAllMocks();
|
||||
__resetPlanningState();
|
||||
__resetSubtaskBreakdownState();
|
||||
__resetMissionInterviewState();
|
||||
|
||||
tmpRoot = mkdtempSync(join(tmpdir(), "kb-session-reconnect-"));
|
||||
store = new TaskStore(tmpRoot);
|
||||
await store.init();
|
||||
aiSessionStore = new AiSessionStore(store.getDatabase());
|
||||
|
||||
setPlanningAiSessionStore(aiSessionStore);
|
||||
setSubtaskAiSessionStore(aiSessionStore);
|
||||
setMissionAiSessionStore(aiSessionStore);
|
||||
|
||||
app = express();
|
||||
app.use(express.json());
|
||||
app.use("/api", createApiRoutes(store, { aiSessionStore }));
|
||||
});
|
||||
|
||||
afterEach(async () => {
|
||||
__setCreateKbAgent(undefined as any);
|
||||
__resetPlanningState();
|
||||
__resetSubtaskBreakdownState();
|
||||
__resetMissionInterviewState();
|
||||
|
||||
try {
|
||||
store.close();
|
||||
} catch {
|
||||
// no-op
|
||||
}
|
||||
await rm(tmpRoot, { recursive: true, force: true });
|
||||
});
|
||||
|
||||
it("replays planning buffered events and supports reconnect catch-up with lastEventId", async () => {
|
||||
const planningResponses = [
|
||||
JSON.stringify({
|
||||
type: "question",
|
||||
data: { id: "q-1", type: "text", question: "What scope?" },
|
||||
}),
|
||||
JSON.stringify({
|
||||
type: "question",
|
||||
data: { id: "q-2", type: "text", question: "Any constraints?" },
|
||||
}),
|
||||
JSON.stringify({
|
||||
type: "complete",
|
||||
data: {
|
||||
title: "Planning complete",
|
||||
description: "Done",
|
||||
suggestedSize: "M",
|
||||
suggestedDependencies: [],
|
||||
keyDeliverables: ["One", "Two"],
|
||||
},
|
||||
}),
|
||||
];
|
||||
|
||||
__setCreateKbAgent(async () => makePlanningAgent(planningResponses));
|
||||
|
||||
const { sessionId } = await createSession("127.0.0.11", "Build reconnect tests", undefined, "/tmp/project");
|
||||
await submitResponse(sessionId, { "q-1": "medium" }, "/tmp/project");
|
||||
await submitResponse(sessionId, { "q-2": "none" }, "/tmp/project");
|
||||
|
||||
const firstStream = await get(app, `/api/planning/${sessionId}/stream?lastEventId=0`);
|
||||
expect(firstStream.status).toBe(200);
|
||||
const firstBody = String(firstStream.body);
|
||||
expect(firstBody).toContain("event: summary");
|
||||
expect(firstBody).toContain("event: complete");
|
||||
|
||||
const completeEventId = extractEventId(firstBody, "complete");
|
||||
|
||||
const reconnect = await get(app, `/api/planning/${sessionId}/stream?lastEventId=${completeEventId}`);
|
||||
expect(reconnect.status).toBe(200);
|
||||
const reconnectBody = String(reconnect.body);
|
||||
expect(reconnectBody).toContain(": connected");
|
||||
expect(reconnectBody).not.toContain("event: summary");
|
||||
expect(reconnectBody).not.toContain("event: complete");
|
||||
});
|
||||
|
||||
it("replays subtask buffered events using Last-Event-ID header", async () => {
|
||||
mockCreateKbAgent.mockImplementation(async () =>
|
||||
makePlanningAgent([
|
||||
JSON.stringify({
|
||||
subtasks: [
|
||||
{
|
||||
id: "subtask-1",
|
||||
title: "First",
|
||||
description: "Do first",
|
||||
suggestedSize: "S",
|
||||
dependsOn: [],
|
||||
},
|
||||
],
|
||||
}),
|
||||
]),
|
||||
);
|
||||
|
||||
const session = await createSubtaskSession("Break this down", store, "/tmp/project");
|
||||
await waitForCondition(() => getSubtaskSession(session.sessionId)?.status === "complete");
|
||||
|
||||
const initial = await get(app, `/api/subtasks/${session.sessionId}/stream`);
|
||||
expect(initial.status).toBe(200);
|
||||
const initialBody = String(initial.body);
|
||||
expect(initialBody).toContain("event: subtasks");
|
||||
expect(initialBody).toContain("event: complete");
|
||||
|
||||
const subtasksEventId = extractEventId(initialBody, "subtasks");
|
||||
|
||||
const replay = await request(
|
||||
app,
|
||||
"GET",
|
||||
`/api/subtasks/${session.sessionId}/stream`,
|
||||
undefined,
|
||||
{ "Last-Event-ID": String(subtasksEventId) },
|
||||
);
|
||||
|
||||
expect(replay.status).toBe(200);
|
||||
const replayBody = String(replay.body);
|
||||
expect(replayBody).toContain("event: complete");
|
||||
expect(replayBody).not.toContain("event: subtasks");
|
||||
});
|
||||
|
||||
it("replays mission interview buffered events using query lastEventId", async () => {
|
||||
mockCreateKbAgent.mockImplementation(async () =>
|
||||
makePlanningAgent([
|
||||
JSON.stringify({
|
||||
type: "question",
|
||||
data: { id: "q-m-1", type: "text", question: "What is the mission?" },
|
||||
}),
|
||||
JSON.stringify({
|
||||
type: "complete",
|
||||
data: {
|
||||
missionTitle: "Mission summary",
|
||||
missionDescription: "Done",
|
||||
milestones: [
|
||||
{
|
||||
title: "Milestone 1",
|
||||
slices: [
|
||||
{
|
||||
title: "Slice 1",
|
||||
features: [{ title: "Feature 1", acceptanceCriteria: "Works" }],
|
||||
},
|
||||
],
|
||||
},
|
||||
],
|
||||
},
|
||||
}),
|
||||
]),
|
||||
);
|
||||
|
||||
const sessionId = await createMissionInterviewSession("127.0.0.22", "Mission reconnect", "/tmp/project");
|
||||
await waitForCondition(() => Boolean(getMissionInterviewSession(sessionId)?.currentQuestion));
|
||||
|
||||
await submitMissionInterviewResponse(sessionId, { "q-m-1": "Build the system" }, "/tmp/project");
|
||||
|
||||
const initial = await get(app, `/api/missions/interview/${sessionId}/stream`);
|
||||
expect(initial.status).toBe(200);
|
||||
const initialBody = String(initial.body);
|
||||
expect(initialBody).toContain("event: summary");
|
||||
expect(initialBody).toContain("event: complete");
|
||||
|
||||
const summaryEventId = extractEventId(initialBody, "summary");
|
||||
|
||||
const replay = await get(app, `/api/missions/interview/${sessionId}/stream?lastEventId=${summaryEventId}`);
|
||||
expect(replay.status).toBe(200);
|
||||
const replayBody = String(replay.body);
|
||||
expect(replayBody).toContain("event: complete");
|
||||
expect(replayBody).not.toContain("event: summary");
|
||||
});
|
||||
|
||||
it("accepts keep-alive ping touches via /api/ai-sessions/:id/ping", async () => {
|
||||
const sessionId = "ping-session-1";
|
||||
const stale = new Date(Date.now() - 90_000).toISOString();
|
||||
|
||||
const row: AiSessionRow = {
|
||||
id: sessionId,
|
||||
type: "planning",
|
||||
status: "awaiting_input",
|
||||
title: "Ping session",
|
||||
inputPayload: JSON.stringify({ initialPlan: "Ping" }),
|
||||
conversationHistory: "[]",
|
||||
currentQuestion: null,
|
||||
result: null,
|
||||
thinkingOutput: "",
|
||||
error: null,
|
||||
projectId: null,
|
||||
createdAt: stale,
|
||||
updatedAt: stale,
|
||||
lockedByTab: null,
|
||||
lockedAt: null,
|
||||
};
|
||||
|
||||
aiSessionStore.upsert(row);
|
||||
store.getDatabase().prepare("UPDATE ai_sessions SET updatedAt = ? WHERE id = ?").run(stale, sessionId);
|
||||
|
||||
const response = await request(app, "POST", `/api/ai-sessions/${sessionId}/ping`);
|
||||
expect(response.status).toBe(200);
|
||||
expect(response.body).toEqual({ ok: true });
|
||||
|
||||
const updated = aiSessionStore.get(sessionId);
|
||||
expect(updated).not.toBeNull();
|
||||
expect(Date.parse(updated!.updatedAt)).toBeGreaterThan(Date.parse(stale));
|
||||
});
|
||||
});
|
||||
298
packages/dashboard/src/__tests__/session-resume-history.test.ts
Normal file
298
packages/dashboard/src/__tests__/session-resume-history.test.ts
Normal file
@@ -0,0 +1,298 @@
|
||||
/**
|
||||
* Covers resume/restore behavior with persisted conversation history,
|
||||
* thinking output continuity, and fresh-session history initialization.
|
||||
*/
|
||||
|
||||
// @vitest-environment node
|
||||
|
||||
import { beforeEach, afterEach, describe, expect, it, vi } from "vitest";
|
||||
import { mkdtempSync } from "node:fs";
|
||||
import { rm } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import { Database } from "@fusion/core";
|
||||
import { AiSessionStore, type AiSessionRow } from "../ai-session-store.js";
|
||||
import {
|
||||
__resetPlanningState,
|
||||
__setCreateKbAgent,
|
||||
createSession,
|
||||
getSession,
|
||||
setAiSessionStore as setPlanningAiSessionStore,
|
||||
submitResponse,
|
||||
} from "../planning.js";
|
||||
import {
|
||||
__resetSubtaskBreakdownState,
|
||||
getSubtaskSession,
|
||||
setAiSessionStore as setSubtaskAiSessionStore,
|
||||
} from "../subtask-breakdown.js";
|
||||
import {
|
||||
__resetMissionInterviewState,
|
||||
createMissionInterviewSession,
|
||||
getMissionInterviewSession,
|
||||
setAiSessionStore as setMissionAiSessionStore,
|
||||
submitMissionInterviewResponse,
|
||||
} from "../mission-interview.js";
|
||||
|
||||
const { mockCreateKbAgent } = vi.hoisted(() => ({
|
||||
mockCreateKbAgent: vi.fn(),
|
||||
}));
|
||||
|
||||
vi.mock("@fusion/engine", () => ({
|
||||
createKbAgent: mockCreateKbAgent,
|
||||
}));
|
||||
|
||||
function makeTmpDir(): string {
|
||||
return mkdtempSync(join(tmpdir(), "kb-session-resume-history-"));
|
||||
}
|
||||
|
||||
function createMockAgent(responses: string[]) {
|
||||
const queue = [...responses];
|
||||
const messages: Array<{ role: string; content: string }> = [];
|
||||
|
||||
return {
|
||||
session: {
|
||||
state: { messages },
|
||||
prompt: vi.fn(async (_input: string) => {
|
||||
const response = queue.shift() ?? queue[queue.length - 1] ?? "{}";
|
||||
messages.push({ role: "assistant", content: response });
|
||||
}),
|
||||
dispose: vi.fn(),
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
async function waitFor(check: () => boolean, timeoutMs = 2000): Promise<void> {
|
||||
const start = Date.now();
|
||||
while (!check()) {
|
||||
if (Date.now() - start > timeoutMs) {
|
||||
throw new Error("Timed out waiting for condition");
|
||||
}
|
||||
await new Promise((resolve) => setTimeout(resolve, 10));
|
||||
}
|
||||
}
|
||||
|
||||
describe("session resume + history restore", () => {
|
||||
let tmpDir: string;
|
||||
let db: Database;
|
||||
let aiSessionStore: AiSessionStore;
|
||||
|
||||
beforeEach(() => {
|
||||
vi.clearAllMocks();
|
||||
__resetPlanningState();
|
||||
__resetSubtaskBreakdownState();
|
||||
__resetMissionInterviewState();
|
||||
|
||||
tmpDir = makeTmpDir();
|
||||
db = new Database(join(tmpDir, ".fusion"));
|
||||
db.init();
|
||||
aiSessionStore = new AiSessionStore(db);
|
||||
|
||||
setPlanningAiSessionStore(aiSessionStore);
|
||||
setSubtaskAiSessionStore(aiSessionStore);
|
||||
setMissionAiSessionStore(aiSessionStore);
|
||||
});
|
||||
|
||||
afterEach(async () => {
|
||||
__setCreateKbAgent(undefined as any);
|
||||
__resetPlanningState();
|
||||
__resetSubtaskBreakdownState();
|
||||
__resetMissionInterviewState();
|
||||
|
||||
try {
|
||||
db.close();
|
||||
} catch {
|
||||
// no-op
|
||||
}
|
||||
|
||||
await rm(tmpDir, { recursive: true, force: true });
|
||||
});
|
||||
|
||||
it("restores planning history/thinking from SQLite and resumes with replayed context", async () => {
|
||||
const now = new Date().toISOString();
|
||||
const row: AiSessionRow = {
|
||||
id: "planning-resume-1",
|
||||
type: "planning",
|
||||
status: "awaiting_input",
|
||||
title: "Planning resume",
|
||||
inputPayload: JSON.stringify({ ip: "127.0.0.1", initialPlan: "Resume planning" }),
|
||||
conversationHistory: JSON.stringify([
|
||||
{
|
||||
question: { id: "q-1", type: "text", question: "What are we building?" },
|
||||
response: { "q-1": "A resume flow" },
|
||||
thinkingOutput: "turn-1-thinking",
|
||||
},
|
||||
]),
|
||||
currentQuestion: JSON.stringify({ id: "q-2", type: "text", question: "Any constraints?" }),
|
||||
result: null,
|
||||
thinkingOutput: "latest-thinking",
|
||||
error: null,
|
||||
projectId: null,
|
||||
createdAt: now,
|
||||
updatedAt: now,
|
||||
lockedByTab: null,
|
||||
lockedAt: null,
|
||||
};
|
||||
aiSessionStore.upsert(row);
|
||||
|
||||
const resumedAgent = createMockAgent([
|
||||
"context-ack",
|
||||
JSON.stringify({
|
||||
type: "question",
|
||||
data: { id: "q-3", type: "text", question: "What timeline?" },
|
||||
}),
|
||||
]);
|
||||
const createKbAgentSpy = vi.fn(async () => resumedAgent);
|
||||
__setCreateKbAgent(createKbAgentSpy as any);
|
||||
|
||||
const restored = getSession(row.id);
|
||||
expect(restored).toBeDefined();
|
||||
expect(restored?.history).toHaveLength(1);
|
||||
expect(restored?.history[0]?.thinkingOutput).toBe("turn-1-thinking");
|
||||
expect(restored?.thinkingOutput).toBe("latest-thinking");
|
||||
expect(restored?.lastGeneratedThinking).toBe("latest-thinking");
|
||||
|
||||
const response = await submitResponse(row.id, { "q-2": "No constraints" }, "/tmp/project");
|
||||
expect(response.type).toBe("question");
|
||||
if (response.type === "question") {
|
||||
expect(response.data.id).toBe("q-3");
|
||||
}
|
||||
|
||||
expect(createKbAgentSpy).toHaveBeenCalledTimes(1);
|
||||
expect(resumedAgent.session.prompt).toHaveBeenCalledTimes(2);
|
||||
expect(resumedAgent.session.prompt.mock.calls[0]?.[0]).toContain("Previous conversation summary");
|
||||
expect(resumedAgent.session.prompt.mock.calls[1]?.[0]).toContain("Any constraints?");
|
||||
});
|
||||
|
||||
it("restores mission interview history/thinking and resumes with replayed context", async () => {
|
||||
const now = new Date().toISOString();
|
||||
const row: AiSessionRow = {
|
||||
id: "mission-resume-1",
|
||||
type: "mission_interview",
|
||||
status: "awaiting_input",
|
||||
title: "Mission resume",
|
||||
inputPayload: JSON.stringify({ ip: "127.0.0.1", missionId: "M-1", missionTitle: "Mission title" }),
|
||||
conversationHistory: JSON.stringify([
|
||||
{
|
||||
question: { id: "q-m-1", type: "text", question: "What is the mission?" },
|
||||
response: { "q-m-1": "Ship a dashboard" },
|
||||
thinkingOutput: "mission-turn-1-thinking",
|
||||
},
|
||||
]),
|
||||
currentQuestion: JSON.stringify({ id: "q-m-2", type: "text", question: "Any technical constraints?" }),
|
||||
result: null,
|
||||
thinkingOutput: "mission-latest-thinking",
|
||||
error: null,
|
||||
projectId: null,
|
||||
createdAt: now,
|
||||
updatedAt: now,
|
||||
lockedByTab: null,
|
||||
lockedAt: null,
|
||||
};
|
||||
aiSessionStore.upsert(row);
|
||||
|
||||
const resumedAgent = createMockAgent([
|
||||
"context-ack",
|
||||
JSON.stringify({
|
||||
type: "question",
|
||||
data: { id: "q-m-3", type: "text", question: "Who are the users?" },
|
||||
}),
|
||||
]);
|
||||
const createKbAgentSpy = vi.fn(async () => resumedAgent);
|
||||
mockCreateKbAgent.mockImplementation(createKbAgentSpy);
|
||||
|
||||
const restored = getMissionInterviewSession(row.id);
|
||||
expect(restored).toBeDefined();
|
||||
expect(restored?.history).toHaveLength(1);
|
||||
expect(restored?.history[0]?.thinkingOutput).toBe("mission-turn-1-thinking");
|
||||
expect(restored?.thinkingOutput).toBe("mission-latest-thinking");
|
||||
expect(restored?.lastGeneratedThinking).toBe("mission-latest-thinking");
|
||||
|
||||
const response = await submitMissionInterviewResponse(
|
||||
row.id,
|
||||
{ "q-m-2": "None" },
|
||||
"/tmp/project",
|
||||
);
|
||||
|
||||
expect(response.type).toBe("question");
|
||||
if (response.type === "question") {
|
||||
expect(response.data.id).toBe("q-m-3");
|
||||
}
|
||||
|
||||
expect(createKbAgentSpy).toHaveBeenCalledTimes(1);
|
||||
expect(resumedAgent.session.prompt).toHaveBeenCalledTimes(2);
|
||||
expect(resumedAgent.session.prompt.mock.calls[0]?.[0]).toContain("Previous conversation summary");
|
||||
expect(resumedAgent.session.prompt.mock.calls[1]?.[0]).toContain("Any technical constraints?");
|
||||
});
|
||||
|
||||
it("restores persisted subtask session state from SQLite", () => {
|
||||
const now = new Date().toISOString();
|
||||
const subtasks = [
|
||||
{
|
||||
id: "subtask-1",
|
||||
title: "Analyze",
|
||||
description: "Analyze requirements",
|
||||
suggestedSize: "S",
|
||||
dependsOn: [],
|
||||
},
|
||||
];
|
||||
|
||||
const row: AiSessionRow = {
|
||||
id: "subtask-resume-1",
|
||||
type: "subtask",
|
||||
status: "generating",
|
||||
title: "Subtask resume",
|
||||
inputPayload: JSON.stringify({ initialDescription: "Break down this task" }),
|
||||
conversationHistory: JSON.stringify([{ thinkingOutput: "subtask-thinking" }]),
|
||||
currentQuestion: null,
|
||||
result: JSON.stringify(subtasks),
|
||||
thinkingOutput: "subtask-latest-thinking",
|
||||
error: null,
|
||||
projectId: null,
|
||||
createdAt: now,
|
||||
updatedAt: now,
|
||||
lockedByTab: null,
|
||||
lockedAt: null,
|
||||
};
|
||||
aiSessionStore.upsert(row);
|
||||
|
||||
const restored = getSubtaskSession(row.id);
|
||||
expect(restored).toBeDefined();
|
||||
expect(restored?.sessionId).toBe(row.id);
|
||||
expect(restored?.status).toBe("generating");
|
||||
expect(restored?.subtasks).toEqual(subtasks);
|
||||
|
||||
const persistedRow = aiSessionStore.get(row.id);
|
||||
expect(JSON.parse(persistedRow?.conversationHistory ?? "[]")).toEqual([
|
||||
{ thinkingOutput: "subtask-thinking" },
|
||||
]);
|
||||
});
|
||||
|
||||
it("starts fresh planning and mission sessions with empty history", async () => {
|
||||
__setCreateKbAgent(
|
||||
async () =>
|
||||
createMockAgent([
|
||||
JSON.stringify({
|
||||
type: "question",
|
||||
data: { id: "q-fresh-plan", type: "text", question: "Plan question" },
|
||||
}),
|
||||
]),
|
||||
);
|
||||
|
||||
mockCreateKbAgent.mockImplementation(async () =>
|
||||
createMockAgent([
|
||||
JSON.stringify({
|
||||
type: "question",
|
||||
data: { id: "q-fresh-mission", type: "text", question: "Mission question" },
|
||||
}),
|
||||
]),
|
||||
);
|
||||
|
||||
const planning = await createSession("127.0.0.88", "Fresh planning", undefined, "/tmp/project");
|
||||
const missionSessionId = await createMissionInterviewSession("127.0.0.89", "Fresh mission", "/tmp/project");
|
||||
|
||||
await waitFor(() => Boolean(getMissionInterviewSession(missionSessionId)?.currentQuestion));
|
||||
|
||||
expect(getSession(planning.sessionId)?.history).toEqual([]);
|
||||
expect(getMissionInterviewSession(missionSessionId)?.history).toEqual([]);
|
||||
});
|
||||
});
|
||||
Reference in New Issue
Block a user