fix(sse): /events keepalive heartbeat — keeps idle agenthub_work alive across short MCP client timeouts (TSK-0042)

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
chahinebrini 2026-07-01 01:52:46 +02:00
parent 10f3184ea2
commit 8b17baae96
2 changed files with 26 additions and 4 deletions

View File

@ -18,6 +18,8 @@ import { eventBus, emitChange } from './events.js';
import type { AgentHubEvent } from './events.js'; import type { AgentHubEvent } from './events.js';
import type { Task, Handoff, Decision, Memory, Message } from '../core/schema.js'; import type { Task, Handoff, Decision, Memory, Message } from '../core/schema.js';
const SSE_KEEPALIVE_MS = 10_000;
function notFound(reply: FastifyReply, resource: string) { function notFound(reply: FastifyReply, resource: string) {
return reply.status(404).send({ error: `${resource} not found` }); return reply.status(404).send({ error: `${resource} not found` });
} }
@ -58,8 +60,9 @@ export async function registerRoutes(app: FastifyInstance, cwd: string): Promise
// GET /events?role=<role> // GET /events?role=<role>
// //
// Keeps the connection open and streams JSON-encoded AgentHubEvent objects // Keeps the connection open and streams JSON-encoded AgentHubEvent objects
// as SSE data lines. Sends a keepalive comment (": \n\n") every 25 s so // as SSE data lines. Sends an immediate keepalive comment and then another
// proxies and clients detect the connection is still alive. // one every 10 s so proxies, fetch(), and MCP clients don't treat an idle
// work-loop request as a dead connection.
// //
// Optional ?role= filter: tasks whose role doesn't match are dropped // Optional ?role= filter: tasks whose role doesn't match are dropped
// server-side. All handoff / decision / memory events are always forwarded. // server-side. All handoff / decision / memory events are always forwarded.
@ -81,6 +84,7 @@ export async function registerRoutes(app: FastifyInstance, cwd: string): Promise
Connection: 'keep-alive', Connection: 'keep-alive',
}); });
raw.flushHeaders(); raw.flushHeaders();
raw.write(': connected\n\n');
const listener = (event: AgentHubEvent) => { const listener = (event: AgentHubEvent) => {
// Server-side role filter: skip tasks that belong to a different role. // Server-side role filter: skip tasks that belong to a different role.
@ -94,8 +98,8 @@ export async function registerRoutes(app: FastifyInstance, cwd: string): Promise
eventBus.on('change', listener); eventBus.on('change', listener);
const keepAliveTimer = setInterval(() => { const keepAliveTimer = setInterval(() => {
raw.write(':\n\n'); raw.write(': keepalive\n\n');
}, 25_000); }, SSE_KEEPALIVE_MS);
// Clean up when the client disconnects (or the server closes). // Clean up when the client disconnects (or the server closes).
request.raw.on('close', () => { request.raw.on('close', () => {

View File

@ -261,6 +261,24 @@ describe('SSE stream e2e', () => {
rmSync(cwd, { recursive: true, force: true }); rmSync(cwd, { recursive: true, force: true });
}); });
it('sends an immediate keepalive comment when /events connects', async () => {
const controller = new AbortController();
const res = await fetch(`${server.url}/events`, { signal: controller.signal });
expect(res.status).toBe(200);
expect(res.headers.get('content-type')).toContain('text/event-stream');
expect(res.body).toBeTruthy();
const reader = res.body!.getReader();
const timeout = new Promise<never>((_, reject) => {
setTimeout(() => reject(new Error('SSE: no keepalive frame within 1 s')), 1000);
});
const chunk = await Promise.race([reader.read(), timeout]);
controller.abort();
expect(chunk.done).toBe(false);
expect(new TextDecoder().decode(chunk.value)).toContain(': connected\n\n');
}, 3000);
it('delivers task/created to a connected SSE subscriber', async () => { it('delivers task/created to a connected SSE subscriber', async () => {
const controller = new AbortController(); const controller = new AbortController();