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

feat(env): add @earendil-works/pi-env, remote execution environments over SSH

Browse source 
- RemoteExecutionEnv: a Durable ExecutionEnv on another machine whose
  results match NodeExecutionEnv there; Connection to the pi-env daemon
- packages/env/daemon: the Rust daemon for Linux, macOS, Android and
  Windows, following Node and libuv semantics
- env.yml runs Durable's env tests, the daemon's checks and pi-env's
  conformance and differential tests on Linux, macOS and Windows
This commit is contained in:
Mario Zechner 2026-10-05 12:39:52 +02:00
parent b9ab918c62
commit ba03e03f2d
33 changed files with 4352 additions and 12 deletions
Download patch file Download diff file Expand all files Collapse all files

3
.github/workflows/ci.yml vendored
View file

@ -38,6 +38,9 @@ jobs:
- name: Check
run: npm run check
- name: Build the pi-env daemon
run: npm run build:daemon --workspace packages/env
- name: Test
run: npm test

39
.github/workflows/durable-env.yml → .github/workflows/env.yml vendored
View file

@ -1,13 +1,18 @@
name: Durable env
name: Environments
# Work-branch checks for Durable ExecutionEnv changes: the full repo check, plus the env tests on every OS.
# Execution environments on every OS: Durable's NodeExecutionEnv and pi-env's daemon and RemoteExecutionEnv, which
# must give the same results. The main CI runs on Linux only.
on:
push:
branches: ["durable-env-*"]
branches: [main, "durable-env-*"]
paths: ["packages/durable/src/env/**", "packages/durable/src/tools/**", "packages/durable/test/env-*", "packages/env/**", ".github/workflows/env.yml"]
pull_request:
branches: [main]
paths: ["packages/durable/src/env/**", "packages/durable/src/tools/**", "packages/durable/test/env-*", "packages/env/**", ".github/workflows/env.yml"]
workflow_dispatch:
concurrency:
group: durable-env-${{ github.ref }}
group: env-${{ github.ref }}
cancel-in-progress: true
jobs:
@ -38,16 +43,15 @@ jobs:
- name: Check
run: npm run check
- name: Durable tests
working-directory: packages/durable
run: npx vitest --run
env-tests:
tests:
strategy:
fail-fast: false
matrix:
os: [ubuntu-latest, ubuntu-24.04-arm, macos-latest, windows-latest, windows-11-arm]
runs-on: ${{ matrix.os }}
defaults:
run:
shell: bash
steps:
- name: Checkout
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
@ -61,6 +65,21 @@ jobs:
- name: Install dependencies
run: npm ci --ignore-scripts
- name: Env tests
- name: Daemon format, lint and unit tests
working-directory: packages/env/daemon
run: |
cargo fmt --check
cargo clippy --locked --all-targets -- -D warnings
cargo test --locked
- name: Build the daemon
working-directory: packages/env
run: npm run build:daemon
- name: Durable env tests
working-directory: packages/durable
run: npx vitest --run test/env-node.test.ts test/env-node-spill.test.ts test/env-node-conformance.test.ts test/env-line-scan.test.ts test/tools.test.ts test/tools-read-differential.test.ts
- name: pi-env tests
working-directory: packages/env
run: npx vitest --run

2
.gitignore vendored
View file

@ -37,6 +37,8 @@ out.jsonl
pi-*.html
out.html
packages/coding-agent/binaries/
packages/env/bin/
packages/env/daemon/target/
todo.md
plans/
.pi/hf-sessions/

20
package-lock.json generated
View file

@ -682,6 +682,10 @@
"resolved": "packages/durable",
"link": true
},
"node_modules/@earendil-works/pi-env": {
"resolved": "packages/env",
"link": true
},
"node_modules/@earendil-works/pi-evals": {
"resolved": "packages/evals",
"link": true
@ -5586,6 +5590,22 @@
"node": ">=22.19.0"
}
},
"packages/env": {
"name": "@earendil-works/pi-env",
"version": "1.0.3",
"license": "MIT",
"dependencies": {
"@earendil-works/chord": "^1.0.3",
"@earendil-works/pi-durable": "^1.0.3"
},
"devDependencies": {
"shx": "0.4.0",
"vitest": "4.1.11"
},
"engines": {
"node": ">=22.19.0"
}
},
"packages/evals": {
"name": "@earendil-works/pi-evals",
"version": "1.0.3",

4
package.json
View file

@ -12,8 +12,8 @@
],
"scripts": {
"clean": "npm run clean --workspaces",
"build": "cd packages/chord && npm run build && cd ../tui && npm run build && cd ../telemetry && npm run build && cd ../codemode && npm run build && cd ../mcp && npm run build && cd ../ai && npm run build && cd ../durable && npm run build && cd ../agent && npm run build && cd ../protocol && npm run build && cd ../client && npm run build && cd ../server && npm run build && cd ../coding-agent && npm run build",
"build:offline": "cd packages/chord && npm run build && cd ../tui && npm run build && cd ../telemetry && npm run build && cd ../codemode && npm run build && cd ../mcp && npm run build && cd ../ai && npm run build:offline && cd ../durable && npm run build && cd ../agent && npm run build && cd ../protocol && npm run build && cd ../client && npm run build && cd ../server && npm run build && cd ../coding-agent && npm run build",
"build": "cd packages/chord && npm run build && cd ../tui && npm run build && cd ../telemetry && npm run build && cd ../codemode && npm run build && cd ../mcp && npm run build && cd ../ai && npm run build && cd ../durable && npm run build && cd ../env && npm run build && cd ../agent && npm run build && cd ../protocol && npm run build && cd ../client && npm run build && cd ../server && npm run build && cd ../coding-agent && npm run build",
"build:offline": "cd packages/chord && npm run build && cd ../tui && npm run build && cd ../telemetry && npm run build && cd ../codemode && npm run build && cd ../mcp && npm run build && cd ../ai && npm run build:offline && cd ../durable && npm run build && cd ../env && npm run build && cd ../agent && npm run build && cd ../protocol && npm run build && cd ../client && npm run build && cd ../server && npm run build && cd ../coding-agent && npm run build",
"build:native:darwin": "npm --prefix packages/tui run build:native:darwin",
"build:native:linux": "npm --prefix packages/tui run build:native:linux",
"build:native:win32": "npm --prefix packages/tui run build:native:win32",

7
packages/env/CHANGELOG.md vendored Normal file
View file

@ -0,0 +1,7 @@
# Changelog
## [Unreleased]
### Added
- Initial release of `@earendil-works/pi-env`: `RemoteExecutionEnv`, a Durable `ExecutionEnv` on another machine, and `Connection` to the `pi-env` daemon that runs there. Results match `NodeExecutionEnv` on that machine on Linux, macOS, Android and Windows.

29
packages/env/README.md vendored Normal file
View file

@ -0,0 +1,29 @@
# @earendil-works/pi-env
Remote execution environments for [Pi Durable](../durable): an agent's tools run on another machine, usually over SSH,
while the Durable worker, its storage and credentials stay local.
- `pi-env` (`daemon/`): a small Rust program that runs on the remote machine. It speaks a framed protocol on stdin and
stdout ([docs/protocol.md](docs/protocol.md)) and performs file operations and commands there.
- `RemoteExecutionEnv`: a Durable `ExecutionEnv` that talks to the daemon through a `Connection`. Its results match
`NodeExecutionEnv` running on the remote machine; only error messages may differ
([docs/semantics.md](docs/semantics.md)).
```ts
import { Connection, RemoteExecutionEnv } from "@earendil-works/pi-env";
const connection = new Connection({ command: ["ssh", "-T", "--", "gpu-box", "~/.pi/mobile/tools/pi-env"] });
const env = new RemoteExecutionEnv({ connection, id: "pi-env:gpu-box", cwd: "/home/me/project" });
```
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: SSH bootstrap and binary deployment, output windowing (all output is transferred), daemon-side file watching
(the client polls).

179
packages/env/daemon/Cargo.lock generated vendored Normal file
View file

@ -0,0 +1,179 @@
# This file is automatically @generated by Cargo.
# It is not intended for manual editing.
version = 4
[[package]]
name = "cfg-if"
version = "1.0.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4e7648175b45a9a48536d676f68d918270699102aa8dab5496df06904c914600"
[[package]]
name = "core_detect"
version = "1.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7f8f80099a98041a3d1622845c271458a2d73e688351bf3cb999266764b81d48"
[[package]]
name = "encoding_rs"
version = "0.8.42"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8e985e0451871ad22fb8d2b6b076e2028a502a0d3950998c2c5c0a4f9b5d9679"
dependencies = [
"cfg-if",
"core_detect",
"multiversion_no_op",
"rustversion",
"scopeguard",
"simdutf8",
]
[[package]]
name = "itoa"
version = "1.0.18"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682"
[[package]]
name = "libc"
version = "0.2.190"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ce5d3ddc6d3fa000eb1536d85e147bfe31aacaba692ed6a876f95cb7c855be78"
[[package]]
name = "memchr"
version = "2.8.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98"
[[package]]
name = "multiversion_no_op"
version = "1.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "743fb55ba31b18fb1ecef6bdc9aa2743314978ac084044301a7eee33fb99a20d"
[[package]]
name = "pi-env"
version = "0.1.0"
dependencies = [
"encoding_rs",
"libc",
"serde_json",
"windows-sys",
]
[[package]]
name = "proc-macro2"
version = "1.0.107"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "985e7ec9bb745e6ce6535b544d84d6cd6f7ad8bd711c398938ae983b91a766d9"
dependencies = [
"unicode-ident",
]
[[package]]
name = "quote"
version = "1.0.47"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1fbf4db142a473a8d80c26bbf18454ed458bf8d26c8219c331daecfdbd079001"
dependencies = [
"proc-macro2",
]
[[package]]
name = "rustversion"
version = "1.0.23"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "cf54715a573b99ac80df0bc206da022bcd442c974952c7b9720069370852e21f"
[[package]]
name = "scopeguard"
version = "1.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49"
[[package]]
name = "serde"
version = "1.0.229"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4148590afebada386688f18773da617792bf2ef03ffc1e4cbd2b1d45b023e0ba"
dependencies = [
"serde_core",
]
[[package]]
name = "serde_core"
version = "1.0.229"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "67dca2c9c51e58a4791a4b1ed58308b39c64224d349a935ab5039aa360942a48"
dependencies = [
"serde_derive",
]
[[package]]
name = "serde_derive"
version = "1.0.229"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e7a5d71263a5a7d47b41f6b3f06ba276f10cc18b0931f1799f710578e2309348"
dependencies = [
"proc-macro2",
"quote",
"syn",
]
[[package]]
name = "serde_json"
version = "1.0.151"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c841b55ecdae098c80dcae9cf767f6f8a0c2cdb3416bbef72181df4d0fe73f14"
dependencies = [
"itoa",
"memchr",
"serde",
"serde_core",
"zmij",
]
[[package]]
name = "simdutf8"
version = "0.1.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e3a9fe34e3e7a50316060351f37187a3f546bce95496156754b601a5fa71b76e"
[[package]]
name = "syn"
version = "3.0.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8593e8e72159ed2257d083c7a454a85cbf854f37a0966d8d483aff8c8a3ebcee"
dependencies = [
"proc-macro2",
"quote",
"unicode-ident",
]
[[package]]
name = "unicode-ident"
version = "1.0.26"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d245f478577f809a851594d02313b640fb437e0bb33866753cff937863096954"
[[package]]
name = "windows-link"
version = "0.2.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5"
[[package]]
name = "windows-sys"
version = "0.61.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ae137229bcbd6cdf0f7b80a31df61766145077ddf49416a728b02cb3921ff3fc"
dependencies = [
"windows-link",
]
[[package]]
name = "zmij"
version = "1.0.23"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "29666d0abbfad1e3dc4dcf6144730dd3a3ab225bbbdac83319345b1b44ccfc1b"

27
packages/env/daemon/Cargo.toml vendored Normal file
View file

@ -0,0 +1,27 @@
[package]
name = "pi-env"
version = "0.1.0"
edition = "2024"
description = "Remote execution environment daemon for Pi Durable"
publish = false
[dependencies]
encoding_rs = "=0.8.42"
serde_json = "=1.0.151"
[target.'cfg(unix)'.dependencies]
libc = "=0.2.190"
[target.'cfg(windows)'.dependencies]
windows-sys = { version = "=0.61.2", features = [
"Win32_Foundation",
"Win32_Security",
"Win32_Storage_FileSystem",
"Win32_System_JobObjects",
"Win32_System_Threading",
] }
[profile.release]
lto = true
codegen-units = 1
strip = true

3
packages/env/daemon/rust-toolchain.toml vendored Normal file
View file

@ -0,0 +1,3 @@
[toolchain]
channel = "1.96.0"
components = ["rustfmt", "clippy"]

83
packages/env/daemon/src/decode.rs vendored Normal file
View file

@ -0,0 +1,83 @@
//! WHATWG UTF-8 decoding that matches Node's `new TextDecoder().decode(bytes)` of a whole input when the input arrives
//! in pieces (Durable `env/decode.ts`): invalid sequences become U+FFFD by the maximal-subpart rule, and only a
//! byte-order mark at the very start is dropped.
use encoding_rs::{Decoder, UTF_8};
/// Decodes one stream chunk by chunk, dropping a leading byte-order mark.
pub struct StreamDecoder {
decoder: Decoder,
}
impl StreamDecoder {
pub fn new() -> StreamDecoder {
// Removes a UTF-8 BOM only; plain `new_decoder()` would also sniff UTF-16 BOMs and switch encodings.
StreamDecoder {
decoder: UTF_8.new_decoder_with_bom_removal(),
}
}
/// Text for `bytes`, holding back an incomplete character; with `last`, the end of the stream.
pub fn decode(&mut self, bytes: &[u8], last: bool) -> String {
decode_into(&mut self.decoder, bytes, last)
}
}
/// A decoder for a byte range that does not start at the beginning of its input: no BOM handling at all.
pub fn range_decoder() -> Decoder {
UTF_8.new_decoder_without_bom_handling()
}
pub fn decode_into(decoder: &mut Decoder, bytes: &[u8], last: bool) -> String {
let capacity = decoder
.max_utf8_buffer_length(bytes.len())
.unwrap_or(bytes.len() * 3 + 16);
let mut text = String::with_capacity(capacity);
let (_result, read, _replaced) = decoder.decode_to_string(bytes, &mut text, last);
debug_assert_eq!(read, bytes.len());
text
}
/// Whether decoding the whole input drops its first three bytes as a byte-order mark.
pub fn starts_with_bom(bytes: &[u8]) -> bool {
bytes.starts_with(&[0xef, 0xbb, 0xbf])
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn keeps_a_feff_that_does_not_start_the_stream() {
let mut decoder = StreamDecoder::new();
let bytes = [0xe2, 0x82, 0xef, 0xbb, 0xbf, 0x61];
let text: String = bytes
.iter()
.map(|byte| decoder.decode(&[*byte], false))
.collect::<String>()
+ &decoder.decode(&[], true);
assert_eq!(text, "\u{fffd}\u{feff}a");
}
#[test]
fn drops_only_a_leading_mark_and_never_sniffs_utf16() {
let mut decoder = StreamDecoder::new();
assert_eq!(
decoder.decode(&[0xef, 0xbb], false) + &decoder.decode(&[0xbf, 0x61], true),
"a"
);
let mut decoder = StreamDecoder::new();
assert_eq!(
decoder.decode(&[0xff, 0xfe, 0x41, 0x00], true),
"\u{fffd}\u{fffd}A\u{0}"
);
}
#[test]
fn replaces_an_incomplete_sequence_once() {
let mut decoder = StreamDecoder::new();
assert_eq!(decoder.decode(&[0xe2, 0x82, 0x41], true), "\u{fffd}A");
let mut decoder = StreamDecoder::new();
assert_eq!(decoder.decode(&[0xe2, 0x82], true), "\u{fffd}");
}
}

126
packages/env/daemon/src/errors.rs vendored Normal file
View file

@ -0,0 +1,126 @@
//! Errors as Node reports them: a code such as `ENOENT` plus a diagnostic message.
use serde_json::{Value, json};
use std::io;
/// A failure with a Node-style code; messages are diagnostic only.
#[derive(Debug)]
pub struct Failure {
pub code: String,
pub message: String,
details: Option<Box<Details>>,
}
#[derive(Debug, Default)]
struct Details {
syscall: Option<String>,
path: Option<String>,
/// Extra fields for the client, such as `spillPath`.
extra: Option<Value>,
}
impl Failure {
pub fn new(code: &str, message: impl Into<String>) -> Failure {
Failure {
code: code.to_string(),
message: message.into(),
details: None,
}
}
fn details(&mut self) -> &mut Details {
self.details.get_or_insert_with(Default::default)
}
/// Add fields to the error JSON, such as `spillPath`.
pub fn extra(mut self, extra: Value) -> Failure {
self.details().extra = Some(extra);
self
}
/// An I/O error of `syscall` on `path`, with libuv's error name as code.
pub fn io(error: &io::Error, syscall: &str, path: &str) -> Failure {
Failure::io_between(error, syscall, path, None)
}
/// Node's message for an error of a two-path call: `CODE: description, syscall 'from' -> 'to'`.
pub fn io_between(
error: &io::Error,
syscall: &str,
path: &str,
destination: Option<&str>,
) -> Failure {
let code = crate::sys::error_name(error).to_string();
let description = describe(&code).map(str::to_string).unwrap_or_else(|| {
// Unknown to libuv's table: the OS text without Rust's " (os error N)".
let text = error.to_string();
text.split(" (os error")
.next()
.unwrap_or(&text)
.to_lowercase()
});
let target = match destination {
Some(destination) => format!("'{path}' -> '{destination}'"),
None => format!("'{path}'"),
};
let mut failure = Failure::new(&code, format!("{code}: {description}, {syscall} {target}"));
failure.details().syscall = Some(syscall.to_string());
failure.path(path)
}
pub fn path(mut self, path: &str) -> Failure {
self.details().path = Some(path.to_string());
self
}
pub fn to_json(&self) -> Value {
let mut value = json!({ "code": self.code, "message": self.message });
let Some(details) = &self.details else {
return value;
};
if let Some(syscall) = &details.syscall {
value["syscall"] = json!(syscall);
}
if let Some(path) = &details.path {
value["path"] = json!(path);
}
if let Some(Value::Object(extra)) = &details.extra {
for (key, field) in extra {
value[key] = field.clone();
}
}
value
}
}
/// libuv's `uv_strerror` texts, which Node uses on every system.
fn describe(code: &str) -> Option<&'static str> {
Some(match code {
"EACCES" => "permission denied",
"EAGAIN" => "resource temporarily unavailable",
"EBADF" => "bad file descriptor",
"EBUSY" => "resource busy or locked",
"EEXIST" => "file already exists",
"EFBIG" => "file too large",
"EINVAL" => "invalid argument",
"EIO" => "i/o error",
"EISDIR" => "illegal operation on a directory",
"ELOOP" => "too many symbolic links encountered",
"EMFILE" => "too many open files",
"EMLINK" => "too many links",
"ENAMETOOLONG" => "name too long",
"ENFILE" => "file table overflow",
"ENOENT" => "no such file or directory",
"ENOMEM" => "not enough memory",
"ENOSPC" => "no space left on device",
"ENOSYS" => "function not implemented",
"ENOTDIR" => "not a directory",
"ENOTEMPTY" => "directory not empty",
"ENOTSUP" => "operation not supported on socket",
"EPERM" => "operation not permitted",
"EROFS" => "read-only file system",
"ETXTBSY" => "text file is busy",
"EXDEV" => "cross-device link not permitted",
_ => return None,
})
}

