Add realtime watcher+SSE transport with tests and lock-retry read path

src/lib/coalescer.ts

@@ -0,0 +1,76 @@
+import { windowsPathKey } from './pathing';
+
+export interface CoalescedEventInput<T> {
+  projectRoot: string;
+  payload: T;
+}
+
+interface PendingEvent<T> {
+  timer: NodeJS.Timeout;
+  projectRoot: string;
+  payload: T;
+}
+
+export class ProjectEventCoalescer<T> {
+  private readonly pending = new Map<string, PendingEvent<T>>();
+
+  private readonly debounceMs: number;
+
+  private readonly onFlush: (event: CoalescedEventInput<T>) => void;
+
+  constructor(debounceMs: number, onFlush: (event: CoalescedEventInput<T>) => void) {
+    this.debounceMs = debounceMs;
+    this.onFlush = onFlush;
+  }
+
+  queue(projectRoot: string, payload: T): void {
+    const projectKey = windowsPathKey(projectRoot);
+    const existing = this.pending.get(projectKey);
+    if (existing) {
+      clearTimeout(existing.timer);
+      existing.projectRoot = projectRoot;
+      existing.payload = payload;
+      existing.timer = setTimeout(() => this.flush(projectKey), this.debounceMs);
+      return;
+    }
+
+    this.pending.set(projectKey, {
+      projectRoot,
+      payload,
+      timer: setTimeout(() => this.flush(projectKey), this.debounceMs),
+    });
+  }
+
+  cancel(projectRoot: string): void {
+    const projectKey = windowsPathKey(projectRoot);
+    const pending = this.pending.get(projectKey);
+    if (!pending) {
+      return;
+    }
+    clearTimeout(pending.timer);
+    this.pending.delete(projectKey);
+  }
+
+  cancelAll(): void {
+    for (const pending of this.pending.values()) {
+      clearTimeout(pending.timer);
+    }
+    this.pending.clear();
+  }
+
+  pendingCount(): number {
+    return this.pending.size;
+  }
+
+  private flush(projectKey: string): void {
+    const pending = this.pending.get(projectKey);
+    if (!pending) {
+      return;
+    }
+    this.pending.delete(projectKey);
+    this.onFlush({
+      projectRoot: pending.projectRoot,
+      payload: pending.payload,
+    });
+  }
+}

src/lib/read-issues.ts

@@ -1,8 +1,8 @@
-import fs from 'node:fs/promises';
 import path from 'node:path';
 
 import { parseIssuesJsonl } from './parser';
 import { canonicalizeWindowsPath } from './pathing';
