From 74c04cc3767d60ba51f3fb72cd3d93904edb3ec0 Mon Sep 17 00:00:00 2001 From: "Joshua J. Bouw" Date: Mon, 8 Jun 2026 00:03:08 +0400 Subject: [PATCH 1/2] test(process): persistent stdin acceptance + un-stub write-stdin/close-stdin WIT The id-keyed persistent write-stdin/close-stdin landed implemented in #867 but were never test-locked and the WIT still tagged them (NOT YET IMPLEMENTED). - Adds a registry test proving the acceptance: write-stdin delivers bytes to a persistent child, a second by-id write (needing only the registry + id, i.e. what a post-reset instance has) still reaches the same child, close-stdin yields a clean EOF exit, over-cap -> too-large, wrong-owner -> no-such-process, post-close -> closed. - Bumps the wit submodule to drop the stale (NOT YET IMPLEMENTED) tags on the id-keyed forms and correct the two persistent-tier banners (registry has landed). Restages wit-staging. The ProcessHandle (ephemeral) form, attach, and watch/unwatch stay deferred and keep their tags. --- CHANGELOG.md | 1 + .../host/process/persistent/registry_tests.rs | 129 ++++++++++++++++++ .../deps/astrid-process/process@1.0.0.wit | 20 ++- wit | 2 +- 4 files changed, 144 insertions(+), 8 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 1a7c1c227..e225fb2c6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,6 +15,7 @@ Changelog tracking starts with 0.2.0. Prior versions were not tracked. ### Changed +- **The `astrid:process` WIT now documents the id-keyed persistent `write-stdin` / `close-stdin` as implemented.** These landed in #867 — the persistent-process registry retains the child's stdin pipe host-side across pooled-instance resets (1 MiB-capped, backpressured, ownership-checked, audited) — but the WIT still tagged them `(NOT YET IMPLEMENTED)` and carried two stale `PERSISTENT TIER` "stubbed until the registry lands" banners. The tags are dropped and the banners corrected; function signatures are unchanged (doc-only). The ephemeral `ProcessHandle` form, `attach`, and `watch` / `unwatch` remain genuinely deferred and stay tagged. An acceptance test now locks the id-keyed path: deliver bytes → a by-id re-write (needing only registry + id, exactly what a post-reset instance holds) reaches the same child → `close-stdin` yields a clean EOF exit → over-cap returns `too-large` → wrong-owner returns `no-such-process` → post-close returns `closed`. Refs #870. - **Host-call concurrency is split into separate blocking vs async-I/O semaphores — the LLM-path throughput gate.** A single `host_semaphore` sized at `cores - 2` previously gated *every* host call: both the blocking ones that `block_in_place` + `block_on` and pin a tokio worker for the whole permit-held wait (KV, identity, sys, fs, the net/process security gates, DNS, sockets) **and** the async-I/O ones that `.await` real I/O and free the worker (HTTP request/stream, `ipc::recv`). The `cores - 2` cap is right for the blocking class — it must not approach the worker-pool size or blocking host work starves the scheduler — but it throttles outbound I/O far below what the host sustains, capping the LLM/HTTP path. `HostState` now carries two gates: `blocking_semaphore` (host-derived `cores - 2`, the same ceiling — but no longer contended by I/O calls, so strictly more headroom) and `io_semaphore` (host-derived, cores-scaled and clamped by half the process `RLIMIT_NOFILE` — floored at 1, so on an fd-scarce host the descriptor budget wins and the ceiling can fall below the `IO_MIN` floor, preventing `EMFILE` — since each in-flight async call may hold a socket fd). The four generic `bounded_*` host helpers are unchanged; each call site passes the semaphore matching its class (the `block_on` helpers take blocking, the `await` helpers take I/O). Net stays on the blocking semaphore for now — reclassifying its socket/accept/DNS paths, which do real I/O but currently use the blocking `block_on` helpers, to the async semaphore is a follow-up. Refs #816. - **Per-Store memory limiting now also meters per-principal peak usage.** The plain `wasmtime::StoreLimits` that capped each capsule instance's linear memory is replaced by a `StoreMemoryMeter` that keeps the same enforcement (the per-invocation `max_memory_bytes` ceiling) **and** records, into a kernel-owned shared `MemoryLedger`, the high-water linear-memory size each invoking principal grows a Store to — the RAM analogue of the per-principal `FuelLedger`. It is keyed by the same invoking principal (caller → owner → default) the fuel ledger charges, and re-targeted per invocation since a pooled Store is leased across principals. Sharded + atomic like the fuel ledger, so recording stays off the lock — and unlike the fuel ledger it is **bounded** (capped at a max principal count, evicting the lowest-peak entry when full) so a flood of ephemeral sub-agent principals cannot grow it without limit (`astrid#827`). Telemetry only — no deny path. Refs #816. diff --git a/crates/astrid-capsule/src/engine/wasm/host/process/persistent/registry_tests.rs b/crates/astrid-capsule/src/engine/wasm/host/process/persistent/registry_tests.rs index 99afe8e9f..ee3c89437 100644 --- a/crates/astrid-capsule/src/engine/wasm/host/process/persistent/registry_tests.rs +++ b/crates/astrid-capsule/src/engine/wasm/host/process/persistent/registry_tests.rs @@ -70,6 +70,39 @@ fn params( } } +/// Like [`spawn_raw`] but with stdin PIPED — the keep-stdin-open path. Returns +/// the write end so the test can hand it to the registry as the retained pipe. +fn spawn_raw_stdin( + script: &str, +) -> ( + tokio::process::Child, + tokio::process::ChildStdout, + tokio::process::ChildStderr, + tokio::process::ChildStdin, + u32, +) { + let mut std_cmd = std::process::Command::new("sh"); + std_cmd + .arg("-c") + .arg(script) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .stdin(Stdio::piped()); + #[cfg(unix)] + { + use std::os::unix::process::CommandExt as _; + std_cmd.process_group(0); + } + let mut cmd = tokio::process::Command::from(std_cmd); + cmd.kill_on_drop(true); + let mut child = cmd.spawn().expect("spawn test child"); + let pid = child.id().expect("child pid"); + let stdout = child.stdout.take().expect("stdout pipe"); + let stderr = child.stderr.take().expect("stderr pipe"); + let stdin = child.stdin.take().expect("stdin pipe"); + (child, stdout, stderr, stdin, pid) +} + #[tokio::test] async fn spawn_wait_read_and_owner_isolation() { let reg = PersistentProcessRegistry::new(tokio::runtime::Handle::current()); @@ -204,3 +237,99 @@ async fn read_since_is_non_draining_with_cursor() { reg.release(&id, &p, "cap").expect("release"); } + +#[tokio::test] +async fn write_stdin_delivers_survives_reset_and_close_eofs() { + use crate::engine::wasm::bindings::astrid::process::host::ErrorCode; + + let reg = PersistentProcessRegistry::new(tokio::runtime::Handle::current()); + let alice = PrincipalId::new("alice").unwrap(); + let bob = PrincipalId::new("bob").unwrap(); + + // A child that echoes each stdin line back on stdout and EXITS when stdin + // closes (`read` fails at EOF, ending the loop). `sh`'s builtin `echo` + // writes unbuffered, so each line surfaces on stdout immediately. + let (child, so, se, stdin, pid) = + spawn_raw_stdin("while IFS= read -r line; do echo \"got:$line\"; done"); + let spawn_params = SpawnParams { + creator: alice.clone(), + capsule_id: Arc::from("cap"), + command: "sh -c ".to_string(), + os_pid: pid, + child, + stdout: so, + stderr: se, + // keep-stdin-open: the registry retains the write end off-instance. + stdin: Some(stdin), + concurrent_cap: 8, + label: None, + overflow: None, + log_ring_bytes: None, + max_lifetime_ms: None, + idle_timeout_ms: None, + exit_retention_ms: None, + }; + let id = reg + .spawn(spawn_params) + .expect("spawn-persistent with stdin"); + + // Over-cap write is rejected before any I/O. + assert!(matches!( + reg.write_stdin(&id, &alice, "cap", &vec![0u8; super::MAX_STDIN_WRITE + 1]) + .await, + Err(ErrorCode::TooLarge) + )); + + // First write reaches the retained pipe; the child echoes it. + let n = reg + .write_stdin(&id, &alice, "cap", b"hello\n") + .await + .expect("write 1"); + assert_eq!(n, 6); + tokio::time::sleep(Duration::from_millis(250)).await; + let logs = reg.read_logs(&id, &alice, "cap").expect("read 1"); + assert!( + logs.stdout.contains("got:hello"), + "stdout: {:?}", + logs.stdout + ); + + // The second write is a standalone by-id call needing ONLY (registry, id) — + // no per-instance state — so it stands in for a write issued from a + // DIFFERENT pooled instance after an instance reset. The pipe lives in the + // host-owned registry, so it still reaches the same child. + let n2 = reg + .write_stdin(&id, &alice, "cap", b"world\n") + .await + .expect("write 2 (post-reset)"); + assert_eq!(n2, 6); + tokio::time::sleep(Duration::from_millis(250)).await; + let logs = reg.read_logs(&id, &alice, "cap").expect("read 2"); + assert!( + logs.stdout.contains("got:world"), + "stdout: {:?}", + logs.stdout + ); + + // A foreign principal cannot write — no oracle. + assert!(matches!( + reg.write_stdin(&id, &bob, "cap", b"x\n").await, + Err(ErrorCode::NoSuchProcess) + )); + + // close-stdin → child observes EOF → the read loop ends → clean exit. + reg.close_stdin(&id, &alice, "cap").expect("close-stdin"); + let exit = reg + .wait(&id, &alice, "cap", Duration::from_secs(5)) + .await + .expect("wait after close-stdin"); + assert_eq!(exit.exit_code, Some(0), "child exits cleanly on stdin EOF"); + + // Writing after close is rejected — the pipe is gone. + assert!(matches!( + reg.write_stdin(&id, &alice, "cap", b"late\n").await, + Err(ErrorCode::Closed) + )); + + reg.release(&id, &alice, "cap").expect("release"); +} diff --git a/crates/astrid-capsule/wit-staging/deps/astrid-process/process@1.0.0.wit b/crates/astrid-capsule/wit-staging/deps/astrid-process/process@1.0.0.wit index cbc27ee96..d8828f112 100644 --- a/crates/astrid-capsule/wit-staging/deps/astrid-process/process@1.0.0.wit +++ b/crates/astrid-capsule/wit-staging/deps/astrid-process/process@1.0.0.wit @@ -257,8 +257,8 @@ interface host { } // ================================================================ - // PERSISTENT TIER value types. (NOT YET IMPLEMENTED — the host stubs - // the persistent functions below; the SHAPES are fixed now.) + // PERSISTENT TIER value types. The persistent registry has landed; these + // shapes are in use. The SHAPES are frozen. // ================================================================ /// Opaque, unforgeable, principal-scoped identity for a PERSISTENT @@ -449,9 +449,11 @@ interface host { spawn-background: func(request: spawn-request) -> result; // ================================================================ - // PERSISTENT TIER functions. (NOT YET IMPLEMENTED — the host stubs - // these, returning `persist-unsupported` / `no-such-process` / `unknown` - // until the registry lands. The SHAPES are fixed now.) + // PERSISTENT TIER functions. The host registry has LANDED — these are + // implemented and audited. Still deferred (the host returns a stub + // `unknown`): `attach` (resource-handle materialisation — the id-keyed ops + // below ARE the documented `attach(id)?.method()` equivalent) and `watch` / + // `unwatch` (the lifecycle-event channel, RFC-open). The SHAPES are frozen. // // Every id-keyed function re-resolves the live calling principal + // capsule and checks them against the recorded creator BEFORE touching @@ -504,10 +506,14 @@ interface host { read-since: func(id: process-id, which-stream: log-stream, cursor: log-cursor, max-bytes: u32) -> result; /// `attach(id)?.write-stdin(data)`. Requires `keep-stdin-open = true`. - /// (NOT YET IMPLEMENTED — stdin pipe capture pending.) + /// Caps at 1 MiB/call (`too-large` above that); returns bytes written. The + /// retained pipe lives in the host registry, so a later instance that + /// re-attached keeps writing to the same child across pooled-instance + /// resets. A non-reading child backpressures the write (all-or-nothing per + /// call, so framed stdin is never torn). write-stdin: func(id: process-id, data: list) -> result; - /// `attach(id)?.close-stdin()`. (NOT YET IMPLEMENTED.) + /// `attach(id)?.close-stdin()`. Child observes EOF; idempotent. close-stdin: func(id: process-id) -> result<_, error-code>; /// `attach(id)?.signal(sig)`. Fire-and-forget signal delivery. diff --git a/wit b/wit index 9742f80e8..2fc197c5d 160000 --- a/wit +++ b/wit @@ -1 +1 @@ -Subproject commit 9742f80e83abb92e9ae954ef82664bdd2fcb89cd +Subproject commit 2fc197c5d3f43bd2536cebe508bb46e532dc4d6c From 1795c598ab6c564cfb139d97d07b7bfe680a55c2 Mon Sep 17 00:00:00 2001 From: "Joshua J. Bouw" Date: Mon, 8 Jun 2026 14:48:21 +0400 Subject: [PATCH 2/2] test(process): poll for stdin echo + correct write-stdin atomicity doc MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review feedback on the persistent-stdin finalize: - Replace the two fixed sleep(250ms) waits on the live child's echo with a bounded poll loop (drain_stdout_until) — robust under slow CI, instant on fast machines. Post-exit reads stay one-shot since wait() already fenced the flush. - Correct the write-stdin contract doc: it claimed "all-or-nothing per call, so framed stdin is never torn", but write_all over a pipe can deliver a prefix before erroring and then marks the stream closed. Pipe writes aren't atomic past PIPE_BUF, so the contract can't promise this. Restated accurately; wit-staging mirror + submodule pointer track the canonical (unicity-astrid/wit de89b14). --- .../host/process/persistent/registry_tests.rs | 45 +++++++++++++------ .../deps/astrid-process/process@1.0.0.wit | 14 +++--- wit | 2 +- 3 files changed, 41 insertions(+), 20 deletions(-) diff --git a/crates/astrid-capsule/src/engine/wasm/host/process/persistent/registry_tests.rs b/crates/astrid-capsule/src/engine/wasm/host/process/persistent/registry_tests.rs index ee3c89437..59765a3e1 100644 --- a/crates/astrid-capsule/src/engine/wasm/host/process/persistent/registry_tests.rs +++ b/crates/astrid-capsule/src/engine/wasm/host/process/persistent/registry_tests.rs @@ -103,6 +103,33 @@ fn spawn_raw_stdin( (child, stdout, stderr, stdin, pid) } +/// Poll the (draining) log reader until `needle` shows up on stdout, or the +/// ~2 s bound elapses; returns the accumulated stdout for the caller to assert +/// on. `read_logs` drains, so we accumulate across polls. This replaces a fixed +/// sleep that would race the reader tasks against the child's echo+flush — flaky +/// under slow CI, needlessly slow on fast machines. Returns the instant the +/// marker lands. Used only for output from a *live* child; post-exit reads stay +/// on a one-shot `read_logs` since `wait()` already fenced the flush. +async fn drain_stdout_until( + reg: &PersistentProcessRegistry, + id: &str, + principal: &PrincipalId, + capsule: &str, + needle: &str, +) -> String { + let mut stdout = String::new(); + for _ in 0..200 { + if let Ok(logs) = reg.read_logs(id, principal, capsule) { + stdout.push_str(&logs.stdout); + if stdout.contains(needle) { + break; + } + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + stdout +} + #[tokio::test] async fn spawn_wait_read_and_owner_isolation() { let reg = PersistentProcessRegistry::new(tokio::runtime::Handle::current()); @@ -286,13 +313,8 @@ async fn write_stdin_delivers_survives_reset_and_close_eofs() { .await .expect("write 1"); assert_eq!(n, 6); - tokio::time::sleep(Duration::from_millis(250)).await; - let logs = reg.read_logs(&id, &alice, "cap").expect("read 1"); - assert!( - logs.stdout.contains("got:hello"), - "stdout: {:?}", - logs.stdout - ); + let stdout = drain_stdout_until(®, &id, &alice, "cap", "got:hello").await; + assert!(stdout.contains("got:hello"), "stdout: {stdout:?}"); // The second write is a standalone by-id call needing ONLY (registry, id) — // no per-instance state — so it stands in for a write issued from a @@ -303,13 +325,8 @@ async fn write_stdin_delivers_survives_reset_and_close_eofs() { .await .expect("write 2 (post-reset)"); assert_eq!(n2, 6); - tokio::time::sleep(Duration::from_millis(250)).await; - let logs = reg.read_logs(&id, &alice, "cap").expect("read 2"); - assert!( - logs.stdout.contains("got:world"), - "stdout: {:?}", - logs.stdout - ); + let stdout = drain_stdout_until(®, &id, &alice, "cap", "got:world").await; + assert!(stdout.contains("got:world"), "stdout: {stdout:?}"); // A foreign principal cannot write — no oracle. assert!(matches!( diff --git a/crates/astrid-capsule/wit-staging/deps/astrid-process/process@1.0.0.wit b/crates/astrid-capsule/wit-staging/deps/astrid-process/process@1.0.0.wit index d8828f112..34aaf4de3 100644 --- a/crates/astrid-capsule/wit-staging/deps/astrid-process/process@1.0.0.wit +++ b/crates/astrid-capsule/wit-staging/deps/astrid-process/process@1.0.0.wit @@ -506,11 +506,15 @@ interface host { read-since: func(id: process-id, which-stream: log-stream, cursor: log-cursor, max-bytes: u32) -> result; /// `attach(id)?.write-stdin(data)`. Requires `keep-stdin-open = true`. - /// Caps at 1 MiB/call (`too-large` above that); returns bytes written. The - /// retained pipe lives in the host registry, so a later instance that - /// re-attached keeps writing to the same child across pooled-instance - /// resets. A non-reading child backpressures the write (all-or-nothing per - /// call, so framed stdin is never torn). + /// Caps at 1 MiB/call (`too-large` above that). The retained pipe lives in + /// the host registry, so a later instance that re-attached keeps writing to + /// the same child across pooled-instance resets. A non-reading child + /// backpressures the call until the OS pipe buffer drains. A successful + /// return wrote every byte of `data`, in order, and yields `len(data)`. An + /// error is NOT all-or-nothing: a prefix may already have reached the child, + /// and the stream is then marked closed (`closed` thereafter) — a caller + /// that needs intact framing must treat a failed write as a torn frame, not + /// a no-op. write-stdin: func(id: process-id, data: list) -> result; /// `attach(id)?.close-stdin()`. Child observes EOF; idempotent. diff --git a/wit b/wit index 2fc197c5d..de89b1454 160000 --- a/wit +++ b/wit @@ -1 +1 @@ -Subproject commit 2fc197c5d3f43bd2536cebe508bb46e532dc4d6c +Subproject commit de89b1454f1aa31c1165d9399da2e0f61b87a45e