feat(server): emit SSE for CLI/file writes via filesystem watcher (TSK-0006)

Closes the documented /events limitation: mutations made through the
local CLI (direct file + SQLite writes, no --server) bypassed the
in-process event bus and were invisible to SSE subscribers. A watcher
(e.g. a Windows client) was therefore NOT triggered when another agent
created a task via plain `agenthub task create`.

- src/server/fsWatch.ts: watch the entity dirs (tasks/handoffs/
  decisions/memory); on any .md change, parse the entity and emit the
  matching AgentHubEvent. Debounced per-file; self-creates dirs so it
  attaches even before the first write.
- src/server/events.ts: shared dedup cache (signature = type🆔stamp)
  + emitChange() so the REST path and the watcher deliver each change
  exactly once (no echo when REST writes the file the watcher sees).
- src/server/routes.ts: route emits go through emitChange(); update the
  stale limitation comment.
- src/server/index.ts: start/stop the watcher with the server lifecycle.
- tests/sse.test.ts: CLI-write -> SSE, CLI-update -> SSE, and REST
  dedup (exactly-once). 102/102 green.

Effect: a watcher is auto-triggered the instant any agent creates a
task, with or without --server. Bump 0.1.0 -> 0.1.1.

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

package.json

@@ -1,6 +1,6 @@
 {
   "name": "agenthub",
-  "version": "0.1.0",
+  "version": "0.1.1",
   "description": "Local coordination layer for AI coding agents",
   "type": "module",
   "main": "./dist/index.js",

src/server/events.ts

@@ -31,3 +31,51 @@ export const eventBus = new AgentHubEventBus();
 // Allow an arbitrary number of SSE clients without triggering the
 // default-listener-count warning.
 eventBus.setMaxListeners(0);
+
+// ─── De-duplication between the REST and filesystem-watch emit paths ────────
+// Both mutating REST routes AND the filesystem watcher (fsWatch.ts) can observe
+// the same change: a REST POST/PATCH writes the entity file, then the watcher
+// sees that very write. Without coordination the subscriber would receive the
+// event twice. We solve it with a short-lived signature cache keyed by
+// `${type}:${id}:${updatedAt|createdAt}`. Whichever path emits first records the
+// signature; the other path sees it via `seenRecently()` and stays silent.
+const DEDUP_TTL_MS = 15_000;
+const recentlyEmitted = new Map<string, number>();
+
+/** Stable key for a single logical mutation of one entity revision. */
+export function signatureOf(type: string, id: string, stamp: string | undefined): string {
+  return `${type}:${id}:${stamp ?? ''}`;
+}
+
+function markEmitted(signature: string): void {
+  const now = Date.now();
+  recentlyEmitted.set(signature, now);
+  // Opportunistic GC so the map can't grow unbounded under heavy churn.
+  if (recentlyEmitted.size > 500) {
+    for (const [key, ts] of recentlyEmitted) {
+      if (now - ts > DEDUP_TTL_MS) recentlyEmitted.delete(key);
+    }
+  }
+}
+
+/** True if a change with this signature was emitted within the dedup window. */
+export function seenRecently(signature: string): boolean {
+  const ts = recentlyEmitted.get(signature);
+  if (ts === undefined) return false;
+  if (Date.now() - ts > DEDUP_TTL_MS) {
+    recentlyEmitted.delete(signature);
+    return false;
+  }
+  return true;
+}
+
+/**
+ * Publish a change and record its signature for cross-path de-duplication.
+ * Used by both the REST routes and the filesystem watcher. `stamp` is the
+ * entity's `updatedAt` (or `createdAt` for handoffs) and must match what the
+ * other path derives from the same file revision.
+ */
+export function emitChange(event: AgentHubEvent, stamp: string | undefined): void {
+  markEmitted(signatureOf(event.type, event.id, stamp));
+  eventBus.publish(event);
+}

src/server/fsWatch.ts

@@ -0,0 +1,139 @@
+import { watch, mkdirSync, type FSWatcher } from 'node:fs';
+import { basename } from 'node:path';
+import { readEntity } from '../core/files.js';
+import { getEntityDir, type EntityType } from '../core/paths.js';
+import { emitChange, seenRecently, signatureOf } from './events.js';
+import type { AgentHubEvent, AgentHubEventType } from './events.js';
+
+/**
+ * Filesystem-watch layer.
+ *
+ * Closes the gap documented on the /events route: mutations made via the local
+ * CLI (direct file + SQLite writes, no --server) bypass the in-process event
+ * bus and were therefore invisible to SSE subscribers. This watcher observes
+ * the entity directories and emits the same AgentHubEvent for any change,
+ * regardless of which path produced it — so a watcher (e.g. a Windows client)
+ * is triggered automatically the instant a task is created, even from a plain
+ * `agenthub task create`.
+ *
+ * REST mutations also touch these files, so the watcher would double-emit; the
+ * dedup cache in events.ts (signature = type:id:stamp) suppresses the echo.
+ */
+
+const WATCHED: { dir: EntityType; type: AgentHubEventType }[] = [
+  { dir: 'tasks', type: 'task' },
+  { dir: 'handoffs', type: 'handoff' },
+  { dir: 'decisions', type: 'decision' },
+  { dir: 'memory', type: 'memory' },
+];
+
+// fs.watch can fire several events (rename + change) for a single write, and a
+// file may be observed mid-write. Coalesce per-path bursts before reading.
+const DEBOUNCE_MS = 40;
+
+function str(value: unknown): string | undefined {
+  return value === undefined || value === null ? undefined : String(value);
+}
+
+/**
+ * Build the SSE event + dedup stamp from parsed frontmatter. Returns undefined
+ * if the file isn't a recognizable entity yet (e.g. caught mid-write).
+ */
+function toEvent(
+  type: AgentHubEventType,
+  fm: Record<string, unknown>,
+): { event: AgentHubEvent; stamp: string | undefined } | undefined {
+  const id = str(fm.id);
+  if (!id) return undefined;
+
+  const createdAt = str(fm.createdAt);
+  const updatedAt = str(fm.updatedAt);
+  // Handoffs have only createdAt; for the rest, an updatedAt that differs from
+  // createdAt means a mutation of an existing entity -> 'updated'. This mirrors
+  // the POST=created / PATCH=updated semantics of the REST routes.
+  const stamp = updatedAt ?? createdAt;
+  const action: AgentHubEvent['action'] =
+    updatedAt && createdAt && updatedAt !== createdAt ? 'updated' : 'created';
+
+  switch (type) {
+    case 'task':
+      return {
+        stamp,
+        event: { type, action, id, title: str(fm.title), status: str(fm.status), role: str(fm.role), assignedTo: str(fm.assignedTo) },
+      };
+    case 'handoff':
+      return {
+        stamp,
+        event: { type, action: 'created', id, title: str(fm.summary), role: str(fm.toRole), assignedTo: str(fm.toAgent) },
+      };
+    case 'decision':
+      return { stamp, event: { type, action, id, title: str(fm.title) } };
+    case 'memory':
+      return { stamp, event: { type, action, id, title: str(fm.title) } };
+  }
+}
+
+/**
+ * Start watching the entity directories. Returns a stop function that tears
+ * down every watcher and pending timer. Safe to call in a project that hasn't
+ * created all entity directories yet — missing ones are simply skipped.
+ */
+export function startEntityWatcher(cwd: string): () => void {
+  const watchers: FSWatcher[] = [];
+  const timers = new Map<string, NodeJS.Timeout>();
+
+  for (const { dir, type } of WATCHED) {
+    const path = getEntityDir(cwd, dir);
+
+    try {
+      // Ensure the directory exists so the watcher attaches even before the
+      // first entity of this type is written (e.g. a fresh project's decisions).
+      mkdirSync(path, { recursive: true });
+      const watcher = watch(path, (_eventType, filename) => {
+        if (!filename) return; // some platforms omit the name; nothing to read
+        const name = basename(filename.toString());
+        if (!name.endsWith('.md')) return;
+
+        // Debounce per file so a rename+change burst reads the settled file once.
+        const key = `${type}:${name}`;
+        const existing = timers.get(key);
+        if (existing) clearTimeout(existing);
+        timers.set(
+          key,
+          setTimeout(() => {
+            timers.delete(key);
+            processFile(getEntityDir(cwd, dir) + '/' + name, type);
+          }, DEBOUNCE_MS),
+        );
+      });
+      watchers.push(watcher);
+    } catch {
+      // A directory that can't be watched (permissions, transient) shouldn't
+      // crash the server; the REST emit path still works for that entity.
+    }
+  }
+
+  return () => {
+    for (const t of timers.values()) clearTimeout(t);
+    timers.clear();
+    for (const w of watchers) w.close();
+  };
+}
+
+function processFile(filePath: string, type: AgentHubEventType): void {
+  let fm: Record<string, unknown>;
+  try {
+    fm = readEntity(filePath).frontmatter;
+  } catch {
+    // Deleted again, or read mid-write -> skip; a later settled write re-fires.
+    return;
+  }
+
+  const built = toEvent(type, fm);
+  if (!built) return;
+
+  // If the REST path already emitted this exact revision, stay silent.
+  if (seenRecently(signatureOf(built.event.type, built.event.id, built.stamp))) return;
+
+  emitChange(built.event, built.stamp);
+}

src/server/index.ts

@@ -1,6 +1,7 @@
 import Fastify from 'fastify';
 import { registerRoutes } from './routes.js';
 import { resolveAdvertiseUrl, startDiscoveryBroadcaster } from '../discovery.js';
+import { startEntityWatcher } from './fsWatch.js';
 
 export function buildApp(cwd: string) {
   const app = Fastify({ logger: false });
@@ -14,8 +15,11 @@ export async function startServer(cwd: string, options: { port?: number; host?:
   const host = options.host ?? '127.0.0.1';
 
   let broadcaster: ReturnType<typeof startDiscoveryBroadcaster> | undefined;
+  // Emit SSE events for CLI/file writes too, not just REST mutations.
+  const stopWatcher = startEntityWatcher(cwd);
   app.addHook('onClose', async () => {
     broadcaster?.stop();
+    stopWatcher();
   });
 
   try {

src/server/routes.ts

@@ -8,7 +8,7 @@ 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 { eventBus, emitChange } from './events.js';
 import type { AgentHubEvent } from './events.js';
 import type { Task, Handoff, Decision, Memory } from '../core/schema.js';
 
@@ -37,10 +37,10 @@ export async function registerRoutes(app: FastifyInstance, cwd: string): Promise
   // 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.
+  // Both REST mutations and direct local-CLI writes emit events: REST routes
+  // publish via emitChange(), and the filesystem watcher (fsWatch.ts) observes
+  // the entity directories and emits for any other write. A shared dedup cache
+  // (events.ts) ensures each change is delivered exactly once.
   app.get('/events', async (request, reply) => {
     const { role } = request.query as { role?: string };
 
@@ -94,15 +94,18 @@ export async function registerRoutes(app: FastifyInstance, cwd: string): Promise
     } 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,
-    });
+    emitChange(
+      {
+        type: 'task',
+        action: 'created',
+        id: task.id,
+        title: task.title,
+        status: task.status,
+        role: task.role,
+        assignedTo: task.assignedTo,
+      },
+      task.updatedAt,
+    );
     return task;
   });
 
@@ -155,15 +158,18 @@ export async function registerRoutes(app: FastifyInstance, cwd: string): Promise
         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,
-    });
+    emitChange(
+      {
+        type: 'task',
+        action: 'updated',
+        id: task.id,
+        title: task.title,
+        status: task.status,
+        role: task.role,
+        assignedTo: task.assignedTo,
+      },
+      task.updatedAt,
+    );
     return task;
   });
 
