fix(env): ENOTDIR for listing a file on Windows, pinned NDK install, protocol docs

.github/workflows/env-daemons.yml

@@ -52,7 +52,8 @@ jobs:
         run: |
           ndk="$ANDROID_HOME/ndk/$NDK_VERSION"
           if [ ! -d "$ndk" ]; then
-            yes | "$ANDROID_HOME/cmdline-tools/latest/bin/sdkmanager" --install "ndk;$NDK_VERSION" > /dev/null
+            # `yes` ends with SIGPIPE once the licenses are accepted.
+            { yes || true; } | "$ANDROID_HOME/cmdline-tools/latest/bin/sdkmanager" --install "ndk;$NDK_VERSION" > /dev/null
           fi
           linker="$ndk/toolchains/llvm/prebuilt/linux-x86_64/bin/${{ matrix.android }}-clang"
           test -x "$linker"

packages/env/README.md

@@ -17,23 +17,23 @@ const target = { host: "gpu-box", knownHostsFile: "/data/ssh/known_hosts", hostK
 const { lines, fingerprints } = await scanHostKey(target);
 await acceptHostKey(target, lines);
 
-// Detects the remote system, deploys this version's daemon if missing (verified by SHA-256 before it runs), and
-// returns a connection that starts it over ssh.
+// Detects the remote system, deploys the daemon if missing (named by its SHA-256 and verified before every start),
+// and returns a connection that starts it over ssh.
 const { connection } = await connectSsh(target);
 const env = new RemoteExecutionEnv({ connection, id: "pi-env:gpu", cwd: "/home/me/project" });
 ```
 
 The package ships the daemon for every supported remote system in `bin/`. `ssh` runs with `BatchMode`, strict host-key
-checking against the application's own known-hosts file under a fixed alias, no forwarding, and without forwarding the
-local locale. On Windows, detection and deployment go through PowerShell.
+checking against the application's own known-hosts file under a fixed alias, no forwarding of any kind, no shared
+connections or configured commands, and without forwarding the local locale. On Windows, detection and deployment go
+through PowerShell, and the daemon starts through the server's default shell (cmd.exe or PowerShell).
 
 Supported remote systems: Linux, macOS, Android (Termux) and Windows, on x86-64 and arm64. On Windows, string commands
 run through Git Bash as `NodeExecutionEnv` runs them there; argv commands run directly.
 
 ## Development
 
-`npm run build:daemon` builds the daemon with Cargo; the tests talk to `daemon/target/debug/pi-env` over a pipe. They
-run Durable's env conformance suite against `RemoteExecutionEnv` and compare random operation sequences and Durable's
-tools against `NodeExecutionEnv` on the same machine.
-
-Not yet: output windowing (all output is transferred) and daemon-side file watching (the client polls).
+`npm run build:daemon` builds the daemon with Cargo; the tests talk to `daemon/target/debug/pi-env` over a pipe, or to
+the binary named by `PI_ENV_DAEMON` (CI tests the release builds this way). They run Durable's env conformance suite
+against `RemoteExecutionEnv` and compare random operation sequences and Durable's tools against `NodeExecutionEnv` on
+the same machine. `test/ssh-external.test.ts` runs the suite over a real SSH server when `PI_ENV_SSH_HOST` is set.

packages/env/daemon/src/main.rs

@@ -228,8 +228,8 @@ impl Server {
             }
             "opendir" => {
                 let path = field(json, "path")?;
-                let entries = std::fs::read_dir(path)
-                    .map_err(|error| Failure::io(&error, "opendir", path))?;
+                let entries =
+                    sys::read_dir(path).map_err(|error| Failure::io(&error, "opendir", path))?;
                 let handle = self.insert(Handle::Dir(Arc::new(Mutex::new(DirHandle {
                     entries,
                     path: path.to_string(),

packages/env/daemon/src/sys/unix.rs

@@ -312,3 +312,7 @@ pub fn raw_name(name: &OsStr) -> Option<Vec<u8>> {
 pub fn drive_cwds() -> Map<String, Value> {
     Map::new()
 }
+
+pub fn read_dir(path: &str) -> io::Result<std::fs::ReadDir> {
+    std::fs::read_dir(path)
+}

packages/env/daemon/src/sys/windows.rs

@@ -591,6 +591,17 @@ pub fn drive_cwds() -> Map<String, Value> {
         .collect()
 }
 
+/// libuv's `scandir` and `opendir` report a file as `ENOTDIR` (elsewhere `ERROR_DIRECTORY` is `ENOENT`).
+pub fn read_dir(path: &str) -> io::Result<std::fs::ReadDir> {
+    std::fs::read_dir(path).map_err(|error| {
+        if error.raw_os_error() == Some(267) {
+            synthetic("ENOTDIR")
+        } else {
+            error
+        }
+    })
+}
+
 #[cfg(test)]
 mod tests {
     use super::quote_argument;

packages/env/docs/protocol.md

@@ -22,33 +22,54 @@ json            UTF-8 JSON object, jsonLength bytes
 bytes           raw payload, the remaining length - 9 - jsonLength bytes
 ```
 
-All integers are big-endian. A frame is at most 16 MiB; larger payloads are split by the operations that carry them.
+All integers are big-endian. A frame is at most 16 MiB; larger transfers are split by the client. A request whose JSON
+does not parse gets an `EINVAL` error; framing errors end the connection.
 
 | type | name | direction | meaning |
 |---|---|---|---|
 | 1 | request | client → daemon | `json.op` names the operation |
 | 2 | result | daemon → client | success of request `id`; the request is finished |
 | 3 | error | daemon → client | failure of request `id`; the request is finished |
-| 4 | event | daemon → client | progress of a running request (`exec` output), not finishing it |
-| 5 | cancel | client → daemon | abort request `id`; it still finishes with `result` or `error`. `{ mode: "kill" }` kills an `exec` without aborting it, so it settles with the killed process's status, as `cleanup()` does |
+| 4 | event | daemon → client | progress of a running request (`exec` output, `watch` changes), not finishing it |
+| 5 | cancel | client → daemon | cancel request `id`; see below |
 | 6 | ping | both | liveness; the receiver ignores it |
 
 Errors are `{ code, message, syscall?, path? }`. `code` is a Node-style error code (`ENOENT`, `EISDIR`,
 `ERR_FS_EISDIR`, ...) or one of the execution codes `shell_unavailable`, `spawn_error`, `timeout`, `aborted`, `unknown`.
 Messages are diagnostic; only codes are part of the contract.
 
-Paths are absolute JSON strings; the client resolves relative paths, `~` and `file://` URLs before sending, and they
-are encoded as UTF-8, as Node encodes string paths. File contents and output travel as raw payload bytes.
+Paths are absolute JSON strings; the client resolves relative paths, `~` and `file://` URLs before sending. Strings are
+sent well-formed (lone surrogates become U+FFFD) and encoded as UTF-8, as Node encodes string paths and arguments. File
+contents and output travel as raw payload bytes.
+
+## Cancellation
+
+A cancel names one request. The daemon registers each request before running it, so a cancel sent right after its
+request is never lost. `exec` kills the command's process group and finishes with `aborted`; `{ mode: "kill" }` kills
+without aborting, so the command finishes with the killed process's status, as `cleanup()` does. `scanLines` stops
+between chunks with `aborted`. `watch` stops and finishes with `{}`. Other operations are short and finish normally.
+
+## Scheduling
+
+File operations run on a fixed pool of worker threads; `exec` and `watch` run on their own threads. Chunk writes to one
+handle run in arrival order. Results, errors, pings and `watch` events are sent before queued command output, and
+results with payloads above 64 KiB queue behind output too, so a slow link delays bulk data, not replies. Output of a
+command without an output window waits while more than 4 MiB of output is unsent, which slows the command as a full
+pipe would.
 
 ## Liveness
 
 Each side sends `ping` every 5 seconds. The daemon kills every process group it started and exits after 30 seconds
-without any frame from the client, or when stdin ends. A dropped connection therefore stops remote commands.
+without receiving any bytes, or when stdin ends. A dropped connection therefore stops remote commands. The client
+fails requests of a lost connection with `{ code: "unknown", lost: true }` and starts a new daemon on the next request;
+handles belong to the daemon that opened them, so requests on them fail instead of reaching the new one.
 
 ## Operations
 
-`hello { protocol }` → `{ protocol, version, os, arch, home, tmpdir, pid }`. The first request. `os` and `arch` are
-Rust's `std::env::consts` values; `tmpdir` follows the remote Node's `os.tmpdir()` rules.
+`hello { protocol }` → `{ protocol, version, os, arch, home, tmpdir, cwd, driveCwds, pid }`. The first request.
+`version` is the npm package version the daemon shipped with; `os` and `arch` are Rust's `std::env::consts` values;
+`tmpdir` follows the remote Node's `os.tmpdir()` rules; `cwd` and `driveCwds` (Windows' `=C:` variables) are what
+Node's `path.resolve` falls back to for drive-relative paths.
 
 File operations take `{ path }` (and the listed fields) and return `{}` unless noted:
 
