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>
feat(board): fix gaps A+B — full task status transitions + handoff who→whom
2026-06-26 00:16:16 +02:00 · 2026-06-25 23:41:54 +02:00 · 2026-06-25 23:35:28 +02:00
12 changed files with 1141 additions and 16 deletions
--- a/src/cli/commands/watch.ts
+++ b/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;
+      }
+    }
+  }
+}
--- a/src/cli/index.ts
+++ b/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')
--- a/src/core/index.ts
+++ b/src/core/index.ts
@ -15,6 +15,11 @@ export interface IndexEntry {
  role?: string;
  assignedTo?: string;
  tags?: string;
+  // Handoff-specific routing fields
+  fromRole?: string;
+  toRole?: string;
+  fromAgent?: string;
+  toAgent?: string;
 }

 export class Index {
@ -36,10 +41,24 @@ export class Index {
        status TEXT,
        role TEXT,
        assignedTo TEXT,
-        tags TEXT
+        tags TEXT,
+        fromRole TEXT,
+        toRole TEXT,
+        fromAgent TEXT,
+        toAgent TEXT
      );
      CREATE VIRTUAL TABLE IF NOT EXISTS search USING fts5(id, title, content);
    `);
+
+    // Migration: add handoff-routing columns to existing DBs that pre-date this schema.
+    const existingCols = new Set(
+      (this.db.pragma('table_info(entities)') as Array<{ name: string }>).map((r) => r.name),
+    );
+    for (const col of ['fromRole', 'toRole', 'fromAgent', 'toAgent']) {
+      if (!existingCols.has(col)) {
+        this.db.exec(`ALTER TABLE entities ADD COLUMN ${col} TEXT`);
+      }
+    }
  }

  upsert(entry: IndexEntry): void {
@ -48,16 +67,21 @@ export class Index {
      role: null,
      assignedTo: null,
      tags: null,
+      fromRole: null,
+      toRole: null,
+      fromAgent: null,
+      toAgent: null,
      ...entry,
    };

    const insert = this.db.prepare(`
-      INSERT INTO entities (id, type, title, content, filePath, createdAt, updatedAt, status, role, assignedTo, tags)
-      VALUES (@id, @type, @title, @content, @filePath, @createdAt, @updatedAt, @status, @role, @assignedTo, @tags)
+      INSERT INTO entities (id, type, title, content, filePath, createdAt, updatedAt, status, role, assignedTo, tags, fromRole, toRole, fromAgent, toAgent)
+      VALUES (@id, @type, @title, @content, @filePath, @createdAt, @updatedAt, @status, @role, @assignedTo, @tags, @fromRole, @toRole, @fromAgent, @toAgent)
      ON CONFLICT(id) DO UPDATE SET
        type=@type, title=@title, content=@content, filePath=@filePath,
        createdAt=@createdAt, updatedAt=@updatedAt, status=@status, role=@role,
-        assignedTo=@assignedTo, tags=@tags
+        assignedTo=@assignedTo, tags=@tags,
+        fromRole=@fromRole, toRole=@toRole, fromAgent=@fromAgent, toAgent=@toAgent
    `);
    insert.run(params);

--- a/src/core/services/handoffService.ts
+++ b/src/core/services/handoffService.ts
@ -34,6 +34,10 @@ export function createHandoff(cwd: string, options: Partial<Handoff> = {}): Hand
    filePath,
    createdAt: handoff.createdAt,
    updatedAt: handoff.createdAt,
+    fromRole: handoff.fromRole,
+    toRole: handoff.toRole,
+    fromAgent: handoff.fromAgent,
+    toAgent: handoff.toAgent,
  });
  index.close();

--- a/src/core/services/taskService.ts
+++ b/src/core/services/taskService.ts
@ -63,6 +63,18 @@ export function doneTask(cwd: string, id: string): Task {
  return updateTask(cwd, id, { status: 'done' });
 }

+export function reviewTask(cwd: string, id: string): Task {
+  return updateTask(cwd, id, { status: 'review' });
+}
+
+export function cancelTask(cwd: string, id: string): Task {
+  return updateTask(cwd, id, { status: 'cancelled' });
+}
+
+export function reopenTask(cwd: string, id: string): Task {
+  return updateTask(cwd, id, { status: 'open' });
+}
+
 function toIndexEntry(task: Task, filePath: string) {
  return {
    id: task.id,
--- a/src/server/board.ts
+++ b/src/server/board.ts
@ -0,0 +1,291 @@
+/**
+ * Self-contained Trello-like task board, served at `GET /board`.
+ *
+ * Design constraints (MVP — maximum simplicity):
+ *  - One static HTML page: inline CSS + JS, no build step, no framework, no npm deps.
+ *  - Reads only the existing same-origin endpoints (`/tasks`, `/handoffs`,
+ *    `/decisions`). It never mutates state — purely an observability surface.
+ *  - Auto-refreshes on a small interval via `setInterval` + `fetch`.
+ *
+ * The column skeleton is rendered server-side so the markup (and its
+ * `data-column` markers) exist even before the client JS runs; JS only fills
+ * the cards.
+ */
+
+export interface BoardColumn {
+  key: string;
+  label: string;
+}
+
+/** Task statuses, in board order. Mirrors `TaskStatus` in core/schema.ts. */
+export const BOARD_COLUMNS: BoardColumn[] = [
+  { key: 'open', label: 'Open' },
+  { key: 'in_progress', label: 'In Progress' },
+  { key: 'review', label: 'Review' },
+  { key: 'done', label: 'Done' },
+  { key: 'cancelled', label: 'Cancelled' },
+];
+
+function columnSkeleton(): string {
+  return BOARD_COLUMNS.map(
+    (c) => `      <section class="column" data-column="${c.key}">
+        <header class="col-head">
+          <span class="col-label">${c.label}</span>
+          <span class="col-count" data-count="${c.key}">0</span>
+        </header>
+        <div class="cards" data-cards="${c.key}"></div>
+      </section>`,
+  ).join('\n');
+}
+
+export function renderBoardHtml(): string {
+  return `<!doctype html>
+<html lang="en">
+<head>
+  <meta charset="utf-8" />
+  <meta name="viewport" content="width=device-width, initial-scale=1" />
+  <meta name="color-scheme" content="dark" />
+  <title>AgentHub Board</title>
+  <style>
+    :root {
+      --bg: #0d1117;
+      --panel: #161b22;
+      --panel-2: #1c2330;
+      --border: #30363d;
+      --text: #e6edf3;
+      --muted: #8b949e;
+      --accent: #58a6ff;
+      --open: #8b949e;
+      --in_progress: #58a6ff;
+      --review: #d29922;
+      --done: #3fb950;
+      --cancelled: #6e7681;
+    }
+    * { box-sizing: border-box; }
+    html, body { margin: 0; height: 100%; }
+    body {
+      background: var(--bg);
+      color: var(--text);
+      font: 14px/1.5 -apple-system, BlinkMacSystemFont, "Segoe UI", Roboto, Helvetica, Arial, sans-serif;
+      padding: 16px 20px 32px;
+    }
+    .topbar {
+      display: flex; align-items: baseline; gap: 12px;
+      margin-bottom: 16px; flex-wrap: wrap;
+    }
+    .topbar h1 { font-size: 18px; margin: 0; font-weight: 600; }
+    .topbar .dot { width: 8px; height: 8px; border-radius: 50%; background: var(--done); display: inline-block; }
+    .topbar .dot.stale { background: var(--review); }
+    .topbar .dot.down { background: #f85149; }
+    .topbar .meta { color: var(--muted); font-size: 12px; }
+    .topbar .spacer { flex: 1; }
+
+    .board {
+      display: grid;
+      grid-template-columns: repeat(5, minmax(180px, 1fr));
+      gap: 12px;
+      align-items: start;
+    }
+    @media (max-width: 1100px) { .board { grid-template-columns: repeat(2, 1fr); } }
+    @media (max-width: 640px) { .board { grid-template-columns: 1fr; } }
+
+    .column {
+      background: var(--panel);
+      border: 1px solid var(--border);
+      border-radius: 10px;
+      padding: 10px;
+      min-height: 80px;
+    }
+    .col-head {
+      display: flex; align-items: center; justify-content: space-between;
+      margin-bottom: 10px; padding: 0 2px;
+    }
+    .col-label { font-weight: 600; font-size: 13px; letter-spacing: .02em; }
+    .col-count {
+      background: var(--panel-2); color: var(--muted);
+      border-radius: 999px; padding: 1px 8px; font-size: 12px;
+    }
+    .column[data-column="open"] .col-label { color: var(--open); }
+    .column[data-column="in_progress"] .col-label { color: var(--in_progress); }
+    .column[data-column="review"] .col-label { color: var(--review); }
+    .column[data-column="done"] .col-label { color: var(--done); }
+    .column[data-column="cancelled"] .col-label { color: var(--cancelled); }
+
+    .cards { display: flex; flex-direction: column; gap: 8px; }
+    .card {
+      background: var(--panel-2);
+      border: 1px solid var(--border);
+      border-left: 3px solid var(--accent);
+      border-radius: 8px;
+      padding: 8px 10px;
+    }
+    .column[data-column="open"] .card { border-left-color: var(--open); }
+    .column[data-column="in_progress"] .card { border-left-color: var(--in_progress); }
+    .column[data-column="review"] .card { border-left-color: var(--review); }
+    .column[data-column="done"] .card { border-left-color: var(--done); }
+    .column[data-column="cancelled"] .card { border-left-color: var(--cancelled); }
+    .card .id { color: var(--muted); font-size: 11px; font-family: ui-monospace, SFMono-Regular, Menlo, monospace; }
+    .card .title { margin: 2px 0 6px; font-weight: 500; }
+    .card .tags { display: flex; gap: 6px; flex-wrap: wrap; }
+    .badge {
+      font-size: 11px; border-radius: 999px; padding: 1px 8px;
+      background: rgba(88,166,255,.12); color: var(--accent);
+      border: 1px solid rgba(88,166,255,.25);
+    }
+    .badge.role { background: rgba(210,153,34,.12); color: var(--review); border-color: rgba(210,153,34,.25); }
+    .badge.agent { background: rgba(63,185,80,.12); color: var(--done); border-color: rgba(63,185,80,.25); }
+    .empty { color: var(--muted); font-size: 12px; padding: 4px 2px; }
+
+    .panels {
+      display: grid; grid-template-columns: 1fr 1fr; gap: 12px;
+      margin-top: 20px;
+    }
+    @media (max-width: 640px) { .panels { grid-template-columns: 1fr; } }
+    .panel {
+      background: var(--panel); border: 1px solid var(--border);
+      border-radius: 10px; padding: 12px 14px;
+    }
+    .panel h2 { font-size: 13px; margin: 0 0 10px; font-weight: 600; }
+    .row {
+      display: flex; gap: 8px; align-items: baseline;
+      padding: 6px 0; border-top: 1px solid var(--border);
+    }
+    .row:first-of-type { border-top: 0; }
+    .row .id { color: var(--muted); font-size: 11px; font-family: ui-monospace, SFMono-Regular, Menlo, monospace; white-space: nowrap; }
+    .row .what { flex: 1; }
+    .row .who-cell { font-size: 12px; white-space: nowrap; display: flex; align-items: baseline; gap: 4px; }
+    .row .who { display: flex; align-items: baseline; gap: 4px; color: var(--accent); font-size: 11px; white-space: nowrap; }
+    .row .when { color: var(--muted); font-size: 11px; white-space: nowrap; }
+  </style>
+</head>
+<body>
+  <div class="topbar">
+    <span class="dot" id="conn"></span>
+    <h1>AgentHub Board</h1>
+    <span class="spacer"></span>
+    <span class="meta" id="meta">loading…</span>
+  </div>
+
+  <main class="board" id="board">
+${columnSkeleton()}
+  </main>
+
+  <div class="panels">
+    <section class="panel">
+      <h2>Handoffs</h2>
+      <div id="handoffs"><div class="empty">none</div></div>
+    </section>
+    <section class="panel">
+      <h2>Decisions</h2>
+      <div id="decisions"><div class="empty">none</div></div>
+    </section>
+  </div>
+
+  <script>
+    var REFRESH_MS = 4000;
+    var COLUMNS = ${JSON.stringify(BOARD_COLUMNS.map((c) => c.key))};
+
+    function esc(s) {
+      return String(s == null ? '' : s)
+        .replace(/&/g, '&amp;').replace(/</g, '&lt;').replace(/>/g, '&gt;')
+        .replace(/"/g, '&quot;');
+    }
+    function ago(iso) {
+      var t = Date.parse(iso);
+      if (isNaN(t)) return '';
+      var s = Math.max(0, (Date.now() - t) / 1000);
+      if (s < 60) return Math.floor(s) + 's ago';
+      if (s < 3600) return Math.floor(s / 60) + 'm ago';
+      if (s < 86400) return Math.floor(s / 3600) + 'h ago';
+      return Math.floor(s / 86400) + 'd ago';
+    }
+    async function getJSON(path) {
+      var res = await fetch(path, { headers: { accept: 'application/json' } });
+      if (!res.ok) throw new Error(path + ' -> ' + res.status);
+      return res.json();
+    }
+    function taskCard(t) {
+      var tags = '';
+      if (t.role) tags += '<span class="badge role">' + esc(t.role) + '</span>';
+      if (t.assignedTo) tags += '<span class="badge agent">@' + esc(t.assignedTo) + '</span>';
+      return '<div class="card">' +
+        '<div class="id">' + esc(t.id) + '</div>' +
+        '<div class="title">' + esc(t.title) + '</div>' +
+        (tags ? '<div class="tags">' + tags + '</div>' : '') +
+        '</div>';
+    }
+    function renderBoard(tasks) {
+      var byCol = {};
+      COLUMNS.forEach(function (k) { byCol[k] = []; });
+      (tasks || []).forEach(function (t) {
+        var k = byCol[t.status] ? t.status : 'open';
+        byCol[k].push(t);
+      });
+      COLUMNS.forEach(function (k) {
+        var list = byCol[k];
+        var cards = document.querySelector('[data-cards="' + k + '"]');
+        var count = document.querySelector('[data-count="' + k + '"]');
+        if (count) count.textContent = String(list.length);
+        if (!cards) return;
+        cards.innerHTML = list.length
+          ? list.map(taskCard).join('')
+          : '<div class="empty">—</div>';
+      });
+    }
+    function handoffRoute(h) {
+      // Build "fromRole[@fromAgent] → toRole[@toAgent]" label
+      var from = esc(h.fromRole || '?');
+      var to   = esc(h.toRole   || '?');
+      if (h.fromAgent) from += '<span class="badge agent">@' + esc(h.fromAgent) + '</span>';
+      if (h.toAgent)   to   += '<span class="badge agent">@' + esc(h.toAgent)   + '</span>';
+      return '<span class="who">' + from + ' &rarr; ' + to + '</span>';
+    }
+    function renderHandoffs(items) {
+      var el = document.getElementById('handoffs');
+      if (!items || !items.length) { el.innerHTML = '<div class="empty">none</div>'; return; }
+      el.innerHTML = items.slice(0, 12).map(function (h) {
+        return '<div class="row">' +
+          '<span class="id">' + esc(h.id) + '</span>' +
+          '<span class="what">' + esc(h.title) + '</span>' +
+          '<span class="who-cell">' + handoffRoute(h) + '</span>' +
+          '<span class="when">' + esc(ago(h.createdAt)) + '</span>' +
+          '</div>';
+      }).join('');
+    }
+    function renderDecisions(items) {
+      var el = document.getElementById('decisions');
+      if (!items || !items.length) { el.innerHTML = '<div class="empty">none</div>'; return; }
+      el.innerHTML = items.slice(0, 12).map(function (d) {
+        var st = d.status ? '<span class="badge">' + esc(d.status) + '</span>' : '';
+        return '<div class="row">' +
+          '<span class="id">' + esc(d.id) + '</span>' +
+          '<span class="what">' + esc(d.title) + ' ' + st + '</span>' +
+          '<span class="when">' + esc(ago(d.createdAt)) + '</span>' +
+          '</div>';
+      }).join('');
+    }
+    function setConn(state) {
+      var dot = document.getElementById('conn');
+      dot.className = 'dot' + (state === 'ok' ? '' : state === 'stale' ? ' stale' : ' down');
+    }
+    async function refresh() {
+      try {
+        var r = await Promise.all([getJSON('/tasks'), getJSON('/handoffs'), getJSON('/decisions')]);
+        renderBoard(r[0]);
+        renderHandoffs(r[1]);
+        renderDecisions(r[2]);
+        setConn('ok');
+        document.getElementById('meta').textContent =
+          (r[0] || []).length + ' tasks · updated ' + new Date().toLocaleTimeString();
+      } catch (e) {
+        setConn('down');
+        document.getElementById('meta').textContent = 'disconnected — retrying…';
+      }
+    }
+    refresh();
+    setInterval(refresh, REFRESH_MS);
+  </script>
+</body>
+</html>
+`;
+}
--- a/src/server/events.ts
+++ b/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);
--- a/src/server/routes.ts
+++ b/src/server/routes.ts
@ -1,11 +1,14 @@
 import { FastifyInstance, FastifyReply } from 'fastify';
-import { createTask, listTasks, getTask, claimTask, doneTask } from '../core/services/taskService.js';
+import { createTask, listTasks, getTask, claimTask, doneTask, reviewTask, cancelTask, reopenTask } from '../core/services/taskService.js';
 import { createHandoff, listHandoffs, getHandoff } from '../core/services/handoffService.js';
 import { createDecision, listDecisions } from '../core/services/decisionService.js';
 import { addMemory, searchMemory, listMemory } from '../core/services/memoryService.js';
 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) {
@ -17,20 +20,89 @@ function badRequest(reply: FastifyReply, message: string) {
 }

 export async function registerRoutes(app: FastifyInstance, cwd: string): Promise<void> {
+  // Static, self-contained Trello-like board. Polls /tasks, /handoffs and
+  // /decisions on the same origin; no build step, no deps. Cached once — the
+  // markup is constant, only the data it fetches changes.
+  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) => {
@ -46,22 +118,59 @@ export async function registerRoutes(app: FastifyInstance, cwd: string): Promise
  app.patch('/tasks/:id', async (request, reply) => {
    const { id } = request.params as { id: string };
    const patch = request.body as Partial<Task>;
-    if (patch.status === 'in_progress' && patch.assignedTo) {
-      return claimTask(cwd, id, patch.assignedTo);
+
+    let task: Task;
+    switch (patch.status) {
+      case 'in_progress':
+        if (!patch.assignedTo) return badRequest(reply, 'assignedTo is required to claim a task (in_progress)');
+        task = claimTask(cwd, id, patch.assignedTo);
+        break;
+      case 'done':
+        task = doneTask(cwd, id);
+        break;
+      case 'review':
+        task = reviewTask(cwd, id);
+        break;
+      case 'cancelled':
+        task = cancelTask(cwd, id);
+        break;
+      case 'open':
+        task = reopenTask(cwd, id);
+        break;
+      default:
+        return badRequest(reply, 'Unsupported patch: status must be one of open, in_progress, review, done, cancelled');
    }
-    if (patch.status === 'done') {
-      return doneTask(cwd, id);
-    }
-    return badRequest(reply, 'Unsupported patch');
+
+    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 };
@ -73,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);
--- a/tests/handoffService.test.ts
+++ b/tests/handoffService.test.ts
@ -16,4 +16,34 @@ describe('handoffService', () => {
    expect(getHandoff(cwd, h.id).handoff.summary).toBe('s');
    expect(listHandoffs(cwd)).toHaveLength(1);
  });
+
+  it('listHandoffs carries fromRole and toRole in the index entry', () => {
+    createHandoff(cwd, {
+      fromRole: 'architect',
+      toRole: 'implementer',
+      summary: 'Hand over design',
+      context: 'done',
+    });
+    const items = listHandoffs(cwd);
+    expect(items).toHaveLength(1);
+    expect(items[0].fromRole).toBe('architect');
+    expect(items[0].toRole).toBe('implementer');
+    // agent names default to undefined when not supplied
+    expect(items[0].fromAgent == null).toBe(true);
+    expect(items[0].toAgent == null).toBe(true);
+  });
+
+  it('listHandoffs carries fromAgent and toAgent when supplied', () => {
+    createHandoff(cwd, {
+      fromRole: 'reviewer',
+      toRole: 'implementer',
+      fromAgent: 'claude',
+      toAgent: 'codex',
+      summary: 'Review complete',
+      context: 'lgtm',
+    });
+    const items = listHandoffs(cwd);
+    expect(items[0].fromAgent).toBe('claude');
+    expect(items[0].toAgent).toBe('codex');
+  });
 });
--- a/tests/server.test.ts
+++ b/tests/server.test.ts
@ -47,6 +47,48 @@ describe('server routes', () => {
    expect(res.statusCode).toBe(400);
  });

+  it('returns 400 when claiming without assignedTo', async () => {
+    await app.inject({ method: 'POST', url: '/tasks', payload: { title: 'A', role: 'implementer' } });
+    const res = await app.inject({ method: 'PATCH', url: '/tasks/TSK-0001', payload: { status: 'in_progress' } });
+    expect(res.statusCode).toBe(400);
+  });
+
+  it('PATCH /tasks/:id → review', async () => {
+    await app.inject({ method: 'POST', url: '/tasks', payload: { title: 'A', role: 'implementer' } });
+    const res = await app.inject({ method: 'PATCH', url: '/tasks/TSK-0001', payload: { status: 'review' } });
+    expect(res.statusCode).toBe(200);
+    expect(JSON.parse(res.payload).status).toBe('review');
+  });
+
+  it('PATCH /tasks/:id → cancelled', async () => {
+    await app.inject({ method: 'POST', url: '/tasks', payload: { title: 'A', role: 'implementer' } });
+    const res = await app.inject({ method: 'PATCH', url: '/tasks/TSK-0001', payload: { status: 'cancelled' } });
+    expect(res.statusCode).toBe(200);
+    expect(JSON.parse(res.payload).status).toBe('cancelled');
+  });
+
+  it('PATCH /tasks/:id → open (reopen)', async () => {
+    await app.inject({ method: 'POST', url: '/tasks', payload: { title: 'A', role: 'implementer' } });
+    // First cancel it, then reopen
+    await app.inject({ method: 'PATCH', url: '/tasks/TSK-0001', payload: { status: 'cancelled' } });
+    const res = await app.inject({ method: 'PATCH', url: '/tasks/TSK-0001', payload: { status: 'open' } });
+    expect(res.statusCode).toBe(200);
+    expect(JSON.parse(res.payload).status).toBe('open');
+  });
+
+  it('review and cancelled tasks appear in GET /tasks list', async () => {
+    await app.inject({ method: 'POST', url: '/tasks', payload: { title: 'A', role: 'implementer' } });
+    await app.inject({ method: 'POST', url: '/tasks', payload: { title: 'B', role: 'implementer' } });
+    await app.inject({ method: 'PATCH', url: '/tasks/TSK-0001', payload: { status: 'review' } });
+    await app.inject({ method: 'PATCH', url: '/tasks/TSK-0002', payload: { status: 'cancelled' } });
+
+    const allRes = await app.inject({ method: 'GET', url: '/tasks' });
+    const all = JSON.parse(allRes.payload) as Array<{ status: string }>;
+    const statuses = all.map((t) => t.status);
+    expect(statuses).toContain('review');
+    expect(statuses).toContain('cancelled');
+  });
+
  it('updates status via POST /status/update', async () => {
    await app.inject({ method: 'POST', url: '/tasks', payload: { title: 'A', role: 'implementer' } });
    const res = await app.inject({ method: 'POST', url: '/status/update' });
@ -60,4 +102,55 @@ describe('server routes', () => {
    expect(res.statusCode).toBe(200);
    expect(JSON.parse(res.payload)).toHaveLength(1);
  });
+
+  it('serves the task board as HTML via GET /board', async () => {
+    const res = await app.inject({ method: 'GET', url: '/board' });
+    expect(res.statusCode).toBe(200);
+    expect(res.headers['content-type']).toContain('text/html');
+
+    const html = res.payload;
+    expect(html).toContain('<title>AgentHub Board</title>');
+    // All five status columns are present in the static markup.
+    for (const col of ['open', 'in_progress', 'review', 'done', 'cancelled']) {
+      expect(html).toContain(`data-column="${col}"`);
+    }
+    // Handoffs + decisions panels and the polling logic are wired in.
+    expect(html).toContain('Handoffs');
+    expect(html).toContain('Decisions');
+    expect(html).toContain("getJSON('/tasks')");
+    expect(html).toContain('setInterval(refresh');
+  });
+
+  it('board HTML contains the who-arrow rendering logic', async () => {
+    const res = await app.inject({ method: 'GET', url: '/board' });
+    const html = res.payload;
+    // The handoffRoute helper and the &rarr; arrow must be present
+    expect(html).toContain('handoffRoute');
+    expect(html).toContain('&rarr;');
+    // The who-cell class must be used in renderHandoffs
+    expect(html).toContain('who-cell');
+  });
+
+  it('GET /handoffs returns fromRole and toRole fields', async () => {
+    await app.inject({
+      method: 'POST',
+      url: '/handoffs',
+      payload: {
+        fromRole: 'architect',
+        toRole: 'implementer',
+        fromAgent: 'claude',
+        toAgent: 'codex',
+        summary: 'Design done',
+        context: 'See decisions',
+      },
+    });
+    const res = await app.inject({ method: 'GET', url: '/handoffs' });
+    expect(res.statusCode).toBe(200);
+    const items = JSON.parse(res.payload) as Array<Record<string, unknown>>;
+    expect(items).toHaveLength(1);
+    expect(items[0].fromRole).toBe('architect');
+    expect(items[0].toRole).toBe('implementer');
+    expect(items[0].fromAgent).toBe('claude');
+    expect(items[0].toAgent).toBe('codex');
+  });
 });
--- a/tests/sse.test.ts
+++ b/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);
+});
--- a/tests/taskService.test.ts
+++ b/tests/taskService.test.ts
@ -2,7 +2,10 @@ import { describe, it, expect, beforeEach, afterEach } from 'vitest';
 import { mkdtempSync, rmSync } from 'fs';
 import { tmpdir } from 'os';
 import { join } from 'path';
-import { createTask, listTasks, getTask, claimTask, doneTask } from '../src/core/services/taskService.js';
+import {
+  createTask, listTasks, getTask,
+  claimTask, doneTask, reviewTask, cancelTask, reopenTask,
+} from '../src/core/services/taskService.js';

 describe('taskService', () => {
  let cwd: string;
@ -35,4 +38,38 @@ describe('taskService', () => {
    const done = doneTask(cwd, task.id);
    expect(done.status).toBe('done');
  });
+
+  it('transitions a task to review', () => {
+    const task = createTask(cwd, { title: 'C', role: 'implementer' });
+    const inReview = reviewTask(cwd, task.id);
+    expect(inReview.status).toBe('review');
+    // Verify index is updated
+    const listed = listTasks(cwd, { status: 'review' });
+    expect(listed).toHaveLength(1);
+    expect(listed[0].id).toBe(task.id);
+  });
+
+  it('cancels a task', () => {
+    const task = createTask(cwd, { title: 'D', role: 'implementer' });
+    const cancelled = cancelTask(cwd, task.id);
+    expect(cancelled.status).toBe('cancelled');
+    const listed = listTasks(cwd, { status: 'cancelled' });
+    expect(listed).toHaveLength(1);
+  });
+
+  it('reopens a task (any status → open)', () => {
+    const task = createTask(cwd, { title: 'E', role: 'implementer' });
+    cancelTask(cwd, task.id);
+    const reopened = reopenTask(cwd, task.id);
+    expect(reopened.status).toBe('open');
+    const listed = listTasks(cwd, { status: 'open' });
+    expect(listed).toHaveLength(1);
+  });
+
+  it('getTask reads back the correct task', () => {
+    const task = createTask(cwd, { title: 'F', role: 'architect' });
+    const { task: read } = getTask(cwd, task.id);
+    expect(read.title).toBe('F');
+    expect(read.role).toBe('architect');
+  });
 });