feat(realtime): SSE event channel + agenthub watch (TSK-0006 part 1)

- src/server/events.ts: in-process EventEmitter bus (AgentHubEventBus);
  singleton per server process, unlimited listeners for SSE fan-out.
- src/server/routes.ts: GET /events SSE endpoint — keepalive every 25 s,
  optional ?role= server-side filter, reply.hijack() for clean streaming;
  eventBus.publish() called on every successful mutating route (POST tasks/
  handoffs/decisions/memory, PATCH tasks/:id).
- src/cli/commands/watch.ts: SSE client using Node fetch + ReadableStream
  reader; exports parseSSEBuffer + formatEvent for testability; --once flag
  exits 0 after first event (turn-based agent use case); --role for client-
  side filtering (additive to server-side filter).
- src/cli/index.ts: register agenthub watch command via resolveContext.
- tests/sse.test.ts: 18 new tests (74 total, all green) — unit tests for
  parseSSEBuffer + formatEvent, integration tests for eventBus on each
  mutation type, E2E SSE stream + role-filter tests over real HTTP.

Limitation (documented): only REST mutations through the server emit events.
Local-CLI writes (direct file+SQLite) are invisible to subscribers — a
filesystem-watch increment is deferred to a later phase.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

src/cli/commands/watch.ts

@@ -0,0 +1,133 @@
+/**
+ * `agenthub watch` — subscribe to live events from a running AgentHub server.
+ *
+ * Connects to GET /events (Server-Sent Events) and prints each change
+ * compactly to stdout. No new dependencies: uses Node's built-in fetch +
+ * ReadableStream reader.
+ *
+ * Flags:
+ *   --once          Exit 0 after the first event is received. Lets a
+ *                   turn-based agent use this as a blocking wait: run in
+ *                   background, get woken up when something changes.
+ *   --role <role>   Client-side filter: suppress events whose `role` field
+ *                   doesn't match. (The server also accepts ?role= for a
+ *                   server-side filter, reducing traffic; both can be used.)
+ */
+
+import type { AgentHubEvent } from '../../server/events.js';
+
+// Re-export so tests can import type + helpers from one place.
+export type { AgentHubEvent } from '../../server/events.js';
+
+/**
+ * Parse all complete SSE events from a text buffer.
+ *
+ * SSE wire format:  "data: <json>\n\n" per event, ": \n\n" for keepalives.
+ * This function splits on double-newline boundaries, extracts the `data:`
+ * line from each complete block, and returns anything that didn't end with
+ * "\n\n" as `remaining` (to be prepended to the next chunk).
+ */
+export function parseSSEBuffer(buffer: string): { events: AgentHubEvent[]; remaining: string } {
+  const parts = buffer.split('\n\n');
+  const remaining = parts.pop() ?? ''; // last segment may be incomplete
+  const events: AgentHubEvent[] = [];
+
+  for (const part of parts) {
+    // A keepalive block looks like ":" — no data line.
+    const dataLine = part.split('\n').find((l) => l.startsWith('data: '));
+    if (!dataLine) continue;
+    try {
+      events.push(JSON.parse(dataLine.slice(6)) as AgentHubEvent);
+    } catch {
+      // Ignore malformed JSON — should never happen in practice.
+    }
+  }
+
+  return { events, remaining };
+}
+
+/**
+ * Format an event as a single compact log line, e.g.:
+ *   [task/updated] TSK-0003 My task title status=done (windows-claude)
+ */
+export function formatEvent(event: AgentHubEvent): string {
+  const parts: string[] = [`[${event.type}/${event.action}]`, event.id];
+  if (event.title) parts.push(event.title);
+  if (event.status) parts.push(`status=${event.status}`);
+  if (event.role) parts.push(`role=${event.role}`);
+  if (event.assignedTo) parts.push(`(${event.assignedTo})`);
+  return parts.join(' ');
+}
+
+/**
+ * Connect to the AgentHub server's SSE endpoint and stream events to stdout.
+ *
+ * Exits the process when:
+ *   - `--once` is set and the first event arrives (exit 0).
+ *   - The server closes the stream (normal exit).
+ *   - A connection error occurs (exit 1).
+ */
+export async function watchEvents(
+  serverUrl: string,
+  options: { once?: boolean; role?: string } = {},
+): Promise<void> {
+  const url = new URL('/events', serverUrl);
+  // Pass role to the server for an additional server-side filter (saves
+  // bandwidth on high-volume setups, optional).
+  if (options.role) url.searchParams.set('role', options.role);
+
+  let response: Response;
+  try {
+    response = await fetch(url.toString(), {
+      headers: { Accept: 'text/event-stream' },
+    });
+  } catch (err) {
+    const msg = err instanceof Error ? err.message : String(err);
+    console.error(`Cannot reach AgentHub server at ${serverUrl}: ${msg}`);
+    process.exit(1);
+    return; // unreachable; satisfies TypeScript
+  }
+
+  if (!response.ok || !response.body) {
+    console.error(`AgentHub server returned ${response.status} for /events`);
+    process.exit(1);
+    return;
+  }
+
+  const reader = response.body.getReader();
+  const decoder = new TextDecoder();
+  let buffer = '';
+
+  while (true) {
+    let done: boolean;
+    let value: Uint8Array | undefined;
+    try {
+      ({ done, value } = await reader.read());
+    } catch {
+      // Connection closed (e.g. AbortController or server shutdown).
+      break;
+    }
+    if (done) break;
+    if (value) {
+      buffer += decoder.decode(value, { stream: true });
+    }
+
+    const { events, remaining } = parseSSEBuffer(buffer);
+    buffer = remaining;
+
+    for (const event of events) {
+      // Client-side role filter: skip tasks that don't match the requested
+      // role. Non-task events (handoffs, decisions, memory) always print.
+      if (options.role && event.type === 'task' && event.role !== undefined && event.role !== options.role) {
+        continue;
+      }
+
+      console.log(formatEvent(event));
+
+      if (options.once) {
+        await reader.cancel();
+        return;
+      }
+    }
+  }
+}

