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

fix(FN-732): fix dashboard real-time updates and SSE pipeline

Browse Source 
- Fix SSE event relay to properly broadcast task store events to dashboard clients
- Use named heartbeat events instead of SSE comments for reliable keep-alive
- Add missing event emission in core task store for state changes
- Add comprehensive tests for SSE pipeline, event emission, and UI hooks
- Remove broken useTerminal hook and AgentLogViewer tests, fix flaky test suites
This commit is contained in:
gsxdsm
2026-04-02 19:34:35 -07:00
parent 24d04ca63a
commit 74c0f8fdc2
7 changed files with 537 additions and 112 deletions
Download Patch File Download Diff File Expand all files Collapse all files

83
packages/core/src/store.test.ts
View File

@@ -4388,4 +4388,87 @@ Task with acceptance criteria
expect(detail.prompt).toMatch(/^# FN-\d+: Generated Task Title\n/);
});
});
describe("event emissions", () => {
it("createTask emits task:created with the new task", async () => {
const events: any[] = [];
store.on("task:created", (t: any) => events.push(t));
const task = await store.createTask({ description: "event test" });
expect(events).toHaveLength(1);
expect(events[0].id).toBe(task.id);
expect(events[0].description).toBe("event test");
});
it("moveTask emits task:moved with from/to columns", async () => {
const task = await createTestTask();
const events: any[] = [];
store.on("task:moved", (data: any) => events.push(data));
await store.moveTask(task.id, "todo");
expect(events).toHaveLength(1);
expect(events[0].from).toBe("triage");
expect(events[0].to).toBe("todo");
expect(events[0].task.id).toBe(task.id);
});
it("updateTask emits task:updated with the updated task", async () => {
const task = await createTestTask();
const events: any[] = [];
store.on("task:updated", (t: any) => events.push(t));
await store.updateTask(task.id, { title: "Updated" });
expect(events.length).toBeGreaterThanOrEqual(1);
expect(events.some((e: any) => e.title === "Updated")).toBe(true);
});
it("pauseTask emits task:updated", async () => {
const task = await createTestTask();
await store.moveTask(task.id, "todo");
const events: any[] = [];
store.on("task:updated", (t: any) => events.push(t));
await store.pauseTask(task.id, true);
expect(events.length).toBeGreaterThanOrEqual(1);
expect(events.some((e: any) => e.paused === true)).toBe(true);
});
it("updateStep emits task:updated", async () => {
const task = await createTaskWithSteps();
await store.moveTask(task.id, "todo");
const events: any[] = [];
store.on("task:updated", (t: any) => events.push(t));
await store.updateStep(task.id, 0, "in-progress");
expect(events.length).toBeGreaterThanOrEqual(1);
});
it("deleteTask emits task:deleted", async () => {
const task = await createTestTask();
const events: any[] = [];
store.on("task:deleted", (t: any) => events.push(t));
await store.deleteTask(task.id);
expect(events).toHaveLength(1);
expect(events[0].id).toBe(task.id);
});
it("logEntry emits task:updated", async () => {
const task = await createTestTask();
const events: any[] = [];
store.on("task:updated", (t: any) => events.push(t));
await store.logEntry(task.id, "test action", "test outcome");
expect(events.length).toBeGreaterThanOrEqual(1);
});
it("cache is updated when polling is active even without fs.watch", async () => {
// Start watching which enables polling
await store.watch();
const task = await createTestTask();
// Verify the cache was updated (internal detail: the isWatching getter)
// Move the task and verify the event fires correctly (not duplicated by poll)
const movedEvents: any[] = [];
store.on("task:moved", (data: any) => movedEvents.push(data));
await store.moveTask(task.id, "todo");
expect(movedEvents).toHaveLength(1);
expect(movedEvents[0].from).toBe("triage");
expect(movedEvents[0].to).toBe("todo");
});
});
});

49
packages/core/src/store.ts
View File

