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

FN-7663: consolidate /automations and /routines run-stream SSE handlers into a shared factory

Browse Source 
Removes duplicated SSE streaming logic between the /automations/:id/run/stream and /routines/:id/run/stream endpoints by extracting a single generic makeRunStreamHandler factory.

- Add makeRunStreamHandler<TStore, TEntity> factory in routes.ts, parameterized by store resolver, entity getter, and not-found message
- Replace the two near-identical inline SSE handlers (connect headers, buffered replay, run subscription, auto-attach, teardown, error handling) with calls to the shared factory
- Import ScopeValue type from ./routes/types.js to type the factory's scope parameter
- Add regression tests in routes-automation.test.ts covering the consolidated handler behavior for both routes

Files changed:
 .../src/__tests__/routes-automation.test.ts        |  35 +++
 packages/dashboard/src/routes.ts                   | 246 +++++++++------------
 2 files changed, 137 insertions(+), 144 deletions(-)

Fusion-Task-Id: FN-7663

Fusion-Task-Lineage: 55ede468-c4ed-4a72-a428-7d19686e8ea5

Co-authored-by: Fusion (runfusion.ai) <noreply@runfusion.ai>
This commit is contained in:
gsxdsm
2026-07-08 00:41:58 -07:00
parent aae6a6eca2
commit d2e38ba920
2 changed files with 137 additions and 144 deletions
Download Patch File Download Diff File Expand all files Collapse all files

35
packages/dashboard/src/__tests__/routes-automation.test.ts
View File

