sscoble/pi-mono
1
0
Fork 0
mirror of https://github.com/badlogic/pi-mono.git synced 2026-11-05 23:37:25 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

fix(durable): watcher coverage, growing reads, output BOM and progress defaults

Browse source
This commit is contained in:
Mario Zechner 2026-10-05 15:47:25 +02:00
parent b223082bb2
commit cd60a5b990
10 changed files with 207 additions and 277 deletions
Download patch file Download diff file Expand all files Collapse all files

7
packages/durable/CHANGELOG.md
View file

@ -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

126
packages/durable/src/env/node-watch.ts vendored
View file

@ -37,6 +37,12 @@ type Entry = {
type Snapshot = Map<string, Entry>;
/** 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<string> };
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<string, FSWatcher>();
readonly #watchers = new Map<string, Installed>();
readonly #events = new Set<string>();
#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<Set<string>> {
const changed = new Set<string>();
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<Snapshot> {
async #scan(): Promise<Scan> {
const snapshot: Snapshot = new Map();
let directories = 0;
const countDirectory = (): void => {
if (++directories > this.#maxDirectories) {
const linkedFiles = new Set<string>();
// 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<string>();
const traversed = new Set<string>();
/** 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<string, Entry["kind"]>();
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<void> => {
const scanDirectory = async (index: number, target: ResolvedTarget, directory: string): Promise<void> => {
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<string, boolean>();
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;

6
packages/durable/src/harness/agent.ts
View file

@ -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",

17
packages/durable/src/harness/output.ts
View file

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

38
packages/durable/src/testing/env-conformance.ts
View file

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

7
packages/durable/src/tools/read.ts
View file

@ -81,12 +81,15 @@ export function createReadTool(): ToolRegistration<typeof readSchema, ReadToolDe
const absolutePath = await resolveReadToolPath(env, path, context);
const reader = getOrThrow(await env.openBinaryReader(absolutePath, undefined, context));
try {
// A concurrent writer can change the file between the scan and the reads; retry once from a fresh scan.
// A concurrent writer can change the file between the scan and the reads. Appending (a growing log) leaves
// the scanned bytes as they were; a file that shrank or was rewritten in place is read again once.
for (let attempt = 0; ; attempt++) {
const before = getOrThrow(await reader.info(context));
const result = await readText(reader, before, path, offset, limit, context);
const after = getOrThrow(await reader.info(context));
if (after.size === before.size && after.mtimeMs === before.mtimeMs) return result;
if (after.size > 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 {

16
packages/durable/test/harness-output.test.ts
View file

@ -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("€");

9
packages/durable/test/harness-registry.test.ts
View file

@ -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");

23
packages/durable/test/tools.test.ts
View file

@ -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(

235
packages/env/src/polling-watch.ts vendored
View file

@ -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<string, string>();
#timer: ReturnType<typeof setTimeout> | undefined;
#running: Promise<void> | 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<PollingWatcher> {
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<void> {
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<void> {
let next: Map<string, string>;
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<string>();
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<RemoteInfo | undefined> {
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<string> {
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<Map<string, string>> {
const snapshot = new Map<string, string>();
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<void> => {
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;
}
}