src/cli/index.ts

@@ -8,6 +8,7 @@ import { decisionCreate, decisionList } from './commands/decision.js';
 import { delegate } from './commands/delegate.js';
 import { serverStart } from './commands/server.js';
 import { update } from './commands/update.js';
+import { watchEvents } from './commands/watch.js';
 import { loadConfig } from '../core/config.js';
 import { findProjectRoot } from '../core/paths.js';
 import { discoverServer } from '../discovery.js';
@@ -351,6 +352,23 @@ export function createProgram(cwd: string): Command {
       await update();
     });
 
+  // ─── watch ───────────────────────────────────────────────────────────────
+  program
+    .command('watch')
+    .description('Stream live AgentHub events from a running server (SSE)')
+    .option('--once', 'Exit 0 after the first event (useful as a blocking wait for agents)')
+    .option('--role <role>', 'Client-side role filter (only show events for this role)')
+    .action(async (options) => {
+      const { serverUrl } = await resolveContext(program, cwd);
+      if (!serverUrl) {
+        console.error('No AgentHub server found. Start one with: agenthub server start --host 0.0.0.0');
+        process.exit(1);
+        return;
+      }
+      await watchEvents(serverUrl, { once: options.once as boolean | undefined, role: options.role as string | undefined });
+    });
+
+  // ─── server ──────────────────────────────────────────────────────────────
   const serverCmd = new Command('server').description('Optional local API server');
   serverCmd
     .command('start')

src/server/events.ts

@@ -0,0 +1,33 @@
+import { EventEmitter } from 'node:events';
+
+export type AgentHubEventType = 'task' | 'handoff' | 'decision' | 'memory';
+export type AgentHubEventAction = 'created' | 'updated';
+
+export interface AgentHubEvent {
+  type: AgentHubEventType;
+  action: AgentHubEventAction;
+  id: string;
+  title?: string;
+  status?: string;
+  role?: string;
+  assignedTo?: string;
+}
+
+/**
+ * In-process event bus. Routes emit here on every successful mutating REST
+ * operation; the SSE /events handler fans them out to connected subscribers.
+ *
+ * Singleton per Node.js process — in server mode that is always exactly one
+ * process, which is the intended topology.
+ */
+class AgentHubEventBus extends EventEmitter {
+  /** Publish a change event to all current SSE subscribers. */
+  publish(event: AgentHubEvent): void {
+    this.emit('change', event);
+  }
+}
+
+export const eventBus = new AgentHubEventBus();
+// Allow an arbitrary number of SSE clients without triggering the
+// default-listener-count warning.
+eventBus.setMaxListeners(0);