@@ -1588,6 +1588,17 @@ describe("Automation routes", () => {
expect(res.body.error).toContain("Schedule not found");
});
// FNXC:AutomationLiveOutput 2026-07-07-08:30 (FN-7663): proves the shared run-stream handler
// preserves scope isolation identically to every other /automations/:id/* route.
it("GET /automations/:id/run/stream returns 404 for schedule with wrong scope", async () => {
const mockStore = createMockAutomationStore();
mockStore.getSchedule.mockResolvedValue({ ...FAKE_SCHEDULE, scope: "global" as const });
const { app } = buildApp(mockStore);
const res = await REQUEST(app, "GET", "/api/automations/sched-001/run/stream?scope=project");
expect(res.status).toBe(404);
expect(res.body.error).toContain("Schedule not found");
});
it("POST /automations/:id/toggle with matching scope returns 200", async () => {
const mockStore = createMockAutomationStore();
mockStore.getSchedule.mockResolvedValue({ ...FAKE_SCHEDULE, scope: "project" as const });
@@ -2094,6 +2105,18 @@ describe("Routine routes", () => {
expect(runRes.body.result.output).toBe("routine-live-output");
});
// FNXC:AutomationLiveOutput 2026-07-07-08:30 (FN-7663): mirrors the automations "unknown manual
// run id" coverage — both routes now share one handler for the requested-runId-not-found branch.
it("returns a terminal SSE error for an unknown manual run id", async () => {
const mockStore = createMockRoutineStore();
const { app } = buildRoutineApp(mockStore);
const streamRes = await performRequest(app, "GET", "/api/routines/routine-001/run/stream?runId=missing-run");
expect(streamRes.status).toBe(200);
expect(String(streamRes.body)).toContain("event: error");
expect(String(streamRes.body)).toContain("Live run not found or expired");
});
// FNXC:AutomationLiveOutput 2026-07-07-01:00 (FN-7652): mirrors the /automations/:id/run/stream
// coverage above — both stream endpoints share AutomationLiveRunRegistry/attachRun, so the
// runId-less-stream invariant (success run never delivers a terminal error) must hold for /routines too.
@@ -2534,6 +2557,18 @@ describe("Routine routes", () => {
expect(res.body.error).toContain("Routine not found");
});
// FNXC:AutomationLiveOutput 2026-07-07-08:30 (FN-7663): mirrors the automations run-stream
// scope test above — both routes now delegate to the same shared handler, so scope isolation
// must hold identically for /routines/:id/run/stream.
it("GET /routines/:id/run/stream returns 404 for routine with wrong scope", async () => {
const mockStore = createMockRoutineStore();
mockStore.getRoutine.mockResolvedValue({ ...FAKE_ROUTINE, scope: "project" as const });
const { app } = buildRoutineApp(mockStore);
const res = await REQUEST(app, "GET", "/api/routines/routine-001/run/stream?scope=global");
expect(res.status).toBe(404);
expect(res.body.error).toContain("Routine not found");
});
it("POST /routines/:id/trigger returns 404 for routine with wrong scope", async () => {
const mockStore = createMockRoutineStore();
mockStore.getRoutine.mockResolvedValue({ ...FAKE_ROUTINE, scope: "project" as const });

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

@@ -140,6 +140,7 @@ function resolveBundledPluginDirInDashboard(pluginId: string): string | null {
import { createSessionDiagnostics } from "./ai-session-diagnostics.js";
import { createApiRoutesContext } from "./routes/context.js";
import type { ScopeValue } from "./routes/types.js";
import { registerTaskWorkflowRoutes } from "./routes/register-task-workflow-routes.js";
import { registerWorkflowRoutes } from "./routes/register-workflow-routes.js";
import { registerPlanningSubtaskRoutes } from "./routes/register-planning-subtask-routes.js";
@@ -2443,84 +2444,110 @@ export function createApiRoutes(store: TaskStore, options?: ServerOptions): Rout
}
});
// GET /automations/:id/run/stream — stream live manual-run output.
router.get("/automations/:id/run/stream", async (req: Request, res: Response) => {
const scope = parseScopeParam(req);
const automationStore = resolveAutomationStore(req, scope);
const id = Array.isArray(req.params.id) ? req.params.id[0] : req.params.id;
/**
* FNXC:AutomationLiveOutput 2026-07-07-08:30 (FN-7663, follow-up from FN-7652):
* `/automations/:id/run/stream` and `/routines/:id/run/stream` are otherwise-identical SSE
* endpoints layered over the same `AutomationLiveRunRegistry` (`automationLiveRuns`). Before
* this consolidation, each route carried its own copy of the header/replay/subscribe/teardown
* logic — the FN-7652 live-output fix had to be applied twice, and any future fix could drift
* between the two copies. This single generic factory is parameterized ONLY by what actually
* differs between the two routes (store resolver, entity getter, not-found message) so a fix
* to the streaming behavior is written once and applies to both endpoints.
*/
function makeRunStreamHandler<TStore, TEntity extends { id: string; scope?: ScopeValue }>(config: {
resolveStore: (req: Request, scope: ScopeValue | undefined) => TStore;
getEntity: (store: TStore, id: string) => Promise<TEntity>;
notFoundMessage: string;
}): (req: Request, res: Response) => Promise<void> {
const { resolveStore, getEntity, notFoundMessage } = config;
return async (req: Request, res: Response) => {
const scope = parseScopeParam(req);
const store = resolveStore(req, scope);
const id = Array.isArray(req.params.id) ? req.params.id[0] : req.params.id;
try {
const schedule = await automationStore.getSchedule(id);
if (scope && schedule.scope !== scope) {
throw notFound("Schedule not found");
}
res.setHeader("Content-Type", "text/event-stream");
res.setHeader("Cache-Control", "no-cache");
res.setHeader("Connection", "keep-alive");
res.setHeader("X-Accel-Buffering", "no");
res.flushHeaders();
res.write(": connected\n\n");
const requestedRunId = typeof req.query.runId === "string" ? req.query.runId : undefined;
const lastEventId = parseLastEventId(req);
let unsubscribeRun: (() => void) | undefined;
let unsubscribeStart: (() => void) | undefined;
const attachRun = (run: AutomationLiveRunRecord) => {
const buffered = automationLiveRuns.getBufferedEvents(run.runId, lastEventId ?? 0);
if (!replayBufferedSSE(res, buffered)) {
res.end();
return;
try {
const entity = await getEntity(store, id);
if (scope && entity.scope !== scope) {
throw notFound(notFoundMessage);
}
if (run.status !== "running") {
res.end();
return;
}
unsubscribeRun = automationLiveRuns.subscribe(run.runId, (event, eventId) => {
if (!writeSSEEvent(res, event.type, JSON.stringify(event.data ?? {}), eventId)) {
unsubscribeRun?.();
res.setHeader("Content-Type", "text/event-stream");
res.setHeader("Cache-Control", "no-cache");
res.setHeader("Connection", "keep-alive");
res.setHeader("X-Accel-Buffering", "no");
res.flushHeaders();
res.write(": connected\n\n");
const requestedRunId = typeof req.query.runId === "string" ? req.query.runId : undefined;
const lastEventId = parseLastEventId(req);
let unsubscribeRun: (() => void) | undefined;
let unsubscribeStart: (() => void) | undefined;
const attachRun = (run: AutomationLiveRunRecord) => {
const buffered = automationLiveRuns.getBufferedEvents(run.runId, lastEventId ?? 0);
if (!replayBufferedSSE(res, buffered)) {
res.end();
return;
}
if (event.type === "complete" || event.type === "error") {
unsubscribeRun?.();
if (run.status !== "running") {
res.end();
return;
}
});
};
unsubscribeRun = automationLiveRuns.subscribe(run.runId, (event, eventId) => {
if (!writeSSEEvent(res, event.type, JSON.stringify(event.data ?? {}), eventId)) {
unsubscribeRun?.();
return;
}
if (event.type === "complete" || event.type === "error") {
unsubscribeRun?.();
res.end();
}
});
};
// FNXC:AutomationLiveOutput 2026-07-07-00:00 (FN-7652): no explicit runId means "attach me to
// this request's own run" — use getForAutoAttach so a stale finished run from before this
// trigger isn't mistaken for it (see AutomationLiveRunRegistry.getForAutoAttach).
const existingRun = requestedRunId
? automationLiveRuns.get(requestedRunId, schedule.id)
: automationLiveRuns.getForAutoAttach(schedule.id);
if (existingRun) {
attachRun(existingRun);
} else if (requestedRunId) {
writeSSEEvent(res, "error", JSON.stringify({ message: "Live run not found or expired", runId: requestedRunId }));
res.end();
} else {
unsubscribeStart = automationLiveRuns.subscribeToScheduleStart(schedule.id, (run) => {
// FNXC:AutomationLiveOutput 2026-07-07-00:00 (FN-7652): no explicit runId means "attach me to
// this request's own run" — use getForAutoAttach so a stale finished run from before this
// trigger isn't mistaken for it (see AutomationLiveRunRegistry.getForAutoAttach).
const existingRun = requestedRunId
? automationLiveRuns.get(requestedRunId, entity.id)
: automationLiveRuns.getForAutoAttach(entity.id);
if (existingRun) {
attachRun(existingRun);
} else if (requestedRunId) {
writeSSEEvent(res, "error", JSON.stringify({ message: "Live run not found or expired", runId: requestedRunId }));
res.end();
} else {
unsubscribeStart = automationLiveRuns.subscribeToScheduleStart(entity.id, (run) => {
unsubscribeStart?.();
attachRun(run);
});
}
req.on("close", () => {
unsubscribeRun?.();
unsubscribeStart?.();
attachRun(run);
});
} catch (err: unknown) {
if (err instanceof ApiError) {
throw err;
}
if ((err as NodeJS.ErrnoException).code === "ENOENT") {
throw notFound(notFoundMessage);
}
rethrowAsApiError(err);
}
};
}
req.on("close", () => {
unsubscribeRun?.();
unsubscribeStart?.();
});
} catch (err: unknown) {
if (err instanceof ApiError) {
throw err;
}
if ((err as NodeJS.ErrnoException).code === "ENOENT") {
throw notFound("Schedule not found");
}
rethrowAsApiError(err);
}
});
// GET /automations/:id/run/stream — stream live manual-run output.
router.get(
"/automations/:id/run/stream",
makeRunStreamHandler({
resolveStore: resolveAutomationStore,
getEntity: (store, id) => store.getSchedule(id),
notFoundMessage: "Schedule not found",
}),
);
// POST /automations/:id/toggle — toggle enabled/disabled
router.post("/automations/:id/toggle", async (req: Request, res: Response) => {
@@ -2937,83 +2964,14 @@ export function createApiRoutes(store: TaskStore, options?: ServerOptions): Rout
});
// GET /routines/:id/run/stream — stream live manual routine output.
router.get("/routines/:id/run/stream", async (req: Request, res: Response) => {
const scope = parseScopeParam(req);
const routineStore = resolveRoutineStore(req, scope);
const id = Array.isArray(req.params.id) ? req.params.id[0] : req.params.id;
try {
const routine = await routineStore.getRoutine(id);
if (scope && routine.scope !== scope) {
throw notFound("Routine not found");
}
res.setHeader("Content-Type", "text/event-stream");
res.setHeader("Cache-Control", "no-cache");
res.setHeader("Connection", "keep-alive");
res.setHeader("X-Accel-Buffering", "no");
res.flushHeaders();
res.write(": connected\n\n");
const requestedRunId = typeof req.query.runId === "string" ? req.query.runId : undefined;
const lastEventId = parseLastEventId(req);
let unsubscribeRun: (() => void) | undefined;
let unsubscribeStart: (() => void) | undefined;
const attachRun = (run: AutomationLiveRunRecord) => {
const buffered = automationLiveRuns.getBufferedEvents(run.runId, lastEventId ?? 0);
if (!replayBufferedSSE(res, buffered)) {
res.end();
return;
}
if (run.status !== "running") {
res.end();
return;
}
unsubscribeRun = automationLiveRuns.subscribe(run.runId, (event, eventId) => {
if (!writeSSEEvent(res, event.type, JSON.stringify(event.data ?? {}), eventId)) {
unsubscribeRun?.();
return;
}
if (event.type === "complete" || event.type === "error") {
unsubscribeRun?.();
res.end();
}
});
};
// FNXC:AutomationLiveOutput 2026-07-07-00:00 (FN-7652): no explicit runId means "attach me to
// this request's own run" — use getForAutoAttach so a stale finished run from before this
// trigger isn't mistaken for it (see AutomationLiveRunRegistry.getForAutoAttach).
const existingRun = requestedRunId
? automationLiveRuns.get(requestedRunId, routine.id)
: automationLiveRuns.getForAutoAttach(routine.id);
if (existingRun) {
attachRun(existingRun);
} else if (requestedRunId) {
writeSSEEvent(res, "error", JSON.stringify({ message: "Live run not found or expired", runId: requestedRunId }));
res.end();
} else {
unsubscribeStart = automationLiveRuns.subscribeToScheduleStart(routine.id, (run) => {
unsubscribeStart?.();
attachRun(run);
});
}
req.on("close", () => {
unsubscribeRun?.();
unsubscribeStart?.();
});
} catch (err: unknown) {
if (err instanceof ApiError) {
throw err;
}
if ((err as NodeJS.ErrnoException).code === "ENOENT") {
throw notFound("Routine not found");
}
rethrowAsApiError(err);
}
});
router.get(
"/routines/:id/run/stream",
makeRunStreamHandler({
resolveStore: resolveRoutineStore,
getEntity: (store, id) => store.getRoutine(id),
notFoundMessage: "Routine not found",
}),
);
// GET /routines/:id/runs — get execution history
router.get("/routines/:id/runs", async (req: Request, res: Response) => {