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

feat(FN-1145): harden dashboard SSE streams with replay and reconnect

Browse Source 
- Add a shared SSE event buffer utility and wire it into planning, subtask, mission interview, and task stream routes
- Support Last-Event-ID replay semantics and robust event serialization so clients can recover missed stream events
- Add a resilient client-side reconnect wrapper and surface reconnecting state in planning, subtask breakdown, and mission interview modals
- Expand test coverage across API routes, SSE buffering behavior, reconnect handling, and mission/planning stream flows
This commit is contained in:
gsxdsm
2026-04-08 06:02:28 -07:00
parent 586d9b33b1
commit b6fbadec4d
20 changed files with 1682 additions and 296 deletions
Download Patch File Download Diff File Expand all files Collapse all files

69
packages/dashboard/src/__tests__/sse-buffer.test.ts Normal file
View File

@@ -0,0 +1,69 @@
import { describe, expect, it } from "vitest";
import { SessionEventBuffer } from "../sse-buffer.js";
describe("SessionEventBuffer", () => {
it("push assigns monotonically increasing ids", () => {
const buffer = new SessionEventBuffer(10);
const id1 = buffer.push("thinking", JSON.stringify("a"));
const id2 = buffer.push("question", JSON.stringify({ id: "q-1" }));
expect(id1).toBe(1);
expect(id2).toBe(2);
});
it("getEventsSince returns only events newer than lastEventId", () => {
const buffer = new SessionEventBuffer(10);
buffer.push("thinking", JSON.stringify("a"));
buffer.push("thinking", JSON.stringify("b"));
buffer.push("question", JSON.stringify({ id: "q-1" }));
const events = buffer.getEventsSince(1);
expect(events.map((event) => event.id)).toEqual([2, 3]);
});
it("drops oldest events when capacity overflows", () => {
const buffer = new SessionEventBuffer(2);
buffer.push("thinking", JSON.stringify("a"));
buffer.push("thinking", JSON.stringify("b"));
buffer.push("question", JSON.stringify({ id: "q-1" }));
const events = buffer.getEventsSince(0);
expect(events).toHaveLength(2);
expect(events[0]?.id).toBe(2);
expect(events[1]?.id).toBe(3);
});
it("returns empty array for an empty buffer", () => {
const buffer = new SessionEventBuffer(10);
expect(buffer.getEventsSince(0)).toEqual([]);
});
it("returns all buffered events for non-finite lastEventId", () => {
const buffer = new SessionEventBuffer(10);
buffer.push("thinking", JSON.stringify("a"));
buffer.push("complete", JSON.stringify({}));
expect(buffer.getEventsSince(Number.NaN)).toHaveLength(2);
});
it("supports interleaved push/read access without duplicate ids", async () => {
const buffer = new SessionEventBuffer(20);
await Promise.all(
Array.from({ length: 10 }, (_, index) =>
Promise.resolve().then(() => {
const eventId = buffer.push("thinking", JSON.stringify(`event-${index + 1}`));
const snapshot = buffer.getEventsSince(Math.max(0, eventId - 1));
expect(snapshot[snapshot.length - 1]?.id).toBe(eventId);
}),
),
);
const events = buffer.getEventsSince(0);
expect(events).toHaveLength(10);
expect(events.map((event) => event.id)).toEqual([1, 2, 3, 4, 5, 6, 7, 8, 9, 10]);
});
});

84
packages/dashboard/src/mission-e2e.test.ts
View File

