agenthub/src/server/fsWatch.ts

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';
import { TaskSchema } from '../core/schema.js';
import { toIndexEntry } from '../core/services/taskService.js';
import { Index } from '../core/index.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' },
  { dir: 'messages', type: 'message' },
  { dir: 'asks', type: 'ask' },
];

// 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;
const RETRY_MS = 100;
const MAX_READ_ATTEMPTS = 3;
const indexErrors = new Map<string, { filePath: string; error: string; at: string }>();

export function getFsWatchErrors(): Array<{ filePath: string; error: string; at: string }> {
  return [...indexErrors.values()];
}

export function resetFsWatchErrors(): void {
  indexErrors.clear();
}

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), claimedBy: str(fm.claimedBy), reviewer: str(fm.reviewer) },
      };
    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) } };
    case 'message':
      return { stamp, event: { type, action, id, title: `${str(fm.from)} → ${str(fm.to)}`, status: str(fm.status), assignedTo: str(fm.to) } };
    case 'ask':
      return { stamp, event: { type, action, id, title: `${str(fm.from)} → ${str(fm.to)}`, status: str(fm.status), assignedTo: str(fm.to) } };
  }
}

/**
 * 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);
        const filePath = getEntityDir(cwd, dir) + '/' + name;
        const schedule = (attempt: number, delay: number) => {
          timers.set(key, setTimeout(() => {
            timers.delete(key);
            const error = processFile(cwd, filePath, type);
            if (error && attempt < MAX_READ_ATTEMPTS) schedule(attempt + 1, RETRY_MS);
          }, delay));
        };
        schedule(1, 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(cwd: string, filePath: string, type: AgentHubEventType): Error | undefined {
  let fm: Record<string, unknown>;
  try {
    const entity = readEntity(filePath);
    fm = entity.frontmatter;
    if (type === 'task') {
      const task = TaskSchema.parse(fm);
      const index = new Index(cwd);
      try {
        index.upsert(toIndexEntry(task, filePath));
      } finally {
        index.close();
      }
    }
    indexErrors.delete(filePath);
  } catch (cause) {
    const error = cause instanceof Error ? cause : new Error(String(cause));
    indexErrors.set(filePath, { filePath, error: error.message, at: new Date().toISOString() });
    console.error(`AgentHub fsWatch: could not reindex ${filePath}: ${error.message}`);
    return error;
  }

  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);
}