@@ -56,15 +77,16 @@ File operations take `{ path }` (and the listed fields) and return `{}` unless n
 |---|---|---|
 | `lstat` | | `info` |
 | `realpath` | | `{ path }` |
-| `write` | `append`, payload | creates missing parent directories like Node's recursive `mkdir` |
+| `write` | `append`, `parents?` (default true), `keep?`, payload | creates missing parents like Node's recursive `mkdir`, opens like `writeFile` (`w`) or `appendFile` (`a`), writes the payload; with `keep`, `{ handle }` for `writeChunk` |
+| `writeChunk` | `handle`, payload | appends to a kept write handle; after a failed chunk, later chunks fail with `EBADF` |
 | `truncate` | `size` | |
 | `fsync` | | |
 | `rename` | `to` | |
 | `mkdir` | `recursive` | |
 | `rm` | `recursive`, `force` | Node `fs.rm` semantics |
-| `mkdtemp` | `prefix` (in `tmpdir`) | `{ path }` |
-| `open` | `noFollow` | `{ handle, info }`; regular files only |
-| `pread` | `handle`, `offset`, `length` | payload |
+| `mkdtemp` | (`path` is the prefix) | `{ path }` |
+| `open` | `noFollow`, `mode?` | `{ handle, info }` for `openBinaryReader`: nonblocking, regular files only. With `mode: "read"`, Node's `open(path, "r")` of any file, blocking: `{ handle, stat }` or `{ handle, statError }` |
+| `pread` | `handle`, `offset?`, `length` | payload; without `offset`, one read at the current position |
 | `fstat` | `handle` | `info` |
 | `scanLines` | `handle`, `startLine`, `endLine?` | `LineScan` |
 | `opendir` | | `{ handle }` |