@@ -176,14 +182,17 @@ export async function registerRoutes(app: FastifyInstance, cwd: string): Promise
     } 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,
-    });
+    emitChange(
+      {
+        type: 'handoff',
+        action: 'created',
+        id: handoff.id,
+        title: handoff.summary,
+        role: handoff.toRole,
+        assignedTo: handoff.toAgent,
+      },
+      handoff.createdAt,
+    );
     return handoff;
   });
   app.get('/handoffs/:id', async (request, reply) => {
@@ -205,12 +214,15 @@ export async function registerRoutes(app: FastifyInstance, cwd: string): Promise
     } catch (err) {
       return badRequest(reply, err instanceof Error ? err.message : 'Invalid decision');
     }
-    eventBus.publish({
-      type: 'decision',
-      action: 'created',
-      id: decision.id,
-      title: decision.title,
-    });
+    emitChange(
+      {
+        type: 'decision',
+        action: 'created',
+        id: decision.id,
+        title: decision.title,
+      },
+      decision.updatedAt,
+    );
     return decision;
   });
 
@@ -223,12 +235,15 @@ export async function registerRoutes(app: FastifyInstance, cwd: string): Promise
     } catch (err) {
       return badRequest(reply, err instanceof Error ? err.message : 'Invalid memory');
     }