+import { readTextFileWithRetry } from './read-text-retry';
 import type { BeadIssue } from './types';
 
 export interface ReadIssuesOptions {
@@ -26,7 +26,7 @@ export async function readIssuesFromDisk(options: ReadIssuesOptions = {}): Promi
 
   for (const issuesPath of candidates) {
     try {
-      const jsonl = await fs.readFile(issuesPath, 'utf8');
+      const jsonl = await readTextFileWithRetry(issuesPath);
       return parseIssuesJsonl(jsonl, {
         includeTombstones: options.includeTombstones ?? false,
       });

src/lib/read-text-retry.ts

@@ -0,0 +1,41 @@
+import fs from 'node:fs/promises';
+
+const DEFAULT_RETRY_CODES = new Set(['EBUSY', 'EPERM']);
+
+export interface ReadTextRetryOptions {
+  retries?: number;
+  delayMs?: number;
+  retryCodes?: Set<string>;
+}
+
+function sleep(delayMs: number): Promise<void> {
+  return new Promise((resolve) => setTimeout(resolve, delayMs));
+}
+
+function shouldRetry(error: unknown, retryCodes: Set<string>): boolean {
+  const code = (error as NodeJS.ErrnoException | undefined)?.code;
+  return typeof code === 'string' && retryCodes.has(code);
+}
+
+export async function readTextFileWithRetry(
+  filePath: string,
+  options: ReadTextRetryOptions = {},
+): Promise<string> {
+  const retries = options.retries ?? 2;
+  const delayMs = options.delayMs ?? 40;
+  const retryCodes = options.retryCodes ?? DEFAULT_RETRY_CODES;
+
+  let attempt = 0;
+  while (true) {
+    try {
+      return await fs.readFile(filePath, 'utf8');
+    } catch (error) {
+      if (attempt >= retries || !shouldRetry(error, retryCodes)) {
+        throw error;
+      }
+
+      attempt += 1;
+      await sleep(delayMs);
+    }
+  }
+}

src/lib/realtime.ts

@@ -0,0 +1,82 @@
+import { canonicalizeWindowsPath, windowsPathKey } from './pathing';
+
+export type IssuesChangeKind = 'changed' | 'renamed';
+
+export interface IssuesChangedEvent {
+  id: number;
+  projectRoot: string;
+  changedPath?: string;
+  kind: IssuesChangeKind;
+  at: string;
+}
+
+interface Subscriber {
+  projectKey?: string;
+  listener: (event: IssuesChangedEvent) => void;
+}
+
+export interface SubscribeOptions {
+  projectRoot?: string;
+}
+
+export class IssuesEventBus {
+  private nextEventId = 1;
+
+  private readonly subscribers = new Map<number, Subscriber>();
+
+  private nextSubscriberId = 1;
+
+  emit(projectRoot: string, changedPath?: string, kind: IssuesChangeKind = 'changed'): IssuesChangedEvent {
+    const canonicalProjectRoot = canonicalizeWindowsPath(projectRoot);
+    const projectKey = windowsPathKey(canonicalProjectRoot);
+    const event: IssuesChangedEvent = {
+      id: this.nextEventId,
+      projectRoot: canonicalProjectRoot,
+      changedPath: changedPath ? canonicalizeWindowsPath(changedPath) : undefined,
+      kind,
+      at: new Date().toISOString(),
+    };
+    this.nextEventId += 1;
+
+    for (const subscriber of this.subscribers.values()) {
+      if (!subscriber.projectKey || subscriber.projectKey === projectKey) {
+        subscriber.listener(event);
+      }
+    }
+
+    return event;
+  }
+
+  subscribe(listener: (event: IssuesChangedEvent) => void, options: SubscribeOptions = {}): () => void {
+    const id = this.nextSubscriberId;
+    this.nextSubscriberId += 1;
+
+    this.subscribers.set(id, {
+      listener,
+      projectKey: options.projectRoot ? windowsPathKey(options.projectRoot) : undefined,
+    });
+
+    return () => {
+      this.subscribers.delete(id);
+    };
+  }
+
+  getSubscriberCount(): number {
+    return this.subscribers.size;
+  }
+
+  resetForTests(): void {
+    this.subscribers.clear();
+    this.nextSubscriberId = 1;
+    this.nextEventId = 1;
+  }
+}
+
+export const issuesEventBus = new IssuesEventBus();
+
+export function toSseFrame(event: IssuesChangedEvent): string {
+  return `id: ${event.id}\nevent: issues\ndata: ${JSON.stringify(event)}\n\n`;
+}
+
+export const SSE_HEARTBEAT_FRAME = ': heartbeat\n\n';
+export const SSE_CONNECTED_FRAME = ': connected\n\n';

src/lib/watcher.ts

@@ -0,0 +1,114 @@
+import chokidar, { type FSWatcher } from 'chokidar';
+
+import { ProjectEventCoalescer } from './coalescer';
+import { windowsPathKey } from './pathing';
+import { issuesEventBus, type IssuesChangeKind, type IssuesEventBus } from './realtime';
+import { resolveIssuesJsonlPathCandidates } from './read-issues';
+
+type FileEventName = 'add' | 'change' | 'unlink';
+
+interface WatchRegistration {
+  projectRoot: string;
+  watcher: FSWatcher;
+}
+
+export interface WatchManagerOptions {
+  debounceMs?: number;
+  eventBus?: IssuesEventBus;
+}
+
+export class IssuesWatchManager {
+  private readonly registrations = new Map<string, WatchRegistration>();
+
+  private readonly eventBus: IssuesEventBus;
+
+  private readonly coalescer: ProjectEventCoalescer<{
+    changedPath?: string;
+    kind: IssuesChangeKind;
+  }>;
+
+  constructor(options: WatchManagerOptions = {}) {
+    const debounceMs = options.debounceMs ?? 150;
+    this.eventBus = options.eventBus ?? issuesEventBus;
+    this.coalescer = new ProjectEventCoalescer(debounceMs, ({ projectRoot, payload }) => {
+      this.eventBus.emit(projectRoot, payload.changedPath, payload.kind);
+    });
+  }
+
+  startWatch(projectRoot: string): void {
+    const projectKey = windowsPathKey(projectRoot);
+    if (this.registrations.has(projectKey)) {
+      return;
+    }
+
+    const watchedPaths = resolveIssuesJsonlPathCandidates(projectRoot);
+    const watcher = chokidar.watch(watchedPaths, {
+      ignoreInitial: true,
+      awaitWriteFinish: {
+        stabilityThreshold: 80,
+        pollInterval: 15,
+      },
+    });
+
+    const onFileEvent = (eventName: FileEventName, changedPath: string) => {
+      const kind: IssuesChangeKind = eventName === 'unlink' ? 'renamed' : 'changed';
+      this.queueCoalescedEvent(projectRoot, changedPath, kind);
+    };
+
+    watcher.on('add', (changedPath) => onFileEvent('add', changedPath));
+    watcher.on('change', (changedPath) => onFileEvent('change', changedPath));
+    watcher.on('unlink', (changedPath) => onFileEvent('unlink', changedPath));
+
+    this.registrations.set(projectKey, {
+      projectRoot,
+      watcher,
+    });
+  }
+
+  async stopWatch(projectRoot: string): Promise<void> {
+    const projectKey = windowsPathKey(projectRoot);
+    const registration = this.registrations.get(projectKey);
+    if (!registration) {
+      return;
+    }
+
+    this.coalescer.cancel(projectRoot);
+    this.registrations.delete(projectKey);
+    await registration.watcher.close();
+  }
+
+  async stopAll(): Promise<void> {
+    const closeOps: Promise<void>[] = [];
+
+    for (const registration of this.registrations.values()) {
+      closeOps.push(registration.watcher.close());
+    }
+
+    this.coalescer.cancelAll();
+    this.registrations.clear();
+    await Promise.all(closeOps);
+  }
+
+  getWatchedProjectCount(): number {
+    return this.registrations.size;
+  }
+
+  private queueCoalescedEvent(projectRoot: string, changedPath: string, kind: IssuesChangeKind): void {
+    this.coalescer.queue(projectRoot, {
+      changedPath,
+      kind,
+    });
+  }
+}
+
+const globalRegistry = globalThis as typeof globalThis & {
+  __beadboardWatchManager?: IssuesWatchManager;
+};
+
+export function getIssuesWatchManager(): IssuesWatchManager {
+  if (!globalRegistry.__beadboardWatchManager) {
+    globalRegistry.__beadboardWatchManager = new IssuesWatchManager();
+  }
+
+  return globalRegistry.__beadboardWatchManager;
+}