@@ -72,11 +94,21 @@ File operations take `{ path }` (and the listed fields) and return `{}` unless n
 | `close` | `handle` | |
 
 `info` is `{ name, kind, size, mtimeSec, mtimeNsec, dev, ino }` with `kind` one of `file`, `directory`, `symlink`, `other`.
-`readdir` entries are `{ name, info }` or `{ name, error }` when the entry's `lstat` failed; names that are not valid UTF-8
-are decoded lossily, as Node does. Handles are numbers, valid until `close` or the end of the connection.
+`readdir` entries are `{ name, raw?, info }` or `{ name, raw?, error }` when the entry's `lstat` failed. Names are
+decoded as UTF-8 with replacement characters, and lstat'ed under that name, as Node does; `raw` holds the original
+bytes of a name that is not valid UTF-8, for sorting. Handles are numbers, valid until `close` or the end of the
+connection; opening more than 4096 fails with `EMFILE`.
 
 `exec { command | argv, cwd, env, inheritEnv, shellPath?, timeoutMs?, spill?, window? }` runs a shell string or an
 argv array. While it runs, `event` frames `{ kind: "output", stream, skipped? }` carry decoded output text as UTF-8
-payload. The result is `{ exitCode?, signal?, spillPath? }`; failures are errors with codes `timeout` or `aborted`
-(carrying `spillPath`), `shell_unavailable`, `spawn_error`, or `unknown` (spill failure). `cancel` kills the command's
-process group.
+payload. With `window { maxBytes, maxLines, minIntervalMs, bytesPerSecond }`, output is coalesced and paced like the
+caller's commits, at most one output frame per command is unsent, and output followed by more than the window is
+replaced by `skipped { bytes, newlines, endsWithNewline }` on the next event, which carries all output after it
+(`ShellOutputInfo.skipped`). The result is `{ exitCode, spillPath? }`; failures are errors with codes `timeout` or
+`aborted` (carrying `spillPath`), `shell_unavailable`, `spawn_error`, or `unknown` (spill failure).
+
+`watch { targets, mode?, pollIntervalMs?, maxDirectories? }` runs Durable's `NodeFileWatcher` in the daemon. `targets`
+are `{ path, recursive, exclude: { hidden, names } }`. Once coverage is established, an event `{ kind: "ready", mode }`
+follows; then `{ kind: "change", paths }` or `{ kind: "change", overflow: true, mode: "polling" }` (native watching
+failed), or `{ kind: "error", code, message }`, after which nothing follows. The request lasts until cancelled.
+Failing to establish coverage is an error of the request.