src/server/routes.ts

@@ -7,6 +7,8 @@ import { getStatus, updateStatus } from '../core/services/statusService.js';
 import { autoDelegate, suggestDelegation } from '../core/services/delegateService.js';
 import { loadConfig } from '../core/config.js';
 import { renderBoardHtml } from './board.js';
+import { eventBus } from './events.js';
+import type { AgentHubEvent } from './events.js';
 import type { Task, Handoff, Decision, Memory } from '../core/schema.js';
 
 function notFound(reply: FastifyReply, resource: string) {
@@ -24,20 +26,83 @@ export async function registerRoutes(app: FastifyInstance, cwd: string): Promise
   const boardHtml = renderBoardHtml();
   app.get('/board', async (_request, reply) => reply.type('text/html; charset=utf-8').send(boardHtml));
 
+  // ─── Server-Sent Events ──────────────────────────────────────────────────
+  // GET /events?role=<role>
+  //
+  // Keeps the connection open and streams JSON-encoded AgentHubEvent objects
+  // as SSE data lines. Sends a keepalive comment (": \n\n") every 25 s so
+  // proxies and clients detect the connection is still alive.
+  //
+  // Optional ?role= filter: tasks whose role doesn't match are dropped
+  // server-side. All handoff / decision / memory events are always forwarded.
+  //
+  // Limitation: only mutations performed through this server emit events.
+  // Direct local-CLI writes (file + SQLite) bypass the event bus and are
+  // therefore invisible to subscribers. A filesystem-watch layer can be
+  // added in a later increment.
+  app.get('/events', async (request, reply) => {
+    const { role } = request.query as { role?: string };
+
+    // Take full control of the raw response so Fastify doesn't interfere.
+    reply.hijack();
+
+    const raw = reply.raw;
+    raw.writeHead(200, {
+      'Content-Type': 'text/event-stream',
+      'Cache-Control': 'no-cache',
+      Connection: 'keep-alive',
+    });
+    raw.flushHeaders();
+
+    const listener = (event: AgentHubEvent) => {
+      // Server-side role filter: skip tasks that belong to a different role.
+      // Handoffs, decisions and memory always pass through.
+      if (role && event.type === 'task' && event.role !== undefined && event.role !== role) {
+        return;
+      }
+      raw.write(`data: ${JSON.stringify(event)}\n\n`);
+    };
+
+    eventBus.on('change', listener);
+
+    const keepAliveTimer = setInterval(() => {
+      raw.write(':\n\n');
+    }, 25_000);
+
+    // Clean up when the client disconnects (or the server closes).
+    request.raw.on('close', () => {
+      clearInterval(keepAliveTimer);
+      eventBus.off('change', listener);
+    });
+  });
+
+  // ─── Status ──────────────────────────────────────────────────────────────
   app.get('/status', async () => ({ body: getStatus(cwd) }));
   app.post('/status/update', async () => ({ body: updateStatus(cwd) }));
 
+  // ─── Tasks ───────────────────────────────────────────────────────────────
   app.get('/tasks', async (request) => {
     const { status, role } = request.query as { status?: string; role?: string };
     return listTasks(cwd, { status, role });
   });
 
   app.post('/tasks', async (request, reply) => {
+    let task: Task;
     try {
-      return createTask(cwd, request.body as Partial<Task>);
+      task = createTask(cwd, request.body as Partial<Task>);
     } catch (err) {
       return badRequest(reply, err instanceof Error ? err.message : 'Invalid task');
     }
+    eventBus.publish({
+      type: 'task',
+      action: 'created',
+      id: task.id,
+      title: task.title,
+      status: task.status,
+      role: task.role,
+      assignedTo: task.assignedTo,
+    });
+    return task;
   });
 
   app.get('/tasks/:id', async (request, reply) => {
@@ -54,30 +119,58 @@ export async function registerRoutes(app: FastifyInstance, cwd: string): Promise
     const { id } = request.params as { id: string };
     const patch = request.body as Partial<Task>;
 
+    let task: Task;
     switch (patch.status) {
       case 'in_progress':
         if (!patch.assignedTo) return badRequest(reply, 'assignedTo is required to claim a task (in_progress)');
-        return claimTask(cwd, id, patch.assignedTo);
+        task = claimTask(cwd, id, patch.assignedTo);
+        break;
       case 'done':
-        return doneTask(cwd, id);
+        task = doneTask(cwd, id);
+        break;
       case 'review':
-        return reviewTask(cwd, id);
+        task = reviewTask(cwd, id);
+        break;
       case 'cancelled':
-        return cancelTask(cwd, id);
+        task = cancelTask(cwd, id);
+        break;
       case 'open':
-        return reopenTask(cwd, id);
+        task = reopenTask(cwd, id);
+        break;
       default:
         return badRequest(reply, 'Unsupported patch: status must be one of open, in_progress, review, done, cancelled');
     }
+
+    eventBus.publish({
+      type: 'task',
+      action: 'updated',
+      id: task.id,
+      title: task.title,
+      status: task.status,
+      role: task.role,
+      assignedTo: task.assignedTo,
+    });
+    return task;
   });
 
+  // ─── Handoffs ────────────────────────────────────────────────────────────
   app.get('/handoffs', async () => listHandoffs(cwd));
   app.post('/handoffs', async (request, reply) => {
+    let handoff: Handoff;
     try {
-      return createHandoff(cwd, request.body as Partial<Handoff>);
+      handoff = createHandoff(cwd, request.body as Partial<Handoff>);
     } catch (err) {
       return badRequest(reply, err instanceof Error ? err.message : 'Invalid handoff');
     }
+    eventBus.publish({
+      type: 'handoff',
+      action: 'created',
+      id: handoff.id,
+      title: handoff.summary,
+      role: handoff.toRole,
+      assignedTo: handoff.toAgent,
+    });
+    return handoff;
   });
   app.get('/handoffs/:id', async (request, reply) => {
     const { id } = request.params as { id: string };
@@ -89,28 +182,47 @@ export async function registerRoutes(app: FastifyInstance, cwd: string): Promise
     }
   });
 
