fix(coding-agent,mom): retry fs watchers on async errors closes #3564

packages/mom/CHANGELOG.md

@@ -2,6 +2,10 @@
 
 ## [Unreleased]
 
+### Fixed
+
+- Fixed Mom event-directory `fs.watch` error handling to retry after transient watcher failures such as `EMFILE`, avoiding startup crashes ([#3564](https://github.com/badlogic/pi-mono/issues/3564))
+
 ## [0.69.0] - 2026-04-22
 
 ### Breaking Changes

packages/mom/src/events.ts

@@ -1,7 +1,8 @@
 import { Cron } from "croner";
-import { existsSync, type FSWatcher, mkdirSync, readdirSync, statSync, unlinkSync, watch } from "fs";
+import { existsSync, type FSWatcher, mkdirSync, readdirSync, statSync, unlinkSync } from "fs";
 import { readFile } from "fs/promises";
 import { join } from "path";
+import { closeWatcher, FS_WATCH_RETRY_DELAY_MS, watchWithErrorHandler } from "./fs-watch.js";
 import * as log from "./log.js";
 import type { SlackBot, SlackEvent } from "./slack.js";
 
@@ -46,7 +47,9 @@ export class EventsWatcher {
 	private debounceTimers: Map<string, NodeJS.Timeout> = new Map();
 	private startTime: number;
 	private watcher: FSWatcher | null = null;
+	private watcherRetryTimer: NodeJS.Timeout | null = null;
 	private knownFiles: Set<string> = new Set();
+	private stopped = true;
 
 	constructor(
 		private eventsDir: string,
@@ -59,6 +62,8 @@ export class EventsWatcher {
 	 * Start watching for events. Call this after SlackBot is ready.
 	 */
 	start(): void {
+		this.stopped = false;
+
 		// Ensure events directory exists
 		if (!existsSync(this.eventsDir)) {
 			mkdirSync(this.eventsDir, { recursive: true });
@@ -69,11 +74,7 @@ export class EventsWatcher {
 		// Scan existing files
 		this.scanExisting();
 
-		// Watch for changes
-		this.watcher = watch(this.eventsDir, (_eventType, filename) => {
-			if (!filename || !filename.endsWith(".json")) return;
-			this.debounce(filename, () => this.handleFileChange(filename));
-		});
+		this.startFsWatcher();
 
 		log.logInfo(`Events watcher started, tracking ${this.knownFiles.size} files`);
 	}
@@ -82,10 +83,14 @@ export class EventsWatcher {
 	 * Stop watching and cancel all scheduled events.
 	 */
 	stop(): void {
+		this.stopped = true;
+
 		// Stop fs watcher
-		if (this.watcher) {
-			this.watcher.close();
-			this.watcher = null;
+		closeWatcher(this.watcher);
+		this.watcher = null;
+		if (this.watcherRetryTimer) {
+			clearTimeout(this.watcherRetryTimer);
+			this.watcherRetryTimer = null;
 		}
 
 		// Cancel all debounce timers
@@ -110,6 +115,40 @@ export class EventsWatcher {
 		log.logInfo("Events watcher stopped");
 	}
 
+	private startFsWatcher(): void {
+		this.watcher = watchWithErrorHandler(
+			this.eventsDir,
+			(_eventType, filename) => {
+				if (!filename || !filename.endsWith(".json")) return;
+				this.debounce(filename, () => this.handleFileChange(filename));
+			},
+			() => this.handleFsWatcherError(),
+		);
+	}
+
+	private handleFsWatcherError(): void {
+		closeWatcher(this.watcher);
+		this.watcher = null;
+		this.scheduleFsWatcherRetry();
+	}
+
+	private scheduleFsWatcherRetry(): void {
+		if (this.stopped || this.watcherRetryTimer) {
+			return;
+		}
+
+		this.watcherRetryTimer = setTimeout(() => {
+			this.watcherRetryTimer = null;
+			if (this.stopped) {
+				return;
+			}
+			this.startFsWatcher();
+			if (this.watcher) {
+				this.rescanExisting();
+			}
+		}, FS_WATCH_RETRY_DELAY_MS);
+	}
+
 	private debounce(filename: string, fn: () => void): void {
 		const existing = this.debounceTimers.get(filename);
 		if (existing) {
@@ -138,6 +177,26 @@ export class EventsWatcher {
 		}
 	}
 
+	private rescanExisting(): void {
+		let files: string[];
+		try {
+			files = readdirSync(this.eventsDir).filter((f) => f.endsWith(".json"));
+		} catch (err) {
+			log.logWarning("Failed to read events directory", String(err));
+			return;
+		}
+
+		const currentFiles = new Set(files);
+		for (const filename of files) {
+			this.handleFileChange(filename);
+		}
+		for (const filename of Array.from(this.knownFiles)) {
+			if (!currentFiles.has(filename)) {
+				this.handleDelete(filename);
+			}
+		}
+	}
+
 	private handleFileChange(filename: string): void {
 		const filePath = join(this.eventsDir, filename);
 

packages/mom/src/fs-watch.ts

@@ -0,0 +1,30 @@
+import { type FSWatcher, type WatchListener, watch } from "node:fs";
+
+export const FS_WATCH_RETRY_DELAY_MS = 5000;
+
+export function closeWatcher(watcher: FSWatcher | null | undefined): void {
+	if (!watcher) {
+		return;
+	}
+
+	try {
+		watcher.close();
+	} catch {
+		// Ignore watcher close errors
+	}
+}
+
+export function watchWithErrorHandler(
+	path: string,
+	listener: WatchListener<string>,
+	onError: () => void,
+): FSWatcher | null {
+	try {
+		const watcher = watch(path, listener);
+		watcher.on("error", onError);
+		return watcher;
+	} catch {
+		onError();
+		return null;
+	}
+}

packages/mom/test/events.test.ts

@@ -0,0 +1,50 @@
+import { existsSync, type FSWatcher, mkdtempSync, rmSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+import { EventsWatcher } from "../src/events.js";
+import type { SlackBot, SlackEvent } from "../src/slack.js";
+
+describe("EventsWatcher fs.watch error handling", () => {
+	let tempDir: string;
+
+	beforeEach(() => {
+		tempDir = mkdtempSync(join(tmpdir(), "mom-events-"));
+	});
+
+	afterEach(() => {
+		if (tempDir && existsSync(tempDir)) {
+			rmSync(tempDir, { recursive: true, force: true });
+		}
+	});
+
+	it("retries the events fs watcher 5 seconds after an async error", async () => {
+		vi.useFakeTimers();
+		const slack = {
+			enqueueEvent: vi.fn((_event: SlackEvent) => true),
+		} as unknown as SlackBot;
+		const eventsDir = join(tempDir, "events");
+		const watcher = new EventsWatcher(eventsDir, slack);
+
+		try {
+			watcher.start();
+			const watcherWithInternals = watcher as unknown as { watcher: FSWatcher | null };
+			const originalWatcher = watcherWithInternals.watcher;
+			expect(originalWatcher).not.toBeNull();
+			expect(originalWatcher?.listenerCount("error")).toBeGreaterThan(0);
+
+			originalWatcher?.emit("error", new Error("simulated EMFILE"));
+			expect(watcherWithInternals.watcher).toBeNull();
+
+			await vi.advanceTimersByTimeAsync(4999);
+			expect(watcherWithInternals.watcher).toBeNull();
+
+			await vi.advanceTimersByTimeAsync(1);
+			expect(watcherWithInternals.watcher).not.toBeNull();
+			expect(watcherWithInternals.watcher).not.toBe(originalWatcher);
+		} finally {
+			watcher.stop();
+			vi.useRealTimers();
+		}
+	});
+});