test(process): persistent stdin acceptance + un-stub write-stdin/close-stdin WIT

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.
 

crates/astrid-capsule/src/engine/wasm/host/process/persistent/registry_tests.rs

@@ -70,6 +70,66 @@ 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)
+}
+
+/// 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());
@@ -204,3 +264,89 @@ 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 <stdin-echo>".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);
+    let stdout = drain_stdout_until(&reg, &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
+    // 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);
+    let stdout = drain_stdout_until(&reg, &id, &alice, "cap", "got:world").await;
+    assert!(stdout.contains("got:world"), "stdout: {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");
+}

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<process-handle, error-code>;
 
     // ================================================================
-    // 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,18 @@ interface host {
     read-since: func(id: process-id, which-stream: log-stream, cursor: log-cursor, max-bytes: u32) -> result<log-chunk, error-code>;
 
     /// `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). 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<u8>) -> result<u32, error-code>;
 
-    /// `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.