378
packages/env/daemon/src/exec.rs vendored Normal file
View file

@ -0,0 +1,378 @@
//! `exec`: Durable's `NodeExecutionEnv.exec` semantics (docs/semantics.md), output streamed as `output` events.
use crate::decode::StreamDecoder;
use crate::errors::Failure;
use crate::frame::{EVENT, Frame};
use crate::sys;
use serde_json::{Map, Value, json};
use std::collections::HashSet;
use std::fs::{File, OpenOptions};
use std::io::{Read, Write};
use std::path::Path;
use std::process::{Child, Command, ExitStatus, Stdio};
use std::sync::mpsc::{Receiver, RecvTimeoutError, Sender};
use std::sync::{Arc, Mutex};
use std::thread;
use std::time::{Duration, Instant};
const EXIT_STDIO_GRACE: Duration = Duration::from_millis(100);
const READ_CHUNK: usize = 64 * 1024;
/// Commands this daemon started and has not reaped; killed when the connection ends.
pub type Groups = Arc<Mutex<HashSet<u32>>>;
pub enum Message {
Data(usize, Vec<u8>),
Eof(usize),
Exit(ExitStatus),
WaitFailed(String),
/// Abort: the command settles as `aborted`.
Cancel,
/// Kill without aborting, as `cleanup()` does: the command settles with the killed process's status.
Kill,
}
pub struct ExecRequest {
pub command: Option<String>,
pub argv: Option<Vec<String>>,
pub cwd: String,
pub env: Map<String, Value>,
pub inherit_env: bool,
pub shell_path: Option<String>,
pub timeout: Option<Duration>,
pub spill: Option<(u64, u64)>,
}
impl ExecRequest {
pub fn from_json(json: &Value) -> Result<ExecRequest, Failure> {
let strings = |value: &Value| -> Option<Vec<String>> {
value
.as_array()?
.iter()
.map(|item| item.as_str().map(str::to_string))
.collect()
};
Ok(ExecRequest {
command: json["command"].as_str().map(str::to_string),
argv: strings(&json["argv"]),
cwd: json["cwd"]
.as_str()
.ok_or_else(|| Failure::new("EINVAL", "exec needs cwd"))?
.to_string(),
env: json["env"].as_object().cloned().unwrap_or_default(),
inherit_env: json["inheritEnv"].as_bool().unwrap_or(true),
shell_path: json["shellPath"].as_str().map(str::to_string),
timeout: json["timeoutMs"]
.as_f64()
.map(|ms| Duration::from_secs_f64(ms / 1000.0)),
spill: json["spill"].as_object().and_then(|spill| {
Some((
spill.get("afterBytes")?.as_u64()?,
spill.get("afterLines")?.as_u64()?,
))
}),
})
}
}
pub fn kill_group(pid: u32) {
sys::kill_tree(pid);
}
fn random_uuid() -> String {
let mut bytes = [0u8; 16];
if let Ok(mut random) = File::open("/dev/urandom") {
let _ = random.read_exact(&mut bytes);
}
bytes[6] = (bytes[6] & 0x0f) | 0x40;
bytes[8] = (bytes[8] & 0x3f) | 0x80;
let hex: String = bytes.iter().map(|byte| format!("{byte:02x}")).collect();
format!(
"{}-{}-{}-{}-{}",
&hex[0..8],
&hex[8..12],
&hex[12..16],
&hex[16..20],
&hex[20..32]
)
}
/// Spill of the complete raw output, started once it crosses either threshold, like Node's.
struct Spill {
after_bytes: u64,
after_lines: u64,
seen_bytes: u64,
seen_newlines: u64,
prefix: Vec<Vec<u8>>,
file: Option<File>,
path: Option<String>,
failed: Option<String>,
}
impl Spill {
fn push(&mut self, chunk: &[u8], tmpdir: &str) {
if self.failed.is_some() || chunk.is_empty() {
return;
}
if self.file.is_none() {
self.seen_bytes += chunk.len() as u64;
self.seen_newlines += chunk.iter().filter(|byte| **byte == b'\n').count() as u64;
let lines = self.seen_newlines + u64::from(chunk.last() != Some(&b'\n'));
if self.seen_bytes <= self.after_bytes && lines <= self.after_lines {
self.prefix.push(chunk.to_vec());
return;
}
if let Err(error) = self.start(tmpdir) {
self.failed = Some(error);
return;
}
}
if let Some(file) = &mut self.file
&& let Err(error) = file.write_all(chunk)
{
self.failed = Some(error.to_string());
}
}
/// `createTempFile({ prefix: "pi-output-", suffix: ".log" })`: a fresh `tmp-` directory holding the file.
fn start(&mut self, tmpdir: &str) -> Result<(), String> {
let directory =
crate::fs::mkdtemp(&format!("{tmpdir}/tmp-")).map_err(|failure| failure.message)?;
let directory = directory["path"].as_str().unwrap_or_default().to_string();
let path = format!("{directory}/pi-output-{}.log", random_uuid());
let mut file = OpenOptions::new()
.append(true)
.create(true)
.open(&path)
.map_err(|error| error.to_string())?;
for chunk in self.prefix.drain(..) {
file.write_all(&chunk).map_err(|error| error.to_string())?;
}
self.file = Some(file);
self.path = Some(path);
Ok(())
}
}
fn spawn_reader(stream: usize, mut pipe: impl Read + Send + 'static, sender: Sender<Message>) {
thread::spawn(move || {
let mut buffer = vec![0u8; READ_CHUNK];
loop {
match pipe.read(&mut buffer) {
Ok(0) | Err(_) => break,
Ok(read) => {
if sender
.send(Message::Data(stream, buffer[..read].to_vec()))
.is_err()
{
break;
}
}
}
}
let _ = sender.send(Message::Eof(stream));
});
}
fn spawn(
request: &ExecRequest,
program: &str,
args: &[String],
piped_stdin: bool,
) -> std::io::Result<Child> {
let mut command = Command::new(program);
command
.current_dir(&request.cwd)
.stdin(if piped_stdin {
Stdio::piped()
} else {
Stdio::null()
})
.stdout(Stdio::piped())
.stderr(Stdio::piped());
sys::configure(&mut command, args, &request.env, request.inherit_env);
let child = command.spawn()?;
sys::after_spawn(&child);
Ok(child)
}
/// Run one command, sending output events for request `id`, until it settles.
pub fn run(
id: u32,
request: ExecRequest,
tmpdir: &str,
out: &Sender<Frame>,
messages: (Sender<Message>, Receiver<Message>),
groups: &Groups,
) -> Result<Value, Failure> {
// A string runs through the shell (as its last argument, or on stdin for legacy WSL bash); argv runs directly.
let (program, args, stdin_command) = match (&request.command, &request.argv) {
(Some(command), _) => {
let shell = sys::shell_config(request.shell_path.as_deref())?;
let mut args = shell.args;
if shell.command_on_stdin {
(shell.program, args, Some(command.clone()))
} else {
args.push(command.clone());
(shell.program, args, None)
}
}
(None, Some(argv)) => match argv.split_first() {
Some((first, rest)) => (first.clone(), rest.to_vec(), None),
None => return Err(Failure::new("spawn_error", "Empty argv: no program to run")),
},
(None, None) => return Err(Failure::new("EINVAL", "exec needs command or argv")),
};
if !Path::new(&request.cwd).exists() {
return Err(Failure::new(
"spawn_error",
format!(
"Working directory does not exist: {}\nCannot execute bash commands.",
request.cwd
),
));
}
let (sender, receiver) = messages;
let mut child = spawn(&request, &program, &args, stdin_command.is_some())
.map_err(|error| Failure::new("spawn_error", error.to_string()))?;
if let (Some(command), Some(mut stdin)) = (stdin_command, child.stdin.take()) {
let _ = stdin.write_all(command.as_bytes());
}
let pid = child.id();
groups.lock().unwrap().insert(pid);
let deadline = request.timeout.map(|timeout| Instant::now() + timeout);
spawn_reader(0, child.stdout.take().unwrap(), sender.clone());
spawn_reader(1, child.stderr.take().unwrap(), sender.clone());
let waiter = sender.clone();
thread::spawn(move || {
let _ = waiter.send(match child.wait() {
Ok(status) => Message::Exit(status),
Err(error) => Message::WaitFailed(error.to_string()),
});
});
drop(sender);
let mut decoders = [StreamDecoder::new(), StreamDecoder::new()];
let mut spill = request.spill.map(|(after_bytes, after_lines)| Spill {
after_bytes,
after_lines,
seen_bytes: 0,
seen_newlines: 0,
prefix: Vec::new(),
file: None,
path: None,
failed: None,
});
let emit = |stream: usize, text: String| {
if !text.is_empty() {
let name = if stream == 0 { "stdout" } else { "stderr" };
let _ = out.send(Frame::with_payload(
EVENT,
id,
json!({ "kind": "output", "stream": name }),
text.into_bytes(),
));
}
};
let mut ended = [false, false];
let mut status: Option<Result<ExitStatus, String>> = None;
let mut idle_until: Option<Instant> = None;
let mut timed_out = false;
let mut aborted = false;
let mut killed = false;
let kill = |killed: &mut bool| {
if !*killed {
*killed = true;
kill_group(pid);
}
};
loop {
if status.is_some() && ended[0] && ended[1] {
break;
}
let now = Instant::now();
let mut wait = Duration::from_secs(3600);
if let Some(deadline) = deadline.filter(|_| !timed_out) {
wait = wait.min(deadline.saturating_duration_since(now));
}
if let Some(idle) = idle_until {
wait = wait.min(idle.saturating_duration_since(now));
}
match receiver.recv_timeout(wait) {
Ok(Message::Data(stream, bytes)) => {
emit(stream, decoders[stream].decode(&bytes, false));
if let Some(spill) = &mut spill {
spill.push(&bytes, tmpdir);
if spill.failed.is_some() {
kill(&mut killed);
}
}
if status.is_some() {
idle_until = Some(Instant::now() + EXIT_STDIO_GRACE);
}
}
Ok(Message::Eof(stream)) => ended[stream] = true,
Ok(Message::Exit(exit)) => {
status = Some(Ok(exit));
idle_until = Some(Instant::now() + EXIT_STDIO_GRACE);
}
Ok(Message::WaitFailed(error)) => {
status = Some(Err(error));
idle_until = Some(Instant::now());
}
Ok(Message::Cancel) => {
aborted = true;
kill(&mut killed);
}
Ok(Message::Kill) => kill(&mut killed),
Err(RecvTimeoutError::Timeout) => {
if deadline.is_some_and(|deadline| Instant::now() >= deadline)
&& !timed_out
&& status.is_none()
{
timed_out = true;
kill(&mut killed);
} else if idle_until.is_some_and(|idle| Instant::now() >= idle) {
// A descendant may hold the pipes open after the process exited; stop waiting for it.
break;
}
}
Err(RecvTimeoutError::Disconnected) => break,
}
}
groups.lock().unwrap().remove(&pid);
emit(0, decoders[0].decode(&[], true));
emit(1, decoders[1].decode(&[], true));
let spill_path = spill.as_ref().and_then(|spill| spill.path.clone());
let with_spill = |failure: Failure| match &spill_path {
Some(path) => failure.extra(json!({ "spillPath": path })),
None => failure,
};
if timed_out {
return Err(with_spill(Failure::new("timeout", "timeout")));
}
if aborted {
return Err(with_spill(Failure::new("aborted", "aborted")));
}
if let Some(error) = spill.as_ref().and_then(|spill| spill.failed.clone()) {
return Err(Failure::new(
"unknown",
format!("Failed to preserve complete shell output: {error}"),
));
}
let status = match status {
Some(Ok(status)) => status,
Some(Err(error)) => return Err(Failure::new("spawn_error", error)),
None => {
return Err(Failure::new(
"spawn_error",
"process did not report its exit",
));
}
};
let mut result = json!({ "exitCode": sys::exit_code(status) });
if let Some(path) = spill_path {
result["spillPath"] = json!(path);
}
Ok(result)
}

116
packages/env/daemon/src/frame.rs vendored Normal file
View file

@ -0,0 +1,116 @@
//! Length-prefixed frames: `u32 length | u8 type | u32 id | u32 jsonLength | json | payload` (docs/protocol.md).
use serde_json::Value;
use std::io::{self, Read, Write};
pub const REQUEST: u8 = 1;
pub const RESULT: u8 = 2;
pub const ERROR: u8 = 3;
pub const EVENT: u8 = 4;
pub const CANCEL: u8 = 5;
pub const PING: u8 = 6;
/// Largest frame either side accepts.
pub const MAX_FRAME: usize = 16 * 1024 * 1024;
/// Largest payload a response carries, leaving room for its JSON.
pub const MAX_PAYLOAD: usize = MAX_FRAME - 64 * 1024;
pub struct Frame {
pub kind: u8,
pub id: u32,
pub json: Value,
pub payload: Vec<u8>,
}
impl Frame {
pub fn new(kind: u8, id: u32, json: Value) -> Frame {
Frame {
kind,
id,
json,
payload: Vec::new(),
}
}
pub fn with_payload(kind: u8, id: u32, json: Value, payload: Vec<u8>) -> Frame {
Frame {
kind,
id,
json,
payload,
}
}
}
fn invalid(message: &str) -> io::Error {
io::Error::new(io::ErrorKind::InvalidData, message.to_string())
}
/// Read one frame; `None` at a clean end of input.
pub fn read_frame(input: &mut impl Read) -> io::Result<Option<Frame>> {
let mut length = [0u8; 4];
match input.read_exact(&mut length) {
Ok(()) => {}
Err(error) if error.kind() == io::ErrorKind::UnexpectedEof => return Ok(None),
Err(error) => return Err(error),
}
let length = u32::from_be_bytes(length) as usize;
if !(9..=MAX_FRAME).contains(&length) {
return Err(invalid("frame length out of range"));
}
let mut body = vec![0u8; length];
input.read_exact(&mut body)?;
let kind = body[0];
let id = u32::from_be_bytes([body[1], body[2], body[3], body[4]]);
let json_length = u32::from_be_bytes([body[5], body[6], body[7], body[8]]) as usize;
if 9 + json_length > length {
return Err(invalid("JSON length out of range"));
}
let json = if json_length == 0 {
Value::Object(Default::default())
} else {
serde_json::from_slice(&body[9..9 + json_length]).map_err(|_| invalid("invalid JSON"))?
};
let payload = body[9 + json_length..].to_vec();
Ok(Some(Frame {
kind,
id,
json,
payload,
}))
}
pub fn write_frame(output: &mut impl Write, frame: &Frame) -> io::Result<()> {
let json = serde_json::to_vec(&frame.json).map_err(|_| invalid("unserializable JSON"))?;
let length = 9 + json.len() + frame.payload.len();
let mut header = Vec::with_capacity(13);
header.extend_from_slice(&(length as u32).to_be_bytes());
header.push(frame.kind);
header.extend_from_slice(&frame.id.to_be_bytes());
header.extend_from_slice(&(json.len() as u32).to_be_bytes());
output.write_all(&header)?;
output.write_all(&json)?;
output.write_all(&frame.payload)?;
output.flush()
}
#[cfg(test)]
mod tests {
use super::*;
use serde_json::json;
#[test]
fn round_trips_json_and_payload() {
let mut buffer = Vec::new();
write_frame(
&mut buffer,
&Frame::with_payload(RESULT, 7, json!({ "a": 1 }), vec![0, 255, 10]),
)
.unwrap();
let frame = read_frame(&mut buffer.as_slice()).unwrap().unwrap();
assert_eq!((frame.kind, frame.id), (RESULT, 7));
assert_eq!(frame.json, json!({ "a": 1 }));
assert_eq!(frame.payload, vec![0, 255, 10]);
assert!(read_frame(&mut &[][..]).unwrap().is_none());
}
}