-    eventBus.publish({
-      type: 'memory',
-      action: 'created',
-      id: memory.id,
-      title: memory.title,
-    });
+    emitChange(
+      {
+        type: 'memory',
+        action: 'created',
+        id: memory.id,
+        title: memory.title,
+      },
+      memory.updatedAt,
+    );
     return memory;
   });
   app.get('/memory/search', async (request) => {

tests/sse.test.ts

@@ -17,6 +17,8 @@ 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';
+import { startEntityWatcher } from '../src/server/fsWatch.js';
+import { createTask, claimTask } from '../src/core/services/taskService.js';
 
 // ─── 1. Unit: SSE buffer parser ──────────────────────────────────────────────
 
@@ -320,3 +322,112 @@ describe('SSE stream e2e', () => {
     expect(received).toHaveLength(0);
   }, 5000);
 });
+
+// ─── 5. fsWatch: CLI / direct file writes also emit (TSK-0006 part 2) ─────────
+// The whole point of the watcher: a `agenthub task create` (no --server) writes
+// the entity file directly, bypassing the REST emit path — yet a connected
+// watcher must still be triggered. Here we drive the services directly (exactly
+// what the CLI does) and assert the eventBus fires.
+
+async function waitFor(
+  collected: AgentHubEvent[],
+  pred: (e: AgentHubEvent) => boolean,
+  ms = 2000,
+): Promise<AgentHubEvent | undefined> {
+  const deadline = Date.now() + ms;
+  while (Date.now() < deadline) {
+    const hit = collected.find(pred);
+    if (hit) return hit;
+    await new Promise((r) => setTimeout(r, 25));
+  }
+  return collected.find(pred);
+}
+
+describe('fsWatch emits for non-REST (CLI/file) writes', () => {
+  let cwd: string;
+  let stop: () => void;
+  const collected: AgentHubEvent[] = [];
+  const onChange = (e: AgentHubEvent) => collected.push(e);
+
+  beforeEach(async () => {
+    cwd = mkdtempSync(join(tmpdir(), 'ah-fswatch-'));
+    init(cwd, { projectName: 'fswatch-test', yes: true });
+    collected.length = 0;
+    eventBus.on('change', onChange);
+    stop = startEntityWatcher(cwd);
+    // fs.watch (FSEvents on macOS) needs a brief moment after watch() before it
+    // reliably delivers; a write fired in that gap is missed. The server starts
+    // its watcher long before any write, so this only matters for the test.
+    await new Promise((r) => setTimeout(r, 250));
+  });
+
+  afterEach(() => {
+    stop();
+    eventBus.off('change', onChange);
+    collected.length = 0;
+    rmSync(cwd, { recursive: true, force: true });
+  });
+
+  it('emits task/created when a task is written via the service (CLI path)', async () => {
+    const task = createTask(cwd, { title: 'CLI-created task', role: 'implementer' });
+    const ev = await waitFor(collected, (e) => e.type === 'task' && e.id === task.id);
+    expect(ev).toBeDefined();
+    expect(ev).toMatchObject({
+      type: 'task',
+      action: 'created',
+      id: task.id,
+      title: 'CLI-created task',
+      role: 'implementer',
+    });
+  }, 4000);
+
+  it('emits task/updated when a task file is mutated via the service (claim)', async () => {
+    const task = createTask(cwd, { title: 'T', role: 'implementer' });
+    await waitFor(collected, (e) => e.id === task.id); // drain the created event
+    collected.length = 0;
+
+    claimTask(cwd, task.id, 'windows-claude');
+    const ev = await waitFor(collected, (e) => e.id === task.id && e.action === 'updated');
+    expect(ev).toBeDefined();
+    expect(ev).toMatchObject({ type: 'task', action: 'updated', status: 'in_progress', assignedTo: 'windows-claude' });
+  }, 4000);
+});
+
+// ─── 6. fsWatch + REST dedup: each change is delivered exactly once ────────────
+
+describe('fsWatch dedup with the REST emit path', () => {
+  let cwd: string;
+  let server: Awaited<ReturnType<typeof startServer>>;
+  const collected: AgentHubEvent[] = [];
+  const onChange = (e: AgentHubEvent) => collected.push(e);
+
+  beforeEach(async () => {
+    cwd = mkdtempSync(join(tmpdir(), 'ah-fswatch-dedup-'));
+    init(cwd, { projectName: 'fswatch-dedup', yes: true });
+    collected.length = 0;
+    eventBus.on('change', onChange);
+    server = await startServer(cwd, { host: '127.0.0.1', port: 0 });
+  });
+
+  afterEach(async () => {
+    eventBus.off('change', onChange);
+    collected.length = 0;
+    await server.app.close();
+    rmSync(cwd, { recursive: true, force: true });
+  });
+
+  it('REST POST /tasks emits exactly once (watcher echo suppressed)', async () => {
+    await fetch(`${server.url}/tasks`, {
+      method: 'POST',
+      headers: { 'Content-Type': 'application/json' },
+      body: JSON.stringify({ title: 'dedup task', role: 'implementer' }),
+    });
+
+    // Wait well past the watcher debounce so any duplicate would have landed.
+    await new Promise((r) => setTimeout(r, 300));
+
+    const forTask = collected.filter((e) => e.type === 'task' && e.id === 'TSK-0001');
+    expect(forTask).toHaveLength(1);
+    expect(forTask[0].action).toBe('created');
+  }, 4000);
+});