@@ -7,7 +7,7 @@
// @vitest-environment node
import { describe, it, expect, vi } from "vitest";
import { beforeEach, describe, expect, it, vi } from "vitest";
import express from "express";
import { createMissionRouter } from "./mission-routes.js";
import { request, get } from "./test-request.js";
@@ -19,6 +19,11 @@ import type {
MissionFeature,
MissionWithHierarchy,
} from "@fusion/core";
import {
__resetMissionInterviewState,
createMissionInterviewSession,
missionInterviewStreamManager,
} from "./mission-interview.js";
// Mock MissionStore factory
function createMockMissionStore() {
@@ -701,6 +706,10 @@ describe("Mission API", () => {
});
describe("Interview endpoints", () => {
beforeEach(() => {
__resetMissionInterviewState();
});
it("should return 400 when missionTitle is missing on interview start", async () => {
const { app } = buildApp();
const res = await request(
@@ -740,6 +749,79 @@ describe("Mission API", () => {
expect(res.body.error).toContain("sessionId");
});
it("replays buffered interview events when Last-Event-ID is provided", async () => {
const { app } = buildApp();
const sessionId = await createMissionInterviewSession("127.0.0.1", "Replay Mission", "/tmp/project");
missionInterviewStreamManager.broadcast(sessionId, { type: "thinking", data: "first" });
missionInterviewStreamManager.broadcast(sessionId, { type: "thinking", data: "second" });
setTimeout(() => {
missionInterviewStreamManager.broadcast(sessionId, { type: "complete" });
}, 0);
const res = await request(
app,
"GET",
`/api/missions/interview/${sessionId}/stream`,
undefined,
{ "last-event-id": "1" },
);
expect(res.status).toBe(200);
expect(res.body).toContain("id: 2");
expect(res.body).toContain("event: thinking");
expect(res.body).toContain("id: 3");
expect(res.body).toContain("event: complete");
expect(res.body).not.toContain("id: 1\nevent: thinking");
});
it("does not replay buffered interview events when Last-Event-ID is missing", async () => {
const { app } = buildApp();
const sessionId = await createMissionInterviewSession("127.0.0.1", "No Replay Mission", "/tmp/project");
missionInterviewStreamManager.broadcast(sessionId, { type: "thinking", data: "first" });
setTimeout(() => {
missionInterviewStreamManager.broadcast(sessionId, { type: "complete" });
}, 0);
const res = await request(
app,
"GET",
`/api/missions/interview/${sessionId}/stream`,
);
expect(res.status).toBe(200);
expect(res.body).not.toContain("id: 1\nevent: thinking");
expect(res.body).toContain("id: 2");
expect(res.body).toContain("event: complete");
});
it("gracefully ignores invalid Last-Event-ID values for interview streams", async () => {
const { app } = buildApp();
const sessionId = await createMissionInterviewSession("127.0.0.1", "Invalid Replay Mission", "/tmp/project");
missionInterviewStreamManager.broadcast(sessionId, { type: "thinking", data: "first" });
setTimeout(() => {
missionInterviewStreamManager.broadcast(sessionId, { type: "complete" });
}, 0);
const res = await request(
app,
"GET",
`/api/missions/interview/${sessionId}/stream`,
undefined,
{ "last-event-id": "not-a-number" },
);
expect(res.status).toBe(200);
expect(res.body).not.toContain("id: 1\nevent: thinking");
expect(res.body).toContain("id: 2");
expect(res.body).toContain("event: complete");
});
it("should return 400 when sessionId is missing on create-mission", async () => {
const { app } = buildApp();
const res = await request(

27
packages/dashboard/src/mission-interview.test.ts
View File

@@ -183,8 +183,9 @@ describe("mission-interview module", () => {
expect(missionInterviewStreamManager.hasSubscribers("session-1")).toBe(true);
missionInterviewStreamManager.broadcast("session-1", { type: "thinking", data: "analyzing" });
expect(callback).toHaveBeenCalledWith({ type: "thinking", data: "analyzing" });
const eventId = missionInterviewStreamManager.broadcast("session-1", { type: "thinking", data: "analyzing" });
expect(eventId).toBe(1);
expect(callback).toHaveBeenCalledWith({ type: "thinking", data: "analyzing" }, 1);
unsubscribe();
expect(missionInterviewStreamManager.hasSubscribers("session-1")).toBe(false);
@@ -192,6 +193,28 @@ describe("mission-interview module", () => {
missionInterviewStreamManager.cleanupSession("session-1");
expect(missionInterviewStreamManager.hasSubscribers("session-1")).toBe(false);
});
it("returns buffered events since last event id", () => {
const sessionId = "session-buffered";
missionInterviewStreamManager.broadcast(sessionId, { type: "thinking", data: "delta-1" });
missionInterviewStreamManager.broadcast(sessionId, { type: "thinking", data: "delta-2" });
missionInterviewStreamManager.broadcast(sessionId, { type: "complete" });
const buffered = missionInterviewStreamManager.getBufferedEvents(sessionId, 1);
expect(buffered).toHaveLength(2);
expect(buffered.map((event) => event.id)).toEqual([2, 3]);
expect(buffered[1]).toMatchObject({ event: "complete", data: "{}" });
});
it("clears buffered events on cleanup", () => {
const sessionId = "session-cleanup";
missionInterviewStreamManager.broadcast(sessionId, { type: "thinking", data: "delta" });
expect(missionInterviewStreamManager.getBufferedEvents(sessionId, 0)).toHaveLength(1);
missionInterviewStreamManager.cleanupSession(sessionId);
expect(missionInterviewStreamManager.getBufferedEvents(sessionId, 0)).toEqual([]);
});
});
describe("response parsing", () => {

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

@@ -18,6 +18,7 @@ import type { PlanningQuestion } from "@fusion/core";
import { randomUUID } from "node:crypto";
import { EventEmitter } from "node:events";
import type { AiSessionStore, AiSessionRow } from "./ai-session-store.js";
import { SessionEventBuffer, type SessionBufferedEvent } from "./sse-buffer.js";
// Dynamic import for @fusion/engine to avoid resolution issues in test environment
// eslint-disable-next-line @typescript-eslint/consistent-type-imports, @typescript-eslint/no-explicit-any
@@ -151,7 +152,7 @@ export type MissionInterviewStreamEvent =
| { type: "complete" };
/** Callback function for streaming events */
export type MissionInterviewStreamCallback = (event: MissionInterviewStreamEvent) => void;
export type MissionInterviewStreamCallback = (event: MissionInterviewStreamEvent, eventId?: number) => void;
/** In-memory interview session */
interface MissionInterviewSession {
@@ -242,7 +243,12 @@ process.on("beforeExit", () => clearInterval(cleanupInterval));
// ── Stream Manager ──────────────────────────────────────────────────────────
export class MissionInterviewStreamManager extends EventEmitter {
private sessions = new Map<string, Set<MissionInterviewStreamCallback>>();
private readonly sessions = new Map<string, Set<MissionInterviewStreamCallback>>();
private readonly buffers = new Map<string, SessionEventBuffer>();
constructor(private readonly bufferSize = 100) {
super();
}
subscribe(sessionId: string, callback: MissionInterviewStreamCallback): () => void {
if (!this.sessions.has(sessionId)) {
@@ -258,16 +264,38 @@ export class MissionInterviewStreamManager extends EventEmitter {
};
}
broadcast(sessionId: string, event: MissionInterviewStreamEvent): void {
private getBuffer(sessionId: string): SessionEventBuffer {
let buffer = this.buffers.get(sessionId);
if (!buffer) {
buffer = new SessionEventBuffer(this.bufferSize);
this.buffers.set(sessionId, buffer);
}
return buffer;
}
broadcast(sessionId: string, event: MissionInterviewStreamEvent): number {
const serialized = JSON.stringify((event as { data?: unknown }).data ?? {});
const eventData = typeof serialized === "string" ? serialized : "{}";
const eventId = this.getBuffer(sessionId).push(event.type, eventData);
const callbacks = this.sessions.get(sessionId);
if (!callbacks) return;
if (!callbacks) return eventId;
for (const callback of callbacks) {
try {
callback(event);
callback(event, eventId);
} catch (err) {
console.error(`[mission-interview] Error broadcasting to client for session ${sessionId}:`, err);
}
}
return eventId;
}
getBufferedEvents(sessionId: string, sinceId: number): SessionBufferedEvent[] {
const buffer = this.buffers.get(sessionId);
if (!buffer) return [];
return buffer.getEventsSince(sinceId);
}
hasSubscribers(sessionId: string): boolean {
@@ -277,6 +305,13 @@ export class MissionInterviewStreamManager extends EventEmitter {
cleanupSession(sessionId: string): void {
this.sessions.delete(sessionId);
this.buffers.delete(sessionId);
}
reset(): void {
this.sessions.clear();
this.buffers.clear();
this.removeAllListeners();
}
}
@@ -807,7 +842,7 @@ export function __resetMissionInterviewState(): void {
}
sessions.clear();
rateLimits.clear();
missionInterviewStreamManager.removeAllListeners();
missionInterviewStreamManager.reset();
}
// ── Custom Errors ───────────────────────────────────────────────────────────

92
packages/dashboard/src/mission-routes.ts
View File

@@ -39,6 +39,7 @@ import {
FEATURE_STATUSES,
INTERVIEW_STATES,
} from "@fusion/core";
import { writeSSEEvent } from "./sse-buffer.js";
// ── Validation Utilities ────────────────────────────────────────────────────
@@ -141,6 +142,34 @@ function asyncHandler(fn: (req: TypedRequest, res: Response, next: NextFunction)
// ── Router Factory ──────────────────────────────────────────────────────────
function parseLastEventId(req: Request): number | undefined {
const rawHeader = req.headers["last-event-id"];
const rawQuery = req.query.lastEventId;
const raw = Array.isArray(rawHeader)
? rawHeader[0]
: (typeof rawHeader === "string" ? rawHeader : Array.isArray(rawQuery) ? rawQuery[0] : rawQuery);
if (raw === undefined || raw === null) return undefined;
const parsed = Number.parseInt(String(raw), 10);
if (!Number.isFinite(parsed) || parsed < 0) return undefined;
return parsed;
}
function replayBufferedSSE(
res: Response,
bufferedEvents: Array<{ id: number; event: string; data: string }>,
): boolean {
for (const bufferedEvent of bufferedEvents) {
if (!writeSSEEvent(res, bufferedEvent.event, bufferedEvent.data, bufferedEvent.id)) {
return false;
}
}
return true;
}
export function createMissionRouter(
store: TaskStore,
missionAutopilot?: {
@@ -407,25 +436,60 @@ export function createMissionRouter(
// Verify session exists
const session = getMissionInterviewSession(sessionId);
if (!session) {
res.write(`event: error\ndata: ${JSON.stringify({ message: "Session not found or expired" })}\n\n`);
writeSSEEvent(res, "error", JSON.stringify({ message: "Session not found or expired" }));
res.end();
return;
}
const lastEventId = parseLastEventId(req);
if (lastEventId !== undefined) {
const buffered = missionInterviewStreamManager.getBufferedEvents(sessionId, lastEventId);
if (!replayBufferedSSE(res, buffered)) {
res.end();
return;
}
}
if (session.summary) {
const existing = missionInterviewStreamManager.getBufferedEvents(sessionId, 0);
const lastSummaryEvent = [...existing].reverse().find((event) => event.event === "summary");
const summaryEventId = lastSummaryEvent?.id
?? missionInterviewStreamManager.broadcast(sessionId, {
type: "summary",
data: session.summary,
});
if (lastEventId === undefined || summaryEventId > lastEventId) {
if (!writeSSEEvent(res, "summary", JSON.stringify(session.summary), summaryEventId)) {
res.end();
return;
}
}
const lastCompleteEvent = [...existing].reverse().find((event) => event.event === "complete");
const completeEventId = lastCompleteEvent?.id
?? missionInterviewStreamManager.broadcast(sessionId, { type: "complete" });
if (lastEventId === undefined || completeEventId > lastEventId) {
writeSSEEvent(res, "complete", JSON.stringify({}), completeEventId);
}
res.end();
return;
}
// Subscribe to session events
const unsubscribe = missionInterviewStreamManager.subscribe(sessionId, (event) => {
try {
const data = (event as { data?: unknown }).data;
res.write(`event: ${event.type}\ndata: ${JSON.stringify(data ?? {})}\n\n`);
// End stream on complete or error
if (event.type === "complete" || event.type === "error") {
unsubscribe();
res.end();
}
} catch {
// Client disconnected
const unsubscribe = missionInterviewStreamManager.subscribe(sessionId, (event, eventId) => {
const data = (event as { data?: unknown }).data;
if (!writeSSEEvent(res, event.type, JSON.stringify(data ?? {}), eventId)) {
unsubscribe();
return;
}
// End stream on complete or error
if (event.type === "complete" || event.type === "error") {
unsubscribe();
res.end();
}
});
@@ -447,7 +511,7 @@ export function createMissionRouter(
clearInterval(heartbeat);
});
} catch (err: any) {
res.write(`event: error\ndata: ${JSON.stringify({ message: err.message || "Stream error" })}\n\n`);
writeSSEEvent(res, "error", JSON.stringify({ message: err.message || "Stream error" }));
res.end();
}
})

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

@@ -8,6 +8,7 @@ import {
getCurrentQuestion,
getSummary,
cleanupSession,
planningStreamManager,
checkRateLimit,
getRateLimitResetTime,
__resetPlanningState,
@@ -702,6 +703,70 @@ describe("planning module", () => {
});
});
describe("PlanningStreamManager buffering", () => {
it("stores broadcast events and returns buffered events since id", () => {
const sessionId = "stream-session-1";
const received: Array<{ type: string; id?: number }> = [];
const unsubscribe = planningStreamManager.subscribe(sessionId, (event, eventId) => {
received.push({ type: event.type, id: eventId });
});
const firstId = planningStreamManager.broadcast(sessionId, {
type: "thinking",
data: "delta-1",
});
const secondId = planningStreamManager.broadcast(sessionId, {
type: "question",
data: {
id: "q-1",
type: "text",
question: "Question?",
description: "desc",
},
});
expect(firstId).toBe(1);
expect(secondId).toBe(2);
expect(received).toEqual([
{ type: "thinking", id: 1 },
{ type: "question", id: 2 },
]);
const buffered = planningStreamManager.getBufferedEvents(sessionId, 1);
expect(buffered).toHaveLength(1);
expect(buffered[0]).toMatchObject({ id: 2, event: "question" });
unsubscribe();
});
it("broadcast buffers events even with no subscribers", () => {
const sessionId = "stream-session-2";
const eventId = planningStreamManager.broadcast(sessionId, {
type: "complete",
});
expect(eventId).toBe(1);
const buffered = planningStreamManager.getBufferedEvents(sessionId, 0);
expect(buffered).toHaveLength(1);
expect(buffered[0]).toMatchObject({ id: 1, event: "complete", data: "{}" });
});
it("cleanupSession clears buffered events", () => {
const sessionId = "stream-session-3";
planningStreamManager.broadcast(sessionId, {
type: "thinking",
data: "delta",
});
expect(planningStreamManager.getBufferedEvents(sessionId, 0)).toHaveLength(1);
planningStreamManager.cleanupSession(sessionId);
expect(planningStreamManager.getBufferedEvents(sessionId, 0)).toEqual([]);
});
});
describe("generateSubtasksFromPlanning", () => {
/** Helper: create a session and complete it to get a summary */
async function createCompletedSession(

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

@@ -22,6 +22,7 @@ import type { SubtaskItem } from "./subtask-breakdown.js";
import { randomUUID } from "node:crypto";
import { EventEmitter } from "node:events";
import type { AiSessionStore, AiSessionRow } from "./ai-session-store.js";
import { SessionEventBuffer, type SessionBufferedEvent } from "./sse-buffer.js";
// Dynamic import for @fusion/engine to avoid resolution issues in test environment
// eslint-disable-next-line @typescript-eslint/consistent-type-imports, @typescript-eslint/no-explicit-any
@@ -117,7 +118,7 @@ export type PlanningStreamEvent =
| { type: "complete" };
/** Callback function for streaming events */
export type PlanningStreamCallback = (event: PlanningStreamEvent) => void;
export type PlanningStreamCallback = (event: PlanningStreamEvent, eventId?: number) => void;
interface Session {
id: string;
@@ -241,7 +242,12 @@ process.on("beforeExit", () => {
* Each session can have multiple connected clients receiving streaming updates.
*/
export class PlanningStreamManager extends EventEmitter {
private sessions = new Map<string, Set<PlanningStreamCallback>>();
private readonly sessions = new Map<string, Set<PlanningStreamCallback>>();
private readonly buffers = new Map<string, SessionEventBuffer>();
constructor(private readonly bufferSize = 100) {
super();
}
/**
* Register a client callback for a planning session.
@@ -251,11 +257,10 @@ export class PlanningStreamManager extends EventEmitter {
if (!this.sessions.has(sessionId)) {
this.sessions.set(sessionId, new Set());
}
const callbacks = this.sessions.get(sessionId)!;
callbacks.add(callback);
// Return unsubscribe function
return () => {
callbacks.delete(callback);
if (callbacks.size === 0) {
@@ -264,20 +269,45 @@ export class PlanningStreamManager extends EventEmitter {
};
}
private getBuffer(sessionId: string): SessionEventBuffer {
let buffer = this.buffers.get(sessionId);
if (!buffer) {
buffer = new SessionEventBuffer(this.bufferSize);
this.buffers.set(sessionId, buffer);
}
return buffer;
}
/**
* Broadcast an event to all clients subscribed to a session.
* Every event is buffered and assigned a monotonically increasing id.
*/
broadcast(sessionId: string, event: PlanningStreamEvent): void {
broadcast(sessionId: string, event: PlanningStreamEvent): number {
const serialized = JSON.stringify((event as { data?: unknown }).data ?? {});
const eventData = typeof serialized === "string" ? serialized : "{}";
const eventId = this.getBuffer(sessionId).push(event.type, eventData);
const callbacks = this.sessions.get(sessionId);
if (!callbacks) return;
if (!callbacks) return eventId;
for (const callback of callbacks) {
try {
callback(event);
callback(event, eventId);
} catch (err) {
console.error(`[planning] Error broadcasting to client for session ${sessionId}:`, err);
}
}
return eventId;
}
/**
* Get buffered events with id > sinceId for the session.
*/
getBufferedEvents(sessionId: string, sinceId: number): SessionBufferedEvent[] {
const buffer = this.buffers.get(sessionId);
if (!buffer) return [];
return buffer.getEventsSince(sinceId);
}
/**
@@ -296,10 +326,20 @@ export class PlanningStreamManager extends EventEmitter {
}
/**
* Clean up all subscriptions for a session.
* Clean up all subscriptions and buffered events for a session.
*/
cleanupSession(sessionId: string): void {
this.sessions.delete(sessionId);
this.buffers.delete(sessionId);
}
/**
* Reset all subscriptions and buffers (test helper).
*/
reset(): void {
this.sessions.clear();
this.buffers.clear();
this.removeAllListeners();
}
}
@@ -1239,7 +1279,7 @@ export function __resetPlanningState(): void {
}
sessions.clear();
rateLimits.clear();
planningStreamManager.removeAllListeners();
planningStreamManager.reset();
}
/**

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

@@ -15,8 +15,8 @@ import type { TaskStore, TaskAttachment } from "@fusion/core";
import type { TaskDetail } from "@fusion/core";
import type { AuthStorageLike, ModelRegistryLike } from "./routes.js";
import { __resetBatchImportRateLimiter } from "./routes.js";
import { __resetPlanningState, __setCreateKbAgent } from "./planning.js";
import { __resetSubtaskBreakdownState } from "./subtask-breakdown.js";
import { __resetPlanningState, __setCreateKbAgent, planningStreamManager } from "./planning.js";
import { __resetSubtaskBreakdownState, subtaskStreamManager } from "./subtask-breakdown.js";
import * as terminalServiceModule from "./terminal-service.js";
import { get as performGet, request as performRequest } from "./test-request.js";
@@ -603,6 +603,136 @@ describe("POST /subtasks/*", () => {
expect(typeof res.body.sessionId).toBe("string");
});
it("replays buffered subtask events using lastEventId query param", async () => {
const start = await REQUEST(
buildApp(),
"POST",
"/api/subtasks/start-streaming",
JSON.stringify({ description: "Replay buffered subtask stream" }),
{ "Content-Type": "application/json" },
);
const sessionId = start.body.sessionId as string;
// Reset any initial stream manager state from background generation.
subtaskStreamManager.cleanupSession(sessionId);
subtaskStreamManager.broadcast(sessionId, { type: "thinking", data: "first" });
subtaskStreamManager.broadcast(sessionId, { type: "thinking", data: "second" });
setTimeout(() => {
subtaskStreamManager.broadcast(sessionId, { type: "complete" });
}, 0);
const streamRes = await REQUEST(
buildApp(),
"GET",
`/api/subtasks/${sessionId}/stream?lastEventId=1`,
);
expect(streamRes.status).toBe(200);
expect(streamRes.body).toContain("id: 2");
expect(streamRes.body).toContain("event: thinking");
expect(streamRes.body).toContain("event: complete");
expect(streamRes.body).not.toContain("id: 1\nevent: thinking");
});
it("replays buffered subtask events using Last-Event-ID header", async () => {
const start = await REQUEST(
buildApp(),
"POST",
"/api/subtasks/start-streaming",
JSON.stringify({ description: "Replay buffered subtask stream from header" }),
{ "Content-Type": "application/json" },
);
const sessionId = start.body.sessionId as string;
subtaskStreamManager.cleanupSession(sessionId);
subtaskStreamManager.broadcast(sessionId, { type: "thinking", data: "first" });
subtaskStreamManager.broadcast(sessionId, { type: "thinking", data: "second" });
setTimeout(() => {
subtaskStreamManager.broadcast(sessionId, { type: "complete" });
}, 0);
const streamRes = await REQUEST(
buildApp(),
"GET",
`/api/subtasks/${sessionId}/stream`,
undefined,
{ "Last-Event-ID": "1" },
);
expect(streamRes.status).toBe(200);
expect(streamRes.body).toContain("id: 2");
expect(streamRes.body).toContain("event: thinking");
expect(streamRes.body).toContain("event: complete");
expect(streamRes.body).not.toContain("id: 1\nevent: thinking");
});
it("skips subtask replay when Last-Event-ID is missing", async () => {
const start = await REQUEST(
buildApp(),
"POST",
"/api/subtasks/start-streaming",
JSON.stringify({ description: "No subtask replay without header" }),
{ "Content-Type": "application/json" },
);
const sessionId = start.body.sessionId as string;
subtaskStreamManager.cleanupSession(sessionId);
subtaskStreamManager.broadcast(sessionId, { type: "thinking", data: "first" });
setTimeout(() => {
subtaskStreamManager.broadcast(sessionId, { type: "complete" });
}, 0);
const streamRes = await REQUEST(
buildApp(),
"GET",
`/api/subtasks/${sessionId}/stream`,
);
expect(streamRes.status).toBe(200);
expect(streamRes.body).not.toContain("id: 1\nevent: thinking");
expect(streamRes.body).toContain("id: 2");
expect(streamRes.body).toContain("event: complete");
});
it("gracefully ignores invalid Last-Event-ID values for subtask streams", async () => {
const start = await REQUEST(
buildApp(),
"POST",
"/api/subtasks/start-streaming",
JSON.stringify({ description: "Invalid subtask last event id" }),
{ "Content-Type": "application/json" },
);
const sessionId = start.body.sessionId as string;
subtaskStreamManager.cleanupSession(sessionId);
subtaskStreamManager.broadcast(sessionId, { type: "thinking", data: "first" });
setTimeout(() => {
subtaskStreamManager.broadcast(sessionId, { type: "complete" });
}, 0);
const streamRes = await REQUEST(
buildApp(),
"GET",
`/api/subtasks/${sessionId}/stream`,
undefined,
{ "Last-Event-ID": "not-a-number" },
);
expect(streamRes.status).toBe(200);
expect(streamRes.body).not.toContain("id: 1\nevent: thinking");
expect(streamRes.body).toContain("id: 2");
expect(streamRes.body).toContain("event: complete");
});
it("creates tasks from a breakdown and resolves dependencies", async () => {
(store.createTask as ReturnType<typeof vi.fn>)
.mockResolvedValueOnce({ ...FAKE_TASK_DETAIL, id: "FN-101", title: "First", column: "triage" })
@@ -5801,6 +5931,99 @@ describe("Git Management endpoints", () => {
});
});
describe("GET /planning/:sessionId/stream", () => {
it("replays buffered events when Last-Event-ID header is provided", async () => {
const startRes = await REQUEST(
buildApp(),
"POST",
"/api/planning/start",
JSON.stringify({ initialPlan: "Reconnect planning stream" }),
{ "Content-Type": "application/json" },
);
const sessionId = startRes.body.sessionId as string;
planningStreamManager.broadcast(sessionId, { type: "thinking", data: "first" });
planningStreamManager.broadcast(sessionId, { type: "thinking", data: "second" });
setTimeout(() => {
planningStreamManager.broadcast(sessionId, { type: "complete" });
}, 0);
const streamRes = await REQUEST(
buildApp(),
"GET",
`/api/planning/${sessionId}/stream`,
undefined,
{ "Last-Event-ID": "1" },
);
expect(streamRes.status).toBe(200);
expect(typeof streamRes.body).toBe("string");
expect(streamRes.body).toContain("id: 2");
expect(streamRes.body).toContain("event: thinking");
expect(streamRes.body).toContain("id: 3");
expect(streamRes.body).toContain("event: complete");
expect(streamRes.body).not.toContain("id: 1\nevent: thinking");
});
it("skips replay when Last-Event-ID is missing", async () => {
const startRes = await REQUEST(
buildApp(),
"POST",
"/api/planning/start",
JSON.stringify({ initialPlan: "No replay planning stream" }),
{ "Content-Type": "application/json" },
);
const sessionId = startRes.body.sessionId as string;
planningStreamManager.broadcast(sessionId, { type: "thinking", data: "first" });
setTimeout(() => {
planningStreamManager.broadcast(sessionId, { type: "complete" });
}, 0);
const streamRes = await REQUEST(buildApp(), "GET", `/api/planning/${sessionId}/stream`);
expect(streamRes.status).toBe(200);
expect(streamRes.body).not.toContain("id: 1\nevent: thinking");
expect(streamRes.body).toContain("id: 2");
expect(streamRes.body).toContain("event: complete");
});
it("gracefully ignores invalid Last-Event-ID values", async () => {
const startRes = await REQUEST(
buildApp(),
"POST",
"/api/planning/start",
JSON.stringify({ initialPlan: "Invalid last event id" }),
{ "Content-Type": "application/json" },
);
const sessionId = startRes.body.sessionId as string;
planningStreamManager.broadcast(sessionId, { type: "thinking", data: "first" });
setTimeout(() => {
planningStreamManager.broadcast(sessionId, { type: "complete" });
}, 0);
const streamRes = await REQUEST(
buildApp(),
"GET",
`/api/planning/${sessionId}/stream`,
undefined,
{ "Last-Event-ID": "not-a-number" },
);
expect(streamRes.status).toBe(200);
expect(streamRes.body).not.toContain("id: 1\nevent: thinking");
expect(streamRes.body).toContain("id: 2");
expect(streamRes.body).toContain("event: complete");
});
});
describe("POST /planning/respond", () => {
it("processes response and returns next question", async () => {
// First create a session

173
packages/dashboard/src/routes.ts
View File

@@ -37,6 +37,7 @@ import {
SessionNotFoundError as AgentGenerationSessionNotFoundError,
} from "./agent-generation.js";
import { getMissionInterviewSession, cleanupMissionInterviewSession } from "./mission-interview.js";
import { writeSSEEvent } from "./sse-buffer.js";
/**
* Minimal interface matching pi-coding-agent's ModelRegistry API surface
@@ -1266,6 +1267,34 @@ export function __resetBatchImportRateLimiter(): void {
}
}
function parseLastEventId(req: Request): number | undefined {
const rawHeader = req.headers["last-event-id"];
const rawQuery = req.query.lastEventId;
const raw = Array.isArray(rawHeader)
? rawHeader[0]
: (typeof rawHeader === "string" ? rawHeader : Array.isArray(rawQuery) ? rawQuery[0] : rawQuery);
if (raw === undefined || raw === null) return undefined;
const parsed = Number.parseInt(String(raw), 10);
if (!Number.isFinite(parsed) || parsed < 0) return undefined;
return parsed;
}
function replayBufferedSSE(
res: Response,
bufferedEvents: Array<{ id: number; event: string; data: string }>,
): boolean {
for (const bufferedEvent of bufferedEvents) {
if (!writeSSEEvent(res, bufferedEvent.event, bufferedEvent.data, bufferedEvent.id)) {
return false;
}
}
return true;
}
export function createApiRoutes(store: TaskStore, options?: ServerOptions): Router {
const router = Router();
@@ -5387,39 +5416,80 @@ export function createApiRoutes(store: TaskStore, options?: ServerOptions): Rout
const { subtaskStreamManager, getSubtaskSession } = await import("./subtask-breakdown.js");
const session = getSubtaskSession(sessionId);
if (!session) {
res.write(`event: error\ndata: ${JSON.stringify("Session not found or expired")}\n\n`);
writeSSEEvent(res, "error", JSON.stringify("Session not found or expired"));
res.end();
return;
}
const unsubscribe = subtaskStreamManager.subscribe(sessionId, (event) => {
try {
const data = (event as { data?: unknown }).data;
res.write(`event: ${event.type}\ndata: ${JSON.stringify(data ?? {})}\n\n`);
if (event.type === "complete" || event.type === "error") {
unsubscribe();
res.end();
}
} catch {
unsubscribe();
const lastEventId = parseLastEventId(req);
if (lastEventId !== undefined) {
const buffered = subtaskStreamManager.getBufferedEvents(sessionId, lastEventId);
if (!replayBufferedSSE(res, buffered)) {
res.end();
return;
}
});
}
if (session.status === "complete") {
res.write(`event: subtasks\ndata: ${JSON.stringify(session.subtasks)}\n\n`);
res.write("event: complete\ndata: {}\n\n");
unsubscribe();
const existing = subtaskStreamManager.getBufferedEvents(sessionId, 0);
const lastSubtasksEvent = [...existing].reverse().find((event) => event.event === "subtasks");
const subtasksEventId = lastSubtasksEvent?.id
?? subtaskStreamManager.broadcast(sessionId, {
type: "subtasks",
data: session.subtasks,
});
if (lastEventId === undefined || subtasksEventId > lastEventId) {
if (!writeSSEEvent(res, "subtasks", JSON.stringify(session.subtasks), subtasksEventId)) {
res.end();
return;
}
}
const lastCompleteEvent = [...existing].reverse().find((event) => event.event === "complete");
const completeEventId = lastCompleteEvent?.id
?? subtaskStreamManager.broadcast(sessionId, { type: "complete" });
if (lastEventId === undefined || completeEventId > lastEventId) {
writeSSEEvent(res, "complete", JSON.stringify({}), completeEventId);
}
res.end();
return;
}
if (session.status === "error") {
res.write(`event: error\ndata: ${JSON.stringify(String(session.error || "Unknown error"))}\n\n`);
unsubscribe();
const errorMessage = String(session.error || "Unknown error");
const existing = subtaskStreamManager.getBufferedEvents(sessionId, 0);
const lastErrorEvent = [...existing].reverse().find((event) => event.event === "error");
const errorEventId = lastErrorEvent?.id
?? subtaskStreamManager.broadcast(sessionId, {
type: "error",
data: errorMessage,
});
if (lastEventId === undefined || errorEventId > lastEventId) {
writeSSEEvent(res, "error", JSON.stringify(errorMessage), errorEventId);
}
res.end();
return;
}
const unsubscribe = subtaskStreamManager.subscribe(sessionId, (event, eventId) => {
const data = (event as { data?: unknown }).data;
if (!writeSSEEvent(res, event.type, JSON.stringify(data ?? {}), eventId)) {
unsubscribe();
return;
}
if (event.type === "complete" || event.type === "error") {
unsubscribe();
res.end();
}
});
const heartbeat = setInterval(() => {
if (res.writableEnded) {
clearInterval(heartbeat);
@@ -5433,7 +5503,7 @@ export function createApiRoutes(store: TaskStore, options?: ServerOptions): Rout
unsubscribe();
});
} catch (err: any) {
res.write(`event: error\ndata: ${JSON.stringify(String(err?.message) || "Unknown error")}\n\n`);
writeSSEEvent(res, "error", JSON.stringify(String(err?.message) || "Unknown error"));
res.end();
}
});
@@ -6002,30 +6072,65 @@ export function createApiRoutes(store: TaskStore, options?: ServerOptions): Rout
res.write(": connected\n\n");
try {
const { planningStreamManager, getSession, SessionNotFoundError } = await import("./planning.js");
const { planningStreamManager, getSession } = await import("./planning.js");
// Verify session exists
const session = getSession(sessionId);
if (!session) {
res.write(`event: error\ndata: ${JSON.stringify({ message: "Session not found or expired" })}\n\n`);
writeSSEEvent(res, "error", JSON.stringify({ message: "Session not found or expired" }));
res.end();
return;
}
const lastEventId = parseLastEventId(req);
if (lastEventId !== undefined) {
const buffered = planningStreamManager.getBufferedEvents(sessionId, lastEventId);
if (!replayBufferedSSE(res, buffered)) {
res.end();
return;
}
}
if (session.summary) {
const existing = planningStreamManager.getBufferedEvents(sessionId, 0);
const lastSummaryEvent = [...existing].reverse().find((event) => event.event === "summary");
const summaryEventId = lastSummaryEvent?.id
?? planningStreamManager.broadcast(sessionId, {
type: "summary",
data: session.summary,
});
if (lastEventId === undefined || summaryEventId > lastEventId) {
if (!writeSSEEvent(res, "summary", JSON.stringify(session.summary), summaryEventId)) {
res.end();
return;
}
}
const lastCompleteEvent = [...existing].reverse().find((event) => event.event === "complete");
const completeEventId = lastCompleteEvent?.id
?? planningStreamManager.broadcast(sessionId, { type: "complete" });
if (lastEventId === undefined || completeEventId > lastEventId) {
writeSSEEvent(res, "complete", JSON.stringify({}), completeEventId);
}
res.end();
return;
}
// Subscribe to session events
const unsubscribe = planningStreamManager.subscribe(sessionId, (event) => {
try {
const data = (event as { data?: unknown }).data;
res.write(`event: ${event.type}\ndata: ${JSON.stringify(data ?? {})}\n\n`);
// End stream on complete or error
if (event.type === "complete" || event.type === "error") {
unsubscribe();
res.end();
}
} catch (err) {
// Client disconnected
const unsubscribe = planningStreamManager.subscribe(sessionId, (event, eventId) => {
const data = (event as { data?: unknown }).data;
if (!writeSSEEvent(res, event.type, JSON.stringify(data ?? {}), eventId)) {
unsubscribe();
return;
}
// End stream on complete or error
if (event.type === "complete" || event.type === "error") {
unsubscribe();
res.end();
}
});
@@ -6047,7 +6152,7 @@ export function createApiRoutes(store: TaskStore, options?: ServerOptions): Rout
clearInterval(heartbeat);
});
} catch (err: any) {
res.write(`event: error\ndata: ${JSON.stringify({ message: err.message || "Stream error" })}\n\n`);
writeSSEEvent(res, "error", JSON.stringify({ message: err.message || "Stream error" }));
res.end();
}
});

89
packages/dashboard/src/sse-buffer.ts Normal file
View File

@@ -0,0 +1,89 @@
import type { Response } from "express";
export interface SessionBufferedEvent {
id: number;
event: string;
data: string;
}
/**
* Per-session in-memory ring buffer for SSE events.
*
* Stores only the last N events and assigns monotonically increasing IDs.
*/
export class SessionEventBuffer {
private events: SessionBufferedEvent[] = [];
private nextId = 1;
constructor(private readonly maxCapacity = 100) {
if (!Number.isFinite(maxCapacity) || maxCapacity <= 0) {
throw new Error("maxCapacity must be a positive finite number");
}
}
/**
* Push an event into the buffer and return the assigned event id.
*/
push(event: string, data: string): number {
const id = this.nextId++;
this.events.push({ id, event, data });
if (this.events.length > this.maxCapacity) {
this.events.splice(0, this.events.length - this.maxCapacity);
}
return id;
}
/**
* Return all buffered events with id > lastEventId.
*/
getEventsSince(lastEventId: number): SessionBufferedEvent[] {
if (!Number.isFinite(lastEventId)) {
return [...this.events];
}
return this.events.filter((event) => event.id > lastEventId);
}
clear(): void {
this.events = [];
}
size(): number {
return this.events.length;
}
}
/**
* Render one SSE event payload (with optional id field).
*/
export function formatSSEEvent(event: string, data: string, id?: number): string {
const idLine = id !== undefined ? `id: ${id}\n` : "";
return `${idLine}event: ${event}\ndata: ${data}\n\n`;
}
/**
* Safely write to an SSE response stream.
*/
export function safeWriteSSE(res: Pick<Response, "write" | "writableEnded" | "destroyed">, payload: string): boolean {
try {
if (res.writableEnded || res.destroyed) return false;
res.write(payload);
return true;
} catch {
return false;
}
}
/**
* Write one SSE event to response with optional id field.
*/
export function writeSSEEvent(
res: Pick<Response, "write" | "writableEnded" | "destroyed">,
event: string,
data: string,
id?: number,
): boolean {
return safeWriteSSE(res, formatSSEEvent(event, data, id));
}

73
packages/dashboard/src/subtask-breakdown.test.ts Normal file
View File

@@ -0,0 +1,73 @@
// @vitest-environment node
import { beforeEach, describe, expect, it, vi } from "vitest";
import {
__resetSubtaskBreakdownState,
subtaskStreamManager,
} from "./subtask-breakdown.js";
describe("subtask-breakdown stream buffering", () => {
beforeEach(() => {
__resetSubtaskBreakdownState();
});
it("buffers broadcast events and forwards ids to subscribers", () => {
const sessionId = "subtask-session-1";
const callback = vi.fn();
const unsubscribe = subtaskStreamManager.subscribe(sessionId, callback);
const firstId = subtaskStreamManager.broadcast(sessionId, {
type: "thinking",
data: "delta-1",
});
const secondId = subtaskStreamManager.broadcast(sessionId, {
type: "subtasks",
data: [
{
id: "subtask-1",
title: "Title",
description: "Description",
suggestedSize: "S",
dependsOn: [],
},
],
});
expect(firstId).toBe(1);
expect(secondId).toBe(2);
expect(callback).toHaveBeenNthCalledWith(1, { type: "thinking", data: "delta-1" }, 1);
expect(callback).toHaveBeenNthCalledWith(2, expect.objectContaining({ type: "subtasks" }), 2);
const buffered = subtaskStreamManager.getBufferedEvents(sessionId, 1);
expect(buffered).toHaveLength(1);
expect(buffered[0]).toMatchObject({ id: 2, event: "subtasks" });
unsubscribe();
});
it("buffers complete events without subscribers", () => {
const sessionId = "subtask-session-2";
const eventId = subtaskStreamManager.broadcast(sessionId, { type: "complete" });
expect(eventId).toBe(1);
expect(subtaskStreamManager.getBufferedEvents(sessionId, 0)).toEqual([
{ id: 1, event: "complete", data: "{}" },
]);
});
it("clears subscriptions and buffered events on cleanupSession", () => {
const sessionId = "subtask-session-3";
const callback = vi.fn();
subtaskStreamManager.subscribe(sessionId, callback);
subtaskStreamManager.broadcast(sessionId, { type: "thinking", data: "delta" });
expect(subtaskStreamManager.getBufferedEvents(sessionId, 0)).toHaveLength(1);
subtaskStreamManager.cleanupSession(sessionId);
expect(subtaskStreamManager.getBufferedEvents(sessionId, 0)).toEqual([]);
});
});

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

@@ -2,6 +2,7 @@ import type { TaskStore } from "@fusion/core";
import { randomUUID } from "node:crypto";
import { EventEmitter } from "node:events";
import type { AiSessionStore, AiSessionRow } from "./ai-session-store.js";
import { SessionEventBuffer, type SessionBufferedEvent } from "./sse-buffer.js";
// eslint-disable-next-line @typescript-eslint/no-explicit-any
let createKbAgent: any;
@@ -43,7 +44,7 @@ export type SubtaskStreamEvent =
| { type: "error"; data: string }
| { type: "complete" };
export type SubtaskStreamCallback = (event: SubtaskStreamEvent) => void;
export type SubtaskStreamCallback = (event: SubtaskStreamEvent, eventId?: number) => void;
const SESSION_TTL_MS = 30 * 60 * 1000;
const CLEANUP_INTERVAL_MS = 5 * 60 * 1000;
@@ -141,7 +142,12 @@ process.on("beforeExit", () => {
});
export class SubtaskStreamManager extends EventEmitter {
private sessions = new Map<string, Set<SubtaskStreamCallback>>();
private readonly sessions = new Map<string, Set<SubtaskStreamCallback>>();
private readonly buffers = new Map<string, SessionEventBuffer>();
constructor(private readonly bufferSize = 100) {
super();
}
subscribe(sessionId: string, callback: SubtaskStreamCallback): () => void {
if (!this.sessions.has(sessionId)) {
@@ -157,20 +163,49 @@ export class SubtaskStreamManager extends EventEmitter {
};
}
broadcast(sessionId: string, event: SubtaskStreamEvent): void {
private getBuffer(sessionId: string): SessionEventBuffer {
let buffer = this.buffers.get(sessionId);
if (!buffer) {
buffer = new SessionEventBuffer(this.bufferSize);
this.buffers.set(sessionId, buffer);
}
return buffer;
}
broadcast(sessionId: string, event: SubtaskStreamEvent): number {
const serialized = JSON.stringify((event as { data?: unknown }).data ?? {});
const eventData = typeof serialized === "string" ? serialized : "{}";
const eventId = this.getBuffer(sessionId).push(event.type, eventData);
const callbacks = this.sessions.get(sessionId);
if (!callbacks) return;
if (!callbacks) return eventId;
for (const callback of callbacks) {
try {
callback(event);
callback(event, eventId);
} catch {
// ignore subscriber failures
}
}
return eventId;
}
getBufferedEvents(sessionId: string, sinceId: number): SessionBufferedEvent[] {
const buffer = this.buffers.get(sessionId);
if (!buffer) return [];
return buffer.getEventsSince(sinceId);
}
cleanupSession(sessionId: string): void {
this.sessions.delete(sessionId);
this.buffers.delete(sessionId);
}
reset(): void {
this.sessions.clear();
this.buffers.clear();
this.removeAllListeners();
}
}
@@ -367,7 +402,7 @@ export function __resetSubtaskBreakdownState(): void {
}
}
sessions.clear();
subtaskStreamManager.removeAllListeners();
subtaskStreamManager.reset();
}
export class SessionNotFoundError extends Error {