221
packages/env/daemon/src/fs.rs vendored Normal file
View file

@ -0,0 +1,221 @@
//! File operations with Node's semantics where the Rust standard library differs (docs/semantics.md).
use crate::errors::Failure;
use crate::sys;
use serde_json::{Value, json};
use std::fs::{self, File, Metadata};
use std::io::{self, Write};
use std::path::Path;
pub type Outcome<T> = Result<T, Failure>;
/// `{ name, kind, size, mtimeSec, mtimeNsec, dev, ino }` of lstat or fstat metadata.
pub fn info(path: &str, metadata: &Metadata) -> Value {
let file_type = metadata.file_type();
let kind = if file_type.is_file() {
"file"
} else if file_type.is_dir() {
"directory"
} else if file_type.is_symlink() {
"symlink"
} else {
"other"
};
let name = Path::new(path)
.file_name()
.map(|name| name.to_string_lossy().into_owned())
.unwrap_or_default();
let (mtime_sec, mtime_nsec) = sys::mtime(metadata);
let (dev, ino) = sys::identity(metadata);
json!({
"name": name,
"kind": kind,
"size": metadata.len(),
"mtimeSec": mtime_sec,
"mtimeNsec": mtime_nsec,
"dev": dev,
"ino": ino,
})
}
pub fn lstat(path: &str) -> Outcome<Value> {
fs::symlink_metadata(path)
.map(|metadata| info(path, &metadata))
.map_err(|error| Failure::io(&error, "lstat", path))
}
pub fn realpath(path: &str) -> Outcome<Value> {
sys::realpath(path)
.map(|resolved| json!({ "path": resolved }))
.map_err(|error| Failure::io(&error, "realpath", path))
}
/// Node's recursive `mkdir`: an existing directory is fine, an existing file is `EEXIST`, a file in the way of a
/// parent is `ENOTDIR`.
pub fn mkdir_recursive(path: &Path) -> io::Result<()> {
let is_dir = |path: &Path| {
fs::metadata(path)
.map(|metadata| metadata.is_dir())
.unwrap_or(false)
};
match fs::create_dir(path) {
Ok(()) => Ok(()),
Err(error) if error.kind() == io::ErrorKind::AlreadyExists => {
if is_dir(path) {
Ok(())
} else {
Err(error)
}
}
Err(error) if error.kind() == io::ErrorKind::NotFound => {
let Some(parent) = path
.parent()
.filter(|parent| !parent.as_os_str().is_empty())
else {
return Err(error);
};
match mkdir_recursive(parent) {
Err(parent_error) if parent_error.kind() == io::ErrorKind::AlreadyExists => {
return Err(sys::not_a_directory());
}
other => other?,
}
match fs::create_dir(path) {
Err(error) if error.kind() == io::ErrorKind::AlreadyExists && is_dir(path) => {
Ok(())
}
other => other,
}
}
Err(error) => Err(error),
}
}
pub fn mkdir(path: &str, recursive: bool) -> Outcome<Value> {
let result = if recursive {
mkdir_recursive(Path::new(path))
} else {
fs::create_dir(path)
};
result
.map(|()| json!({}))
.map_err(|error| Failure::io(&error, "mkdir", path))
}
/// `writeFile`/`appendFile`: create missing parents like Node's recursive `mkdir`, then write.
pub fn write(path: &str, append: bool, content: &[u8]) -> Outcome<Value> {
if let Some(parent) = Path::new(path)
.parent()
.filter(|parent| !parent.as_os_str().is_empty())
{
mkdir_recursive(parent)
.map_err(|error| Failure::io(&error, "mkdir", &parent.to_string_lossy()))?;
}
let mode = if append {
sys::OpenMode::Append
} else {
sys::OpenMode::Write
};
let mut file = sys::open(path, mode).map_err(|error| Failure::io(&error, "open", path))?;
file.write_all(content)
.map_err(|error| Failure::io(&error, "write", path))?;
Ok(json!({}))
}
fn open_existing_for_write(path: &str) -> Outcome<File> {
sys::open(path, sys::OpenMode::ReadWrite).map_err(|error| Failure::io(&error, "open", path))
}
pub fn truncate(path: &str, size: u64) -> Outcome<Value> {
let file = open_existing_for_write(path)?;
file.set_len(size)
.map_err(|error| Failure::io(&error, "ftruncate", path))?;
Ok(json!({}))
}
pub fn fsync(path: &str) -> Outcome<Value> {
// POSIX refuses to open a directory for writing; check first so Windows reports the same.
if fs::metadata(path).is_ok_and(|metadata| metadata.is_dir()) {
return Err(Failure::new(
"EISDIR",
format!("EISDIR: illegal operation on a directory, open '{path}'"),
)
.path(path));
}
let file = open_existing_for_write(path)?;
file.sync_all()
.map_err(|error| Failure::io(&error, "fsync", path))?;
Ok(json!({}))
}
pub fn rename(from: &str, to: &str) -> Outcome<Value> {
sys::rename(from, to)
.map(|()| json!({}))
.map_err(|error| Failure::io_between(&error, "rename", from, Some(to)))
}
/// Node's `fs.rm`: missing paths fail unless `force`; a directory without `recursive` fails with `ERR_FS_EISDIR`.
pub fn rm(path: &str, recursive: bool, force: bool) -> Outcome<Value> {
let metadata = match fs::symlink_metadata(path) {
Ok(metadata) => metadata,
Err(error) if force && error.kind() == io::ErrorKind::NotFound => return Ok(json!({})),
Err(error) => return Err(Failure::io(&error, "lstat", path)),
};
let result = if metadata.is_dir() {
if !recursive {
return Err(Failure::new(
"ERR_FS_EISDIR",
format!("Path is a directory: rm returned EISDIR (is a directory) {path}"),
)
.path(path));
}
fs::remove_dir_all(path)
} else {
sys::remove_file(path)
};
result
.map(|()| json!({}))
.map_err(|error| Failure::io(&error, "rm", path))
}
/// A new directory named `prefix` plus six random characters, as libuv's `mkdtemp` creates it.
pub fn mkdtemp(prefix: &str) -> Outcome<Value> {
sys::mkdtemp(prefix)
.map(|path| json!({ "path": path }))
.map_err(|error| Failure::io(&error, "mkdtemp", &format!("{prefix}XXXXXX")))
}
/// Open a regular file for positional reads; directories fail with `EISDIR`, other non-regular files with
/// `NOT_REGULAR`, and with `no_follow` a final symbolic link with `SYMLINK`.
pub fn open_reader(path: &str, no_follow: bool) -> Outcome<(File, Value)> {
let file = sys::open_reader(path, no_follow)?;
let metadata = file
.metadata()
.map_err(|error| Failure::io(&error, "fstat", path))?;
if metadata.is_dir() {
return Err(
Failure::new("EISDIR", "EISDIR: illegal operation on a directory, read").path(path),
);
}
if !metadata.is_file() {
return Err(Failure::new("NOT_REGULAR", "Not a regular file").path(path));
}
let info = info(path, &metadata);
Ok((file, info))
}
/// Up to `length` bytes at `offset`; fewer only at the end of the file.
pub fn pread(file: &File, path: &str, offset: u64, length: usize) -> Outcome<Vec<u8>> {
let mut buffer = vec![0u8; length];
let mut filled = 0;
while filled < length {
match sys::read_at(file, &mut buffer[filled..], offset + filled as u64) {
Ok(0) => break,
Ok(read) => filled += read,
Err(error) if error.kind() == io::ErrorKind::Interrupted => {}
Err(error) => return Err(Failure::io(&error, "read", path)),
}
}
buffer.truncate(filled);
Ok(buffer)
}

356
packages/env/daemon/src/main.rs vendored Normal file
View file

@ -0,0 +1,356 @@
//! pi-env: a small daemon that gives a Pi Durable host an execution environment on this machine over stdin/stdout
//! (docs/protocol.md). Started as `pi-env serve --token <hex>`, usually through `ssh`.
mod decode;
mod errors;
mod exec;
mod frame;
mod fs;
mod scan;
mod sys;
use errors::Failure;
use frame::{CANCEL, ERROR, Frame, PING, REQUEST, RESULT};
use serde_json::{Value, json};
use std::collections::HashMap;
use std::env;
use std::fs::{File, ReadDir};
use std::io::{self, BufReader, BufWriter, Write};
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::mpsc::{self, Sender};
use std::sync::{Arc, Mutex};
use std::thread;
use std::time::{Duration, SystemTime, UNIX_EPOCH};
const PROTOCOL: u64 = 1;
const PING_INTERVAL: Duration = Duration::from_secs(5);
const SILENCE_LIMIT_MS: u64 = 30_000;
const SCAN_CHUNK: usize = 64 * 1024;
enum Handle {
File(Arc<File>, String),
Dir(Box<ReadDir>, String),
}
enum Cancel {
Flag(Arc<AtomicBool>),
Exec(Sender<exec::Message>),
}
struct Server {
out: Sender<Frame>,
tmpdir: String,
handles: Mutex<HashMap<u64, Handle>>,
next_handle: AtomicU64,
cancels: Mutex<HashMap<u32, Cancel>>,
groups: exec::Groups,
}
fn now_ms() -> u64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|elapsed| elapsed.as_millis() as u64)
.unwrap_or(0)
}
fn field<'a>(json: &'a Value, key: &str) -> Result<&'a str, Failure> {
json[key]
.as_str()
.ok_or_else(|| Failure::new("EINVAL", format!("missing field {key}")))
}
fn number(json: &Value, key: &str) -> Result<u64, Failure> {
json[key]
.as_u64()
.ok_or_else(|| Failure::new("EINVAL", format!("missing or invalid field {key}")))
}
impl Server {
fn handle(&self, json: &Value) -> Result<Handle, Failure> {
let id = number(json, "handle")?;
let handles = self.handles.lock().unwrap();
match handles.get(&id) {
Some(Handle::File(file, path)) => Ok(Handle::File(file.clone(), path.clone())),
Some(Handle::Dir(..)) => Err(Failure::new("EBADF", "not a file handle")),
None => Err(Failure::new("EBADF", "unknown handle")),
}
}
fn file(&self, json: &Value) -> Result<(Arc<File>, String), Failure> {
match self.handle(json)? {
Handle::File(file, path) => Ok((file, path)),
Handle::Dir(..) => Err(Failure::new("EBADF", "not a file handle")),
}
}
fn insert(&self, handle: Handle) -> u64 {
let id = self.next_handle.fetch_add(1, Ordering::Relaxed);
self.handles.lock().unwrap().insert(id, handle);
id
}
fn dispatch(
&self,
id: u32,
json: &Value,
payload: Vec<u8>,
) -> Result<(Value, Vec<u8>), Failure> {
let op = json["op"].as_str().unwrap_or_default();
let plain = |value: Result<Value, Failure>| value.map(|value| (value, Vec::new()));
match op {
"hello" => plain(Ok(json!({
"protocol": PROTOCOL,
"version": env!("CARGO_PKG_VERSION"),
"os": env::consts::OS,
"arch": env::consts::ARCH,
"home": sys::home(),
"separator": sys::PATH_SEPARATOR,
"tmpdir": self.tmpdir,
"pid": std::process::id(),
}))),
"lstat" => plain(fs::lstat(field(json, "path")?)),
"realpath" => plain(fs::realpath(field(json, "path")?)),
"write" => plain(fs::write(
field(json, "path")?,
json["append"].as_bool().unwrap_or(false),
&payload,
)),
"truncate" => plain(fs::truncate(field(json, "path")?, number(json, "size")?)),
"fsync" => plain(fs::fsync(field(json, "path")?)),
"rename" => plain(fs::rename(field(json, "path")?, field(json, "to")?)),
"mkdir" => plain(fs::mkdir(
field(json, "path")?,
json["recursive"].as_bool().unwrap_or(false),
)),
"rm" => plain(fs::rm(
field(json, "path")?,
json["recursive"].as_bool().unwrap_or(false),
json["force"].as_bool().unwrap_or(false),
)),
"mkdtemp" => plain(fs::mkdtemp(field(json, "path")?)),
"open" => {
let path = field(json, "path")?;
let (file, info) =
fs::open_reader(path, json["noFollow"].as_bool().unwrap_or(false))?;
let handle = self.insert(Handle::File(Arc::new(file), path.to_string()));
plain(Ok(json!({ "handle": handle, "info": info })))
}
"pread" => {
let (file, path) = self.file(json)?;
let length = (number(json, "length")? as usize).min(frame::MAX_PAYLOAD);
Ok((
json!({}),
fs::pread(&file, &path, number(json, "offset")?, length)?,
))
}
"fstat" => {
let (file, path) = self.file(json)?;
let metadata = file
.metadata()
.map_err(|error| Failure::io(&error, "fstat", &path))?;
plain(Ok(fs::info(&path, &metadata)))
}
"scanLines" => {
let (file, path) = self.file(json)?;
plain(self.scan_lines(id, &file, &path, json))
}
"opendir" => {
let path = field(json, "path")?;
let dir = std::fs::read_dir(path)
.map_err(|error| Failure::io(&error, "opendir", path))?;
let handle = self.insert(Handle::Dir(Box::new(dir), path.to_string()));
plain(Ok(json!({ "handle": handle })))
}
"readdir" => plain(self.read_dir(json)),
"close" => {
self.handles
.lock()
.unwrap()
.remove(&number(json, "handle")?);
plain(Ok(json!({})))
}
"exec" => {
let request = exec::ExecRequest::from_json(json)?;
let (sender, receiver) = mpsc::channel();
self.cancels
.lock()
.unwrap()
.insert(id, Cancel::Exec(sender.clone()));
plain(exec::run(
id,
request,
&self.tmpdir,
&self.out,
(sender, receiver),
&self.groups,
))
}
_ => Err(Failure::new("EINVAL", format!("unknown operation {op}"))),
}
}
fn scan_lines(&self, id: u32, file: &File, path: &str, json: &Value) -> Result<Value, Failure> {
let start_line = number(json, "startLine")?;
let end_line = json["endLine"].as_u64();
if end_line.is_some_and(|end| end <= start_line) {
return Err(Failure::new("EINVAL", "Invalid line range"));
}
let cancelled = Arc::new(AtomicBool::new(false));
self.cancels
.lock()
.unwrap()
.insert(id, Cancel::Flag(cancelled.clone()));
let mut scanner = scan::LineScanner::new(start_line, end_line);
let mut buffer = vec![0u8; SCAN_CHUNK];
let mut position = 0u64;
loop {
if cancelled.load(Ordering::Relaxed) {
return Err(Failure::new("aborted", "aborted"));
}
let read = sys::read_at(file, &mut buffer, position)
.map_err(|error| Failure::io(&error, "read", path))?;
if read == 0 {
return Ok(scanner.finish());
}
scanner.push(&buffer[..read]);
position += read as u64;
}
}
fn read_dir(&self, json: &Value) -> Result<Value, Failure> {
let id = number(json, "handle")?;
let max = number(json, "max")?.max(1) as usize;
let mut handles = self.handles.lock().unwrap();
let Some(Handle::Dir(dir, path)) = handles.get_mut(&id) else {
return Err(Failure::new("EBADF", "not a directory handle"));
};
let mut entries = Vec::new();
let mut done = false;
while entries.len() < max {
match dir.next() {
None => {
done = true;
break;
}
Some(Err(error)) => return Err(Failure::io(&error, "readdir", path)),
Some(Ok(entry)) => {
let name = entry.file_name().to_string_lossy().into_owned();
let entry_path = entry.path().to_string_lossy().into_owned();
entries.push(match std::fs::symlink_metadata(entry.path()) {
Ok(metadata) => json!({ "name": name, "info": fs::info(&entry_path, &metadata) }),
Err(error) => json!({ "name": name, "error": Failure::io(&error, "lstat", &entry_path).to_json() }),
});
}
}
}
Ok(json!({ "entries": entries, "done": done }))
}
}
fn kill_all(groups: &exec::Groups) {
for pid in groups.lock().unwrap().iter() {
exec::kill_group(*pid);
}
}
fn serve(token: &str) -> io::Result<()> {
let stdout = io::stdout();
{
let mut lock = stdout.lock();
writeln!(lock, "PI-ENV {token}")?;
lock.flush()?;
}
let (out, frames) = mpsc::channel::<Frame>();
thread::spawn(move || {
let mut writer = BufWriter::with_capacity(1 << 20, io::stdout().lock());
for frame in frames {
if frame::write_frame(&mut writer, &frame).is_err() {
break;
}
}
});
let server = Arc::new(Server {
out: out.clone(),
tmpdir: sys::tmpdir(),
handles: Mutex::new(HashMap::new()),
next_handle: AtomicU64::new(1),
cancels: Mutex::new(HashMap::new()),
groups: Arc::new(Mutex::new(Default::default())),
});
let last_seen = Arc::new(AtomicU64::new(now_ms()));
{
let out = out.clone();
let last_seen = last_seen.clone();
let groups = server.groups.clone();
thread::spawn(move || {
loop {
thread::sleep(PING_INTERVAL);
let _ = out.send(Frame::new(PING, 0, json!({})));
// A client that went silent (a phone that lost its network) leaves nothing running.
if now_ms().saturating_sub(last_seen.load(Ordering::Relaxed)) > SILENCE_LIMIT_MS {
kill_all(&groups);
std::process::exit(0);
}
}
});
}
let mut input = BufReader::with_capacity(1 << 20, io::stdin().lock());
while let Some(request) = frame::read_frame(&mut input)? {
last_seen.store(now_ms(), Ordering::Relaxed);
match request.kind {
REQUEST => {
let server = server.clone();
thread::spawn(move || {
let id = request.id;
let reply = match server.dispatch(id, &request.json, request.payload) {
Ok((json, payload)) => Frame::with_payload(RESULT, id, json, payload),
Err(failure) => Frame::new(ERROR, id, failure.to_json()),
};
server.cancels.lock().unwrap().remove(&id);
let _ = server.out.send(reply);
});
}
CANCEL => match server.cancels.lock().unwrap().get(&request.id) {
Some(Cancel::Exec(sender)) => {
let kill = request.json["mode"].as_str() == Some("kill");
let _ = sender.send(if kill {
exec::Message::Kill
} else {
exec::Message::Cancel
});
}
Some(Cancel::Flag(flag)) => flag.store(true, Ordering::Relaxed),
None => {}
},
_ => {}
}
}
// The client is gone: stop everything it started.
kill_all(&server.groups);
Ok(())
}
fn main() {
let args: Vec<String> = env::args().collect();
match args.get(1).map(String::as_str) {
Some("serve") => {
let token = args
.iter()
.position(|arg| arg == "--token")
.and_then(|index| args.get(index + 1));
let Some(token) = token else {
eprintln!("usage: pi-env serve --token <hex>");
std::process::exit(2);
};
if let Err(error) = serve(token) {
eprintln!("pi-env: {error}");
std::process::exit(1);
}
}
Some("--version") => println!("pi-env {}", env!("CARGO_PKG_VERSION")),
_ => {
eprintln!("usage: pi-env serve --token <hex>");
std::process::exit(2);
}
}
}