+  // ─── Decisions ───────────────────────────────────────────────────────────
   app.get('/decisions', async () => listDecisions(cwd));
   app.post('/decisions', async (request, reply) => {
+    let decision: Decision;
     try {
-      return createDecision(cwd, request.body as Partial<Decision>);
+      decision = createDecision(cwd, request.body as Partial<Decision>);
     } catch (err) {
       return badRequest(reply, err instanceof Error ? err.message : 'Invalid decision');
     }
+    eventBus.publish({
+      type: 'decision',
+      action: 'created',
+      id: decision.id,
+      title: decision.title,
+    });
+    return decision;
   });
 
+  // ─── Memory ──────────────────────────────────────────────────────────────
   app.get('/memory', async () => listMemory(cwd));
   app.post('/memory', async (request, reply) => {
+    let memory: Memory;
     try {
-      return addMemory(cwd, request.body as Partial<Memory>);
+      memory = addMemory(cwd, request.body as Partial<Memory>);
     } catch (err) {
       return badRequest(reply, err instanceof Error ? err.message : 'Invalid memory');
     }
+    eventBus.publish({
+      type: 'memory',
+      action: 'created',
+      id: memory.id,
+      title: memory.title,
+    });
+    return memory;
   });
   app.get('/memory/search', async (request) => {
     const { q } = request.query as { q: string };
     return searchMemory(cwd, q ?? '');
   });
 