packages/env/docs/semantics.md

@@ -1,17 +1,19 @@
 # Semantics
 
 The reference is Durable's `NodeExecutionEnv` (`packages/durable/src/env/node.ts`) on the remote machine. Durable's
-env conformance suite and `client/test/differential.test.ts` check each rule below.
+env conformance suite and `test/differential.test.ts` check each rule below.
 
 ## Split of work
 
 | Concern | Where |
 |---|---|
-| Path resolution: `~` (remote home), `file://`, relative to `cwd` | client, POSIX rules |
+| Path resolution: `~` (remote home), `file://`, relative to `cwd`; on Windows drive-relative paths against the remote's working directories | client, the remote system's `path` rules |
 | Abort checkpoints of each method | client, as `NodeExecutionEnv` |
 | Error code mapping (`ENOENT` → `not_found`, ...) | client, as `NodeExecutionEnv`'s `toFileError` |
-| `readTextFile` decoding (byte-order mark kept, like `readFile(path, "utf8")`) | client |
+| `readFile`'s read sizes, size limit and decoding (`StringDecoder`, byte-order mark kept) over the daemon's `open(path, "r")` | client |
+| `writeFile` in 512 KiB writes with an abort check before each; `appendFile` without checks | client, over one open file |
 | Line reader | client, positional reads and Durable's `StreamDecoder` |
+| Watching: snapshots, native events, polling | daemon, a port of Durable's `NodeFileWatcher` |
 | Timeout validation, result precedence (callback error, timeout, abort, spill failure, exit code) | client |
 | System calls, processes, spill files, line scans | daemon |
 | Command output decoding: per stream, WHATWG UTF-8, only a leading byte-order mark dropped | daemon (`encoding_rs`) |
@@ -25,22 +27,39 @@ env conformance suite and `client/test/differential.test.ts` check each rule bel
 - `mkdtemp`: `mkdtemp(3)` of `<tmpdir>/<prefix>XXXXXX`; the prefix is joined with POSIX `path.join`, so `../x` works.
 - `tmpdir`: `TMPDIR`, `TMP`, `TEMP`, then `/tmp` (Termux: `$PREFIX/tmp`), trailing slash removed.
 - `listDir`: names in byte order, like libuv's `scandir`; fails if any entry cannot be lstat'ed. `openDirReader` keeps