215
packages/env/daemon/src/scan.rs vendored Normal file
View file

@ -0,0 +1,215 @@
//! Port of Durable's `LineScanner` (`env/line-scan.ts`): locates lines `[start_line, end_line)` of a file fed in order
//! and measures their decoded UTF-8 size as `new TextDecoder().decode(file)` would produce it.
use crate::decode::{decode_into, range_decoder, starts_with_bom};
use encoding_rs::Decoder;
use serde_json::{Value, json};
const NEWLINE: u8 = 0x0a;
pub struct LineScanner {
start_line: u64,
end_line: Option<u64>,
position: u64,
newlines: u64,
line_start: u64,
start: Option<u64>,
end: Option<u64>,
first_line_end: Option<u64>,
last_line_start: Option<u64>,
selected_bytes: u64,
first_line_bytes: u64,
selection: Option<Decoder>,
first_line: Option<Decoder>,
head: Option<Vec<u8>>,
bom: bool,
}
impl LineScanner {
/// `end_line`, when given, must be greater than `start_line`.
pub fn new(start_line: u64, end_line: Option<u64>) -> LineScanner {
let mut scanner = LineScanner {
start_line,
end_line,
position: 0,
newlines: 0,
line_start: 0,
start: None,
end: None,
first_line_end: None,
last_line_start: None,
selected_bytes: 0,
first_line_bytes: 0,
selection: None,
first_line: None,
head: Some(Vec::with_capacity(3)),
bom: false,
};
if start_line == 0 {
scanner.begin(0);
}
scanner
}
fn is_last_line(&self, line: u64) -> bool {
self.end_line.is_some_and(|end| line + 1 == end)
}
pub fn push(&mut self, mut chunk: &[u8]) {
if let Some(head) = &mut self.head {
let take = (3 - head.len()).min(chunk.len());
head.extend_from_slice(&chunk[..take]);
if head.len() < 3 {
return;
}
chunk = &chunk[take..];
self.release_head();
}
self.process(chunk);
}
fn release_head(&mut self) {
let head = self.head.take().unwrap_or_default();
self.bom = starts_with_bom(&head);
self.process(&head);
}
fn process(&mut self, chunk: &[u8]) {
let base = self.position;
let mut from = 0;
let mut index = 0;
while let Some(found) = chunk[index..].iter().position(|byte| *byte == NEWLINE) {
let newline = index + found;
// The newline ends line `self.newlines`; it belongs to the selection between selected lines only.
self.feed(chunk, base, from, newline);
let line = self.newlines;
let position = base + newline as u64;
if line == self.start_line {
self.end_first_line(position);
}
if self.is_last_line(line) {
self.end_selection(position);
}
self.feed(chunk, base, newline, newline + 1);
from = newline + 1;
self.newlines += 1;
self.line_start = position + 1;
if self.newlines == self.start_line {
self.begin(self.line_start);
}
if self.is_last_line(self.newlines) {
self.last_line_start = Some(self.line_start);
}
index = newline + 1;
}
self.feed(chunk, base, from, chunk.len());
self.position += chunk.len() as u64;
}
pub fn finish(mut self) -> Value {
if self.head.is_some() {
self.release_head();
}
let size = self.position;
let Some(start) = self.start else {
return json!({
"newlines": self.newlines, "start": size, "end": size, "firstLineEnd": size,
"lastLineStart": size, "selectedBytes": 0, "firstLineBytes": 0,
});
};
if self.first_line_end.is_none() {
self.end_first_line(size);
}
if self.end.is_none() {
self.end_selection(size);
}
json!({
"newlines": self.newlines,
"start": start,
"end": self.end.unwrap_or(size),
"firstLineEnd": self.first_line_end.unwrap_or(size),
// A selection that reaches past the last line ends with the last line.
"lastLineStart": self.last_line_start.unwrap_or(self.line_start),
"selectedBytes": self.selected_bytes,
"firstLineBytes": self.first_line_bytes,
})
}
fn begin(&mut self, start: u64) {
self.start = Some(start);
if self.is_last_line(self.start_line) {
self.last_line_start = Some(start);
}
self.selection = Some(range_decoder());
self.first_line = Some(range_decoder());
}
fn end_first_line(&mut self, position: u64) {
self.first_line_end = Some(position);
if let Some(mut decoder) = self.first_line.take() {
self.first_line_bytes += decode_into(&mut decoder, &[], true).len() as u64;
}
}
fn end_selection(&mut self, position: u64) {
self.end = Some(position);
if let Some(mut decoder) = self.selection.take() {
self.selected_bytes += decode_into(&mut decoder, &[], true).len() as u64;
}
}
fn feed(&mut self, chunk: &[u8], base: u64, mut from: usize, to: usize) {
// Decoding the whole file drops a leading byte-order mark.
if self.bom && base + (from as u64) < 3 {
from = to.min((3 - base) as usize);
}
if to <= from {
return;
}
let bytes = &chunk[from..to];
if let Some(decoder) = &mut self.selection {
self.selected_bytes += decode_into(decoder, bytes, false).len() as u64;
}
if let Some(decoder) = &mut self.first_line {
self.first_line_bytes += decode_into(decoder, bytes, false).len() as u64;
}
}
}
#[cfg(test)]
mod tests {
use super::*;
fn scan(file: &[u8], start: u64, end: Option<u64>, chunk: usize) -> Value {
let mut scanner = LineScanner::new(start, end);
for piece in file.chunks(chunk.max(1)) {
scanner.push(piece);
}
scanner.finish()
}
#[test]
fn locates_and_measures_lines_like_whole_file_decoding() {
// BOM, "a", invalid sequence, empty line, a later U+FEFF, and no final newline.
let file = [
0xef, 0xbb, 0xbf, 0x61, 0x0a, 0xe2, 0x82, 0x0a, 0x0a, 0xef, 0xbb, 0xbf, 0x62, 0x0a,
0xc3, 0xa9,
];
for chunk in 1..6 {
let all = scan(&file, 0, None, chunk);
assert_eq!(all["newlines"], 4);
// "a\n\u{fffd}\n\n\u{feff}b\né" is 1+1+3+1+1+3+1+1+2 = 14 bytes.
assert_eq!(all["selectedBytes"], 14);
assert_eq!(all["firstLineBytes"], 1);
let middle = scan(&file, 1, Some(3), chunk);
assert_eq!(
(middle["start"].as_u64(), middle["end"].as_u64()),
(Some(5), Some(8))
);
assert_eq!(middle["selectedBytes"], 4);
assert_eq!(middle["lastLineStart"], 8);
let past = scan(&file, 9, None, chunk);
assert_eq!(past["start"], 16);
}
}
}

28
packages/env/daemon/src/sys/mod.rs vendored Normal file
View file

@ -0,0 +1,28 @@
//! What differs between operating systems: error names, positional reads, temporary directories, opening readers,
//! and starting and killing commands. Each implementation follows what Node (libuv) does on that system.
#[cfg(unix)]
mod unix;
#[cfg(unix)]
pub use unix::*;
#[cfg(windows)]
mod windows;
#[cfg(windows)]
pub use windows::*;
/// How a string command runs: the shell, its arguments before the command, and whether the command goes to its
/// stdin instead of its arguments (legacy WSL `bash.exe`).
pub struct ShellConfig {
pub program: String,
pub args: Vec<String>,
pub command_on_stdin: bool,
}
/// How Node opens files for writing: `writeFile` (`w`), `appendFile` (`a`), and `r+` for truncate and fsync.
#[derive(Clone, Copy)]
pub enum OpenMode {
Write,
Append,
ReadWrite,
}

254
packages/env/daemon/src/sys/unix.rs vendored Normal file
View file

@ -0,0 +1,254 @@
//! Linux, macOS and Android.
use super::{OpenMode, ShellConfig};
use crate::errors::Failure;
use serde_json::{Map, Value};
use std::ffi::{CString, OsStr};
use std::fs::{File, Metadata, OpenOptions};
use std::io;
use std::os::unix::ffi::OsStrExt;
use std::os::unix::fs::{FileExt, MetadataExt, OpenOptionsExt};
use std::os::unix::process::{CommandExt, ExitStatusExt};
use std::path::Path;
use std::process::{Child, Command, ExitStatus, Stdio};
/// libuv's name for an OS error.
pub fn error_name(error: &io::Error) -> &'static str {
let Some(errno) = error.raw_os_error() else {
return "UNKNOWN";
};
match errno {
libc::EPERM => "EPERM",
libc::ENOENT => "ENOENT",
libc::EIO => "EIO",
libc::EBADF => "EBADF",
libc::EAGAIN => "EAGAIN",
libc::ENOMEM => "ENOMEM",
libc::EACCES => "EACCES",
libc::EBUSY => "EBUSY",
libc::EEXIST => "EEXIST",
libc::EXDEV => "EXDEV",
libc::ENOTDIR => "ENOTDIR",
libc::EISDIR => "EISDIR",
libc::EINVAL => "EINVAL",
libc::ENFILE => "ENFILE",
libc::EMFILE => "EMFILE",
libc::ETXTBSY => "ETXTBSY",
libc::EFBIG => "EFBIG",
libc::ENOSPC => "ENOSPC",
libc::EROFS => "EROFS",
libc::EMLINK => "EMLINK",
libc::ENAMETOOLONG => "ENAMETOOLONG",
libc::ENOTEMPTY => "ENOTEMPTY",
libc::ELOOP => "ELOOP",
libc::ENOSYS => "ENOSYS",
libc::ENOTSUP => "ENOTSUP",
_ => "UNKNOWN",
}
}
pub const PATH_SEPARATOR: &str = "/";
pub fn read_at(file: &File, buffer: &mut [u8], offset: u64) -> io::Result<usize> {
file.read_at(buffer, offset)
}
/// Modification time as seconds and nanoseconds since the epoch.
pub fn mtime(metadata: &Metadata) -> (i64, i64) {
(metadata.mtime(), metadata.mtime_nsec())
}
/// Device and inode, which identify a file across renames.
pub fn identity(metadata: &Metadata) -> (u64, u64) {
(metadata.dev(), metadata.ino())
}
/// Node's `os.tmpdir()`; Termux's Node falls back to `$PREFIX/tmp`.
pub fn tmpdir() -> String {
let configured = ["TMPDIR", "TMP", "TEMP"]
.iter()
.find_map(|key| std::env::var(key).ok().filter(|value| !value.is_empty()));
let mut path = configured.unwrap_or_else(|| {
if cfg!(target_os = "android") {
format!(
"{}/tmp",
std::env::var("PREFIX")
.unwrap_or_else(|_| "/data/data/com.termux/files/usr".into())
)
} else {
"/tmp".into()
}
});
if path.len() > 1 && path.ends_with('/') {
path.pop();
}
path
}
pub fn home() -> String {
std::env::var("HOME").unwrap_or_default()
}
/// `mkdtemp(3)` of `prefix`, as libuv calls it: six random characters from `[A-Za-z0-9]`.
pub fn mkdtemp(prefix: &str) -> io::Result<String> {
let mut template = prefix.as_bytes().to_vec();
template.extend_from_slice(b"XXXXXX");
let template =
CString::new(template).map_err(|_| io::Error::from_raw_os_error(libc::EINVAL))?;
let mut buffer = template.into_bytes_with_nul();
// SAFETY: `buffer` is a NUL-terminated, writable template that mkdtemp fills in place.
let created = unsafe { libc::mkdtemp(buffer.as_mut_ptr().cast()) };
if created.is_null() {
return Err(io::Error::last_os_error());
}
buffer.pop();
Ok(OsStr::from_bytes(&buffer).to_string_lossy().into_owned())
}
/// Open a regular file for positional reads, without blocking on FIFOs and, with `no_follow`, without following a
/// final symbolic link.
pub fn open_reader(path: &str, no_follow: bool) -> Result<File, Failure> {
let mut flags = libc::O_NONBLOCK;
if no_follow {
flags |= libc::O_NOFOLLOW;
}
OpenOptions::new()
.read(true)
.custom_flags(flags)
.open(path)
.map_err(|error| {
if no_follow && matches!(error.raw_os_error(), Some(libc::ELOOP) | Some(libc::EMLINK)) {
Failure::new("SYMLINK", "Refusing to follow a symbolic link").path(path)
} else {
Failure::io(&error, "open", path)
}
})
}
/// Open a file in one of Node's write modes.
pub fn open(path: &str, mode: OpenMode) -> io::Result<File> {
let mut options = OpenOptions::new();
match mode {
OpenMode::Write => options.write(true).create(true).truncate(true),
OpenMode::Append => options.append(true).create(true),
OpenMode::ReadWrite => options.read(true).write(true),
};
options.open(path)
}
pub fn realpath(path: &str) -> io::Result<String> {
std::fs::canonicalize(path).map(|resolved| resolved.to_string_lossy().into_owned())
}
pub fn rename(from: &str, to: &str) -> io::Result<()> {
std::fs::rename(from, to)
}
pub fn remove_file(path: &str) -> io::Result<()> {
std::fs::remove_file(path)
}
/// The error a recursive `mkdir` reports when a file is in the way of a parent.
pub fn not_a_directory() -> io::Error {
io::Error::from_raw_os_error(libc::ENOTDIR)
}
fn path_exists(path: &str) -> bool {
Path::new(path).symlink_metadata().is_ok()
}
fn which_bash() -> Option<String> {
let output = Command::new("which")
.arg("bash")
.stdin(Stdio::null())
.stderr(Stdio::null())
.output()
.ok()?;
if !output.status.success() {
return None;
}
let first = String::from_utf8_lossy(&output.stdout)
.trim()
.lines()
.next()?
.to_string();
path_exists(&first).then_some(first)
}
/// Node's shell resolution: a configured shell must exist; otherwise `/bin/bash`, `which bash`, then `sh`.
pub fn shell_config(shell_path: Option<&str>) -> Result<ShellConfig, Failure> {
let shell = |program: &str| ShellConfig {
program: program.into(),
args: vec!["-c".into()],
command_on_stdin: false,
};
if let Some(configured) = shell_path {
if path_exists(configured) {
return Ok(shell(configured));
}
return Err(Failure::new(
"shell_unavailable",
format!("Custom shell path not found: {configured}"),
));
}
if path_exists("/bin/bash") {
return Ok(shell("/bin/bash"));
}
Ok(shell(&which_bash().unwrap_or_else(|| "sh".into())))
}
/// Arguments, environment and process setup as libuv applies them.
pub fn configure(
command: &mut Command,
args: &[String],
env: &Map<String, Value>,
inherit_env: bool,
) {
command.args(args);
if !inherit_env {
command.env_clear();
}
for (key, value) in env {
if let Some(value) = value.as_str() {
command.env(key, value);
}
}
// SAFETY: only async-signal-safe calls between fork and exec.
unsafe {
command.pre_exec(|| {
// A new session, so a kill reaches every descendant; default signal handling and an empty mask, as libuv sets.
libc::setsid();
for signal in 1..32 {
if signal != libc::SIGKILL && signal != libc::SIGSTOP {
libc::signal(signal, libc::SIG_DFL);
}
}
let mut mask: libc::sigset_t = std::mem::zeroed();
libc::sigemptyset(&mut mask);
libc::sigprocmask(libc::SIG_SETMASK, &mask, std::ptr::null_mut());
Ok(())
});
}
}
pub fn after_spawn(_child: &Child) {}
/// Kill the command's process group, or the process if the group is gone.
pub fn kill_tree(pid: u32) {
let pid = pid as i32;
// SAFETY: plain signal delivery.
unsafe {
if libc::kill(-pid, libc::SIGKILL) != 0 {
libc::kill(pid, libc::SIGKILL);
}
}
}
/// A process killed by a signal has no exit code; report 128 + the signal, as the shell does.
pub fn exit_code(status: ExitStatus) -> i64 {
i64::from(
status
.code()
.unwrap_or_else(|| 128 + status.signal().unwrap_or(0)),
)
}

492
packages/env/daemon/src/sys/windows.rs vendored Normal file
View file

