From 722388cb00ac659d8b0383a5992d48c229f0c5e6 Mon Sep 17 00:00:00 2001 From: chahinebrini Date: Sat, 27 Jun 2026 15:43:33 +0200 Subject: [PATCH] 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:id: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 --- package.json | 2 +- src/server/events.ts | 48 +++++++++++++++ src/server/fsWatch.ts | 139 ++++++++++++++++++++++++++++++++++++++++++ src/server/index.ts | 4 ++ src/server/routes.ts | 101 +++++++++++++++++------------- tests/sse.test.ts | 111 +++++++++++++++++++++++++++++++++ 6 files changed, 361 insertions(+), 44 deletions(-) create mode 100644 src/server/fsWatch.ts diff --git a/package.json b/package.json index 522f473..ea8c4ad 100644 --- a/package.json +++ b/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", diff --git a/src/server/events.ts b/src/server/events.ts index c335ff9..e800ff1 100644 --- a/src/server/events.ts +++ b/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(); + +/** 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); +} diff --git a/src/server/fsWatch.ts b/src/server/fsWatch.ts new file mode 100644 index 0000000..8352d32 --- /dev/null +++ b/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, +): { 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(); + + 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; + 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); +} diff --git a/src/server/index.ts b/src/server/index.ts index 26214ed..bf90d01 100644 --- a/src/server/index.ts +++ b/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 | 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 { diff --git a/src/server/routes.ts b/src/server/routes.ts index cda6d2a..3d0eb8a 100644 --- a/src/server/routes.ts +++ b/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) => { diff --git a/tests/sse.test.ts b/tests/sse.test.ts index 08b2613..46d110c 100644 --- a/tests/sse.test.ts +++ b/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 { + 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>; + 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); +});