-  directory order and skips entries removed meanwhile.
+  directory order, skips entries removed meanwhile, and fills each page after skipping. Names that are not UTF-8 are
+  decoded with replacement characters and lstat'ed under that name, so they fail with `ENOENT` as in Node.
 - `openBinaryReader`: nonblocking open; directories fail with `is_directory`, other non-regular files with `invalid`;
   `noFollow` refuses a final symbolic link (`O_NOFOLLOW`).
+- `readBinaryFile`, `readTextFile` and `openTextLineReader` open any file like `open(path, "r")`: `/dev/null` reads
+  empty, a FIFO blocks until it has a writer, and a directory fails when read (a line reader at `readLine`). A regular
+  file is read up to its size at open; other files to their end.
 - `readBinaryFile` returns a `Buffer`, reader reads a plain `Uint8Array`, as Node does.
+- Windows: error codes are libuv's translations; `rm` of a symbolic link or junction to a directory removes the link;
+  creating a file where a directory exists fails with `EISDIR`; modification times before 1970 are negative.
 
 ## Commands
 
-- A string runs through the shell: a configured `shellPath` that does not exist fails with `shell_unavailable`;
-  otherwise `/bin/bash`, `which bash`, then `sh`, with `-c`. An argv array runs its program directly.
+- A string runs through the shell: a configured `shellPath` that does not exist (following links) fails with
+  `shell_unavailable`; otherwise `/bin/bash`, `which bash`, then `sh`, with `-c`. An argv array runs its program
+  directly; on Windows it is searched like libuv does (working directory, then `PATH`, adding `.com` and `.exe`), and
+  batch files are refused.
 - The working directory must exist (`spawn_error`).
 - Environment: with `inheritEnv` (default), the daemon's environment, then `shellEnv`, then `env`; without it, only
   `env`.
 - Each command gets a new session (process group), default signal dispositions and an empty signal mask. Timeout,
   abort and `cleanup()` kill the group with `SIGKILL`.
-- After the process exits, output is still collected until both pipes end, or 100 ms pass without output.
+- After the process exits, output is still collected until both pipes end, or 100 ms pass without output; the timeout
+  still applies meanwhile. The pipes are closed when the command settles, even if a descendant keeps them open.
 - An abort settles as `aborted`; `cleanup()` kills without aborting, so the command settles with exit code 137.
 - A process killed by a signal reports `128 + signal`.
 - Spill: once the output crosses `spill.afterBytes` or `spill.afterLines` (counted over raw chunks in arrival order, as
   Node counts), the complete raw output goes to `pi-output-<uuid>.log` in a fresh `tmp-` directory.
+
+## Watching
+
+- The daemon runs Durable's `NodeFileWatcher` next to the files: native events (inotify, FSEvents) only trigger a
+  debounced rescan, and changes are the difference between snapshots plus the event paths. Reads are not events.
+- Windows and network or FUSE file systems poll, as `NodeExecutionEnv` does; running out of native watches switches to
+  polling and reports `overflow`.
+- When the connection is lost, the client opens the watcher again once a new daemon starts and reports `overflow`.

packages/env/test/ssh.test.ts

@@ -170,9 +170,10 @@ describe.skipIf(sshd === undefined)("SSH bootstrap", () => {
 		const file = join(home, ".pi/mobile/tools", deployedName);
 		try {
 			const { pid } = await connection.info();
-			writeFileSync(file, "tampered");
 			process.kill(pid, "SIGKILL");
 			await new Promise((done) => setTimeout(done, 300));
+			// Linux refuses to write a running program's file.
+			writeFileSync(file, "tampered");
 			// The next start redeploys the verified binary instead of running whatever is there.
 			expect((await connection.info()).pid).not.toBe(pid);
 			expect(readFileSync(file).equals(readFileSync(daemon))).toBe(true);