@ -0,0 +1,492 @@
//! Windows, following libuv (`src/win/error.c`, `src/win/process.c`, `src/win/fs.c`) and Node's `child_process`.
use super::{OpenMode, ShellConfig};
use crate::errors::Failure;
use serde_json::{Map, Value};
use std::collections::hash_map::RandomState;
use std::ffi::OsStr;
use std::fs::{File, Metadata};
use std::hash::{BuildHasher, Hasher};
use std::io;
use std::os::windows::ffi::OsStrExt;
use std::os::windows::fs::FileExt;
use std::os::windows::io::{AsRawHandle, FromRawHandle};
use std::os::windows::process::CommandExt;
use std::path::Path;
use std::process::{Child, Command, ExitStatus, Stdio};
use std::sync::OnceLock;
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::UNIX_EPOCH;
use windows_sys::Win32::Foundation::INVALID_HANDLE_VALUE;
use windows_sys::Win32::Storage::FileSystem::{
CREATE_ALWAYS, CreateFileW, FILE_APPEND_DATA, FILE_ATTRIBUTE_NORMAL,
FILE_FLAG_BACKUP_SEMANTICS, FILE_GENERIC_READ, FILE_GENERIC_WRITE, FILE_SHARE_DELETE,
FILE_SHARE_READ, FILE_SHARE_WRITE, FILE_WRITE_DATA, MOVEFILE_REPLACE_EXISTING, MoveFileExW,
OPEN_ALWAYS, OPEN_EXISTING,
};
use windows_sys::Win32::System::JobObjects::{
AssignProcessToJobObject, CreateJobObjectW, JOB_OBJECT_LIMIT_BREAKAWAY_OK,
JOB_OBJECT_LIMIT_DIE_ON_UNHANDLED_EXCEPTION, JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE,
JOB_OBJECT_LIMIT_SILENT_BREAKAWAY_OK, JOBOBJECT_EXTENDED_LIMIT_INFORMATION,
JobObjectExtendedLimitInformation, SetInformationJobObject,
};
use windows_sys::Win32::System::Threading::CREATE_NO_WINDOW;
/// libuv's `uv_translate_sys_error` for the errors file and process operations produce.
pub fn error_name(error: &io::Error) -> &'static str {
let Some(code) = error.raw_os_error() else {
return "UNKNOWN";
};
match code {
1 => "EISDIR", // ERROR_INVALID_FUNCTION
2 | 3 | 15 | 123 | 161 | 4392 => "ENOENT", // FILE_NOT_FOUND, PATH_NOT_FOUND, INVALID_DRIVE, INVALID_NAME, BAD_PATHNAME, INVALID_REPARSE_DATA
4 => "EMFILE", // ERROR_TOO_MANY_OPEN_FILES
5 | 1314 => "EPERM", // ACCESS_DENIED, PRIVILEGE_NOT_HELD
6 => "EBADF", // ERROR_INVALID_HANDLE
8 | 14 => "ENOMEM", // NOT_ENOUGH_MEMORY, OUTOFMEMORY
17 => "EXDEV", // ERROR_NOT_SAME_DEVICE
19 => "EROFS", // ERROR_WRITE_PROTECT
32 | 33 => "EBUSY", // SHARING_VIOLATION, LOCK_VIOLATION
39 | 82 | 112 => "ENOSPC", // HANDLE_DISK_FULL, CANNOT_MAKE, DISK_FULL
50 => "ENOTSUP", // ERROR_NOT_SUPPORTED
80 | 183 => "EEXIST", // FILE_EXISTS, ALREADY_EXISTS
87 | 131 | 1464 => "EINVAL", // INVALID_PARAMETER, NEGATIVE_SEEK, SYMLINK_NOT_SUPPORTED
145 => "ENOTEMPTY", // ERROR_DIR_NOT_EMPTY
206 => "ENAMETOOLONG", // ERROR_FILENAME_EXCED_RANGE
267 => "ENOTDIR", // ERROR_DIRECTORY
740 | 998 | 1920 => "EACCES", // ELEVATION_REQUIRED, NOACCESS, CANT_ACCESS_FILE
1921 => "ELOOP", // ERROR_CANT_RESOLVE_FILENAME
_ => "UNKNOWN",
}
}
pub const PATH_SEPARATOR: &str = "\\";
pub fn read_at(file: &File, buffer: &mut [u8], offset: u64) -> io::Result<usize> {
file.seek_read(buffer, offset)
}
pub fn mtime(metadata: &Metadata) -> (i64, i64) {
match metadata
.modified()
.ok()
.and_then(|time| time.duration_since(UNIX_EPOCH).ok())
{
Some(elapsed) => (elapsed.as_secs() as i64, i64::from(elapsed.subsec_nanos())),
None => (0, 0),
}
}
/// Windows file identities need a handle per file; watching compares kind, size and time instead.
pub fn identity(_metadata: &Metadata) -> (u64, u64) {
(0, 0)
}
/// Node's `os.tmpdir()` on Windows.
pub fn tmpdir() -> String {
let variable = |key: &str| std::env::var(key).ok().filter(|value| !value.is_empty());
let mut path = variable("TEMP")
.or_else(|| variable("TMP"))
.unwrap_or_else(|| {
format!(
"{}\\temp",
variable("SystemRoot")
.or_else(|| variable("windir"))
.unwrap_or_default()
)
});
if path.len() > 1 && path.ends_with('\\') && !path.ends_with(":\\") {
path.pop();
}
path
}
/// libuv's `uv_os_homedir`: `USERPROFILE` first.
pub fn home() -> String {
std::env::var("USERPROFILE").unwrap_or_default()
}
fn wide(text: &str) -> Vec<u16> {
OsStr::new(text)
.encode_wide()
.chain(std::iter::once(0))
.collect()
}
/// libuv's `fs__mkdtemp`: `prefix` plus six characters from `[a-zA-Z0-9]`, retried while the name exists.
pub fn mkdtemp(prefix: &str) -> io::Result<String> {
const CHARACTERS: &[u8] = b"abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789";
static COUNTER: AtomicU64 = AtomicU64::new(0);
let state = RandomState::new();
for _ in 0..1000 {
let mut hasher = state.build_hasher();
hasher.write_u64(COUNTER.fetch_add(1, Ordering::Relaxed));
let mut random = hasher.finish();
let mut path = prefix.to_string();
for _ in 0..6 {
path.push(CHARACTERS[(random % CHARACTERS.len() as u64) as usize] as char);
random /= CHARACTERS.len() as u64;
}
match std::fs::create_dir(&path) {
Ok(()) => return Ok(path),
Err(error) if error.kind() == io::ErrorKind::AlreadyExists => continue,
Err(error) => return Err(error),
}
}
Err(io::Error::from_raw_os_error(80))
}
/// Windows cannot open a directory as a file and has no `O_NOFOLLOW`: check the path first, as `NodeExecutionEnv` does
/// there (not race-free).
pub fn open_reader(path: &str, no_follow: bool) -> Result<File, Failure> {
if no_follow
&& std::fs::symlink_metadata(path).is_ok_and(|metadata| metadata.file_type().is_symlink())
{
return Err(Failure::new("SYMLINK", "Refusing to follow a symbolic link").path(path));
}
if std::fs::metadata(path).is_ok_and(|metadata| metadata.is_dir()) {
return Err(
Failure::new("EISDIR", "EISDIR: illegal operation on a directory, read").path(path),
);
}
File::open(path).map_err(|error| Failure::io(&error, "open", path))
}
/// libuv's `fs__open` for Node's write modes: `CreateFileW` with its access, sharing and disposition, always with
/// `FILE_FLAG_BACKUP_SEMANTICS`, so a directory opens and later calls on it fail as in Node; creating a file where a
/// directory exists reports `EISDIR`. (Rust's `File::create` opens with `OPEN_ALWAYS` and truncates afterwards.)
pub fn open(path: &str, mode: OpenMode) -> io::Result<File> {
let (access, disposition) = match mode {
OpenMode::Write => (FILE_GENERIC_WRITE, CREATE_ALWAYS),
OpenMode::Append => (
(FILE_GENERIC_WRITE & !FILE_WRITE_DATA) | FILE_APPEND_DATA,
OPEN_ALWAYS,
),
OpenMode::ReadWrite => (FILE_GENERIC_READ | FILE_GENERIC_WRITE, OPEN_EXISTING),
};
let name = wide(path);
// SAFETY: a NUL-terminated wide path; the returned handle is owned by the File on success.
let handle = unsafe {
CreateFileW(
name.as_ptr(),
access,
FILE_SHARE_READ | FILE_SHARE_WRITE | FILE_SHARE_DELETE,
std::ptr::null(),
disposition,
FILE_ATTRIBUTE_NORMAL | FILE_FLAG_BACKUP_SEMANTICS,
std::ptr::null_mut(),
)
};
if handle == INVALID_HANDLE_VALUE {
let error = io::Error::last_os_error();
// ERROR_FILE_EXISTS while creating means the path is a directory; ERROR_INVALID_FUNCTION names EISDIR.
if error.raw_os_error() == Some(80) && !matches!(mode, OpenMode::ReadWrite) {
return Err(io::Error::from_raw_os_error(1));
}
return Err(error);
}
// SAFETY: `handle` is a valid, owned file handle.
Ok(unsafe { File::from_raw_handle(handle as _) })
}
/// libuv's `uv_fs_realpath`: the final path without the `\\?\` prefix.
pub fn realpath(path: &str) -> io::Result<String> {
let resolved = std::fs::canonicalize(path)?.to_string_lossy().into_owned();
Ok(if let Some(unc) = resolved.strip_prefix("\\\\?\\UNC\\") {
format!("\\\\{unc}")
} else {
resolved
.strip_prefix("\\\\?\\")
.map(str::to_string)
.unwrap_or(resolved)
})
}
/// libuv's `fs__rename`: `MoveFileExW` replacing an existing destination.
pub fn rename(from: &str, to: &str) -> io::Result<()> {
let (from, to) = (wide(from), wide(to));
// SAFETY: both arguments are NUL-terminated wide strings that outlive the call.
if unsafe { MoveFileExW(from.as_ptr(), to.as_ptr(), MOVEFILE_REPLACE_EXISTING) } == 0 {
return Err(io::Error::last_os_error());
}
Ok(())
}
/// Node's `rm` on Windows retries a read-only file after making it writable.
pub fn remove_file(path: &str) -> io::Result<()> {
match std::fs::remove_file(path) {
Err(error) if error.kind() == io::ErrorKind::PermissionDenied => {
let mut permissions = std::fs::metadata(path)?.permissions();
#[allow(clippy::permissions_set_readonly_false)]
permissions.set_readonly(false);
std::fs::set_permissions(path, permissions)?;
std::fs::remove_file(path)
}
other => other,
}
}
/// The error a recursive `mkdir` reports when a file is in the way of a parent.
pub fn not_a_directory() -> io::Error {
io::Error::from_raw_os_error(267)
}
fn path_exists(path: &str) -> bool {
Path::new(path).symlink_metadata().is_ok()
}
/// Node's `isLegacyWslBashPath`: `^[a-z]:\\windows\\(system32|sysnative)\\bash\.exe$`, case-insensitive.
fn is_legacy_wsl_bash(path: &str) -> bool {
let normalized = path.replace('/', "\\").to_lowercase();
let mut characters = normalized.chars();
characters
.next()
.is_some_and(|drive| drive.is_ascii_lowercase())
&& matches!(
characters.as_str(),
":\\windows\\system32\\bash.exe" | ":\\windows\\sysnative\\bash.exe"
)
}
fn bash_config(program: &str) -> ShellConfig {
if is_legacy_wsl_bash(program) {
ShellConfig {
program: program.into(),
args: vec!["-s".into()],
command_on_stdin: true,
}
} else {
ShellConfig {
program: program.into(),
args: vec!["-c".into()],
command_on_stdin: false,
}
}
}
fn where_bash() -> Option<String> {
let output = Command::new("where")
.arg("bash.exe")
.creation_flags(CREATE_NO_WINDOW)
.stdin(Stdio::null())
.stderr(Stdio::null())
.output()
.ok()?;
if !output.status.success() {
return None;
}
let first = String::from_utf8_lossy(&output.stdout)
.trim()
.lines()
.next()?
.trim()
.to_string();
path_exists(&first).then_some(first)
}
/// Node's shell resolution on Windows: a configured shell, Git Bash, `where bash.exe`, or `shell_unavailable`.
pub fn shell_config(shell_path: Option<&str>) -> Result<ShellConfig, Failure> {
if let Some(configured) = shell_path {
if path_exists(configured) {
return Ok(bash_config(configured));
}
return Err(Failure::new(
"shell_unavailable",
format!("Custom shell path not found: {configured}"),
));
}
let mut candidates = Vec::new();
for variable in ["ProgramFiles", "ProgramFiles(x86)"] {
if let Ok(directory) = std::env::var(variable) {
candidates.push(format!("{directory}\\Git\\bin\\bash.exe"));
}
}
if let Some(found) = candidates.iter().find(|candidate| path_exists(candidate)) {
return Ok(bash_config(found));
}
if let Some(found) = where_bash() {
return Ok(bash_config(&found));
}
let searched: String = candidates
.iter()
.map(|candidate| format!(" {candidate}\n"))
.collect();
Err(Failure::new(
"shell_unavailable",
format!(
"No bash shell found. Options:\n 1. Install Git for Windows: https://git-scm.com/download/win\n 2. Add your bash to PATH (Cygwin, MSYS2, etc.)\n 3. Configure an explicit shellPath\n\nSearched Git Bash in:\n{}",
searched.trim_end()
),
))
}
/// libuv's `quote_cmd_arg`: how one argument is written into the command line the child splits again.
pub fn quote_argument(argument: &str) -> String {
if argument.is_empty() {
return "\"\"".into();
}
if !argument.contains([' ', '\t', '"']) {
return argument.into();
}
if !argument.contains(['"', '\\']) {
return format!("\"{argument}\"");
}
// Walk backwards: backslashes before a quote (or the closing quote) are doubled, and quotes are escaped.
let mut reversed = Vec::new();
let mut quote_hit = true;
for character in argument.chars().rev() {
reversed.push(character);
if quote_hit && character == '\\' {
reversed.push('\\');
} else if character == '"' {
quote_hit = true;
reversed.push('\\');
} else {
quote_hit = false;
}
}
let mut quoted = String::from("\"");
quoted.extend(reversed.into_iter().rev());
quoted.push('"');
quoted
}
/// Variables libuv copies from the parent into every child environment that lacks them.
const REQUIRED_VARIABLES: [&str; 11] = [
"HOMEDRIVE",
"HOMEPATH",
"LOGONSERVER",
"PATH",
"SYSTEMDRIVE",
"SYSTEMROOT",
"TEMP",
"USERDOMAIN",
"USERNAME",
"USERPROFILE",
"WINDIR",
];
/// The child environment as Node builds it on Windows: the variables in object order, sorted by UTF-16 code units,
/// keeping the first of names that differ only in case; then libuv adds required variables the result lacks.
fn child_environment(env: &Map<String, Value>, inherit_env: bool) -> Vec<(String, String)> {
let mut entries: Vec<(String, String)> = Vec::new();
let mut set =
|key: String, value: String| match entries.iter_mut().find(|(name, _)| *name == key) {
Some(entry) => entry.1 = value,
None => entries.push((key, value)),
};
if inherit_env {
for (key, value) in std::env::vars_os() {
set(
key.to_string_lossy().into_owned(),
value.to_string_lossy().into_owned(),
);
}
}
for (key, value) in env {
if let Some(value) = value.as_str() {
set(key.clone(), value.to_string());
}
}
entries.sort_by(|(a, _), (b, _)| a.encode_utf16().cmp(b.encode_utf16()));
let mut seen = std::collections::HashSet::new();
entries.retain(|(name, _)| seen.insert(name.to_uppercase()));
for required in REQUIRED_VARIABLES {
if !entries
.iter()
.any(|(name, _)| name.eq_ignore_ascii_case(required))
&& let Ok(value) = std::env::var(required)
{
entries.push((required.to_string(), value));
}
}
entries
}
/// Arguments quoted as libuv quotes them, Node's environment, and a hidden console.
pub fn configure(
command: &mut Command,
args: &[String],
env: &Map<String, Value>,
inherit_env: bool,
) {
for argument in args {
command.raw_arg(quote_argument(argument));
}
command.env_clear();
for (key, value) in child_environment(env, inherit_env) {
command.env(key, value);
}
command.creation_flags(CREATE_NO_WINDOW);
}
/// libuv puts every child it starts into one job that ends its processes when the daemon exits.
fn job() -> Option<isize> {
static JOB: OnceLock<Option<isize>> = OnceLock::new();
*JOB.get_or_init(|| {
// SAFETY: plain Win32 calls with a zeroed, correctly sized information struct.
unsafe {
let job = CreateJobObjectW(std::ptr::null(), std::ptr::null());
if job.is_null() {
return None;
}
let mut information: JOBOBJECT_EXTENDED_LIMIT_INFORMATION = std::mem::zeroed();
information.BasicLimitInformation.LimitFlags = JOB_OBJECT_LIMIT_BREAKAWAY_OK
| JOB_OBJECT_LIMIT_SILENT_BREAKAWAY_OK
| JOB_OBJECT_LIMIT_DIE_ON_UNHANDLED_EXCEPTION
| JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE;
SetInformationJobObject(
job,
JobObjectExtendedLimitInformation,
(&information as *const JOBOBJECT_EXTENDED_LIMIT_INFORMATION).cast(),
std::mem::size_of::<JOBOBJECT_EXTENDED_LIMIT_INFORMATION>() as u32,
);
Some(job as isize)
}
})
}
pub fn after_spawn(child: &Child) {
if let Some(job) = job() {
// SAFETY: both handles are valid for the duration of the call.
unsafe {
AssignProcessToJobObject(job as _, child.as_raw_handle() as _);
}
}
}
/// Node's `killProcessTree` on Windows: `taskkill /F /T`, not waited for.
pub fn kill_tree(pid: u32) {
let system_root = std::env::var("SystemRoot").unwrap_or_else(|_| "C:\\Windows".into());
let _ = Command::new(format!("{system_root}\\System32\\taskkill.exe"))
.args(["/F", "/T", "/PID", &pid.to_string()])
.creation_flags(CREATE_NO_WINDOW)
.stdin(Stdio::null())
.stdout(Stdio::null())
.stderr(Stdio::null())
.spawn();
}
/// Exit codes are unsigned 32-bit values on Windows (an access violation is 3221225477).
pub fn exit_code(status: ExitStatus) -> i64 {
i64::from(status.code().unwrap_or(1) as u32)
}
#[cfg(test)]
mod tests {
use super::quote_argument;
#[test]
fn quotes_arguments_like_libuv() {
assert_eq!(quote_argument(""), "\"\"");
assert_eq!(quote_argument("plain"), "plain");
assert_eq!(quote_argument("a b"), "\"a b\"");
assert_eq!(quote_argument("*.txt"), "*.txt");
assert_eq!(quote_argument("say \"hi\""), "\"say \\\"hi\\\"\"");
assert_eq!(
quote_argument("C:\\dir with space\\"),
"\"C:\\dir with space\\\\\""
);
assert_eq!(quote_argument("a\\\\b c"), "\"a\\\\b c\"");
}
}

