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

fix: agent log SSE always listens on default store, not project-scoped store

Browse Source 
The agent log SSE endpoint was calling getOrCreateProjectStore(projectId)
and attaching the listener to the returned store. But the in-process
TaskExecutor is always bound to the default store passed to createServer,
so agent:log events were emitted on that store, never reaching the
project-scoped store's EventEmitter.

Fix: remove the getOrCreateProjectStore call and always listen directly
on the default store. Also simplifies the handler by removing the
unnecessary async IIFE (the listener can now be attached synchronously)
and eliminates the detachListener indirection.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
gsxdsm
2026-04-03 07:20:01 -07:00
parent 5148d4670b
commit d8f57e0d31
1 changed files with 13 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -130,7 +130,6 @@ export function createServer(store: TaskStore, options?: ServerOptions): ReturnT
// Per-task SSE endpoint for live agent log streaming
app.get("/api/tasks/:id/logs/stream", (req, res) => {
const taskId = req.params.id;
const projectId = typeof req.query.projectId === "string" ? req.query.projectId : undefined;
res.setHeader("Content-Type", "text/event-stream");
res.setHeader("Cache-Control", "no-cache");
@@ -140,27 +139,19 @@ export function createServer(store: TaskStore, options?: ServerOptions): ReturnT
res.write(": connected\n\n");
let activeStore = store;
let detachListener: (() => void) | null = null;
// agent:log events are emitted by the in-process TaskExecutor via
// store.appendAgentLog(). The executor is always bound to the default
// `store` passed to createServer — never to a project-scoped store
// created by getOrCreateProjectStore — so we must listen on `store`
// directly. Using getOrCreateProjectStore here would attach the listener
// to a different EventEmitter instance that the executor never writes to,
// breaking real-time log streaming.
const onAgentLog = (entry: { taskId: string; text: string; type: string; timestamp: string }) => {
if (entry.taskId !== taskId) return;
res.write(`event: agent:log\ndata: ${JSON.stringify(entry)}\n\n`);
};
void (async () => {
if (projectId) {
activeStore = await getOrCreateProjectStore(projectId);
}
const onAgentLog = (entry: { taskId: string; text: string; type: string; timestamp: string }) => {
if (entry.taskId !== taskId) return;
res.write(`event: agent:log\ndata: ${JSON.stringify(entry)}\n\n`);
};
activeStore.on("agent:log", onAgentLog);
detachListener = () => {
activeStore.off("agent:log", onAgentLog);
};
})().catch(() => {
res.write("event: error\ndata: \"Failed to attach log stream\"\n\n");
res.end();
});
store.on("agent:log", onAgentLog);
const heartbeat = setInterval(() => {
res.write(": heartbeat\n\n");
@@ -168,7 +159,7 @@ export function createServer(store: TaskStore, options?: ServerOptions): ReturnT
req.on("close", () => {
clearInterval(heartbeat);
detachListener?.();
store.off("agent:log", onAgentLog);
});
});