diff --git a/packages/durable/CHANGELOG.md b/packages/durable/CHANGELOG.md index efc13e4b8..7494ba865 100644 --- a/packages/durable/CHANGELOG.md +++ b/packages/durable/CHANGELOG.md @@ -2,6 +2,13 @@ ## [Unreleased] +### Fixed + +- The `read` tool reads a file that grows while it is read (an active log) instead of failing with "changed while it was read"; it reads again only when the file shrank or was rewritten in place. +- `NodeExecutionEnv.watch()`: a directory replaced at the same path gets a new native watcher; a target that is a symbolic link to a file reports changes to that file; a non-recursive target no longer stops an overlapping recursive target from covering subdirectories; closing during a rescan no longer leaks watchers; a target that cannot be read for lack of permission fails with `permission_denied` instead of being watched as missing. +- Bounded tool output keeps a U+FEFF at the start of the retained tail, and drops a byte-order mark only at the very start of byte output. +- `settings.progress` fields given as `undefined` keep their defaults. + ## [1.0.3] - 2026-10-05 ### Breaking Changes diff --git a/packages/durable/src/env/node-watch.ts b/packages/durable/src/env/node-watch.ts index c0000d62e..7f3b8ab55 100644 --- a/packages/durable/src/env/node-watch.ts +++ b/packages/durable/src/env/node-watch.ts @@ -37,6 +37,12 @@ type Entry = { type Snapshot = Map; +/** A native watcher and the identity of what it was installed on: a path replaced by another file needs a new one. */ +type Installed = { readonly watcher: FSWatcher; readonly dev: number; readonly ino: number }; + +/** A scan: the snapshot, and targets that are symbolic links to files, whose files need their own watchers. */ +type Scan = { readonly snapshot: Snapshot; readonly linkedFiles: ReadonlySet }; + const DEBOUNCE_MS = 50; /** * On macOS, `fs.watch` returns before libuv's FSEvents stream is live, and changes in between are never reported @@ -73,6 +79,10 @@ function isNodeError(error: unknown): error is NodeJS.ErrnoException { return error instanceof Error && "code" in error; } +function isDenied(error: unknown): boolean { + return isNodeError(error) && (error.code === "EACCES" || error.code === "EPERM"); +} + function entryOf(stats: Stats, hash?: string): Entry { const kind = stats.isFile() ? "file" @@ -131,7 +141,7 @@ export class NodeFileWatcher implements FileWatcher { readonly #onChange: (change: WatchChange) => void; readonly #pollIntervalMs: number; readonly #maxDirectories: number; - readonly #watchers = new Map(); + readonly #watchers = new Map(); readonly #events = new Set(); #mode: "native" | "polling"; #snapshot: Snapshot = new Map(); @@ -202,7 +212,7 @@ export class NodeFileWatcher implements FileWatcher { this.#timer = undefined; clearTimeout(this.#settleTimer); this.#settleTimer = undefined; - for (const watcher of this.#watchers.values()) watcher.close(); + for (const { watcher } of this.#watchers.values()) watcher.close(); this.#watchers.clear(); } @@ -256,10 +266,11 @@ export class NodeFileWatcher implements FileWatcher { if (changed.size > 0) this.#deliver({ paths: [...changed].sort() }); } while (this.#dirty && !this.#closed); } catch (error) { + const message = error instanceof Error ? error.message : String(error); const fileError = error instanceof FileError ? error - : new FileError("invalid", error instanceof Error ? error.message : String(error)); + : new FileError(isDenied(error) ? "permission_denied" : "invalid", message); this.#deliver({ error: fileError }); this.#stop(); } finally { @@ -273,10 +284,13 @@ export class NodeFileWatcher implements FileWatcher { async #sync(report: boolean): Promise> { const changed = new Set(); for (let round = 0; round < 10 && !this.#closed; round++) { - const next = await this.#scan(); + const scan = await this.#scan(); + // Closed during the scan: installing watchers now would leak them. + if (this.#closed) break; + const next = scan.snapshot; if (report) for (const path of this.#diff(this.#snapshot, next)) changed.add(path); this.#snapshot = next; - if (this.#mode === "polling" || !this.#reconcileWatchers(next)) break; + if (this.#mode === "polling" || !this.#reconcileWatchers(scan)) break; if (process.platform === "darwin") this.#scheduleSettle(); // Something written into a new directory before its watcher existed shows up in the next round. report = true; @@ -300,11 +314,19 @@ export class NodeFileWatcher implements FileWatcher { return this.#targets.find((target) => isWithin(target.path, path))?.path ?? path; } - async #scan(): Promise { + async #scan(): Promise { const snapshot: Snapshot = new Map(); - let directories = 0; - const countDirectory = (): void => { - if (++directories > this.#maxDirectories) { + const linkedFiles = new Set(); + // Directories are counted once, but traversed once per target: overlapping targets differ in recursion and + // exclusions, and an entry recorded for one target must still be descended into for another. + const counted = new Set(); + const traversed = new Set(); + /** Kinds of listed entries, not following links: a target that links to a directory is recorded by `stat`, but + * symbolic links below a recursive target are not followed. */ + const listed = new Map(); + const countDirectory = (path: string): void => { + counted.add(path); + if (counted.size > this.#maxDirectories) { throw new BudgetExceeded(`Watched paths exceed ${this.#maxDirectories} directories`); } }; @@ -323,52 +345,73 @@ export class NodeFileWatcher implements FileWatcher { } snapshot.set(path, entryOf(stats, hash)); }; - const scanDirectory = async (target: ResolvedTarget, directory: string): Promise => { + const scanDirectory = async (index: number, target: ResolvedTarget, directory: string): Promise => { + const key = `${index}\0${directory}`; + if (traversed.has(key)) return; + traversed.add(key); let names: string[]; try { names = await readdir(directory); } catch (error) { + // The watched directory itself must be readable; below it, unreadable directories are skipped. + if (directory === target.path && isDenied(error)) throw error; if (isNodeError(error) && ["ENOENT", "EACCES", "EPERM", "ENOTDIR"].includes(error.code ?? "")) return; throw error; } for (const name of names) { if (excluded(target, name)) continue; const path = `${directory.endsWith(sep) ? directory : directory + sep}${name}`; - if (snapshot.has(path)) continue; - const stats = await lstat(path).catch(() => undefined); - if (stats === undefined) continue; - await record(path, stats); - if (target.recursive && stats.isDirectory()) { - countDirectory(); - await scanDirectory(target, path); + let kind = listed.get(path); + if (kind === undefined) { + const stats = await lstat(path).catch(() => undefined); + if (stats === undefined) continue; + kind = entryOf(stats).kind; + listed.set(path, kind); + // A target's own entry (following links) wins over its listing by another target. + if (!snapshot.has(path)) await record(path, stats); + } + if (target.recursive && kind === "directory") { + countDirectory(path); + await scanDirectory(index, target, path); } } }; - for (const target of this.#targets) { + for (const [index, target] of this.#targets.entries()) { for (const ancestor of ancestorsOf(target.path)) { if (snapshot.has(ancestor)) continue; const stats = await lstat(ancestor).catch(() => undefined); // Identity only: an ancestor's own timestamps change with every unrelated sibling. if (stats !== undefined) snapshot.set(ancestor, { ...entryOf(stats), size: 0, mtimeMs: 0 }); } - // The target itself may be a symbolic link to what is watched; follow it. - const stats = await stat(target.path).catch(() => undefined); - if (stats === undefined) continue; + // The target itself may be a symbolic link to what is watched; follow it. A missing target is watched for its + // creation; one that cannot be reached for lack of permission fails. + let stats: Stats; + try { + stats = await stat(target.path); + } catch (error) { + if (isDenied(error)) throw error; + continue; + } await record(target.path, stats); + if (stats.isFile() && (await lstat(target.path).catch(() => undefined))?.isSymbolicLink()) { + linkedFiles.add(target.path); + } if (stats.isDirectory()) { - countDirectory(); - await scanDirectory(target, target.path); + countDirectory(target.path); + await scanDirectory(index, target, target.path); } } - return snapshot; + return { snapshot, linkedFiles }; } /** - * Watch every existing ancestor of each target, each target directory, and on Linux each directory below a recursive - * target (elsewhere one recursive watcher per target). Returns whether a watcher was added. + * Watch every existing ancestor of each target, each target directory, each target that is a symbolic link to a file + * (changes to that file are not events of the link's directory), and on Linux each directory below a recursive target + * (elsewhere one recursive watcher per target). Returns whether a watcher was added. */ - #reconcileWatchers(snapshot: Snapshot): boolean { + #reconcileWatchers({ snapshot, linkedFiles }: Scan): boolean { const wanted = new Map(); + for (const path of linkedFiles) wanted.set(path, false); const perDirectory = process.platform === "linux" || process.platform === "android"; for (const target of this.#targets) { for (const ancestor of ancestorsOf(target.path)) { @@ -385,26 +428,32 @@ export class NodeFileWatcher implements FileWatcher { } } } - for (const [path, watcher] of this.#watchers) { - if (!wanted.has(path)) { - watcher.close(); + for (const [path, installed] of this.#watchers) { + const entry = snapshot.get(path); + // Gone, or replaced: a watcher follows the directory it was installed on, not the path. + if (!wanted.has(path) || entry === undefined || entry.dev !== installed.dev || entry.ino !== installed.ino) { + installed.watcher.close(); this.#watchers.delete(path); } } let added = false; for (const [path, recursive] of wanted) { - if (this.#watchers.has(path)) continue; + const entry = snapshot.get(path); + if (this.#watchers.has(path) || entry === undefined) continue; + const linked = linkedFiles.has(path); try { const watcher = fsWatch(path, { recursive, persistent: false }, (_event, filename) => - this.#onEvent(path, filename === null ? undefined : String(filename)), + linked + ? this.#onLinkedFileEvent(path) + : this.#onEvent(path, filename === null ? undefined : String(filename)), ); // A watched directory that disappears or fails: rescan, which also replaces the watcher. watcher.on("error", () => { watcher.close(); - if (this.#watchers.get(path) === watcher) this.#watchers.delete(path); + if (this.#watchers.get(path)?.watcher === watcher) this.#watchers.delete(path); this.#scheduleFlush(); }); - this.#watchers.set(path, watcher); + this.#watchers.set(path, { watcher, dev: entry.dev, ino: entry.ino }); added = true; } catch (error) { // Out of watches or unsupported: compare snapshots from now on, and say coverage was uncertain. @@ -420,7 +469,7 @@ export class NodeFileWatcher implements FileWatcher { #switchToPolling(): void { if (this.#mode === "polling") return; this.#mode = "polling"; - for (const watcher of this.#watchers.values()) watcher.close(); + for (const { watcher } of this.#watchers.values()) watcher.close(); this.#watchers.clear(); this.#deliver({ overflow: true }); clearTimeout(this.#timer); @@ -438,6 +487,13 @@ export class NodeFileWatcher implements FileWatcher { if (relevant || filename === undefined) this.#scheduleFlush(); } + /** The file a target links to changed: report the target. */ + #onLinkedFileEvent(target: string): void { + if (this.#closed) return; + this.#events.add(target); + this.#scheduleFlush(); + } + #inScope(path: string): boolean { for (const target of this.#targets) { if (isWithin(target.path, path)) return true; diff --git a/packages/durable/src/harness/agent.ts b/packages/durable/src/harness/agent.ts index a080eca7d..fdbe08859 100644 --- a/packages/durable/src/harness/agent.ts +++ b/packages/durable/src/harness/agent.ts @@ -57,7 +57,11 @@ export function resolveSettings(settings: HarnessSettings | undefined): Settings stream: { ...settings?.stream }, retry: { ...DEFAULT_RETRY_POLICY, ...settings?.retry }, compaction: { ...DEFAULT_COMPACTION_POLICY, ...settings?.compaction }, - progress: { ...DEFAULT_PROGRESS_POLICY, ...settings?.progress }, + // Field by field, so an explicitly undefined interval keeps its default. + progress: { + partialIntervalMs: settings?.progress?.partialIntervalMs ?? DEFAULT_PROGRESS_POLICY.partialIntervalMs, + outputIntervalMs: settings?.progress?.outputIntervalMs ?? DEFAULT_PROGRESS_POLICY.outputIntervalMs, + }, toolExecution: settings?.toolExecution ?? "parallel", steeringMode: settings?.steeringMode ?? "one-at-a-time", followUpMode: settings?.followUpMode ?? "one-at-a-time", diff --git a/packages/durable/src/harness/output.ts b/packages/durable/src/harness/output.ts index 429c5c931..f0e85d363 100644 --- a/packages/durable/src/harness/output.ts +++ b/packages/durable/src/harness/output.ts @@ -1,3 +1,4 @@ +import { rangeDecoder } from "../env/decode.ts"; import type { ShellOutputSkip } from "../env/index.ts"; import { utf8ByteLength } from "../truncate.ts"; @@ -18,7 +19,8 @@ export type OutputSlice = { const NEWLINE = 0x0a; const INVALID_OUTPUT = /[\x00-\x08\x0b-\x1f\ufff9-\ufffb]/g; const encoder = new TextEncoder(); -const decoder = new TextDecoder(); +/** Slices decode exactly: a U+FEFF at a slice's start is text, not a byte-order mark. */ +const decoder = rangeDecoder(); /** Remove control characters that break display and transcripts; tabs and newlines stay. */ export function sanitizeOutput(text: string): string { @@ -110,7 +112,13 @@ function lineCount(bytes: Uint8Array): number { */ export class OutputBuffer { readonly #limits: OutputLimits; - readonly #decoder = new TextDecoder(); + /** + * Decoder of byte chunks. A string chunk or a skip ends an incomplete character of earlier bytes (it becomes U+FFFD); + * the next byte chunk then starts a new character. As in `StreamDecoder`, only a byte-order mark at the very start of + * the output is dropped, never a U+FEFF later in it. + */ + readonly #decoder = rangeDecoder(); + #started = false; /** Stored chunks: for head the start of the stream, for tail a suffix that still contains the next window. */ #chunks: { readonly text: string; readonly bytes: number; readonly newlines: number }[] = []; #storedBytes = 0; @@ -137,7 +145,10 @@ export class OutputBuffer { push(chunk: string | Uint8Array, skipped?: ShellOutputSkip): boolean { // Bytes of an incomplete character from an earlier byte chunk come first. const pending = typeof chunk === "string" || skipped !== undefined ? this.#decoder.decode() : ""; - const text = typeof chunk === "string" ? chunk : this.#decoder.decode(chunk, { stream: true }); + let text = typeof chunk === "string" ? chunk : this.#decoder.decode(chunk, { stream: true }); + const first = !this.#started && pending === "" && skipped === undefined; + if (pending !== "" || text !== "" || skipped !== undefined) this.#started = true; + if (first && typeof chunk !== "string" && text.startsWith("\ufeff")) text = text.slice(1); if (skipped === undefined) return this.#accept(pending + text); if (this.#limits.retain !== "tail") throw new Error("Skipped output requires tail retention"); this.#accept(pending); diff --git a/packages/durable/src/testing/env-conformance.ts b/packages/durable/src/testing/env-conformance.ts index 0b98ce42c..442b9a444 100644 --- a/packages/durable/src/testing/env-conformance.ts +++ b/packages/durable/src/testing/env-conformance.ts @@ -341,6 +341,32 @@ export function createEnvConformance(options: EnvConformanceOptions): readonly E }); }), + watchCase("watch keeps recursive coverage where a non-recursive target overlaps", async (env) => { + getOrThrow(await env.writeFile("skills/a/one.md", "one", context)); + const targets: WatchTarget[] = [{ path: "skills" }, { path: "skills", recursive: true }]; + await watching(env, targets, async ({ expectChange }) => { + await expectChange("skills/a/two.md", async () => { + getOrThrow(await env.writeFile("skills/a/two.md", "two", context)); + }); + }); + }), + + watchCase("watch follows a directory replaced at the same path", async (env) => { + getOrThrow(await env.writeFile("skills/a/x.md", "x", context)); + await watching(env, [{ path: "skills", recursive: true }], async ({ expectChange }) => { + await expectChange("skills/a", async () => { + getOrThrow(await env.renameFile("skills/a", "skills-old", context)); + getOrThrow(await env.createDir("skills/a", undefined, context)); + }); + await expectChange("skills/a/y.md", async () => { + getOrThrow(await env.writeFile("skills/a/y.md", "y", context)); + }); + await expectChange("skills/a/y.md", async () => { + getOrThrow(await env.writeFile("skills/a/y.md", "yy", context)); + }); + }); + }), + watchCase("watch stops reporting once closed", async (env) => { const changes: WatchChange[] = []; const watcher = getOrThrow(await env.watch([{ path: "file.txt" }], (change) => changes.push(change), context)); @@ -466,6 +492,18 @@ export function createEnvConformance(options: EnvConformanceOptions): readonly E assert.strictEqual(decoder.decode(getOrThrow(await inner.read(0, 10, context))), "inner"); await inner.close(context); }), + + watchCase("watch reports changes to the file a watched symbolic link points to", async (env) => { + getOrThrow(await env.writeFile("data/real.md", "one", context)); + getOrThrow(await env.createDir("config", undefined, context)); + const linked = await env.exec([...shell, "ln -s ../data/real.md config/AGENTS.md"], undefined, context); + assert.strictEqual(getOrThrow(linked).exitCode, 0); + await watching(env, [{ path: "config/AGENTS.md" }], async ({ expectChange }) => { + await expectChange("config/AGENTS.md", async () => { + getOrThrow(await env.writeFile("data/real.md", "two!", context)); + }); + }); + }), ); } diff --git a/packages/durable/src/tools/read.ts b/packages/durable/src/tools/read.ts index f08b88fba..2667b2b58 100644 --- a/packages/durable/src/tools/read.ts +++ b/packages/durable/src/tools/read.ts @@ -81,12 +81,15 @@ export function createReadTool(): ToolRegistration before.size || (after.size === before.size && after.mtimeMs === before.mtimeMs)) { + return result; + } if (attempt === 1) throw new Error(`${path} changed while it was read`); } } finally { diff --git a/packages/durable/test/harness-output.test.ts b/packages/durable/test/harness-output.test.ts index f8a822bac..a64b456ac 100644 --- a/packages/durable/test/harness-output.test.ts +++ b/packages/durable/test/harness-output.test.ts @@ -87,6 +87,10 @@ describe("tool output bounds", () => { expect(bound("x\néééé", tail(10, 5))).toEqual({ kept: "éé", droppedBytes: 6, droppedLines: 1 }); }); + it("keeps a U+FEFF at the start of a kept slice", () => { + expect(bound("x\ufeffa", tail(10, 4))).toEqual({ kept: "\ufeffa", droppedBytes: 1, droppedLines: 0 }); + }); + it("cuts tails of surrogate edge cases exactly like Buffer", () => { const inputs = ["a\ud83d", "\ude42b", "a\ude42b", "\ud83d\ud83d\ude42", "\ud83d\ude42\ude42", "👩‍💻"]; for (const input of inputs) assertMatchesBufferTail(input); @@ -148,6 +152,18 @@ describe("OutputBuffer", () => { expect(buffer.snapshot()).toEqual({ text: "b\n", droppedBytes: 3, droppedLines: 1 }); }); + it("drops a byte-order mark only at the start of the output", () => { + const bom = Uint8Array.of(0xef, 0xbb, 0xbf); + const buffer = new OutputBuffer(tail(10)); + buffer.push(bom.subarray(0, 1)); + buffer.push(Uint8Array.of(0xbb, 0xbf, 0x61)); + buffer.push("b"); + // A U+FEFF after a string chunk is text. + buffer.push(Uint8Array.of(...bom, 0x63)); + buffer.end(); + expect(buffer.snapshot().text).toBe("ab\ufeffc"); + }); + it("flushes an incomplete character before a string chunk and at the end", () => { const buffer = new OutputBuffer(tail(10)); const euro = new TextEncoder().encode("€"); diff --git a/packages/durable/test/harness-registry.test.ts b/packages/durable/test/harness-registry.test.ts index a42db5ea7..3def50429 100644 --- a/packages/durable/test/harness-registry.test.ts +++ b/packages/durable/test/harness-registry.test.ts @@ -137,6 +137,15 @@ describe("registry", () => { }); }); +describe("settings", () => { + it("keeps the default of a progress interval given as undefined", () => { + expect(resolveSettings({ progress: { partialIntervalMs: undefined, outputIntervalMs: 250 } }).progress).toEqual({ + partialIntervalMs: 100, + outputIntervalMs: 250, + }); + }); +}); + describe("agent resolution", () => { const read = tool("read"); const bash = tool("bash"); diff --git a/packages/durable/test/tools.test.ts b/packages/durable/test/tools.test.ts index bdf72ea63..a75b56b99 100644 --- a/packages/durable/test/tools.test.ts +++ b/packages/durable/test/tools.test.ts @@ -1,6 +1,6 @@ // Ported from packages/agent/test/harness/tools.test.ts and adapted to ToolRegistration: tools take the environment // from `api.env`, stream through `api.output()`, and report notices as diagnostics instead of content text. -import { mkdirSync, rmSync } from "node:fs"; +import { appendFileSync, mkdirSync, rmSync } from "node:fs"; import { symlink } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; @@ -275,6 +275,27 @@ describe("durable tools", () => { ); }); + it("reads a log that grows while it is read", async () => { + const env = createEnv(); + getOrThrow(await env.writeFile("app.log", "one\ntwo\n", BACKGROUND_CONTEXT)); + const logPath = join(env.cwd, "app.log"); + // Every read appends, as a busy writer would; the scanned lines stay as they were. + const growing = Object.create(env) as NodeExecutionEnv; + growing.openBinaryReader = async (path, options, context) => { + const opened = await env.openBinaryReader(path, options, context); + if (!opened.ok) return opened; + const reader = opened.value; + const read = reader.read.bind(reader); + reader.read = async (position, length, readContext) => { + appendFileSync(logPath, "more\n"); + return read(position, length, readContext); + }; + return opened; + }; + const result = await run(createReadTool(), { path: "app.log", limit: 2 }, growing); + expect(textOutput(result)).toBe("one\ntwo"); + }); + it("reports images by content as unsupported", async () => { const env = createEnv(); const png = Uint8Array.from( diff --git a/packages/env/src/polling-watch.ts b/packages/env/src/polling-watch.ts deleted file mode 100644 index c15b8556d..000000000 --- a/packages/env/src/polling-watch.ts +++ /dev/null @@ -1,235 +0,0 @@ -import { createHash } from "node:crypto"; -import type { PlatformPath } from "node:path"; -import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context"; -import { FileError, type FileWatcher, type WatchChange, type WatchTarget } from "@earendil-works/pi-durable/env"; -import { type Json, RemoteError } from "./connection.ts"; -import { type RemoteExecutionEnv, remotePath } from "./remote-env.ts"; - -const DEFAULT_INTERVAL_MS = 2000; -const MAX_DIRECTORIES = 10_000; -/** Recently modified small files are also compared by content: a second write within the file system's timestamp - * granularity can keep size and modification time. */ -const HASH_MAX_BYTES = 256 * 1024; -const HASH_RECENT_MS = 5000; - -type RemoteInfo = { - kind: "file" | "directory" | "symlink" | "other"; - size: number; - mtimeSec: number; - mtimeNsec: number; - dev: number; - ino: number; -}; - -class BudgetExceeded extends Error {} - -function ancestorsOf(path: string, paths: PlatformPath): string[] { - const result: string[] = []; - for (let current = paths.dirname(path); ; current = paths.dirname(current)) { - result.push(current); - if (paths.dirname(current) === current) return result; - } -} - -function isWithin(path: string, ancestor: string, paths: PlatformPath): boolean { - return path === ancestor || path.startsWith(ancestor.endsWith(paths.sep) ? ancestor : ancestor + paths.sep); -} - -/** - * A polling `FileWatcher` over the daemon's `lstat` and `readdir`: it compares snapshots of the watched paths at an - * interval, with the same rules as `NodeExecutionEnv`'s watcher. A change undone between two snapshots can be missed; - * a lost connection is reported as `overflow`. - */ -export class PollingWatcher implements FileWatcher { - readonly mode = "polling" as const; - readonly #env: RemoteExecutionEnv; - readonly #targets: readonly WatchTarget[]; - readonly #onChange: (change: WatchChange) => void; - readonly #intervalMs: number; - readonly #paths: PlatformPath; - #snapshot = new Map(); - #timer: ReturnType | undefined; - #running: Promise | undefined; - #uncertain = false; - #closed = false; - - private constructor( - env: RemoteExecutionEnv, - targets: readonly WatchTarget[], - onChange: (change: WatchChange) => void, - intervalMs: number, - paths: PlatformPath, - ) { - this.#paths = paths; - this.#env = env; - this.#targets = targets; - this.#onChange = onChange; - this.#intervalMs = intervalMs; - } - - /** `targets` hold absolute paths. Coverage is established when this resolves. */ - static async open( - env: RemoteExecutionEnv, - targets: readonly WatchTarget[], - onChange: (change: WatchChange) => void, - intervalMs = DEFAULT_INTERVAL_MS, - ): Promise { - const watcher = new PollingWatcher(env, targets, onChange, intervalMs, await remotePath(env.connection)); - try { - watcher.#snapshot = await watcher.#scan(); - } catch (error) { - if (error instanceof BudgetExceeded) throw new FileError("invalid", error.message); - throw error; - } - watcher.#schedule(); - return watcher; - } - - async close(): Promise { - if (this.#closed) return; - this.#closed = true; - clearTimeout(this.#timer); - await this.#running; - } - - #deliver(change: WatchChange): void { - if (this.#closed) return; - try { - this.#onChange(change); - } catch { - // A throwing callback must not stop watching. - } - } - - #schedule(): void { - if (this.#closed) return; - this.#timer = setTimeout(() => { - this.#running = this.#poll().finally(() => { - this.#running = undefined; - this.#schedule(); - }); - }, this.#intervalMs); - } - - async #poll(): Promise { - let next: Map; - try { - next = await this.#scan(); - } catch (error) { - if (error instanceof BudgetExceeded) { - this.#deliver({ error: new FileError("invalid", error.message) }); - this.#closed = true; - return; - } - // Unreachable for now (a lost connection): changes meanwhile are unknown. - if (!this.#uncertain) this.#deliver({ overflow: true }); - this.#uncertain = true; - return; - } - if (this.#uncertain) { - this.#uncertain = false; - this.#snapshot = next; - this.#deliver({ overflow: true }); - return; - } - const changed = new Set(); - for (const [path, key] of next) if (this.#snapshot.get(path) !== key) changed.add(this.#reported(path)); - for (const path of this.#snapshot.keys()) if (!next.has(path)) changed.add(this.#reported(path)); - this.#snapshot = next; - if (changed.size > 0) this.#deliver({ paths: [...changed].sort() }); - } - - /** An ancestor that changed identity moved every target below it; report those targets. */ - #reported(path: string): string { - if (this.#targets.some((target) => isWithin(path, target.path, this.#paths))) return path; - return this.#targets.find((target) => isWithin(target.path, path, this.#paths))?.path ?? path; - } - - async #lstat(path: string): Promise { - try { - const { json } = await this.#env.connection.request("lstat", { path }); - return json as unknown as RemoteInfo; - } catch (error) { - if (error instanceof RemoteError && ["ENOENT", "ENOTDIR", "EACCES", "EPERM"].includes(error.code)) - return undefined; - throw error; - } - } - - async #key(path: string, info: RemoteInfo, identityOnly: boolean): Promise { - const identity = `${info.kind}:${info.dev}:${info.ino}`; - if (identityOnly || info.kind === "directory") return identity; - let hash = ""; - const mtimeMs = info.mtimeSec * 1000 + info.mtimeNsec / 1e6; - if (info.kind === "file" && info.size <= HASH_MAX_BYTES && Date.now() - mtimeMs < HASH_RECENT_MS) { - const bytes = await this.#env.readBinaryFile(path, BACKGROUND_CONTEXT); - if (bytes.ok) hash = createHash("sha256").update(bytes.value).digest("hex"); - } - return `${identity}:${info.size}:${info.mtimeSec}.${info.mtimeNsec}:${hash}`; - } - - async #scan(): Promise> { - const snapshot = new Map(); - let directories = 0; - const countDirectory = () => { - if (++directories > MAX_DIRECTORIES) - throw new BudgetExceeded(`Watched paths exceed ${MAX_DIRECTORIES} directories`); - }; - const scanDirectory = async (target: WatchTarget, directory: string): Promise => { - let handle: number; - try { - handle = (await this.#env.connection.request("opendir", { path: directory })).json.handle as number; - } catch (error) { - if (error instanceof RemoteError && ["ENOENT", "ENOTDIR", "EACCES", "EPERM"].includes(error.code)) return; - throw error; - } - const entries: { name: string; info?: RemoteInfo; error?: Json }[] = []; - try { - for (let done = false; !done; ) { - const { json } = await this.#env.connection.request("readdir", { handle, max: 1000 }); - entries.push(...(json.entries as typeof entries)); - done = json.done === true; - } - } finally { - await this.#env.connection.request("close", { handle }).catch(() => undefined); - } - for (const entry of entries) { - if (entry.info === undefined) continue; - if ((target.exclude?.hidden && entry.name.startsWith(".")) || target.exclude?.names?.includes(entry.name)) - continue; - const path = this.#paths.join(directory, entry.name); - if (snapshot.has(path)) continue; - snapshot.set(path, await this.#key(path, entry.info, false)); - if (target.recursive && entry.info.kind === "directory") { - countDirectory(); - await scanDirectory(target, path); - } - } - }; - for (const target of this.#targets) { - for (const ancestor of ancestorsOf(target.path, this.#paths)) { - if (snapshot.has(ancestor)) continue; - const info = await this.#lstat(ancestor); - if (info !== undefined) snapshot.set(ancestor, await this.#key(ancestor, info, true)); - } - // The target itself may be a symbolic link to what is watched; follow it. - let resolved = target.path; - let info = await this.#lstat(target.path); - if (info?.kind === "symlink") { - try { - resolved = (await this.#env.connection.request("realpath", { path: target.path })).json.path as string; - info = await this.#lstat(resolved); - } catch { - info = undefined; - } - } - if (info === undefined) continue; - snapshot.set(target.path, await this.#key(resolved, info, false)); - if (info.kind === "directory") { - countDirectory(); - await scanDirectory(target, target.path); - } - } - return snapshot; - } -}