82
packages/env/docs/protocol.md vendored Normal file
View file

@ -0,0 +1,82 @@
# pi-env wire protocol
Version 1. The client starts `pi-env serve --token <hex>` (over `ssh`, or directly in tests) and talks to it over the
process's stdin and stdout. stderr is diagnostic text for logs only.
## Sync
Shell startup files may print to stdout before the daemon runs. The daemon's first output is the line
`PI-ENV <token>\n`, with the token from its command line. The client discards everything before that line. After it,
stdout carries only frames.
## Frames
Every frame, in both directions:
```text
u32 length bytes after this field
u8 type
u32 id request id; 0 for frames that belong to no request
u32 jsonLength
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.
| 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 |
| 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.
## 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.
## 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.
File operations take `{ path }` (and the listed fields) and return `{}` unless noted:
| op | fields | result |
|---|---|---|
| `lstat` | | `info` |
| `realpath` | | `{ path }` |
| `write` | `append`, payload | creates missing parent directories like Node's recursive `mkdir` |
| `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 |
| `fstat` | `handle` | `info` |
| `scanLines` | `handle`, `startLine`, `endLine?` | `LineScan` |
| `opendir` | | `{ handle }` |
| `readdir` | `handle`, `max` | `{ entries, done }` |
| `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.
`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.

46
packages/env/docs/semantics.md vendored Normal file
View file

@ -0,0 +1,46 @@
# 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.
## Split of work
| Concern | Where |
|---|---|
| Path resolution: `~` (remote home), `file://`, relative to `cwd` | client, POSIX 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 |
| Line reader | client, positional reads and Durable's `StreamDecoder` |
| 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`) |
## Rules the standard libraries do not follow
- Recursive `mkdir`: an existing directory is fine, an existing file fails with `EEXIST`, a file in the way of a parent
fails with `ENOTDIR`. Writes create missing parents this way.
- `rm`: a missing path fails unless `force`; a directory without `recursive` fails with `ERR_FS_EISDIR`
(`FileError` code `unknown`).
- `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.
- `openBinaryReader`: nonblocking open; directories fail with `is_directory`, other non-regular files with `invalid`;
`noFollow` refuses a final symbolic link (`O_NOFOLLOW`).
- `readBinaryFile` returns a `Buffer`, reader reads a plain `Uint8Array`, as Node does.
## 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.
- 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.
- 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.

55
packages/env/package.json vendored Normal file
View file

@ -0,0 +1,55 @@
{
"name": "@earendil-works/pi-env",
"version": "1.0.3",
"description": "Remote execution environments for Pi Durable: a small daemon deployed over SSH and its ExecutionEnv client",
"type": "module",
"main": "./dist/index.js",
"types": "./dist/index.d.ts",
"exports": {
".": {
"source": "./src/index.ts",
"types": "./dist/index.d.ts",
"import": "./dist/index.js",
"default": "./dist/index.js"
}
},
"sideEffects": false,
"files": [
"dist",
"bin",
"docs",
"README.md",
"CHANGELOG.md"
],
"scripts": {
"clean": "shx rm -rf dist",
"build": "tsc -p tsconfig.build.json",
"build:daemon": "cargo build --manifest-path daemon/Cargo.toml --locked",
"test": "vitest --run",
"prepublishOnly": "npm run clean && npm run build"
},
"keywords": [
"durable-execution",
"agents",
"ssh",
"remote"
],
"author": "Earendil Works",
"license": "MIT",
"repository": {
"type": "git",
"url": "git+https://github.com/earendil-works/pi.git",
"directory": "packages/env"
},
"engines": {
"node": ">=22.19.0"
},
"dependencies": {
"@earendil-works/chord": "^1.0.3",
"@earendil-works/pi-durable": "^1.0.3"
},
"devDependencies": {
"shx": "0.4.0",
"vitest": "4.1.11"
}
}

249
packages/env/src/connection.ts vendored Normal file
View file