@@ -87,6 +87,11 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
private pollInterval: ReturnType<typeof setInterval> | null = null;
/** Last known modification timestamp for change detection */
private lastKnownModified: number = 0;
/** Whether the store is actively watching for changes (watcher or polling). */
private get isWatching(): boolean {
return this.watcher !== null || this.pollInterval !== null;
}
/** Cached MissionStore instance */
private missionStore: MissionStore | null = null;
@@ -697,7 +702,7 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
await this.atomicWriteTaskJson(dir, task);
// Update cache if watcher is active
if (this.watcher) this.taskCache.set(id, { ...task });
if (this.isWatching) this.taskCache.set(id, { ...task });
const heading = task.title ? `${id}: ${task.title}` : id;
const prompt = task.column === "triage"
@@ -751,7 +756,7 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
await writeFile(join(newDir, "PROMPT.md"), sourcePrompt);
// Update cache if watcher is active
if (this.watcher) this.taskCache.set(newId, { ...newTask });
if (this.isWatching) this.taskCache.set(newId, { ...newTask });
this.emit("task:created", newTask);
return newTask;
@@ -826,7 +831,7 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
}
// Update cache if watcher is active
if (this.watcher) this.taskCache.set(newId, { ...newTask });
if (this.isWatching) this.taskCache.set(newId, { ...newTask });
this.emit("task:created", newTask);
return newTask;
@@ -922,7 +927,7 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
await this.atomicWriteTaskJson(dir, task);
// Update cache if watcher is active
if (this.watcher) this.taskCache.set(id, { ...task });
if (this.isWatching) this.taskCache.set(id, { ...task });
this.emit("task:moved", { task, from: fromColumn, to: toColumn });
return task;
@@ -1041,7 +1046,7 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
await this.atomicWriteTaskJson(dir, task);
// Update cache if watcher is active
if (this.watcher) this.taskCache.set(id, { ...task });
if (this.isWatching) this.taskCache.set(id, { ...task });
if (updates.prompt !== undefined) {
await mkdir(dir, { recursive: true });
@@ -1104,7 +1109,7 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
});
await this.atomicWriteTaskJson(dir, task);
if (this.watcher) this.taskCache.set(id, { ...task });
if (this.isWatching) this.taskCache.set(id, { ...task });
this.emit("task:updated", task);
return task;
@@ -1161,7 +1166,7 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
});
await this.atomicWriteTaskJson(dir, task);
if (this.watcher) this.taskCache.set(id, { ...task });
if (this.isWatching) this.taskCache.set(id, { ...task });
this.emit("task:updated", task);
return task;
@@ -1189,7 +1194,7 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
task.updatedAt = new Date().toISOString();
await this.atomicWriteTaskJson(dir, task);
if (this.watcher) this.taskCache.set(id, { ...task });
if (this.isWatching) this.taskCache.set(id, { ...task });
this.emit("task:updated", task);
return task;
@@ -1296,7 +1301,7 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
this.db.bumpLastModified();
// Remove from cache if watcher is active
if (this.watcher) this.taskCache.delete(id);
if (this.isWatching) this.taskCache.delete(id);
// Delete directory from disk
const dir = this.taskDir(id);
@@ -1579,7 +1584,7 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
await rm(dir, { recursive: true, force: true });
// Remove from cache if watcher is active
if (this.watcher) {
if (this.isWatching) {
this.taskCache.delete(id);
}
} else {
@@ -1587,7 +1592,7 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
await this.atomicWriteTaskJson(dir, task);
// Update cache if watcher is active
if (this.watcher) this.taskCache.set(id, { ...task });
if (this.isWatching) this.taskCache.set(id, { ...task });
}
this.emit("task:moved", { task, from: "done" as Column, to: "archived" as Column });
@@ -1651,7 +1656,7 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
await this.atomicWriteTaskJson(dir, task);
// Update cache if watcher is active
if (this.watcher) this.taskCache.set(id, { ...task });
if (this.isWatching) this.taskCache.set(id, { ...task });
this.emit("task:moved", { task, from: "archived" as Column, to: "done" as Column });
return task;
@@ -1669,7 +1674,7 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
await this.atomicWriteTaskJson(dir, task);
// Update cache if watcher is active
if (this.watcher) this.taskCache.set(task.id, { ...task });
if (this.isWatching) this.taskCache.set(task.id, { ...task });
this.emit("task:moved", { task, from: "in-review" as Column, to: "done" as Column });
}
@@ -1924,7 +1929,7 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
task.updatedAt = new Date().toISOString();
await this.atomicWriteTaskJson(dir, task);
if (this.watcher) this.taskCache.set(id, { ...task });
if (this.isWatching) this.taskCache.set(id, { ...task });
this.emit("task:updated", task);
return attachment;
@@ -1979,7 +1984,7 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
task.updatedAt = new Date().toISOString();
await this.atomicWriteTaskJson(dir, task);
if (this.watcher) this.taskCache.set(id, { ...task });
if (this.isWatching) this.taskCache.set(id, { ...task });
this.emit("task:updated", task);
return task;
@@ -2054,7 +2059,7 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
currentTask.updatedAt = new Date().toISOString();
await this.atomicWriteTaskJson(dir, currentTask);
if (this.watcher) this.taskCache.set(id, { ...currentTask });
if (this.isWatching) this.taskCache.set(id, { ...currentTask });
this.emit("task:updated", currentTask);
return currentTask;
@@ -2084,7 +2089,7 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
});
await this.atomicWriteTaskJson(dir, task);
if (this.watcher) this.taskCache.set(id, { ...task });
if (this.isWatching) this.taskCache.set(id, { ...task });
this.emit("task:updated", task);
return task;
@@ -2110,7 +2115,7 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
});
await this.atomicWriteTaskJson(dir, task);
if (this.watcher) this.taskCache.set(id, { ...task });
if (this.isWatching) this.taskCache.set(id, { ...task });
this.emit("task:updated", task);
return task;
@@ -2163,7 +2168,7 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
});
await this.atomicWriteTaskJson(dir, task);
if (this.watcher) this.taskCache.set(id, { ...task });
if (this.isWatching) this.taskCache.set(id, { ...task });
this.emit("task:updated", task);
return task;
@@ -2242,7 +2247,7 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
task.updatedAt = new Date().toISOString();
await this.atomicWriteTaskJson(dir, task);
if (this.watcher) this.taskCache.set(id, { ...task });
if (this.isWatching) this.taskCache.set(id, { ...task });
if (badgeChanged) {
this.emit("task:updated", task);
@@ -2306,7 +2311,7 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
task.updatedAt = new Date().toISOString();
await this.atomicWriteTaskJson(dir, task);
if (this.watcher) this.taskCache.set(id, { ...task });
if (this.isWatching) this.taskCache.set(id, { ...task });
if (badgeChanged) {
this.emit("task:updated", task);
@@ -2425,7 +2430,7 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
await rm(dir, { recursive: true, force: true });
// Remove from cache if watcher is active
if (this.watcher) {
if (this.isWatching) {
this.taskCache.delete(task.id);
}