+  // ─── Delegate ────────────────────────────────────────────────────────────
   app.post('/delegate', async (request) => {
     const { auto } = request.query as { auto?: string };
     const config = loadConfig(cwd);

tests/sse.test.ts

@@ -0,0 +1,322 @@
+/**
+ * Tests for the SSE realtime channel (TSK-0006 part 1).
+ *
+ * Three concerns:
+ *   1. Unit: parseSSEBuffer + formatEvent (pure functions, no I/O).
+ *   2. Integration: eventBus emits on REST mutations (via app.inject).
+ *   3. E2E: a real server delivers a task/created event over the SSE stream.
+ */
+
+import { describe, it, expect, beforeEach, afterEach } from 'vitest';
+import { mkdtempSync, rmSync } from 'fs';
+import { tmpdir } from 'os';
+import { join } from 'path';
+import { buildApp } from '../src/server/index.js';
+import { startServer } from '../src/server/index.js';
+import { eventBus } from '../src/server/events.js';
+import type { AgentHubEvent } from '../src/server/events.js';
+import { parseSSEBuffer, formatEvent } from '../src/cli/commands/watch.js';
+import { init } from '../src/cli/commands/init.js';
+
+// ─── 1. Unit: SSE buffer parser ──────────────────────────────────────────────
+
+describe('parseSSEBuffer', () => {
+  it('parses a single complete event', () => {
+    const buf = 'data: {"type":"task","action":"created","id":"TSK-0001"}\n\n';
+    const { events, remaining } = parseSSEBuffer(buf);
+    expect(events).toHaveLength(1);
+    expect(events[0].type).toBe('task');
+    expect(events[0].action).toBe('created');
+    expect(events[0].id).toBe('TSK-0001');
+    expect(remaining).toBe('');
+  });
+
+  it('parses multiple complete events in one buffer', () => {
+    const buf =
+      'data: {"type":"task","action":"created","id":"TSK-0001"}\n\n' +
+      'data: {"type":"memory","action":"created","id":"MEM-001"}\n\n';
+    const { events, remaining } = parseSSEBuffer(buf);
+    expect(events).toHaveLength(2);
+    expect(events[0].type).toBe('task');
+    expect(events[1].type).toBe('memory');
+    expect(remaining).toBe('');
+  });
+
+  it('ignores SSE keepalive comment blocks', () => {
+    const buf =
+      ':\n\n' +
+      'data: {"type":"handoff","action":"created","id":"HOF-0001"}\n\n';
+    const { events } = parseSSEBuffer(buf);
+    expect(events).toHaveLength(1);
+    expect(events[0].id).toBe('HOF-0001');
+  });
+
+  it('keeps an incomplete tail in `remaining`', () => {
+    const buf =
+      'data: {"type":"task","action":"created","id":"TSK-0001"}\n\n' +
+      'data: {"type":"deci'; // truncated chunk
+    const { events, remaining } = parseSSEBuffer(buf);
+    expect(events).toHaveLength(1);
+    expect(remaining).toBe('data: {"type":"deci');
+  });
+
+  it('returns empty events and the full buffer when there is no complete event', () => {
+    const buf = 'data: {"type":"task"'; // no \n\n yet
+    const { events, remaining } = parseSSEBuffer(buf);
+    expect(events).toHaveLength(0);
+    expect(remaining).toBe('data: {"type":"task"');
+  });
+
+  it('silently skips malformed JSON without throwing', () => {
+    const buf = 'data: NOT_JSON\n\n';
+    const { events } = parseSSEBuffer(buf);
+    expect(events).toHaveLength(0);
+  });
+});
+
+// ─── 2. Unit: formatEvent ────────────────────────────────────────────────────
+
+describe('formatEvent', () => {
+  it('formats a minimal event (no optional fields)', () => {
+    const ev: AgentHubEvent = { type: 'decision', action: 'created', id: 'DEC-0001' };
+    expect(formatEvent(ev)).toBe('[decision/created] DEC-0001');
+  });
+
+  it('formats a full task event matching the spec example', () => {
+    const ev: AgentHubEvent = {
+      type: 'task',
+      action: 'updated',
+      id: 'TSK-0003',
+      title: 'Implement auth',
+      status: 'done',
+      role: 'implementer',
+      assignedTo: 'windows-claude',
+    };
+    expect(formatEvent(ev)).toBe('[task/updated] TSK-0003 Implement auth status=done role=implementer (windows-claude)');
+  });
+
+  it('omits absent optional fields', () => {
+    const ev: AgentHubEvent = { type: 'task', action: 'created', id: 'TSK-0001', status: 'open' };
+    expect(formatEvent(ev)).toBe('[task/created] TSK-0001 status=open');
+  });
+});
+
+// ─── 3. Integration: eventBus fires on REST mutations ────────────────────────
+
+describe('eventBus mutations', () => {
+  let cwd: string;
+  let app: ReturnType<typeof buildApp>;
+  const collected: AgentHubEvent[] = [];
+
+  beforeEach(() => {
+    cwd = mkdtempSync(join(tmpdir(), 'ah-sse-bus-'));
+    init(cwd, { projectName: 'sse-bus-test', yes: true });
+    app = buildApp(cwd);
+    eventBus.on('change', (e: AgentHubEvent) => collected.push(e));
+  });
+
+  afterEach(async () => {
+    eventBus.removeAllListeners('change');
+    collected.length = 0;
+    await app.close();
+    rmSync(cwd, { recursive: true, force: true });
+  });
+
+  it('emits task/created when POST /tasks succeeds', async () => {
+    await app.inject({ method: 'POST', url: '/tasks', payload: { title: 'Bus task', role: 'implementer' } });
+    expect(collected).toHaveLength(1);
+    expect(collected[0]).toMatchObject({ type: 'task', action: 'created', id: 'TSK-0001', title: 'Bus task', role: 'implementer' });
+  });
+
+  it('emits task/updated when PATCH /tasks/:id changes status', async () => {
+    await app.inject({ method: 'POST', url: '/tasks', payload: { title: 'T', role: 'implementer' } });
+    collected.length = 0; // clear the created event
+
+    await app.inject({ method: 'PATCH', url: '/tasks/TSK-0001', payload: { status: 'done' } });
+    expect(collected).toHaveLength(1);
+    expect(collected[0]).toMatchObject({ type: 'task', action: 'updated', id: 'TSK-0001', status: 'done' });
+  });
+
+  it('emits task/updated with assignedTo when a task is claimed', async () => {
+    await app.inject({ method: 'POST', url: '/tasks', payload: { title: 'T', role: 'implementer' } });
+    collected.length = 0;
+
+    await app.inject({ method: 'PATCH', url: '/tasks/TSK-0001', payload: { status: 'in_progress', assignedTo: 'mac-claude' } });
+    expect(collected).toHaveLength(1);
+    expect(collected[0]).toMatchObject({ type: 'task', action: 'updated', status: 'in_progress', assignedTo: 'mac-claude' });
+  });
+
+  it('emits handoff/created when POST /handoffs succeeds', async () => {
+    await app.inject({
+      method: 'POST',
+      url: '/handoffs',
+      payload: { fromRole: 'architect', toRole: 'implementer', summary: 'Design complete', context: 'See decisions' },
+    });
+    expect(collected).toHaveLength(1);
+    expect(collected[0]).toMatchObject({ type: 'handoff', action: 'created' });
+  });
+
+  it('emits decision/created when POST /decisions succeeds', async () => {
+    await app.inject({
+      method: 'POST',
+      url: '/decisions',
+      payload: { title: 'Use SQLite', decision: 'SQLite for local storage', context: 'Low overhead' },
+    });
+    expect(collected).toHaveLength(1);
+    expect(collected[0]).toMatchObject({ type: 'decision', action: 'created', title: 'Use SQLite' });
+  });
+
+  it('emits memory/created when POST /memory succeeds', async () => {
+    await app.inject({
+      method: 'POST',
+      url: '/memory',
+      payload: { title: 'DNS cache fact', category: 'technical', content: 'TTL matters' },
+    });
+    expect(collected).toHaveLength(1);
+    expect(collected[0]).toMatchObject({ type: 'memory', action: 'created', title: 'DNS cache fact' });
+  });
+
+  it('does NOT emit when task creation fails (400)', async () => {
+    // title: '' fails TaskSchema min(1) — createTask throws, route returns 400.
+    const res = await app.inject({ method: 'POST', url: '/tasks', payload: { title: '' } });
+    expect(res.statusCode).toBe(400);
+    // No event should be emitted on failure.
+    expect(collected).toHaveLength(0);
+  });
+});
+
+// ─── 4. E2E: SSE stream delivers events over a real HTTP connection ───────────
+
+describe('SSE stream e2e', () => {
+  let cwd: string;
+  let server: Awaited<ReturnType<typeof startServer>>;
+
+  beforeEach(async () => {
+    cwd = mkdtempSync(join(tmpdir(), 'ah-sse-e2e-'));
+    init(cwd, { projectName: 'sse-e2e-test', yes: true });
+    server = await startServer(cwd, { host: '127.0.0.1', port: 0 });
+  });
+
+  afterEach(async () => {
+    await server.app.close();
+    rmSync(cwd, { recursive: true, force: true });
+  });
+
+  it('delivers task/created to a connected SSE subscriber', async () => {
+    const controller = new AbortController();
+
+    // Start listening BEFORE posting the task.
+    const firstEventPromise = new Promise<AgentHubEvent>((resolve, reject) => {
+      const timeout = setTimeout(
+        () => reject(new Error('SSE: no event received within 3 s')),
+        3000,
+      );
+
+      fetch(`${server.url}/events`, { signal: controller.signal })
+        .then(async (res) => {
+          if (!res.body) {
+            clearTimeout(timeout);
+            reject(new Error('SSE response has no body'));
+            return;
+          }
+          const reader = res.body.getReader();
+          const dec = new TextDecoder();
+          let buf = '';
+
+          while (true) {
+            let done: boolean;
+            let value: Uint8Array | undefined;
+            try {
+              ({ done, value } = await reader.read());
+            } catch {
+              break; // AbortError when controller.abort() is called
+            }
+            if (done) break;
+            if (value) buf += dec.decode(value, { stream: true });
+
+            const { events, remaining } = parseSSEBuffer(buf);
+            buf = remaining;
+
+            for (const ev of events) {
+              clearTimeout(timeout);
+              resolve(ev);
+              return;
+            }
+          }
+        })
+        .catch((err: unknown) => {
+          // AbortError is expected when we abort after receiving the event.
+          if (err instanceof Error && err.name === 'AbortError') return;
+          reject(err);
+        });
+    });
+
+    // Give the SSE connection a moment to be established before posting.
+    await new Promise((r) => setTimeout(r, 80));
+
+    // Trigger a mutation through the REST API.
+    await fetch(`${server.url}/tasks`, {
+      method: 'POST',
+      headers: { 'Content-Type': 'application/json' },
+      body: JSON.stringify({ title: 'SSE e2e task', role: 'implementer' }),
+    });
+
+    const event = await firstEventPromise;
+    controller.abort(); // stop the SSE listener
+
+    expect(event.type).toBe('task');
+    expect(event.action).toBe('created');
+    expect(event.id).toBe('TSK-0001');
+    expect(event.title).toBe('SSE e2e task');
+    expect(event.role).toBe('implementer');
+  }, 5000);
+
+  it('role filter on /events?role= suppresses non-matching tasks', async () => {
+    const controller = new AbortController();
+    const received: AgentHubEvent[] = [];
+
+    // Subscribe with role=architect — should NOT see the implementer task.
+    const listenDone = fetch(`${server.url}/events?role=architect`, { signal: controller.signal })
+      .then(async (res) => {
+        if (!res.body) return;
+        const reader = res.body.getReader();
+        const dec = new TextDecoder();
+        let buf = '';
+        while (true) {
+          let done: boolean;
+          let value: Uint8Array | undefined;
+          try {
+            ({ done, value } = await reader.read());
+          } catch {
+            break;
+          }
+          if (done) break;
+          if (value) buf += dec.decode(value, { stream: true });
+          const { events, remaining } = parseSSEBuffer(buf);
+          buf = remaining;
+          received.push(...events);
+        }
+      })
+      .catch((err: unknown) => {
+        if (err instanceof Error && err.name === 'AbortError') return;
+        throw err;
+      });
+
+    await new Promise((r) => setTimeout(r, 80));
+
+    // Post a task for 'implementer' role — should be filtered out.
+    await fetch(`${server.url}/tasks`, {
+      method: 'POST',
+      headers: { 'Content-Type': 'application/json' },
+      body: JSON.stringify({ title: 'Implementer task', role: 'implementer' }),
+    });
+
+    // Give time for any event to arrive (it shouldn't).
+    await new Promise((r) => setTimeout(r, 150));
+
+    controller.abort();
+    await listenDone;
+
+    expect(received).toHaveLength(0);
+  }, 5000);
+});