@ -0,0 +1,249 @@
import { type ChildProcessWithoutNullStreams, spawn } from "node:child_process";
import { randomBytes } from "node:crypto";
/** Frame types (docs/protocol.md). */
const REQUEST = 1;
const RESULT = 2;
const ERROR = 3;
const EVENT = 4;
const CANCEL = 5;
const PING = 6;
const MAX_FRAME = 16 * 1024 * 1024;
const PING_INTERVAL_MS = 5000;
/** The daemon pings every five seconds; this much silence means the connection is gone. */
const SILENCE_LIMIT_MS = 30_000;
export type Json = Record<string, unknown>;
/** A failure reported by the daemon, with its Node-style code. */
export class RemoteError extends Error {
readonly code: string;
readonly path?: string;
readonly fields: Json;
constructor(fields: Json) {
super(typeof fields.message === "string" ? fields.message : "Remote operation failed");
this.name = "RemoteError";
this.code = typeof fields.code === "string" ? fields.code : "unknown";
if (typeof fields.path === "string") this.path = fields.path;
this.fields = fields;
}
}
/** What `hello` reports about the remote machine. */
export interface RemoteInfo {
protocol: number;
version: string;
os: string;
arch: string;
home: string;
tmpdir: string;
/** Path separator: `/`, or `\\` on Windows. */
separator: string;
pid: number;
}
export interface ConnectionOptions {
/** Command that starts the daemon, before its `serve --token <hex>` arguments, e.g. `["ssh", "-T", "--", "host", "~/.pi/mobile/tools/pi-env"]`. */
command: readonly string[];
/** Receives the daemon's and the transport's diagnostic output. */
onLog?: (text: string) => void;
}
type Pending = {
resolve: (value: { json: Json; payload: Uint8Array }) => void;
reject: (error: Error) => void;
onEvent?: (json: Json, payload: Uint8Array) => void;
};
export interface RequestOptions {
payload?: Uint8Array;
/** Aborting sends `cancel`; the request still settles with the daemon's result or error. */
signal?: AbortSignal;
/** Progress events of the request, such as `exec` output. */
onEvent?: (json: Json, payload: Uint8Array) => void;
/** The request's id, for `kill`. */
onStart?: (id: number) => void;
}
function frame(type: number, id: number, json: Json, payload: Uint8Array = new Uint8Array(0)): Buffer {
const body = Buffer.from(JSON.stringify(json), "utf8");
const header = Buffer.alloc(13);
header.writeUInt32BE(9 + body.length + payload.length, 0);
header.writeUInt8(type, 4);
header.writeUInt32BE(id, 5);
header.writeUInt32BE(body.length, 9);
return Buffer.concat([header, body, payload]);
}
/**
* One connection to a pi-env daemon: started lazily on the first request and started again after it is lost. Requests
* in flight when it is lost fail with code `unknown`; for mutations their outcome is then unknown.
*/
export class Connection {
readonly #options: ConnectionOptions;
#child: ChildProcessWithoutNullStreams | undefined;
#ready: Promise<RemoteInfo> | undefined;
#pending = new Map<number, Pending>();
#nextId = 1;
#buffer: Buffer = Buffer.alloc(0);
#synced = false;
#token = "";
#lastSeen = 0;
#timer: ReturnType<typeof setInterval> | undefined;
#closed = false;
constructor(options: ConnectionOptions) {
this.#options = options;
}
/** Connect if needed and return what the daemon reported about its machine. */
info(): Promise<RemoteInfo> {
if (this.#closed) return Promise.reject(new RemoteError({ code: "unknown", message: "Connection closed" }));
this.#ready ??= this.#start();
return this.#ready;
}
async request(op: string, json: Json, options: RequestOptions = {}): Promise<{ json: Json; payload: Uint8Array }> {
await this.info();
return this.#send(op, json, options);
}
/** Stop the daemon; it kills everything it started. */
close(): void {
this.#closed = true;
this.#teardown(new RemoteError({ code: "unknown", message: "Connection closed" }));
}
/** Kill one `exec` without aborting it. */
kill(id: number): void {
this.#write(frame(CANCEL, id, { mode: "kill" }));
}
#send(op: string, json: Json, options: RequestOptions): Promise<{ json: Json; payload: Uint8Array }> {
const id = this.#nextId++;
return new Promise((resolve, reject) => {
const onAbort = () => this.#write(frame(CANCEL, id, {}));
const done = () => options.signal?.removeEventListener("abort", onAbort);
this.#pending.set(id, {
resolve: (value) => {
done();
resolve(value);
},
reject: (error) => {
done();
reject(error);
},
...(options.onEvent ? { onEvent: options.onEvent } : {}),
});
options.signal?.addEventListener("abort", onAbort, { once: true });
options.onStart?.(id);
this.#write(frame(REQUEST, id, { ...json, op }, options.payload));
if (options.signal?.aborted) onAbort();
});
}
#write(data: Buffer): void {
this.#child?.stdin.write(data);
}
async #start(): Promise<RemoteInfo> {
this.#token = randomBytes(16).toString("hex");
this.#synced = false;
this.#buffer = Buffer.alloc(0);
const [program, ...args] = this.#options.command;
if (program === undefined) throw new RemoteError({ code: "spawn_error", message: "No daemon command" });
const child = spawn(program, [...args, "serve", "--token", this.#token], { stdio: ["pipe", "pipe", "pipe"] });
this.#child = child;
const failed = new Promise<never>((_resolve, reject) => {
child.once("error", (error) => reject(new RemoteError({ code: "spawn_error", message: error.message })));
child.once("exit", (code) =>
reject(
new RemoteError({ code: "unknown", message: `pi-env exited with code ${code} before it was ready` }),
),
);
});
failed.catch(() => {});
child.stdout.on("data", (chunk: Buffer) => this.#onData(chunk));
child.stderr.on("data", (chunk: Buffer) => this.#options.onLog?.(chunk.toString("utf8")));
child.on("exit", () => {
if (this.#child === child)
this.#teardown(new RemoteError({ code: "unknown", message: "pi-env connection lost" }));
});
child.stdin.on("error", () => {});
this.#lastSeen = Date.now();
this.#timer = setInterval(() => {
this.#write(frame(PING, 0, {}));
if (Date.now() - this.#lastSeen > SILENCE_LIMIT_MS) {
this.#teardown(new RemoteError({ code: "unknown", message: "pi-env connection timed out" }));
}
}, PING_INTERVAL_MS);
this.#timer.unref();
const hello = this.#send("hello", { protocol: 1 }, {});
const { json } = await Promise.race([hello, failed]);
if (json.protocol !== 1)
throw new RemoteError({ code: "unknown", message: `Unsupported protocol ${json.protocol}` });
return json as unknown as RemoteInfo;
}
#teardown(error: Error): void {
clearInterval(this.#timer);
this.#timer = undefined;
const child = this.#child;
this.#child = undefined;
this.#ready = undefined;
child?.stdin.end();
child?.kill();
const pending = [...this.#pending.values()];
this.#pending.clear();
for (const request of pending) request.reject(error);
}
#onData(chunk: Buffer): void {
this.#lastSeen = Date.now();
this.#buffer = this.#buffer.length === 0 ? chunk : Buffer.concat([this.#buffer, chunk]);
if (!this.#synced) {
// Shell startup files may print before the daemon runs; skip everything before its sync line.
const marker = Buffer.from(`PI-ENV ${this.#token}\n`, "utf8");
const index = this.#buffer.indexOf(marker);
if (index === -1) {
if (this.#buffer.length > 1024 * 1024) this.#buffer = this.#buffer.subarray(-marker.length);
return;
}
const noise = this.#buffer.subarray(0, index).toString("utf8").trim();
if (noise !== "") this.#options.onLog?.(noise);
this.#buffer = this.#buffer.subarray(index + marker.length);
this.#synced = true;
}
while (this.#buffer.length >= 4) {
const length = this.#buffer.readUInt32BE(0);
if (length < 9 || length > MAX_FRAME) {
this.#teardown(new RemoteError({ code: "unknown", message: "Corrupt frame from pi-env" }));
return;
}
if (this.#buffer.length < 4 + length) return;
const type = this.#buffer.readUInt8(4);
const id = this.#buffer.readUInt32BE(5);
const jsonLength = this.#buffer.readUInt32BE(9);
const json = JSON.parse(this.#buffer.subarray(13, 13 + jsonLength).toString("utf8")) as Json;
// A plain Uint8Array copy, like `NodeExecutionEnv`'s reader results, not a view of the receive buffer.
const payload = new Uint8Array(this.#buffer.subarray(13 + jsonLength, 4 + length));
this.#buffer = this.#buffer.subarray(4 + length);
this.#dispatch(type, id, json, payload);
}
}
#dispatch(type: number, id: number, json: Json, payload: Uint8Array): void {
const pending = this.#pending.get(id);
if (type === EVENT) {
pending?.onEvent?.(json, payload);
return;
}
if (type !== RESULT && type !== ERROR) return;
if (pending === undefined) return;
this.#pending.delete(id);
if (type === RESULT) pending.resolve({ json, payload });
else pending.reject(new RemoteError(json));
}
}

3
packages/env/src/index.ts vendored Normal file
View file

@ -0,0 +1,3 @@
export { Connection, type ConnectionOptions, RemoteError, type RemoteInfo, type RequestOptions } from "./connection.ts";
export { PollingWatcher } from "./polling-watch.ts";
export { RemoteExecutionEnv, type RemoteExecutionEnvOptions } from "./remote-env.ts";

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

@ -0,0 +1,235 @@
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;
}
}

755
packages/env/src/remote-env.ts vendored Normal file
View file

@ -0,0 +1,755 @@
import { randomUUID } from "node:crypto";
import { type PlatformPath, posix, win32 } from "node:path";
import { fileURLToPath } from "node:url";
import type { Context } from "@earendil-works/chord";
import {
type BinaryReader,
type DirReader,
type ExecutionEnv,
ExecutionError,
err,
FileError,
type FileInfo,
type FileKind,
type FileWatcher,
type LineScan,
ok,
type Result,
type ShellExecOptions,
type ShellExecResult,
StreamDecoder,
type TextLine,
type TextLineReader,
type WatchChange,
type WatchTarget,
} from "@earendil-works/pi-durable/env";
import { type Connection, type Json, RemoteError, type RemoteInfo } from "./connection.ts";
import { PollingWatcher } from "./polling-watch.ts";
const MAX_TIMEOUT_MS = 2_147_483_647;
const MAX_TIMEOUT_SECONDS = MAX_TIMEOUT_MS / 1000;
/** Bytes per `pread` or `write` request; frames are at most 16 MiB. */
const TRANSFER_CHUNK = 8 * 1024 * 1024;
const LINE_CHUNK = 64 * 1024;
export interface RemoteExecutionEnvOptions {
connection: Connection;
/** The file namespace: equal ids see the same files (`FileSystem.id`), e.g. `pi-env:<env name>`. */
id: string;
cwd: string;
shellPath?: string;
/** Added to the remote environment of every command that inherits it, like `NodeExecutionEnv`'s `shellEnv`. */
shellEnv?: Record<string, string>;
/** Interval of the polling watcher. */
watchIntervalMs?: number;
}
/** An info record from the daemon. */
type RemoteFileInfo = { name: string; kind: FileKind | "other"; size: number; mtimeSec: number; mtimeNsec: number };
function abortResult<T>(signal: AbortSignal | undefined, path?: string): Result<T, FileError> | undefined {
return signal?.aborted ? err(new FileError("aborted", "aborted", path)) : undefined;
}
/** `NodeExecutionEnv`'s mapping of Node error codes to `FileError` codes. */
function toFileError(error: unknown, fallbackPath?: string): FileError {
if (error instanceof FileError) return error;
if (!(error instanceof RemoteError)) {
const cause = error instanceof Error ? error : new Error(String(error));
return new FileError("unknown", cause.message, fallbackPath, cause);
}
const path = error.path ?? fallbackPath;
switch (error.code) {
case "aborted":
return new FileError("aborted", error.message, path, error);
case "ENOENT":
return new FileError("not_found", error.message, path, error);
case "EACCES":
case "EPERM":
return new FileError("permission_denied", error.message, path, error);
case "ENOTDIR":
return new FileError("not_directory", error.message, path, error);
case "EISDIR":
return new FileError("is_directory", error.message, path, error);
case "EINVAL":
case "SYMLINK":
case "NOT_REGULAR":
return new FileError("invalid", error.message, path, error);
}
return new FileError("unknown", error.message, path, error);
}
/** Node's path rules of the remote system. */
export async function remotePath(connection: Connection): Promise<PlatformPath> {
return (await connection.info()).os === "windows" ? win32 : posix;
}
function toInfo(path: string, remote: RemoteFileInfo, paths: PlatformPath): Result<FileInfo, FileError> {
if (remote.kind === "other") return err(new FileError("invalid", "Unsupported file type", path));
return ok({
name: paths.basename(path),
path,
kind: remote.kind,
size: remote.size,
mtimeMs: remote.mtimeSec * 1000 + remote.mtimeNsec / 1e6,
});
}
class RemoteBinaryReader implements BinaryReader {
readonly #env: RemoteExecutionEnv;
readonly #handle: number;
readonly #path: string;
#closed = false;
constructor(env: RemoteExecutionEnv, handle: number, path: string) {
this.#env = env;
this.#handle = handle;
this.#path = path;
}
#closedResult<T>(): Result<T, FileError> | undefined {
return this.#closed ? err(new FileError("invalid", "Binary reader is closed", this.#path)) : undefined;
}
async info(context: Context): Promise<Result<FileInfo, FileError>> {
const early = abortResult<FileInfo>(context.abortSignal, this.#path) ?? this.#closedResult<FileInfo>();
if (early) return early;
try {
const { json } = await this.#env.connection.request("fstat", { handle: this.#handle });
return toInfo(this.#path, json as unknown as RemoteFileInfo, await remotePath(this.#env.connection));
} catch (error) {
return err(toFileError(error, this.#path));
}
}
async read(offset: number, length: number, context: Context): Promise<Result<Uint8Array, FileError>> {
const early = abortResult<Uint8Array>(context.abortSignal, this.#path) ?? this.#closedResult<Uint8Array>();
if (early) return early;
if (!Number.isSafeInteger(offset) || offset < 0 || !Number.isSafeInteger(length) || length < 0) {
return err(new FileError("invalid", "Offset and length must be non-negative safe integers", this.#path));
}
const chunks: Uint8Array[] = [];
let total = 0;
try {
while (total < length) {
const { payload } = await this.#env.connection.request("pread", {
handle: this.#handle,
offset: offset + total,
length: Math.min(length - total, TRANSFER_CHUNK),
});
const aborted = abortResult<Uint8Array>(context.abortSignal, this.#path);
if (aborted) return aborted;
if (payload.length === 0) break;
chunks.push(payload);
total += payload.length;
}
} catch (error) {
return err(toFileError(error, this.#path));
}
if (chunks.length === 1) return ok(chunks[0]!);
const bytes = new Uint8Array(total);
let position = 0;
for (const chunk of chunks) {
bytes.set(chunk, position);
position += chunk.length;
}
return ok(bytes);
}
async scanLines(
options: { startLine: number; endLine?: number },
context: Context,
): Promise<Result<LineScan, FileError>> {
const early = abortResult<LineScan>(context.abortSignal, this.#path) ?? this.#closedResult<LineScan>();
if (early) return early;
const { startLine, endLine } = options;
if (
!Number.isSafeInteger(startLine) ||
startLine < 0 ||
(endLine !== undefined && (!Number.isSafeInteger(endLine) || endLine <= startLine))
) {
return err(new FileError("invalid", "Invalid line range", this.#path));
}
try {
const { json } = await this.#env.connection.request(
"scanLines",
{ handle: this.#handle, startLine, ...(endLine === undefined ? {} : { endLine }) },
context.abortSignal ? { signal: context.abortSignal } : {},
);
return ok(json as unknown as LineScan);
} catch (error) {
return err(toFileError(error, this.#path));
}
}
async close(_context: Context): Promise<void> {
if (this.#closed) return;
this.#closed = true;
await this.#env.connection.request("close", { handle: this.#handle }).catch(() => undefined);
}
}
class RemoteDirReader implements DirReader {
readonly #env: RemoteExecutionEnv;
readonly #handle: number;
readonly #path: string;
#done = false;
#closed = false;
constructor(env: RemoteExecutionEnv, handle: number, path: string) {
this.#env = env;
this.#handle = handle;
this.#path = path;
}
async next(
maxEntries: number,
context: Context,
): Promise<Result<{ entries: FileInfo[]; done: boolean }, FileError>> {
const aborted = abortResult<{ entries: FileInfo[]; done: boolean }>(context.abortSignal, this.#path);
if (aborted) return aborted;
if (this.#closed) return err(new FileError("invalid", "Directory reader is closed", this.#path));
if (!Number.isSafeInteger(maxEntries) || maxEntries <= 0) {
return err(new FileError("invalid", "maxEntries must be a positive safe integer", this.#path));
}
if (this.#done) return ok({ entries: [], done: true });
try {
const { json } = await this.#env.connection.request("readdir", { handle: this.#handle, max: maxEntries });
const entries: FileInfo[] = [];
for (const entry of json.entries as { name: string; info?: RemoteFileInfo; error?: Json }[]) {
const paths = await remotePath(this.#env.connection);
const path = paths.resolve(this.#path, entry.name);
if (entry.error !== undefined) {
// Removed between enumeration and lstat: not part of the listing any more.
if (entry.error.code === "ENOENT") continue;
return err(toFileError(new RemoteError(entry.error), path));
}
const info = toInfo(path, entry.info!, paths);
if (info.ok) entries.push(info.value);
}
this.#done = json.done === true;
return ok({ entries, done: this.#done });
} catch (error) {
return err(toFileError(error, this.#path));
}
}
async close(_context: Context): Promise<void> {
if (this.#closed) return;
this.#closed = true;
await this.#env.connection.request("close", { handle: this.#handle }).catch(() => undefined);
}
}
/** `NodeTextLineReader` over positional reads of a remote file. */
class RemoteTextLineReader implements TextLineReader {
readonly #reader: RemoteBinaryReader;
readonly #path: string;
readonly #decoder = new StreamDecoder();
#offset = 0;
#buffered = "";
#ended = false;
#closed = false;
constructor(reader: RemoteBinaryReader, path: string) {
this.#reader = reader;
this.#path = path;
}
async readLine(context: Context): Promise<Result<TextLine | undefined, FileError>> {
const aborted = abortResult<TextLine | undefined>(context.abortSignal, this.#path);
if (aborted) return aborted;
if (this.#closed) return err(new FileError("invalid", "Text line reader is closed", this.#path));
while (true) {
const newline = this.#buffered.indexOf("\n");
if (newline !== -1) {
const text = this.#buffered.slice(0, newline);
this.#buffered = this.#buffered.slice(newline + 1);
return ok({ text, terminated: true });
}
if (this.#ended) {
if (this.#buffered.length === 0) return ok(undefined);
const text = this.#buffered;
this.#buffered = "";
return ok({ text, terminated: false });
}
const bytes = await this.#reader.read(this.#offset, LINE_CHUNK, context);
if (!bytes.ok) return bytes;
this.#offset += bytes.value.length;
if (bytes.value.length === 0) {
this.#buffered += this.#decoder.decode();
this.#ended = true;
} else {
this.#buffered += this.#decoder.decode(bytes.value);
}
}
}
async close(context: Context): Promise<void> {
if (this.#closed) return;
this.#closed = true;
this.#buffered = "";
await this.#reader.close(context);
}
}
/**
* An `ExecutionEnv` on another machine, reached through a pi-env daemon. Results match `NodeExecutionEnv` running on
* that machine: the daemon performs the system calls, and this class applies Node's path, error, and result rules.
*/
export class RemoteExecutionEnv implements ExecutionEnv {
readonly id: string;
cwd: string;
readonly connection: Connection;
readonly #shellPath: string | undefined;
readonly #shellEnv: Record<string, string> | undefined;
readonly #watchIntervalMs: number | undefined;
/** Running commands this environment started, for `cleanup()`. */
readonly #running = new Set<number>();
constructor(options: RemoteExecutionEnvOptions) {
this.id = options.id;
this.cwd = options.cwd;
this.connection = options.connection;
this.#shellPath = options.shellPath;
this.#shellEnv = options.shellEnv;
this.#watchIntervalMs = options.watchIntervalMs;
}
/** Node's `resolvePath` on the remote system, with its home directory. */
async #resolve(path: string): Promise<string> {
const { home, os } = await this.connection.info();
const windows = os === "windows";
const paths = windows ? win32 : posix;
let normalized = path;
if (normalized === "~") {
normalized = home;
} else if (normalized.startsWith("~/") || (windows && normalized.startsWith("~\\"))) {
normalized = paths.join(home, normalized.slice(2));
} else if (normalized.startsWith("file://")) {
try {
normalized = fileURLToPath(normalized, { windows });
} catch {
// Keep malformed URLs as ordinary paths, as Node does.
}
}
return paths.isAbsolute(normalized) ? paths.resolve(normalized) : paths.resolve(this.cwd, normalized);
}
async #fileOp<T>(
path: string,
context: Context,
run: (resolved: string) => Promise<T>,
check: "before" | "both" = "before",
): Promise<Result<T, FileError>> {
let resolved: string;
try {
resolved = await this.#resolve(path);
} catch (error) {
return err(toFileError(error, path));
}
const aborted = abortResult<T>(context.abortSignal, resolved);
if (aborted) return aborted;
try {
const value = await run(resolved);
if (check === "both") {
const after = abortResult<T>(context.abortSignal, resolved);
if (after) return after;
}
return ok(value);
} catch (error) {
return err(toFileError(error, resolved));
}
}
async absolutePath(path: string, _context: Context): Promise<Result<string, FileError>> {
try {
return ok(await this.#resolve(path));
} catch (error) {
return err(toFileError(error, path));
}
}
async joinPath(parts: string[], _context: Context): Promise<Result<string, FileError>> {
return ok((await remotePath(this.connection)).join(...parts));
}
async readTextFile(path: string, context: Context): Promise<Result<string, FileError>> {
const bytes = await this.readBinaryFile(path, context);
// Node's `readFile(path, "utf8")` keeps a byte-order mark, unlike `TextDecoder`.
return bytes.ok
? ok(Buffer.from(bytes.value.buffer, bytes.value.byteOffset, bytes.value.length).toString("utf8"))
: bytes;
}
async readBinaryFile(path: string, context: Context): Promise<Result<Uint8Array, FileError>> {
const opened = await this.openBinaryReader(path, undefined, context);
if (!opened.ok) return opened;
try {
const chunks: Uint8Array[] = [];
for (let offset = 0; ; ) {
const bytes = await opened.value.read(offset, TRANSFER_CHUNK, context);
if (!bytes.ok) return bytes;
if (bytes.value.length === 0) break;
chunks.push(bytes.value);
offset += bytes.value.length;
}
// Node's `readFile` returns a Buffer.
return ok(Buffer.concat(chunks));
} finally {
await opened.value.close(context);
}
}
async openBinaryReader(
path: string,
options: { noFollow?: boolean } | undefined,
context: Context,
): Promise<Result<BinaryReader, FileError>> {
const opened = await this.#fileOp(path, context, async (resolved) => {
const { json } = await this.connection.request("open", {
path: resolved,
noFollow: options?.noFollow === true,
});
return { handle: json.handle as number, resolved };
});
if (!opened.ok) return opened;
const reader = new RemoteBinaryReader(this, opened.value.handle, opened.value.resolved);
const aborted = abortResult<BinaryReader>(context.abortSignal, opened.value.resolved);
if (aborted) {
await reader.close(context);
return aborted;
}
return ok(reader);
}
async openTextLineReader(path: string, context: Context): Promise<Result<TextLineReader, FileError>> {
const opened = await this.openBinaryReader(path, undefined, context);
if (!opened.ok) return opened;
const reader = opened.value as RemoteBinaryReader;
return ok(new RemoteTextLineReader(reader, await this.#resolve(path)));
}
async readTextLines(
path: string,
options: { maxLines?: number } | undefined,
context: Context,
): Promise<Result<string[], FileError>> {
if (options?.maxLines !== undefined && options.maxLines <= 0) return ok([]);
const opened = await this.openTextLineReader(path, context);
if (!opened.ok) return opened;
const lines: string[] = [];
try {
while (options?.maxLines === undefined || lines.length < options.maxLines) {
const line = await opened.value.readLine(context);
if (!line.ok) return line;
if (line.value === undefined) break;
lines.push(line.value.text);
}
return ok(lines);
} finally {
await opened.value.close(context);
}
}
async #write(path: string, content: string | Uint8Array, append: boolean, context: Context) {
const bytes = typeof content === "string" ? Buffer.from(content, "utf8") : content;
return this.#fileOp(
path,
context,
async (resolved) => {
// Large contents go in pieces: the first write truncates (or appends), the rest append.
for (let offset = 0, first = true; first || offset < bytes.length; first = false) {
const piece = bytes.subarray(offset, offset + TRANSFER_CHUNK);
await this.connection.request("write", { path: resolved, append: append || !first }, { payload: piece });
offset += piece.length;
}
},
append ? "both" : "before",
);
}
writeFile(path: string, content: string | Uint8Array, context: Context): Promise<Result<void, FileError>> {
return this.#write(path, content, false, context);
}
appendFile(path: string, content: string | Uint8Array, context: Context): Promise<Result<void, FileError>> {
return this.#write(path, content, true, context);
}
async truncateFile(path: string, size: number, context: Context): Promise<Result<void, FileError>> {
const resolved = await this.#resolve(path);
const aborted = abortResult<void>(context.abortSignal, resolved);
if (aborted) return aborted;
if (!Number.isSafeInteger(size) || size < 0) {
return err(new FileError("invalid", "File size must be a non-negative safe integer", resolved));
}
return this.#fileOp(
path,
context,
async (target) => {
await this.connection.request("truncate", { path: target, size });
},
"both",
);
}
flushFile(path: string, context: Context): Promise<Result<void, FileError>> {
return this.#fileOp(
path,
context,
async (resolved) => {
await this.connection.request("fsync", { path: resolved });
},
"both",
);
}
async renameFile(sourcePath: string, destinationPath: string, context: Context): Promise<Result<void, FileError>> {
const source = await this.#resolve(sourcePath);
const destination = await this.#resolve(destinationPath);
const aborted = abortResult<void>(context.abortSignal, destination);
if (aborted) return aborted;
try {
await this.connection.request("rename", { path: source, to: destination });
return ok(undefined);
} catch (error) {
return err(toFileError(error, source));
}
}
async fileInfo(path: string, context: Context): Promise<Result<FileInfo, FileError>> {
const result = await this.#fileOp(path, context, async (resolved) => {
const { json } = await this.connection.request("lstat", { path: resolved });
return toInfo(resolved, json as unknown as RemoteFileInfo, await remotePath(this.connection));
});
return result.ok ? result.value : result;
}
listDir(path: string, context: Context): Promise<Result<FileInfo[], FileError>> {
return this.#fileOp(path, context, async (resolved) => {
const { json: opened } = await this.connection.request("opendir", { path: resolved });
const handle = opened.handle as number;
const entries: { name: string; info?: RemoteFileInfo; error?: Json }[] = [];
try {
for (let done = false; !done; ) {
if (context.abortSignal?.aborted) throw new FileError("aborted", "aborted", resolved);
const { json } = await this.connection.request("readdir", { handle, max: 1000 });
entries.push(...(json.entries as typeof entries));
done = json.done === true;
}
} finally {
await this.connection.request("close", { handle }).catch(() => undefined);
}
// Node's `readdir` fails on any entry it cannot lstat. libuv sorts names by bytes on POSIX; on Windows it keeps
// the file system's order, which the daemon reports.
const paths = await remotePath(this.connection);
if (paths === posix) {
entries.sort((a, b) => Buffer.compare(Buffer.from(a.name, "utf8"), Buffer.from(b.name, "utf8")));
}
const infos: FileInfo[] = [];
for (const entry of entries) {
const entryPath = paths.resolve(resolved, entry.name);
if (entry.error !== undefined) throw toFileError(new RemoteError(entry.error), entryPath);
const info = toInfo(entryPath, entry.info!, paths);
if (info.ok) infos.push(info.value);
}
return infos;
});
}
async openDirReader(path: string, context: Context): Promise<Result<DirReader, FileError>> {
const opened = await this.#fileOp(path, context, async (resolved) => {
const { json } = await this.connection.request("opendir", { path: resolved });
return { handle: json.handle as number, resolved };
});
if (!opened.ok) return opened;
const reader = new RemoteDirReader(this, opened.value.handle, opened.value.resolved);
const aborted = abortResult<DirReader>(context.abortSignal, opened.value.resolved);
if (aborted) {
await reader.close(context);
return aborted;
}
return ok(reader);
}
canonicalPath(path: string, context: Context): Promise<Result<string, FileError>> {
return this.#fileOp(path, context, async (resolved) => {
const { json } = await this.connection.request("realpath", { path: resolved });
return json.path as string;
});
}
async exists(path: string, context: Context): Promise<Result<boolean, FileError>> {
const result = await this.fileInfo(path, context);
if (result.ok) return ok(true);
if (result.error.code === "not_found") return ok(false);
return err(result.error);
}
createDir(
path: string,
options: { recursive?: boolean } | undefined,
context: Context,
): Promise<Result<void, FileError>> {
return this.#fileOp(path, context, async (resolved) => {
await this.connection.request("mkdir", { path: resolved, recursive: options?.recursive ?? true });
});
}
remove(
path: string,
options: { recursive?: boolean; force?: boolean } | undefined,
context: Context,
): Promise<Result<void, FileError>> {
return this.#fileOp(path, context, async (resolved) => {
await this.connection.request("rm", {
path: resolved,
recursive: options?.recursive ?? false,
force: options?.force ?? false,
});
});
}
async createTempDir(prefix: string | undefined, context: Context): Promise<Result<string, FileError>> {
const aborted = abortResult<string>(context.abortSignal);
if (aborted) return aborted;
try {
const { tmpdir } = await this.connection.info();
const paths = await remotePath(this.connection);
const { json } = await this.connection.request("mkdtemp", { path: paths.join(tmpdir, prefix ?? "tmp-") });
return ok(json.path as string);
} catch (error) {
return err(toFileError(error));
}
}
async createTempFile(
options: { prefix?: string; suffix?: string } | undefined,
context: Context,
): Promise<Result<string, FileError>> {
const dir = await this.createTempDir("tmp-", context);
if (!dir.ok) return dir;
const paths = await remotePath(this.connection);
const filePath = paths.join(dir.value, `${options?.prefix ?? ""}${randomUUID()}${options?.suffix ?? ""}`);
try {
await this.connection.request("write", { path: filePath, append: false }, { payload: new Uint8Array(0) });
return ok(filePath);
} catch (error) {
return err(toFileError(error, filePath));
}
}
async watch(
targets: readonly WatchTarget[],
onChange: (change: WatchChange) => void,
context: Context,
): Promise<Result<FileWatcher, FileError>> {
const aborted = abortResult<FileWatcher>(context.abortSignal);
if (aborted) return aborted;
try {
const resolved = await Promise.all(
targets.map(async (target) => ({ ...target, path: await this.#resolve(target.path) })),
);
return ok(await PollingWatcher.open(this, resolved, onChange, this.#watchIntervalMs));
} catch (error) {
return err(toFileError(error));
}
}
async exec(
command: string | readonly string[],
options: ShellExecOptions | undefined,
context: Context,
): Promise<Result<ShellExecResult, ExecutionError>> {
const signal = context.abortSignal;
if (signal?.aborted) return err(new ExecutionError("aborted", "aborted"));
const timeout = options?.timeout;
if (timeout !== undefined) {
if (!Number.isFinite(timeout) || timeout <= 0) {
return err(new ExecutionError("timeout", "Invalid timeout: must be a finite number of seconds"));
}
if (timeout * 1000 > MAX_TIMEOUT_MS) {
return err(new ExecutionError("timeout", `Invalid timeout: maximum is ${MAX_TIMEOUT_SECONDS} seconds`));
}
}
const inheritEnv = options?.inheritEnv ?? true;
let cwd: string;
try {
cwd = options?.cwd ? await this.#resolve(options.cwd) : this.cwd;
} catch (error) {
return err(new ExecutionError("unknown", error instanceof Error ? error.message : String(error)));
}
let callbackError: ExecutionError | undefined;
let settled = false;
let id: number | undefined;
const controller = new AbortController();
const onAbort = () => controller.abort();
signal?.addEventListener("abort", onAbort, { once: true });
try {
const { json } = await this.connection.request(
"exec",
{
...(typeof command === "string" ? { command } : { argv: [...command] }),
cwd,
env: inheritEnv ? { ...this.#shellEnv, ...options?.env } : { ...options?.env },
inheritEnv,
...(this.#shellPath === undefined ? {} : { shellPath: this.#shellPath }),
...(timeout === undefined ? {} : { timeoutMs: timeout * 1000 }),
...(options?.spill === undefined ? {} : { spill: options.spill }),
},
{
signal: controller.signal,
onStart: (requestId) => {
id = requestId;
this.#running.add(requestId);
},
onEvent: (event, payload) => {
if (settled || callbackError !== undefined || event.kind !== "output") return;
const text = Buffer.from(payload.buffer, payload.byteOffset, payload.length).toString("utf8");
if (text === "" || options?.onOutput === undefined) return;
try {
options.onOutput(text, context, { stream: event.stream === "stderr" ? "stderr" : "stdout" });
} catch (error) {
const cause = error instanceof Error ? error : new Error(String(error));
callbackError = new ExecutionError("callback_error", cause.message, cause);
controller.abort();
}
},
},
);
if (callbackError) return err(callbackError);
return ok({
exitCode: json.exitCode as number,
...(typeof json.spillPath === "string" ? { spillPath: json.spillPath } : {}),
});
} catch (error) {
if (callbackError) return err(callbackError);
if (!(error instanceof RemoteError)) {
return err(new ExecutionError("unknown", error instanceof Error ? error.message : String(error)));
}
const failure =
error.code === "timeout"
? new ExecutionError("timeout", `timeout:${timeout}`)
: error.code === "aborted"
? new ExecutionError("aborted", "aborted")
: error.code === "shell_unavailable" || error.code === "spawn_error"
? new ExecutionError(error.code, error.message)
: new ExecutionError("unknown", error.message);
if (typeof error.fields.spillPath === "string") failure.spillPath = error.fields.spillPath;
return err(failure);
} finally {
settled = true;
signal?.removeEventListener("abort", onAbort);
if (id !== undefined) this.#running.delete(id);
}
}
async cleanup(_context: Context): Promise<void> {
// Kill without aborting, so the commands settle with their killed status, as `NodeExecutionEnv` does.
for (const id of this.#running) this.connection.kill(id);
this.#running.clear();
}
}
export type { RemoteInfo };

37
packages/env/test/conformance.test.ts vendored Normal file
View file

@ -0,0 +1,37 @@
import { existsSync, mkdtempSync, rmSync } from "node:fs";
import { tmpdir } from "node:os";
import { join, resolve } from "node:path";
import { registerEnvConformance } from "@earendil-works/pi-durable/testing";
import { afterAll, describe, expect, it } from "vitest";
import { Connection } from "../src/connection.ts";
import { RemoteExecutionEnv } from "../src/remote-env.ts";
/** The daemon built by `cargo build` in ../daemon; tests talk to it over a pipe instead of SSH. */
const daemon = resolve(
import.meta.dirname,
`../daemon/target/debug/pi-env${process.platform === "win32" ? ".exe" : ""}`,
);
if (!existsSync(daemon))
throw new Error(`Build the daemon first: npm run build:daemon in packages/env (missing ${daemon})`);
const connection = new Connection({ command: [daemon] });
afterAll(() => connection.close());
const windows = process.platform === "win32";
const gitBash = join(process.env.ProgramFiles ?? "C:\\Program Files", "Git", "bin", "bash.exe");
registerEnvConformance(
{ describe, expect, it },
"RemoteExecutionEnv over a pipe",
async (use) => {
const cwd = mkdtempSync(join(tmpdir(), "pi-env-conformance-"));
try {
await use(new RemoteExecutionEnv({ connection, id: "pi-env:test", cwd, watchIntervalMs: 100 }));
} finally {
// On Windows a killed command's processes can hold the directory for a moment.
rmSync(cwd, { recursive: true, force: true, maxRetries: 50, retryDelay: 100 });
}
},
// Git Bash's `ln -s` copies instead of linking unless native symlinks are enabled.
windows ? { shell: [gitBash, "-c"], symlinks: false } : {},
);

293
packages/env/test/differential.test.ts vendored Normal file
View file

@ -0,0 +1,293 @@
import { mkdtempSync, realpathSync, rmSync } from "node:fs";
import { tmpdir } from "node:os";
import { join, resolve } from "node:path";
import type { Context } from "@earendil-works/chord";
import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context";
import type { ToolExecutionApi } from "@earendil-works/pi-durable";
import type { ExecutionEnv, FileError, Result } from "@earendil-works/pi-durable/env";
import { NodeExecutionEnv } from "@earendil-works/pi-durable/env/node";
import { createBashTool, createReadTool } from "@earendil-works/pi-durable/tools";
import { afterAll, describe, expect, it } from "vitest";
import { Connection } from "../src/connection.ts";
import { RemoteExecutionEnv } from "../src/remote-env.ts";
const daemon = resolve(
import.meta.dirname,
`../daemon/target/debug/pi-env${process.platform === "win32" ? ".exe" : ""}`,
);
const connection = new Connection({ command: [daemon] });
const context: Context = BACKGROUND_CONTEXT;
const dirs: string[] = [];
afterAll(() => {
connection.close();
for (const dir of dirs) rmSync(dir, { recursive: true, force: true, maxRetries: 50, retryDelay: 100 });
});
function pair(): { local: ExecutionEnv; remote: ExecutionEnv; roots: [string, string] } {
const localRoot = mkdtempSync(join(tmpdir(), "pi-env-local-"));
const remoteRoot = mkdtempSync(join(tmpdir(), "pi-env-remote-"));
dirs.push(localRoot, remoteRoot);
return {
local: new NodeExecutionEnv({ cwd: localRoot }),
remote: new RemoteExecutionEnv({ connection, id: "pi-env:test", cwd: remoteRoot }),
roots: [localRoot, remoteRoot],
};
}
function random(seed: number): () => number {
let state = seed >>> 0;
return () => {
state = (state + 0x6d2b79f5) >>> 0;
let t = state;
t = Math.imul(t ^ (t >>> 15), t | 1);
t ^= t + Math.imul(t ^ (t >>> 7), t | 61);
return ((t ^ (t >>> 14)) >>> 0) / 4294967296;
};
}
/** A result with the environment's root replaced and only the error code of failures, for comparison. */
function normalize(value: unknown, root: string): unknown {
const json = JSON.stringify(value, (key, field: unknown) => {
if (field instanceof Uint8Array) return { bytes: Buffer.from(field).toString("hex") };
if (key === "mtimeMs") return "mtime";
if (field && typeof field === "object" && "ok" in field && (field as Result<unknown, FileError>).ok === false) {
return { ok: false, code: (field as unknown as { error: FileError }).error.code };
}
return field;
});
if (json === undefined) return undefined;
// Roots appear JSON-escaped (backslashes doubled on Windows), and resolved paths spell them without Windows' 8.3
// short names (RUNNER~1).
const escaped = (path: string) => JSON.stringify(path).slice(1, -1);
return JSON.parse(json.replaceAll(escaped(realpathSync.native(root)), "<root>").replaceAll(escaped(root), "<root>"));
}
const NAMES = ["a.txt", "b", "dir", "dir/c.txt", "dir/sub", "dir/sub/d.bin", "missing", "a.txt/x"];
const CONTENTS: (string | Uint8Array)[] = [
"",
"hello\n",
"\ufeffbom\r\nline",
Uint8Array.from([0xe2, 0x82, 0x0a, 0xff, 0xef, 0xbb, 0xbf]),
"x".repeat(70_000),
];
type Operation = (env: ExecutionEnv) => Promise<unknown>;
/** A random operation with its arguments fixed, and a label for failure messages. */
function randomOperation(next: () => number): { label: string; run: Operation } {
const args: unknown[] = [];
const pick = <T>(value: T): T => {
args.push(
value instanceof Uint8Array
? "<bytes>"
: typeof value === "string" && value.length > 20
? `<${value.length} chars>`
: value,
);
return value;
};
const nameValue = () => pick(NAMES[Math.floor(next() * NAMES.length)]!);
const contentValue = () => pick(CONTENTS[Math.floor(next() * CONTENTS.length)]!);
const flagValue = () => pick(next() < 0.5);
const small = (limit: number) => pick(Math.floor(next() * limit));
const choices: [string, () => Operation][] = [
[
"writeFile",
() => {
const n = nameValue();
const c = contentValue();
return (env) => env.writeFile(n, c, context);
},
],
[
"appendFile",
() => {
const n = nameValue();
const c = contentValue();
return (env) => env.appendFile(n, c, context);
},
],
[
"readTextFile",
() => {
const n = nameValue();
return (env) => env.readTextFile(n, context);
},
],
[
"readBinaryFile",
() => {
const n = nameValue();
return (env) => env.readBinaryFile(n, context);
},
],
[
"readTextLines",
() => {
const n = nameValue();
const m = small(4);
return (env) => env.readTextLines(n, { maxLines: m }, context);
},
],
[
"fileInfo",
() => {
const n = nameValue();
return (env) => env.fileInfo(n, context);
},
],
[
"exists",
() => {
const n = nameValue();
return (env) => env.exists(n, context);
},
],
[
"listDir",
() => {
const n = nameValue();
return (env) => env.listDir(n, context);
},
],
[
"createDir",
() => {
const n = nameValue();
const r = flagValue();
return (env) => env.createDir(n, { recursive: r }, context);
},
],
[
"remove",
() => {
const n = nameValue();
const r = flagValue();
const f = flagValue();
return (env) => env.remove(n, { recursive: r, force: f }, context);
},
],
[
"renameFile",
() => {
const a = nameValue();
const b = nameValue();
return (env) => env.renameFile(a, b, context);
},
],
[
"truncateFile",
() => {
const n = nameValue();
const size = small(10);
return (env) => env.truncateFile(n, size, context);
},
],
[
"flushFile",
() => {
const n = nameValue();
return (env) => env.flushFile(n, context);
},
],
[
"canonicalPath",
() => {
const n = nameValue();
return (env) => env.canonicalPath(n, context);
},
],
[
"openBinaryReader",
() => {
const n = nameValue();
const noFollow = flagValue();
const offset = small(8);
const length = small(8);
return async (env) => {
const reader = await env.openBinaryReader(n, { noFollow }, context);
if (!reader.ok) return reader;
try {
return [
await reader.value.read(offset, length, context),
await reader.value.scanLines({ startLine: 0 }, context),
];
} finally {
await reader.value.close(context);
}
};
},
],
];
const [label, make] = choices[Math.floor(next() * choices.length)]!;
const run = make();
return { label: `${label}(${args.map((arg) => JSON.stringify(arg)).join(", ")})`, run };
}
describe("RemoteExecutionEnv against NodeExecutionEnv", () => {
it("gives the same results for random file operation sequences", async () => {
for (let seed = 1; seed <= 60; seed++) {
const next = random(seed);
const { local, remote, roots } = pair();
for (let step = 0; step < 25; step++) {
const operation = randomOperation(next);
const expected = normalize(await operation.run(local), roots[0]);
const actual = normalize(await operation.run(remote), roots[1]);
expect(actual, `seed ${seed} step ${step} ${operation.label}`).toEqual(expected);
}
}
}, 120_000);
it("gives the same read and bash tool results", async () => {
const { local, remote, roots } = pair();
const files: [string, string | Uint8Array][] = [
["text.txt", "one\ntwo\n\nthree"],
["big.txt", "line\n".repeat(3000)],
["long.txt", `${"é".repeat(40_000)}\nend`],
["bom.txt", "\ufeffhello"],
["bad.txt", Uint8Array.from([0x61, 0xe2, 0x82, 0x0a, 0x62])],
];
for (const [path, content] of files) {
await local.writeFile(path, content, context);
await remote.writeFile(path, content, context);
}
const api = (env: ExecutionEnv, output: string[]) =>
({
env,
outputWindow: undefined,
output: (chunk: string | Uint8Array) => output.push(String(chunk)),
diagnostic: () => {},
details: async () => {},
}) as unknown as ToolExecutionApi;
const run = async (
env: ExecutionEnv,
tool: ReturnType<typeof createReadTool> | ReturnType<typeof createBashTool>,
args: object,
) => {
const output: string[] = [];
try {
const result = await tool.execute(args as never, api(env, output), context);
return { result, output: output.join("") };
} catch (error) {
return { error: (error as Error).message, output: output.join("") };
}
};
const cases: [ReturnType<typeof createReadTool> | ReturnType<typeof createBashTool>, object][] = [
[createReadTool(), { path: "text.txt" }],
[createReadTool(), { path: "text.txt", offset: 2, limit: 1 }],
[createReadTool(), { path: "big.txt" }],
[createReadTool(), { path: "long.txt" }],
[createReadTool(), { path: "bom.txt" }],
[createReadTool(), { path: "bad.txt" }],
[createReadTool(), { path: "missing.txt" }],
[createBashTool(), { command: "cat text.txt; echo err >&2; exit 0" }],
[createBashTool(), { command: "exit 3" }],
[createBashTool(), { command: "printf 'a\\nb'; ls" }],
];
for (const [tool, args] of cases) {
const expected = normalize(await run(local, tool, args), roots[0]);
const actual = normalize(await run(remote, tool, args), roots[1]);
expect(actual, JSON.stringify(args)).toEqual(expected);
}
}, 60_000);
});

14
packages/env/tsconfig.build.json vendored Normal file
View file

@ -0,0 +1,14 @@
{
"extends": "../../tsconfig.base.json",
"compilerOptions": {
"outDir": "./dist",
"paths": {
"@earendil-works/chord": ["../chord/dist/index.d.ts"],
"@earendil-works/chord/context": ["../chord/dist/context/index.d.ts"],
"@earendil-works/pi-durable/env": ["../durable/dist/env/index.d.ts"]
},
"rootDir": "./src"
},
"include": ["src/**/*.ts"],
"exclude": ["node_modules", "dist", "**/*.d.ts", "src/**/*.d.ts"]
}

11
packages/env/vitest.config.ts vendored Normal file
View file

@ -0,0 +1,11 @@
import { defineConfig } from "vitest/config";
export default defineConfig({
test: {
environment: "node",
},
resolve: {
conditions: ["source"],
},
ssr: { resolve: { conditions: ["source"] } },
});

1
scripts/local-release.mjs
View file

@ -13,6 +13,7 @@ const packages = [
{ directory: "packages/mcp", name: "@earendil-works/pi-mcp" },
{ directory: "packages/ai", name: "@earendil-works/pi-ai" },
{ directory: "packages/durable", name: "@earendil-works/pi-durable" },
{ directory: "packages/env", name: "@earendil-works/pi-env" },
{ directory: "packages/tui", name: "@earendil-works/pi-tui" },
{ directory: "packages/agent", name: "@earendil-works/pi-agent-core" },
{ directory: "packages/protocol", name: "@earendil-works/pi-protocol" },

1
tsconfig.json
View file

@ -30,6 +30,7 @@
"./packages/durable/src/*/index.ts",
"./packages/durable/src/*"
],
"@earendil-works/pi-env": ["./packages/env/src/index.ts"],
"@earendil-works/pi-agent-core": ["./packages/agent/src/index.ts"],
"@earendil-works/pi-coding-agent": ["./packages/coding-agent/src/index.ts"],
"@earendil-works/pi-coding-agent/experimental/plugin": [