From a94d297fdf35ea12add961c9037f14e5d59e5ad9 Mon Sep 17 00:00:00 2001 From: rNoz Date: Sun, 19 Jul 2026 22:02:03 +0200 Subject: [PATCH 1/5] fix(annotate): watch open source files exactly --- .../server/file-browser-watch.test.ts | 272 ++++++++++++++++- .../pi-extension/server/file-browser-watch.ts | 104 +++++-- packages/editor/App.tsx | 25 +- packages/editor/sourceDocumentPaths.test.ts | 42 ++- packages/editor/sourceDocumentPaths.ts | 23 ++ packages/server/reference-watch.test.ts | 282 ++++++++++++++++-- packages/server/reference-watch.ts | 101 +++++-- 7 files changed, 755 insertions(+), 94 deletions(-) diff --git a/apps/pi-extension/server/file-browser-watch.test.ts b/apps/pi-extension/server/file-browser-watch.test.ts index b976e4254..f4c68a76f 100644 --- a/apps/pi-extension/server/file-browser-watch.test.ts +++ b/apps/pi-extension/server/file-browser-watch.test.ts @@ -1,7 +1,126 @@ -import { describe, expect, test } from "bun:test"; +import { afterEach, describe, expect, test } from "bun:test"; +import { mkdirSync, mkdtempSync, renameSync, rmSync, unlinkSync, writeFileSync } from "node:fs"; +import { createServer, type Server } from "node:http"; import { tmpdir } from "node:os"; import { dirname, join } from "node:path"; -import { isFileBrowserWatchIgnoredPath } from "./file-browser-watch.ts"; +import { handleFileBrowserStreamRequest, isFileBrowserWatchIgnoredPath } from "./file-browser-watch.ts"; + +const tempDirs: string[] = []; +const servers: Server[] = []; +const WATCH_READY_MS = 250; + +function makeTempDir(prefix: string): string { + const dir = mkdtempSync(join(tmpdir(), prefix)); + tempDirs.push(dir); + return dir; +} + +async function startWatchServer(): Promise { + const server = createServer((req, res) => { + const url = new URL(req.url ?? "/", "http://localhost"); + if (!handleFileBrowserStreamRequest(req, res, url)) { + res.writeHead(404).end(); + } + }); + servers.push(server); + await new Promise((resolve, reject) => { + server.once("error", reject); + server.listen(0, "127.0.0.1", resolve); + }); + const address = server.address(); + if (!address || typeof address === "string") throw new Error("Missing test server address"); + return `http://127.0.0.1:${address.port}`; +} + +interface SSEEvent { + type?: string; + dirPath?: string; + reason?: string; +} + +interface SSECollector { + next(timeoutMs?: number): Promise; + close(): Promise; +} + +function collectSSE(response: Response): SSECollector { + const reader = response.body?.getReader(); + if (!reader) throw new Error("Missing response body"); + const decoder = new TextDecoder(); + const events: SSEEvent[] = []; + const waiters: Array<(event: SSEEvent) => void> = []; + let pending = ""; + + const pump = async () => { + try { + while (true) { + const result = await reader.read(); + if (result.done) break; + pending += decoder.decode(result.value, { stream: true }); + const blocks = pending.split("\n\n"); + pending = blocks.pop() ?? ""; + for (const block of blocks) { + const line = block.split("\n").find((item) => item.startsWith("data: ")); + if (!line) continue; + const event = JSON.parse(line.slice("data: ".length)) as SSEEvent; + const waiter = waiters.shift(); + if (waiter) waiter(event); + else events.push(event); + } + } + } catch { + // Reader cancellation closes the collector. + } + }; + void pump(); + + return { + next(timeoutMs = 2_000) { + const event = events.shift(); + if (event) return Promise.resolve(event); + return new Promise((resolve, reject) => { + let timeout: ReturnType; + const waiter = (nextEvent: SSEEvent) => { + clearTimeout(timeout); + resolve(nextEvent); + }; + waiters.push(waiter); + timeout = setTimeout(() => { + const index = waiters.indexOf(waiter); + if (index >= 0) waiters.splice(index, 1); + reject(new Error("Timed out waiting for SSE event")); + }, timeoutMs); + }); + }, + close() { + return reader.cancel(); + }, + }; +} + +async function expectNoEvent(collector: SSECollector, timeoutMs = 500): Promise { + try { + const event = await collector.next(timeoutMs); + throw new Error(`Unexpected SSE event: ${JSON.stringify(event)}`); + } catch (error) { + if (error instanceof Error && error.message === "Timed out waiting for SSE event") return; + throw error; + } +} + +function waitForWatcher(): Promise { + return Bun.sleep(WATCH_READY_MS); +} + +afterEach(async () => { + for (const server of servers.splice(0)) { + server.closeAllConnections(); + await new Promise((resolve) => server.close(() => resolve())); + } + for (const dir of tempDirs.splice(0)) { + rmSync(dir, { recursive: true, force: true }); + } +}); describe("Pi file browser watcher", () => { test("ignores nested excluded folders for watcher paths", () => { @@ -14,4 +133,153 @@ describe("Pi file browser watcher", () => { expect(isFileBrowserWatchIgnoredPath(root, root)).toBe(false); expect(isFileBrowserWatchIgnoredPath(join(dirname(root), "outside", "node_modules"), root)).toBe(false); }); + + test("requires exactly one valid watch parameter mode", async () => { + const root = makeTempDir("plannotator-pi-watch-mode-"); + const target = join(root, "plan.md"); + writeFileSync(target, "initial"); + const origin = await startWatchServer(); + const missing = await fetch(`${origin}/api/reference/files/stream`); + const mixedUrl = new URL(`${origin}/api/reference/files/stream`); + mixedUrl.searchParams.append("dirPath", root); + mixedUrl.searchParams.append("filePath", target); + const emptyUrl = new URL(`${origin}/api/reference/files/stream`); + emptyUrl.searchParams.append("filePath", ""); + const invalidParentUrl = new URL(`${origin}/api/reference/files/stream`); + invalidParentUrl.searchParams.append("filePath", join(root, "missing", "plan.md")); + const directoryUrl = new URL(`${origin}/api/reference/files/stream`); + directoryUrl.searchParams.append("filePath", root); + + expect(missing.status).toBe(400); + expect((await fetch(mixedUrl)).status).toBe(400); + expect((await fetch(emptyUrl)).status).toBe(400); + expect((await fetch(invalidParentUrl)).status).toBe(400); + expect((await fetch(directoryUrl)).status).toBe(400); + }); + + test("watches an exact file through writes, deletion, recreation, and rename-away", async () => { + const root = makeTempDir("plannotator-pi-watch-file-"); + const target = join(root, "plan.md"); + const nested = join(root, "nested"); + writeFileSync(target, "initial"); + mkdirSync(nested); + await waitForWatcher(); + const origin = await startWatchServer(); + const url = new URL(`${origin}/api/reference/files/stream`); + url.searchParams.append("filePath", target); + const response = await fetch(url); + const collector = collectSSE(response); + + try { + expect(response.status).toBe(200); + expect(await collector.next()).toMatchObject({ type: "ready", dirPath: root, reason: "initial" }); + await waitForWatcher(); + + writeFileSync(target, "written"); + expect(await collector.next()).toMatchObject({ type: "changed", dirPath: root, reason: "files" }); + + writeFileSync(join(root, "sibling.md"), "sibling"); + writeFileSync(join(nested, "nested.md"), "nested"); + await expectNoEvent(collector); + + unlinkSync(target); + expect(await collector.next()).toMatchObject({ type: "changed", dirPath: root, reason: "files" }); + + writeFileSync(target, "recreated"); + expect(await collector.next()).toMatchObject({ type: "changed", dirPath: root, reason: "files" }); + + renameSync(target, join(root, "renamed.md")); + expect(await collector.next()).toMatchObject({ type: "changed", dirPath: root, reason: "files" }); + } finally { + await collector.close(); + } + }); + + test("allows a missing leaf when its parent exists", async () => { + const root = makeTempDir("plannotator-pi-watch-missing-"); + const target = join(root, "future.md"); + const origin = await startWatchServer(); + const url = new URL(`${origin}/api/reference/files/stream`); + url.searchParams.append("filePath", target); + const collector = collectSSE(await fetch(url)); + + try { + expect(await collector.next()).toMatchObject({ type: "ready", dirPath: root }); + await waitForWatcher(); + writeFileSync(target, "created"); + expect(await collector.next()).toMatchObject({ type: "changed", dirPath: root, reason: "files" }); + } finally { + await collector.close(); + } + }); + + test("deduplicates equivalent exact-file targets", async () => { + const root = makeTempDir("plannotator-pi-watch-dedupe-"); + const target = join(root, "plan.md"); + writeFileSync(target, "initial"); + await waitForWatcher(); + const origin = await startWatchServer(); + const url = new URL(`${origin}/api/reference/files/stream`); + url.searchParams.append("filePath", target); + url.searchParams.append("filePath", join(root, ".", "plan.md")); + const collector = collectSSE(await fetch(url)); + + try { + expect(await collector.next()).toMatchObject({ type: "ready", dirPath: root }); + await expectNoEvent(collector); + } finally { + await collector.close(); + } + }); + + test("keeps directory and file caches separate while preserving recursive directory watching", async () => { + const root = makeTempDir("plannotator-pi-watch-cache-"); + const target = join(root, "plan.md"); + const nested = join(root, "nested"); + writeFileSync(target, "initial"); + mkdirSync(nested); + await waitForWatcher(); + const origin = await startWatchServer(); + const dirUrl = new URL(`${origin}/api/reference/files/stream`); + dirUrl.searchParams.append("dirPath", root); + const fileUrl = new URL(`${origin}/api/reference/files/stream`); + fileUrl.searchParams.append("filePath", target); + const dirCollector = collectSSE(await fetch(dirUrl)); + const fileCollector = collectSSE(await fetch(fileUrl)); + + try { + await dirCollector.next(); + await fileCollector.next(); + await waitForWatcher(); + writeFileSync(join(nested, "sibling.md"), "nested"); + expect(await dirCollector.next()).toMatchObject({ type: "changed", dirPath: root, reason: "files" }); + await expectNoEvent(fileCollector); + } finally { + await dirCollector.close(); + await fileCollector.close(); + } + }); + + test("does not report adjacent Git metadata for an exact file", async () => { + const root = makeTempDir("plannotator-pi-watch-git-"); + const target = join(root, "plan.md"); + const gitDir = join(root, ".git"); + mkdirSync(gitDir); + writeFileSync(join(gitDir, "HEAD"), "ref: refs/heads/main\n"); + writeFileSync(target, "initial"); + await waitForWatcher(); + const origin = await startWatchServer(); + const url = new URL(`${origin}/api/reference/files/stream`); + url.searchParams.append("filePath", target); + const collector = collectSSE(await fetch(url)); + + try { + await collector.next(); + await waitForWatcher(); + writeFileSync(join(gitDir, "HEAD"), "ref: refs/heads/other\n"); + await expectNoEvent(collector); + } finally { + await collector.close(); + } + }); }); diff --git a/apps/pi-extension/server/file-browser-watch.ts b/apps/pi-extension/server/file-browser-watch.ts index 8f79ed9a0..8204f316e 100644 --- a/apps/pi-extension/server/file-browser-watch.ts +++ b/apps/pi-extension/server/file-browser-watch.ts @@ -1,7 +1,7 @@ import chokidar, { type FSWatcher } from "chokidar"; import { existsSync, statSync } from "node:fs"; import type { IncomingMessage, ServerResponse } from "node:http"; -import { isAbsolute, relative } from "node:path"; +import { dirname, isAbsolute, relative } from "node:path"; import { isFileBrowserExcludedPath } from "../generated/reference-common.ts"; import { resolveUserPath } from "../generated/resolve-file.ts"; @@ -16,13 +16,21 @@ interface FileBrowserChangeEvent { } interface WatchEntry { - dirPath: string; + key: string; subscribers: Map; contentWatcher: FSWatcher | null; gitWatcher: FSWatcher | null; debounceTimer: ReturnType | null; } +interface WatchTarget { + key: string; + watchPath: string; + clientDirPath: string; + watchGit: boolean; + ignored?: (path: string) => boolean; +} + const HEARTBEAT_MS = 30_000; const DEBOUNCE_MS = 180; const watchers = new Map(); @@ -73,8 +81,8 @@ function closeWatcher(entry: WatchEntry): void { if (entry.debounceTimer) clearTimeout(entry.debounceTimer); void entry.contentWatcher?.close(); void entry.gitWatcher?.close(); - if (watchers.get(entry.dirPath) === entry) { - watchers.delete(entry.dirPath); + if (watchers.get(entry.key) === entry) { + watchers.delete(entry.key); } } @@ -83,22 +91,22 @@ function releaseSubscriber(entry: WatchEntry, res: ServerResponse): void { if (entry.subscribers.size === 0) closeWatcher(entry); } -function ensureWatcher(dirPath: string): WatchEntry { - const existing = watchers.get(dirPath); +function ensureWatcher(target: WatchTarget): WatchEntry { + const existing = watchers.get(target.key); if (existing) return existing; const entry: WatchEntry = { - dirPath, + key: target.key, subscribers: new Map(), contentWatcher: null, gitWatcher: null, debounceTimer: null, }; - entry.contentWatcher = chokidar.watch(dirPath, { + entry.contentWatcher = chokidar.watch(target.watchPath, { ignoreInitial: true, persistent: true, - ignored: (path) => isFileBrowserWatchIgnoredPath(path, dirPath), + ignored: target.ignored, awaitWriteFinish: { stabilityThreshold: 120, pollInterval: 30, @@ -107,7 +115,9 @@ function ensureWatcher(dirPath: string): WatchEntry { entry.contentWatcher.on("all", () => scheduleBroadcast(entry, "files")); entry.contentWatcher.on("error", () => scheduleBroadcast(entry, "files")); - const gitWatchPaths = getGitMetadataWatchPaths(dirPath); + const gitWatchPaths = target.watchGit + ? getGitMetadataWatchPaths(target.watchPath) + : []; if (gitWatchPaths.length > 0) { entry.gitWatcher = chokidar.watch(gitWatchPaths, { ignoreInitial: true, @@ -121,43 +131,79 @@ function ensureWatcher(dirPath: string): WatchEntry { entry.gitWatcher.on("error", () => scheduleBroadcast(entry, "git")); } - watchers.set(dirPath, entry); + watchers.set(target.key, entry); return entry; } +function isValidFileTarget(filePath: string): boolean { + if (!filePath) return false; + try { + if (existsSync(filePath)) return !statSync(filePath).isDirectory(); + return isValidDirectory(dirname(filePath)); + } catch { + return false; + } +} + export function handleFileBrowserStreamRequest(req: IncomingMessage, res: ServerResponse, url: URL): boolean { if (url.pathname !== "/api/reference/files/stream" || req.method !== "GET") return false; const rawDirPaths = url.searchParams.getAll("dirPath"); - if (rawDirPaths.length === 0) { - json(res, { error: "Missing dirPath parameter" }, 400); + const rawFilePaths = url.searchParams.getAll("filePath"); + if ((rawDirPaths.length > 0) === (rawFilePaths.length > 0)) { + json(res, { error: "Provide exactly one of dirPath or filePath" }, 400); return true; } - const dirPaths: string[] = []; - const clientDirPaths: string[] = []; - for (const rawDirPath of rawDirPaths) { - const dirPath = resolveUserPath(rawDirPath); - if (!isValidDirectory(dirPath)) { - json(res, { error: "Invalid directory path" }, 400); - return true; + const targets = new Map(); + if (rawDirPaths.length > 0) { + for (const rawDirPath of rawDirPaths) { + const dirPath = resolveUserPath(rawDirPath); + if (!isValidDirectory(dirPath)) { + json(res, { error: "Invalid directory path" }, 400); + return true; + } + const key = `dir:${dirPath}`; + if (!targets.has(key)) { + targets.set(key, { + key, + watchPath: dirPath, + clientDirPath: rawDirPath, + watchGit: true, + ignored: (path) => isFileBrowserWatchIgnoredPath(path, dirPath), + }); + } } - if (!dirPaths.includes(dirPath)) { - dirPaths.push(dirPath); - clientDirPaths.push(rawDirPath); + } else { + for (const rawFilePath of rawFilePaths) { + const filePath = resolveUserPath(rawFilePath); + if (!isValidFileTarget(filePath)) { + json(res, { error: "Invalid file path" }, 400); + return true; + } + const key = `file:${filePath}`; + if (!targets.has(key)) { + targets.set(key, { + key, + watchPath: filePath, + clientDirPath: dirname(rawFilePath), + watchGit: false, + }); + } } } - const entries = dirPaths.map((dirPath) => ensureWatcher(dirPath)); + const subscriptions = [...targets.values()].map((target) => ({ + entry: ensureWatcher(target), + clientDirPath: target.clientDirPath, + })); res.writeHead(200, { "Content-Type": "text/event-stream", "Cache-Control": "no-cache", Connection: "keep-alive", }); res.setTimeout(0); - for (let i = 0; i < entries.length; i++) { - const entry = entries[i]!; - const clientDirPath = clientDirPaths[i] ?? entry.dirPath; + for (const { entry, clientDirPath } of subscriptions) { res.write(serialize({ type: "ready", dirPath: clientDirPath, @@ -171,14 +217,14 @@ export function handleFileBrowserStreamRequest(req: IncomingMessage, res: Server try { res.write(": heartbeat\n\n"); } catch { - for (const entry of entries) releaseSubscriber(entry, res); + for (const { entry } of subscriptions) releaseSubscriber(entry, res); clearInterval(heartbeat); } }, HEARTBEAT_MS); res.on("close", () => { clearInterval(heartbeat); - for (const entry of entries) releaseSubscriber(entry, res); + for (const { entry } of subscriptions) releaseSubscriber(entry, res); }); return true; } diff --git a/packages/editor/App.tsx b/packages/editor/App.tsx index 3bebfc258..c4e5dfe91 100644 --- a/packages/editor/App.tsx +++ b/packages/editor/App.tsx @@ -149,7 +149,11 @@ import { } from './savedFileChangeValidation'; import { fetchSourceDocumentSnapshot, probeSourceSave } from './sourceDocumentClient'; import { reconcileSourceDocuments, type SourceDocumentReconcileEvent } from './sourceDocumentReconciliation'; -import { dirnameBrowserPath, normalizeBrowserPath, pathIsInsideDir } from './sourceDocumentPaths'; +import { + buildSourceWatchSubscription, + normalizeBrowserPath, + pathIsInsideDir, +} from './sourceDocumentPaths'; import { pickRestoredSingleFileDraftToDisplay } from './draftRestoreSelection'; type NoteAutoSaveResults = { @@ -2179,20 +2183,17 @@ const App: React.FC = () => { reconcileOpenSourceDocumentsRef.current = reconcileOpenSourceDocuments; }, [reconcileOpenSourceDocuments]); - const sourceWatchDirsKey = useMemo(() => { - const dirs = new Set(); - for (const doc of openSourceDocuments) dirs.add(dirnameBrowserPath(doc.sourceSave.path)); - return [...dirs].sort().join('\n'); - }, [openSourceDocuments]); + const sourceWatchSubscription = useMemo( + () => buildSourceWatchSubscription(openSourceDocuments.map((doc) => doc.sourceSave.path)), + [openSourceDocuments], + ); useEffect(() => { - if (!sourceWatchDirsKey || typeof EventSource === 'undefined') return; + if (!sourceWatchSubscription.key || typeof EventSource === 'undefined') return; - const dirs = sourceWatchDirsKey.split('\n').filter(Boolean); + const dirs = sourceWatchSubscription.dirs; const timers = new Map>(); - const params = new URLSearchParams(); - for (const dir of dirs) params.append('dirPath', dir); - const source = new EventSource(`/api/reference/files/stream?${params.toString()}`); + const source = new EventSource(`/api/reference/files/stream?${sourceWatchSubscription.query}`); const schedule = (dir?: string) => { const key = dir ?? '*'; @@ -2223,7 +2224,7 @@ const App: React.FC = () => { for (const timer of timers.values()) clearTimeout(timer); source.close(); }; - }, [sourceWatchDirsKey]); + }, [sourceWatchSubscription.key]); const handleTaterModeChange = useCallback((enabled: boolean) => { setTaterMode(enabled); diff --git a/packages/editor/sourceDocumentPaths.test.ts b/packages/editor/sourceDocumentPaths.test.ts index e1eb76790..219b95d27 100644 --- a/packages/editor/sourceDocumentPaths.test.ts +++ b/packages/editor/sourceDocumentPaths.test.ts @@ -1,5 +1,10 @@ import { describe, expect, test } from 'bun:test'; -import { dirnameBrowserPath, normalizeBrowserPath, pathIsInsideDir } from './sourceDocumentPaths'; +import { + buildSourceWatchSubscription, + dirnameBrowserPath, + normalizeBrowserPath, + pathIsInsideDir, +} from './sourceDocumentPaths'; describe('source document path helpers', () => { test('normalizes separators and trailing slashes', () => { @@ -26,4 +31,39 @@ describe('source document path helpers', () => { expect(pathIsInsideDir('C:\\note.md', 'C:/')).toBe(true); expect(pathIsInsideDir('/repo/docs/a.md', '')).toBe(false); }); + + test('builds a stable exact-file watch subscription from Unix paths', () => { + const subscription = buildSourceWatchSubscription([ + '/repo/docs/b.md', + '', + '/repo/docs/a.md', + '/repo/docs/a.md', + '/repo/notes/c.md', + ]); + + expect(subscription).toEqual({ + query: [ + 'filePath=%2Frepo%2Fdocs%2Fa.md', + 'filePath=%2Frepo%2Fdocs%2Fb.md', + 'filePath=%2Frepo%2Fnotes%2Fc.md', + ].join('&'), + dirs: ['/repo/docs', '/repo/notes'], + key: ['/repo/docs/a.md', '/repo/docs/b.md', '/repo/notes/c.md'].join('\n'), + }); + }); + + test('normalizes and deduplicates Windows paths before building a subscription', () => { + const subscription = buildSourceWatchSubscription([ + 'C:\\repo\\notes\\b.md', + 'C:/repo/docs/a.md', + 'C:\\repo\\docs\\a.md', + 'C:\\repo\\notes\\b.md', + ]); + + expect(subscription).toEqual({ + query: 'filePath=C%3A%2Frepo%2Fdocs%2Fa.md&filePath=C%3A%2Frepo%2Fnotes%2Fb.md', + dirs: ['C:/repo/docs', 'C:/repo/notes'], + key: 'C:/repo/docs/a.md\nC:/repo/notes/b.md', + }); + }); }); diff --git a/packages/editor/sourceDocumentPaths.ts b/packages/editor/sourceDocumentPaths.ts index f3b79138d..4f318abcc 100644 --- a/packages/editor/sourceDocumentPaths.ts +++ b/packages/editor/sourceDocumentPaths.ts @@ -1,5 +1,28 @@ +import { + dirnameBrowserPath, + normalizeBrowserPath, +} from '@plannotator/shared/browser-paths'; + export { dirnameBrowserPath, normalizeBrowserPath, pathIsInsideDir, } from '@plannotator/shared/browser-paths'; + +export interface SourceWatchSubscription { + query: string; + dirs: string[]; + key: string; +} + +export function buildSourceWatchSubscription(paths: string[]): SourceWatchSubscription { + const normalizedPaths = [...new Set(paths.map(normalizeBrowserPath).filter(Boolean))].sort(); + const params = new URLSearchParams(); + for (const path of normalizedPaths) params.append('filePath', path); + + return { + query: params.toString(), + dirs: [...new Set(normalizedPaths.map(dirnameBrowserPath))].sort(), + key: normalizedPaths.join('\n'), + }; +} diff --git a/packages/server/reference-watch.test.ts b/packages/server/reference-watch.test.ts index f7606a8af..fec9f59bd 100644 --- a/packages/server/reference-watch.test.ts +++ b/packages/server/reference-watch.test.ts @@ -1,10 +1,12 @@ import { afterEach, describe, expect, test } from "bun:test"; -import { mkdtempSync, rmSync } from "node:fs"; +import { mkdirSync, mkdtempSync, renameSync, rmSync, unlinkSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; import { basename, dirname, join } from "node:path"; import { handleFileBrowserFilesStream, isFileBrowserWatchIgnoredPath } from "./reference-watch"; const tempDirs: string[] = []; +const WATCH_READY_MS = 250; +const EVENT_TIMEOUT_MS = 2_000; function makeTempDir(prefix: string): string { const dir = mkdtempSync(join(tmpdir(), prefix)); @@ -12,39 +14,97 @@ function makeTempDir(prefix: string): string { return dir; } -async function readSSEEvents(response: Response, count: number): Promise> { +interface SSEEvent { + type?: string; + dirPath?: string; + reason?: string; +} + +interface SSECollector { + next(timeoutMs?: number): Promise; + close(): Promise; +} + +function collectSSE(response: Response): SSECollector { const reader = response.body?.getReader(); if (!reader) throw new Error("Missing response body"); const decoder = new TextDecoder(); - const events: Array<{ type?: string; dirPath?: string }> = []; + const events: SSEEvent[] = []; + const waiters: Array<(event: SSEEvent) => void> = []; let pending = ""; - try { - while (events.length < count) { - let timeout: ReturnType | null = null; - const result = await Promise.race([ - reader.read(), - new Promise((_, reject) => { - timeout = setTimeout(() => reject(new Error("Timed out waiting for SSE event")), 1000); - }), - ]); - if (timeout) clearTimeout(timeout); - if (result.done) break; - pending += decoder.decode(result.value, { stream: true }); - const blocks = pending.split("\n\n"); - pending = blocks.pop() ?? ""; - for (const block of blocks) { - const line = block.split("\n").find((item) => item.startsWith("data: ")); - if (!line) continue; - events.push(JSON.parse(line.slice("data: ".length))); + const pump = async () => { + try { + while (true) { + const result = await reader.read(); + if (result.done) break; + pending += decoder.decode(result.value, { stream: true }); + const blocks = pending.split("\n\n"); + pending = blocks.pop() ?? ""; + for (const block of blocks) { + const line = block.split("\n").find((item) => item.startsWith("data: ")); + if (!line) continue; + const event = JSON.parse(line.slice("data: ".length)) as SSEEvent; + const waiter = waiters.shift(); + if (waiter) waiter(event); + else events.push(event); + } } + } catch { + // Reader cancellation closes the collector. } + }; + void pump(); + + return { + next(timeoutMs = EVENT_TIMEOUT_MS) { + const event = events.shift(); + if (event) return Promise.resolve(event); + return new Promise((resolve, reject) => { + let timeout: ReturnType; + const waiter = (nextEvent: SSEEvent) => { + clearTimeout(timeout); + resolve(nextEvent); + }; + waiters.push(waiter); + timeout = setTimeout(() => { + const index = waiters.indexOf(waiter); + if (index >= 0) waiters.splice(index, 1); + reject(new Error("Timed out waiting for SSE event")); + }, timeoutMs); + }); + }, + close() { + return reader.cancel(); + }, + }; +} + +async function readSSEEvents(response: Response, count: number): Promise { + const collector = collectSSE(response); + const events: SSEEvent[] = []; + try { + while (events.length < count) events.push(await collector.next()); return events; } finally { - await reader.cancel(); + await collector.close(); } } +async function expectNoEvent(collector: SSECollector, timeoutMs = 500): Promise { + try { + const event = await collector.next(timeoutMs); + throw new Error(`Unexpected SSE event: ${JSON.stringify(event)}`); + } catch (error) { + if (error instanceof Error && error.message === "Timed out waiting for SSE event") return; + throw error; + } +} + +function waitForWatcher(): Promise { + return Bun.sleep(WATCH_READY_MS); +} + afterEach(() => { for (const dir of tempDirs.splice(0)) { rmSync(dir, { recursive: true, force: true }); @@ -94,4 +154,182 @@ describe("handleFileBrowserFilesStream", () => { expect(events[0]?.type).toBe("ready"); expect(events[0]?.dirPath).toBe(nonCanonicalRoot); }); + + test("requires exactly one watch parameter mode", () => { + const root = makeTempDir("plannotator-watch-mode-"); + const filePath = join(root, "plan.md"); + writeFileSync(filePath, "initial"); + + const missing = handleFileBrowserFilesStream( + new Request("http://localhost/api/reference/files/stream"), + ); + const mixedUrl = new URL("http://localhost/api/reference/files/stream"); + mixedUrl.searchParams.append("dirPath", root); + mixedUrl.searchParams.append("filePath", filePath); + const mixed = handleFileBrowserFilesStream(new Request(mixedUrl)); + + expect(missing.status).toBe(400); + expect(mixed.status).toBe(400); + }); + + test("watches an exact file through writes, deletion, recreation, and rename-away", async () => { + const root = makeTempDir("plannotator-watch-file-"); + const target = join(root, "plan.md"); + const sibling = join(root, "sibling.md"); + const nestedDir = join(root, "nested"); + writeFileSync(target, "initial"); + mkdirSync(nestedDir); + await waitForWatcher(); + + const url = new URL("http://localhost/api/reference/files/stream"); + url.searchParams.append("filePath", target); + const response = handleFileBrowserFilesStream(new Request(url)); + const collector = collectSSE(response); + + try { + expect(response.status).toBe(200); + expect(await collector.next()).toMatchObject({ + type: "ready", + dirPath: root, + reason: "initial", + }); + await waitForWatcher(); + + writeFileSync(target, "written"); + expect(await collector.next()).toMatchObject({ type: "changed", dirPath: root, reason: "files" }); + + writeFileSync(sibling, "sibling"); + writeFileSync(join(nestedDir, "nested.md"), "nested"); + await expectNoEvent(collector); + + unlinkSync(target); + expect(await collector.next()).toMatchObject({ type: "changed", dirPath: root, reason: "files" }); + + writeFileSync(target, "recreated"); + expect(await collector.next()).toMatchObject({ type: "changed", dirPath: root, reason: "files" }); + + renameSync(target, join(root, "renamed.md")); + expect(await collector.next()).toMatchObject({ type: "changed", dirPath: root, reason: "files" }); + } finally { + await collector.close(); + } + }); + + test("allows a missing leaf when its parent exists and reports its creation", async () => { + const root = makeTempDir("plannotator-watch-missing-"); + const target = join(root, "future.md"); + const url = new URL("http://localhost/api/reference/files/stream"); + url.searchParams.append("filePath", target); + const response = handleFileBrowserFilesStream(new Request(url)); + const collector = collectSSE(response); + + try { + expect(response.status).toBe(200); + expect(await collector.next()).toMatchObject({ type: "ready", dirPath: root }); + await waitForWatcher(); + writeFileSync(target, "created"); + expect(await collector.next()).toMatchObject({ type: "changed", dirPath: root, reason: "files" }); + } finally { + await collector.close(); + } + }); + + test("rejects file targets with an invalid parent or a directory leaf", () => { + const root = makeTempDir("plannotator-watch-invalid-"); + const emptyUrl = new URL("http://localhost/api/reference/files/stream"); + emptyUrl.searchParams.append("filePath", ""); + const missingParentUrl = new URL("http://localhost/api/reference/files/stream"); + missingParentUrl.searchParams.append("filePath", join(root, "missing", "plan.md")); + const directoryUrl = new URL("http://localhost/api/reference/files/stream"); + directoryUrl.searchParams.append("filePath", root); + + expect(handleFileBrowserFilesStream(new Request(emptyUrl)).status).toBe(400); + expect(handleFileBrowserFilesStream(new Request(missingParentUrl)).status).toBe(400); + expect(handleFileBrowserFilesStream(new Request(directoryUrl)).status).toBe(400); + }); + + test("deduplicates equivalent exact-file targets", async () => { + const root = makeTempDir("plannotator-watch-dedupe-"); + const target = join(root, "plan.md"); + writeFileSync(target, "initial"); + await waitForWatcher(); + const url = new URL("http://localhost/api/reference/files/stream"); + url.searchParams.append("filePath", target); + url.searchParams.append("filePath", join(root, ".", "plan.md")); + const response = handleFileBrowserFilesStream(new Request(url)); + const collector = collectSSE(response); + + try { + expect(await collector.next()).toMatchObject({ type: "ready", dirPath: root }); + await expectNoEvent(collector); + } finally { + await collector.close(); + } + }); + + test("keeps directory and exact-file watchers as separate cache entries", async () => { + const root = makeTempDir("plannotator-watch-cache-"); + const target = join(root, "plan.md"); + const sibling = join(root, "sibling.md"); + writeFileSync(target, "initial"); + await waitForWatcher(); + const dirUrl = new URL("http://localhost/api/reference/files/stream"); + dirUrl.searchParams.append("dirPath", root); + const fileUrl = new URL("http://localhost/api/reference/files/stream"); + fileUrl.searchParams.append("filePath", target); + const dirCollector = collectSSE(handleFileBrowserFilesStream(new Request(dirUrl))); + const fileCollector = collectSSE(handleFileBrowserFilesStream(new Request(fileUrl))); + + try { + await dirCollector.next(); + await fileCollector.next(); + await waitForWatcher(); + writeFileSync(sibling, "sibling"); + expect(await dirCollector.next()).toMatchObject({ type: "changed", reason: "files" }); + await expectNoEvent(fileCollector); + } finally { + await dirCollector.close(); + await fileCollector.close(); + } + }); + + test("preserves recursive directory watching", async () => { + const root = makeTempDir("plannotator-watch-recursive-"); + const nested = join(root, "docs", "nested"); + mkdirSync(nested, { recursive: true }); + const url = new URL("http://localhost/api/reference/files/stream"); + url.searchParams.append("dirPath", root); + const collector = collectSSE(handleFileBrowserFilesStream(new Request(url))); + + try { + await collector.next(); + await waitForWatcher(); + writeFileSync(join(nested, "plan.md"), "nested"); + expect(await collector.next()).toMatchObject({ type: "changed", dirPath: root, reason: "files" }); + } finally { + await collector.close(); + } + }); + + test("does not report adjacent Git metadata changes for an exact file", async () => { + const root = makeTempDir("plannotator-watch-git-"); + const target = join(root, "plan.md"); + const gitDir = join(root, ".git"); + mkdirSync(gitDir); + writeFileSync(join(gitDir, "HEAD"), "ref: refs/heads/main\n"); + writeFileSync(target, "initial"); + await waitForWatcher(); + const url = new URL("http://localhost/api/reference/files/stream"); + url.searchParams.append("filePath", target); + const collector = collectSSE(handleFileBrowserFilesStream(new Request(url))); + + try { + await collector.next(); + await waitForWatcher(); + writeFileSync(join(gitDir, "HEAD"), "ref: refs/heads/other\n"); + await expectNoEvent(collector); + } finally { + await collector.close(); + } + }); }); diff --git a/packages/server/reference-watch.ts b/packages/server/reference-watch.ts index 0cca22ffd..d4b0e6b01 100644 --- a/packages/server/reference-watch.ts +++ b/packages/server/reference-watch.ts @@ -1,6 +1,6 @@ import chokidar, { type FSWatcher } from "chokidar"; import { existsSync, statSync } from "fs"; -import { isAbsolute, relative } from "path"; +import { dirname, isAbsolute, relative } from "path"; import { isFileBrowserExcludedPath } from "@plannotator/shared/reference-common"; import { resolveUserPath } from "@plannotator/shared/resolve-file"; import { getGitMetadataWatchPaths } from "@plannotator/shared/workspace-status"; @@ -13,13 +13,21 @@ interface FileBrowserChangeEvent { } interface WatchEntry { - dirPath: string; + key: string; subscribers: Map; contentWatcher: FSWatcher | null; gitWatcher: FSWatcher | null; debounceTimer: ReturnType | null; } +interface WatchTarget { + key: string; + watchPath: string; + clientDirPath: string; + watchGit: boolean; + ignored?: (path: string) => boolean; +} + const HEARTBEAT_MS = 30_000; const DEBOUNCE_MS = 180; const watchers = new Map(); @@ -71,8 +79,8 @@ function closeWatcher(entry: WatchEntry): void { if (entry.debounceTimer) clearTimeout(entry.debounceTimer); void entry.contentWatcher?.close(); void entry.gitWatcher?.close(); - if (watchers.get(entry.dirPath) === entry) { - watchers.delete(entry.dirPath); + if (watchers.get(entry.key) === entry) { + watchers.delete(entry.key); } } @@ -81,22 +89,22 @@ function releaseSubscriber(entry: WatchEntry, controller: ReadableStreamDefaultC if (entry.subscribers.size === 0) closeWatcher(entry); } -function ensureWatcher(dirPath: string): WatchEntry { - const existing = watchers.get(dirPath); +function ensureWatcher(target: WatchTarget): WatchEntry { + const existing = watchers.get(target.key); if (existing) return existing; const entry: WatchEntry = { - dirPath, + key: target.key, subscribers: new Map(), contentWatcher: null, gitWatcher: null, debounceTimer: null, }; - entry.contentWatcher = chokidar.watch(dirPath, { + entry.contentWatcher = chokidar.watch(target.watchPath, { ignoreInitial: true, persistent: true, - ignored: (path) => isFileBrowserWatchIgnoredPath(path, dirPath), + ignored: target.ignored, awaitWriteFinish: { stabilityThreshold: 120, pollInterval: 30, @@ -105,7 +113,9 @@ function ensureWatcher(dirPath: string): WatchEntry { entry.contentWatcher.on("all", () => scheduleBroadcast(entry, "files")); entry.contentWatcher.on("error", () => scheduleBroadcast(entry, "files")); - const gitWatchPaths = getGitMetadataWatchPaths(dirPath); + const gitWatchPaths = target.watchGit + ? getGitMetadataWatchPaths(target.watchPath) + : []; if (gitWatchPaths.length > 0) { entry.gitWatcher = chokidar.watch(gitWatchPaths, { ignoreInitial: true, @@ -119,44 +129,79 @@ function ensureWatcher(dirPath: string): WatchEntry { entry.gitWatcher.on("error", () => scheduleBroadcast(entry, "git")); } - watchers.set(dirPath, entry); + watchers.set(target.key, entry); return entry; } +function isValidFileTarget(filePath: string): boolean { + if (!filePath) return false; + try { + if (existsSync(filePath)) return !statSync(filePath).isDirectory(); + return isValidDirectory(dirname(filePath)); + } catch { + return false; + } +} + export function handleFileBrowserFilesStream( req: Request, options?: { disableIdleTimeout?: () => void }, ): Response { const url = new URL(req.url); const rawDirPaths = url.searchParams.getAll("dirPath"); - if (rawDirPaths.length === 0) { - return Response.json({ error: "Missing dirPath parameter" }, { status: 400 }); + const rawFilePaths = url.searchParams.getAll("filePath"); + if ((rawDirPaths.length > 0) === (rawFilePaths.length > 0)) { + return Response.json({ error: "Provide exactly one of dirPath or filePath" }, { status: 400 }); } - const dirPaths: string[] = []; - const clientDirPaths: string[] = []; - for (const rawDirPath of rawDirPaths) { - const dirPath = resolveUserPath(rawDirPath); - if (!isValidDirectory(dirPath)) { - return Response.json({ error: "Invalid directory path" }, { status: 400 }); + const targets = new Map(); + if (rawDirPaths.length > 0) { + for (const rawDirPath of rawDirPaths) { + const dirPath = resolveUserPath(rawDirPath); + if (!isValidDirectory(dirPath)) { + return Response.json({ error: "Invalid directory path" }, { status: 400 }); + } + const key = `dir:${dirPath}`; + if (!targets.has(key)) { + targets.set(key, { + key, + watchPath: dirPath, + clientDirPath: rawDirPath, + watchGit: true, + ignored: (path) => isFileBrowserWatchIgnoredPath(path, dirPath), + }); + } } - if (!dirPaths.includes(dirPath)) { - dirPaths.push(dirPath); - clientDirPaths.push(rawDirPath); + } else { + for (const rawFilePath of rawFilePaths) { + const filePath = resolveUserPath(rawFilePath); + if (!isValidFileTarget(filePath)) { + return Response.json({ error: "Invalid file path" }, { status: 400 }); + } + const key = `file:${filePath}`; + if (!targets.has(key)) { + targets.set(key, { + key, + watchPath: filePath, + clientDirPath: dirname(rawFilePath), + watchGit: false, + }); + } } } options?.disableIdleTimeout?.(); - const entries = dirPaths.map((dirPath) => ensureWatcher(dirPath)); + const subscriptions = [...targets.values()].map((target) => ({ + entry: ensureWatcher(target), + clientDirPath: target.clientDirPath, + })); let controllerRef: ReadableStreamDefaultController | null = null; let heartbeatTimer: ReturnType | null = null; const stream = new ReadableStream({ start(controller) { controllerRef = controller; - for (let i = 0; i < entries.length; i++) { - const entry = entries[i]!; - const clientDirPath = clientDirPaths[i] ?? entry.dirPath; + for (const { entry, clientDirPath } of subscriptions) { entry.subscribers.set(controller, clientDirPath); controller.enqueue(serialize({ type: "ready", @@ -169,7 +214,7 @@ export function handleFileBrowserFilesStream( try { controller.enqueue(encoder.encode(": heartbeat\n\n")); } catch { - for (const entry of entries) releaseSubscriber(entry, controller); + for (const { entry } of subscriptions) releaseSubscriber(entry, controller); if (heartbeatTimer) clearInterval(heartbeatTimer); } }, HEARTBEAT_MS); @@ -177,7 +222,7 @@ export function handleFileBrowserFilesStream( cancel() { if (heartbeatTimer) clearInterval(heartbeatTimer); if (controllerRef) { - for (const entry of entries) releaseSubscriber(entry, controllerRef); + for (const { entry } of subscriptions) releaseSubscriber(entry, controllerRef); } }, }); From 1a52c7b7380de1ceb9e3892edfdd17873ee62434 Mon Sep 17 00:00:00 2001 From: rNoz Date: Fri, 24 Jul 2026 09:25:20 +0200 Subject: [PATCH 2/5] test(annotate): cover atomic watcher saves --- .../server/file-browser-watch.test.ts | 28 +++++++++++++++++++ packages/server/reference-watch.test.ts | 27 ++++++++++++++++++ 2 files changed, 55 insertions(+) diff --git a/apps/pi-extension/server/file-browser-watch.test.ts b/apps/pi-extension/server/file-browser-watch.test.ts index f4c68a76f..9f02ae81f 100644 --- a/apps/pi-extension/server/file-browser-watch.test.ts +++ b/apps/pi-extension/server/file-browser-watch.test.ts @@ -195,6 +195,34 @@ describe("Pi file browser watcher", () => { } }); + test("keeps watching after atomic rename-over saves", async () => { + const root = makeTempDir("plannotator-pi-watch-atomic-"); + const target = join(root, "plan.md"); + writeFileSync(target, "initial"); + await waitForWatcher(); + const origin = await startWatchServer(); + const url = new URL(`${origin}/api/reference/files/stream`); + url.searchParams.append("filePath", target); + const collector = collectSSE(await fetch(url)); + + try { + await collector.next(); + await waitForWatcher(); + for (const content of ["first", "second"]) { + const replacement = join(root, `plan-${content}.tmp`); + writeFileSync(replacement, content); + renameSync(replacement, target); + expect(await collector.next()).toMatchObject({ + type: "changed", + dirPath: root, + reason: "files", + }); + } + } finally { + await collector.close(); + } + }); + test("allows a missing leaf when its parent exists", async () => { const root = makeTempDir("plannotator-pi-watch-missing-"); const target = join(root, "future.md"); diff --git a/packages/server/reference-watch.test.ts b/packages/server/reference-watch.test.ts index fec9f59bd..19ddcd14a 100644 --- a/packages/server/reference-watch.test.ts +++ b/packages/server/reference-watch.test.ts @@ -215,6 +215,33 @@ describe("handleFileBrowserFilesStream", () => { } }); + test("keeps watching after atomic rename-over saves", async () => { + const root = makeTempDir("plannotator-watch-atomic-"); + const target = join(root, "plan.md"); + writeFileSync(target, "initial"); + await waitForWatcher(); + const url = new URL("http://localhost/api/reference/files/stream"); + url.searchParams.append("filePath", target); + const collector = collectSSE(handleFileBrowserFilesStream(new Request(url))); + + try { + await collector.next(); + await waitForWatcher(); + for (const content of ["first", "second"]) { + const replacement = join(root, `plan-${content}.tmp`); + writeFileSync(replacement, content); + renameSync(replacement, target); + expect(await collector.next()).toMatchObject({ + type: "changed", + dirPath: root, + reason: "files", + }); + } + } finally { + await collector.close(); + } + }); + test("allows a missing leaf when its parent exists and reports its creation", async () => { const root = makeTempDir("plannotator-watch-missing-"); const target = join(root, "future.md"); From e22d145675c4b08ab845df9ae64793cf7d402f6c Mon Sep 17 00:00:00 2001 From: rNoz Date: Fri, 24 Jul 2026 18:14:11 +0200 Subject: [PATCH 3/5] fix(watch): survive atomic file replacement --- .../pi-extension/server/file-browser-watch.ts | 20 ++++++++++++++++--- packages/server/reference-watch.ts | 20 ++++++++++++++++--- 2 files changed, 34 insertions(+), 6 deletions(-) diff --git a/apps/pi-extension/server/file-browser-watch.ts b/apps/pi-extension/server/file-browser-watch.ts index 8204f316e..dc3895c7f 100644 --- a/apps/pi-extension/server/file-browser-watch.ts +++ b/apps/pi-extension/server/file-browser-watch.ts @@ -1,7 +1,7 @@ import chokidar, { type FSWatcher } from "chokidar"; import { existsSync, statSync } from "node:fs"; import type { IncomingMessage, ServerResponse } from "node:http"; -import { dirname, isAbsolute, relative } from "node:path"; +import { dirname, isAbsolute, relative, resolve } from "node:path"; import { isFileBrowserExcludedPath } from "../generated/reference-common.ts"; import { resolveUserPath } from "../generated/resolve-file.ts"; @@ -28,6 +28,8 @@ interface WatchTarget { watchPath: string; clientDirPath: string; watchGit: boolean; + depth?: number; + matchesContentEvent?: (path: string) => boolean; ignored?: (path: string) => boolean; } @@ -106,13 +108,18 @@ function ensureWatcher(target: WatchTarget): WatchEntry { entry.contentWatcher = chokidar.watch(target.watchPath, { ignoreInitial: true, persistent: true, + depth: target.depth, ignored: target.ignored, awaitWriteFinish: { stabilityThreshold: 120, pollInterval: 30, }, }); - entry.contentWatcher.on("all", () => scheduleBroadcast(entry, "files")); + entry.contentWatcher.on("all", (_event, path) => { + if (!target.matchesContentEvent || target.matchesContentEvent(path)) { + scheduleBroadcast(entry, "files"); + } + }); entry.contentWatcher.on("error", () => scheduleBroadcast(entry, "files")); const gitWatchPaths = target.watchGit @@ -183,11 +190,18 @@ export function handleFileBrowserStreamRequest(req: IncomingMessage, res: Server } const key = `file:${filePath}`; if (!targets.has(key)) { + const parentPath = dirname(filePath); targets.set(key, { key, - watchPath: filePath, + watchPath: parentPath, clientDirPath: dirname(rawFilePath), watchGit: false, + depth: 0, + matchesContentEvent: (path) => resolve(path) === filePath, + ignored: (path) => { + const resolvedPath = resolve(path); + return resolvedPath !== parentPath && resolvedPath !== filePath; + }, }); } } diff --git a/packages/server/reference-watch.ts b/packages/server/reference-watch.ts index d4b0e6b01..701c98fa7 100644 --- a/packages/server/reference-watch.ts +++ b/packages/server/reference-watch.ts @@ -1,6 +1,6 @@ import chokidar, { type FSWatcher } from "chokidar"; import { existsSync, statSync } from "fs"; -import { dirname, isAbsolute, relative } from "path"; +import { dirname, isAbsolute, relative, resolve } from "path"; import { isFileBrowserExcludedPath } from "@plannotator/shared/reference-common"; import { resolveUserPath } from "@plannotator/shared/resolve-file"; import { getGitMetadataWatchPaths } from "@plannotator/shared/workspace-status"; @@ -25,6 +25,8 @@ interface WatchTarget { watchPath: string; clientDirPath: string; watchGit: boolean; + depth?: number; + matchesContentEvent?: (path: string) => boolean; ignored?: (path: string) => boolean; } @@ -104,13 +106,18 @@ function ensureWatcher(target: WatchTarget): WatchEntry { entry.contentWatcher = chokidar.watch(target.watchPath, { ignoreInitial: true, persistent: true, + depth: target.depth, ignored: target.ignored, awaitWriteFinish: { stabilityThreshold: 120, pollInterval: 30, }, }); - entry.contentWatcher.on("all", () => scheduleBroadcast(entry, "files")); + entry.contentWatcher.on("all", (_event, path) => { + if (!target.matchesContentEvent || target.matchesContentEvent(path)) { + scheduleBroadcast(entry, "files"); + } + }); entry.contentWatcher.on("error", () => scheduleBroadcast(entry, "files")); const gitWatchPaths = target.watchGit @@ -180,11 +187,18 @@ export function handleFileBrowserFilesStream( } const key = `file:${filePath}`; if (!targets.has(key)) { + const parentPath = dirname(filePath); targets.set(key, { key, - watchPath: filePath, + watchPath: parentPath, clientDirPath: dirname(rawFilePath), watchGit: false, + depth: 0, + matchesContentEvent: (path) => resolve(path) === filePath, + ignored: (path) => { + const resolvedPath = resolve(path); + return resolvedPath !== parentPath && resolvedPath !== filePath; + }, }); } } From 7682c5a9a80abeca10df48b3beb66fcf97306222 Mon Sep 17 00:00:00 2001 From: rNoz Date: Fri, 24 Jul 2026 18:19:27 +0200 Subject: [PATCH 4/5] fix(watch): disable exact-file coalescing --- apps/pi-extension/server/file-browser-watch.ts | 3 +++ packages/server/reference-watch.ts | 3 +++ 2 files changed, 6 insertions(+) diff --git a/apps/pi-extension/server/file-browser-watch.ts b/apps/pi-extension/server/file-browser-watch.ts index dc3895c7f..c9adc46ab 100644 --- a/apps/pi-extension/server/file-browser-watch.ts +++ b/apps/pi-extension/server/file-browser-watch.ts @@ -28,6 +28,7 @@ interface WatchTarget { watchPath: string; clientDirPath: string; watchGit: boolean; + atomic?: boolean; depth?: number; matchesContentEvent?: (path: string) => boolean; ignored?: (path: string) => boolean; @@ -108,6 +109,7 @@ function ensureWatcher(target: WatchTarget): WatchEntry { entry.contentWatcher = chokidar.watch(target.watchPath, { ignoreInitial: true, persistent: true, + atomic: target.atomic, depth: target.depth, ignored: target.ignored, awaitWriteFinish: { @@ -196,6 +198,7 @@ export function handleFileBrowserStreamRequest(req: IncomingMessage, res: Server watchPath: parentPath, clientDirPath: dirname(rawFilePath), watchGit: false, + atomic: false, depth: 0, matchesContentEvent: (path) => resolve(path) === filePath, ignored: (path) => { diff --git a/packages/server/reference-watch.ts b/packages/server/reference-watch.ts index 701c98fa7..d1beac849 100644 --- a/packages/server/reference-watch.ts +++ b/packages/server/reference-watch.ts @@ -25,6 +25,7 @@ interface WatchTarget { watchPath: string; clientDirPath: string; watchGit: boolean; + atomic?: boolean; depth?: number; matchesContentEvent?: (path: string) => boolean; ignored?: (path: string) => boolean; @@ -106,6 +107,7 @@ function ensureWatcher(target: WatchTarget): WatchEntry { entry.contentWatcher = chokidar.watch(target.watchPath, { ignoreInitial: true, persistent: true, + atomic: target.atomic, depth: target.depth, ignored: target.ignored, awaitWriteFinish: { @@ -193,6 +195,7 @@ export function handleFileBrowserFilesStream( watchPath: parentPath, clientDirPath: dirname(rawFilePath), watchGit: false, + atomic: false, depth: 0, matchesContentEvent: (path) => resolve(path) === filePath, ignored: (path) => { From d8a4e772a4f7674ff2175bbb4cbd63d79c99115b Mon Sep 17 00:00:00 2001 From: rNoz Date: Fri, 24 Jul 2026 18:32:10 +0200 Subject: [PATCH 5/5] fix(watch): track exact file signatures --- .../pi-extension/server/file-browser-watch.ts | 74 +++++++++++-------- packages/server/reference-watch.ts | 74 +++++++++++-------- 2 files changed, 88 insertions(+), 60 deletions(-) diff --git a/apps/pi-extension/server/file-browser-watch.ts b/apps/pi-extension/server/file-browser-watch.ts index c9adc46ab..0fc28c756 100644 --- a/apps/pi-extension/server/file-browser-watch.ts +++ b/apps/pi-extension/server/file-browser-watch.ts @@ -1,5 +1,5 @@ -import chokidar, { type FSWatcher } from "chokidar"; -import { existsSync, statSync } from "node:fs"; +import chokidar, { type FSWatcher as ChokidarWatcher } from "chokidar"; +import { existsSync, statSync, watch, type FSWatcher as NodeWatcher } from "node:fs"; import type { IncomingMessage, ServerResponse } from "node:http"; import { dirname, isAbsolute, relative, resolve } from "node:path"; @@ -18,8 +18,8 @@ interface FileBrowserChangeEvent { interface WatchEntry { key: string; subscribers: Map; - contentWatcher: FSWatcher | null; - gitWatcher: FSWatcher | null; + contentWatcher: ChokidarWatcher | NodeWatcher | null; + gitWatcher: ChokidarWatcher | null; debounceTimer: ReturnType | null; } @@ -28,9 +28,7 @@ interface WatchTarget { watchPath: string; clientDirPath: string; watchGit: boolean; - atomic?: boolean; - depth?: number; - matchesContentEvent?: (path: string) => boolean; + exactFilePath?: string; ignored?: (path: string) => boolean; } @@ -56,6 +54,17 @@ function isValidDirectory(dirPath: string): boolean { } } +function getFileSignature(filePath: string): string { + try { + const stats = statSync(filePath, { bigint: true }); + return stats.isDirectory() + ? "directory" + : `${stats.dev}:${stats.ino}:${stats.size}:${stats.mtimeNs}:${stats.ctimeNs}`; + } catch { + return "missing"; + } +} + function broadcast(entry: WatchEntry, reason: FileBrowserChangeEvent["reason"]): void { for (const [res, clientDirPath] of entry.subscribers) { const payload = serialize({ @@ -106,23 +115,34 @@ function ensureWatcher(target: WatchTarget): WatchEntry { debounceTimer: null, }; - entry.contentWatcher = chokidar.watch(target.watchPath, { - ignoreInitial: true, - persistent: true, - atomic: target.atomic, - depth: target.depth, - ignored: target.ignored, - awaitWriteFinish: { - stabilityThreshold: 120, - pollInterval: 30, - }, - }); - entry.contentWatcher.on("all", (_event, path) => { - if (!target.matchesContentEvent || target.matchesContentEvent(path)) { + if (target.exactFilePath) { + const exactFilePath = target.exactFilePath; + let signature = getFileSignature(exactFilePath); + entry.contentWatcher = watch(target.watchPath, { persistent: true }, (_event, filename) => { + const nextSignature = getFileSignature(exactFilePath); + const eventMatches = filename === null + || resolve(target.watchPath, filename.toString()) === exactFilePath; + if (eventMatches || nextSignature !== signature) { + signature = nextSignature; + scheduleBroadcast(entry, "files"); + } + }); + entry.contentWatcher.on("error", () => scheduleBroadcast(entry, "files")); + } else { + entry.contentWatcher = chokidar.watch(target.watchPath, { + ignoreInitial: true, + persistent: true, + ignored: target.ignored, + awaitWriteFinish: { + stabilityThreshold: 120, + pollInterval: 30, + }, + }); + entry.contentWatcher.on("all", () => { scheduleBroadcast(entry, "files"); - } - }); - entry.contentWatcher.on("error", () => scheduleBroadcast(entry, "files")); + }); + entry.contentWatcher.on("error", () => scheduleBroadcast(entry, "files")); + } const gitWatchPaths = target.watchGit ? getGitMetadataWatchPaths(target.watchPath) @@ -198,13 +218,7 @@ export function handleFileBrowserStreamRequest(req: IncomingMessage, res: Server watchPath: parentPath, clientDirPath: dirname(rawFilePath), watchGit: false, - atomic: false, - depth: 0, - matchesContentEvent: (path) => resolve(path) === filePath, - ignored: (path) => { - const resolvedPath = resolve(path); - return resolvedPath !== parentPath && resolvedPath !== filePath; - }, + exactFilePath: filePath, }); } } diff --git a/packages/server/reference-watch.ts b/packages/server/reference-watch.ts index d1beac849..88fe17d14 100644 --- a/packages/server/reference-watch.ts +++ b/packages/server/reference-watch.ts @@ -1,5 +1,5 @@ -import chokidar, { type FSWatcher } from "chokidar"; -import { existsSync, statSync } from "fs"; +import chokidar, { type FSWatcher as ChokidarWatcher } from "chokidar"; +import { existsSync, statSync, watch, type FSWatcher as NodeWatcher } from "fs"; import { dirname, isAbsolute, relative, resolve } from "path"; import { isFileBrowserExcludedPath } from "@plannotator/shared/reference-common"; import { resolveUserPath } from "@plannotator/shared/resolve-file"; @@ -15,8 +15,8 @@ interface FileBrowserChangeEvent { interface WatchEntry { key: string; subscribers: Map; - contentWatcher: FSWatcher | null; - gitWatcher: FSWatcher | null; + contentWatcher: ChokidarWatcher | NodeWatcher | null; + gitWatcher: ChokidarWatcher | null; debounceTimer: ReturnType | null; } @@ -25,9 +25,7 @@ interface WatchTarget { watchPath: string; clientDirPath: string; watchGit: boolean; - atomic?: boolean; - depth?: number; - matchesContentEvent?: (path: string) => boolean; + exactFilePath?: string; ignored?: (path: string) => boolean; } @@ -54,6 +52,17 @@ function isValidDirectory(dirPath: string): boolean { } } +function getFileSignature(filePath: string): string { + try { + const stats = statSync(filePath, { bigint: true }); + return stats.isDirectory() + ? "directory" + : `${stats.dev}:${stats.ino}:${stats.size}:${stats.mtimeNs}:${stats.ctimeNs}`; + } catch { + return "missing"; + } +} + function broadcast(entry: WatchEntry, reason: FileBrowserChangeEvent["reason"]): void { for (const [subscriber, clientDirPath] of entry.subscribers) { const payload = serialize({ @@ -104,23 +113,34 @@ function ensureWatcher(target: WatchTarget): WatchEntry { debounceTimer: null, }; - entry.contentWatcher = chokidar.watch(target.watchPath, { - ignoreInitial: true, - persistent: true, - atomic: target.atomic, - depth: target.depth, - ignored: target.ignored, - awaitWriteFinish: { - stabilityThreshold: 120, - pollInterval: 30, - }, - }); - entry.contentWatcher.on("all", (_event, path) => { - if (!target.matchesContentEvent || target.matchesContentEvent(path)) { + if (target.exactFilePath) { + const exactFilePath = target.exactFilePath; + let signature = getFileSignature(exactFilePath); + entry.contentWatcher = watch(target.watchPath, { persistent: true }, (_event, filename) => { + const nextSignature = getFileSignature(exactFilePath); + const eventMatches = filename === null + || resolve(target.watchPath, filename.toString()) === exactFilePath; + if (eventMatches || nextSignature !== signature) { + signature = nextSignature; + scheduleBroadcast(entry, "files"); + } + }); + entry.contentWatcher.on("error", () => scheduleBroadcast(entry, "files")); + } else { + entry.contentWatcher = chokidar.watch(target.watchPath, { + ignoreInitial: true, + persistent: true, + ignored: target.ignored, + awaitWriteFinish: { + stabilityThreshold: 120, + pollInterval: 30, + }, + }); + entry.contentWatcher.on("all", () => { scheduleBroadcast(entry, "files"); - } - }); - entry.contentWatcher.on("error", () => scheduleBroadcast(entry, "files")); + }); + entry.contentWatcher.on("error", () => scheduleBroadcast(entry, "files")); + } const gitWatchPaths = target.watchGit ? getGitMetadataWatchPaths(target.watchPath) @@ -195,13 +215,7 @@ export function handleFileBrowserFilesStream( watchPath: parentPath, clientDirPath: dirname(rawFilePath), watchGit: false, - atomic: false, - depth: 0, - matchesContentEvent: (path) => resolve(path) === filePath, - ignored: (path) => { - const resolvedPath = resolve(path); - return resolvedPath !== parentPath && resolvedPath !== filePath; - }, + exactFilePath: filePath, }); } }