diff --git a/CHANGELOG.md b/CHANGELOG.md index e714817a1..7b3c3034d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -28,6 +28,7 @@ Changelog tracking starts with 0.2.0. Prior versions were not tracked. ### Added +- **`astrid:process@1.0.0` persistent-process tier — implemented.** A capsule can now spawn a background child that **outlives the pooled, stateless WASM instance** that started it. Previously an ephemeral `process-handle` is reaped when its instance resets on return to the dynamic pool, so a process started in one tool invocation could not survive to the next — the split `spawn → read → stop` pattern was impossible. A new host-owned `PersistentProcessRegistry` — cloned into every pooled `HostState` exactly like the cancellation `ProcessTracker`, so a `process-id` survives instance churn — owns the child (spawned on the daemon runtime under the same `bwrap`/Seatbelt sandbox as the ephemeral tier), its per-stream log rings, and its stdin pipe. **Implemented:** `spawn-persistent` (returns a 256-bit host-minted CSPRNG `process-id`, lowercase base32 so it doubles as an IPC topic suffix; the registry stores only a keyed BLAKE3 hash, never the raw token), `status` / `status-many` / `list-processes`, `read-logs` (drain) + `read-since` (non-draining, cursor-addressed, byte-faithful `list`), `signal` (incl. `stop`/`cont`), bounded `wait`, `stop` (SIGTERM→grace→SIGKILL, frees the slot), `release-process`, and `write-stdin` / `close-stdin` (via `keep-stdin-open` capture). Every id-keyed call re-resolves the live `(principal, capsule)` and checks it against the recorded creator, so a leaked id is inert across the principal/capsule boundary — unknown / wrong-owner / wrong-capsule / reaped all collapse to `no-such-process` with no oracle; `spawn-persistent` refuses the owner-fallback principal (`persist-unsupported`) so tenants never share a `default` namespace. Lifecycle is enforced by a per-capsule reaper task: per-principal concurrent + retained-id caps, idle / max-lifetime / exit-retention TTLs (guest values clamped DOWN to host ceilings), and a kill-all on capsule unload / daemon graceful shutdown. Works on Linux and macOS (the macOS caveat — a daemon *hard crash*, not a graceful shutdown, can orphan a still-sandboxed child because Seatbelt has no `die-with-parent` — is a weaker cleanup guarantee, not a containment gap). **Still deferred, honestly:** `attach` (the resource-handle composition sugar; the id-keyed ops are its documented `attach(id)?.method()` equivalent), `watch` / `unwatch` (host-published lifecycle events — an OPEN publish-authority question in RFC host_abi, with `status` + bounded `wait` polling as the working alternative), and the WIT's own `(NOT YET …)` items (resource-limit enforcement, `cpu-ms` / `mem-bytes-peak`, instance-local pollables). Contract: unicity-astrid/wit#12. Design: unicity-astrid/rfcs#22. Closes #866. - **Per-principal peak memory in usage reporting.** `ResourceUsage` gains `memory_bytes_peak_total: Option` — the cross-capsule high-water linear memory a principal has driven (max across every capsule it invokes), read from the shared memory ledger and surfaced by the admin `UsageGet`, `GET /api/sys/principals/{id}/usage` (a new field on the OpenAPI `ResourceUsageView`), and `astrid quota show` (a new "memory peak" row). This fills the memory side of per-principal usage, which previously reported only the per-instance ceiling. Under pooled, shared Stores a live "current" total is not cleanly attributable, so the **peak** is the reported signal — the principal that grows a Store owns the peak; `memory_bytes_current_total` stays `None`. Refs #816. - **Operator overrides for capsule runtime sizing (config + env + CLI).** New `[capsule]` config section with `host_blocking_concurrency`, `host_io_concurrency`, and `instance_pool_size` (all optional; unset → the host-derived default), the matching `ASTRID_CAPSULE_HOST_BLOCKING_CONCURRENCY` / `ASTRID_CAPSULE_HOST_IO_CONCURRENCY` / `ASTRID_CAPSULE_INSTANCE_POOL_SIZE` env vars, and `astrid-daemon --host-blocking-concurrency` / `--host-io-concurrency` / `--instance-pool-size` flags. Precedence is CLI flag > config file > env > host-derived default; the daemon resolves the values once at boot and the kernel forwards them, unmodified, to every capsule's `WasmEngine` (the same handle-plumbing shape as `FuelLedger`). A zero override is rejected at config-validation time (it would wedge a host-call class or leave a capsule with no instance to lease, rather than throttle). Refs #816. diff --git a/crates/astrid-capsule/src/engine/wasm/host/process/handle.rs b/crates/astrid-capsule/src/engine/wasm/host/process/handle.rs index 7763d0923..dc5934fd6 100644 --- a/crates/astrid-capsule/src/engine/wasm/host/process/handle.rs +++ b/crates/astrid-capsule/src/engine/wasm/host/process/handle.rs @@ -111,6 +111,8 @@ impl HostProcessHandle for HostState { ProcessSignal::Usr1 => nix::sys::signal::Signal::SIGUSR1, ProcessSignal::Usr2 => nix::sys::signal::Signal::SIGUSR2, ProcessSignal::Int => nix::sys::signal::Signal::SIGINT, + ProcessSignal::Stop => nix::sys::signal::Signal::SIGSTOP, + ProcessSignal::Cont => nix::sys::signal::Signal::SIGCONT, }; let raw = i32::try_from(pid).map_err(|_| ErrorCode::InvalidInput)?; nix::sys::signal::kill(nix::unistd::Pid::from_raw(raw), nix_sig) diff --git a/crates/astrid-capsule/src/engine/wasm/host/process/mod.rs b/crates/astrid-capsule/src/engine/wasm/host/process/mod.rs index a8dcbf01b..8b00b7f97 100644 --- a/crates/astrid-capsule/src/engine/wasm/host/process/mod.rs +++ b/crates/astrid-capsule/src/engine/wasm/host/process/mod.rs @@ -19,6 +19,7 @@ mod handle; mod managed; +mod persistent; mod tracker; use std::collections::VecDeque; @@ -30,12 +31,14 @@ use tracing::warn; use wasmtime::component::Resource; use crate::engine::wasm::bindings::astrid::process::host::{ - self as process, EnvVar, ErrorCode, ExitInfo, ProcessHandle, ProcessResult, SpawnRequest, + self as process, EnvVar, ErrorCode, ExitInfo, LogChunk, LogCursor, LogStream, ProcessHandle, + ProcessInfo, ProcessResult, ProcessSignal, ReadLogsResult, SpawnRequest, }; use crate::engine::wasm::host::util; use crate::engine::wasm::host_state::HostState; use managed::{ManagedProcess, attach_pipes, configure_piped, prepare_sandboxed_command}; +pub use persistent::PersistentProcessRegistry; pub use tracker::ProcessTracker; // Public so other crates (engine/init, hooks) can reference the type // even though the field has moved off HostState. @@ -44,6 +47,10 @@ pub use managed::ManagedProcess as PublicManagedProcess; /// Per-capsule hard ceiling on concurrent background processes. pub(crate) const MAX_BACKGROUND_PROCESSES: usize = 8; +/// Per-spawn stdin prelude cap (the WIT: `spawn-request.stdin` "Capped at +/// 4 MiB per spawn"). Oversized preludes are rejected with `too-large`. +const MAX_SPAWN_STDIN_BYTES: usize = 4 * 1024 * 1024; + /// Audit a process host fn invocation. fn audit_process( state: &HostState, @@ -74,6 +81,40 @@ fn audit_process( } } +/// Audit an id-keyed persistent-process op. Logs a short, non-reversible +/// hash of the `process-id` — never the raw token, per the WIT ("never the +/// raw id") — plus the op, principal, and capsule. +fn audit_process_id( + state: &HostState, + op: &'static str, + id: &str, + result: &Result, +) { + let id_hash = blake3::hash(id.as_bytes()).to_hex(); + let id = &id_hash[..16]; + let capsule_id = state.capsule_id.as_str(); + let principal = state.effective_principal(); + match result { + Ok(_) => tracing::debug!( + target: "astrid.audit.process", + %capsule_id, + %principal, + fn = op, + id, + "audit", + ), + Err(e) => tracing::debug!( + target: "astrid.audit.process", + %capsule_id, + %principal, + fn = op, + id, + error = ?e, + "audit", + ), + } +} + /// Extract the call_id from the caller's IPC context if it carried a /// `ToolExecuteRequest` payload. fn extract_call_id(state: &HostState) -> Option { @@ -96,6 +137,43 @@ fn env_summary(env: &[EnvVar]) -> String { .join(",") } +/// The AUTHENTICATED calling principal, or `None` when the call resolves to +/// the capsule-owner fallback (no caller in scope). `spawn-persistent` +/// refuses the fallback: a persistent id MUST be scoped to a real principal, +/// or unauthenticated paths would share one `default` namespace that +/// `list-processes` would enumerate across tenants. +fn authenticated_principal(state: &HostState) -> Option { + state + .caller_context + .as_ref() + .and_then(|m| m.principal.as_deref()) + .and_then(|p| astrid_core::principal::PrincipalId::new(p).ok()) +} + +/// Build the sandboxed `Child` for a persistent spawn: stdout/stderr piped, +/// stdin piped only when a prelude or `keep-stdin-open` needs it, own process +/// group (so signals reach descendants), `kill_on_drop` as the reap backstop. +fn build_persistent_child( + request: &SpawnRequest, + workspace_root: &std::path::Path, + want_stdin: bool, +) -> Result { + let mut sandboxed = prepare_sandboxed_command(&request.cmd, &request.args, workspace_root) + .map_err(|_| ErrorCode::InvalidInput)?; + // `configure_piped` sets the process group + stdout/stderr pipes. + configure_piped(&mut sandboxed); + if want_stdin { + sandboxed.stdin(Stdio::piped()); + } else { + sandboxed.stdin(Stdio::null()); + } + let mut tokio_cmd = TokioCommand::from(sandboxed); + tokio_cmd.kill_on_drop(true); + tokio_cmd + .spawn() + .map_err(|e| ErrorCode::Unknown(format!("spawn-persistent failed: {e}"))) +} + impl process::Host for HostState { fn spawn(&mut self, request: SpawnRequest) -> Result { let workspace_root = self.workspace_root.clone(); @@ -192,7 +270,13 @@ impl process::Host for HostState { .get(&principal) .copied() .unwrap_or(0); - if by_principal >= effective_cap || self.process_count_total >= MAX_BACKGROUND_PROCESSES { + // The per-principal concurrent cap is SHARED with the persistent tier: + // count this principal's live persistent processes too, so mixing the + // two tiers cannot exceed the cap. + let persistent_live = self.persistent_processes.live_count(&principal); + if by_principal + persistent_live >= effective_cap + || self.process_count_total >= MAX_BACKGROUND_PROCESSES + { return Err(ErrorCode::Quota); } @@ -288,4 +372,392 @@ impl process::Host for HostState { ); result } + + // ================================================================ + // PERSISTENT TIER — `astrid:process@1.0.0`. + // + // Backed by the host-owned `PersistentProcessRegistry` + // (`self.persistent_processes`), shared across the capsule's pooled + // instances so an id survives instance reset. Every id-keyed op + // re-resolves the live `(principal, capsule)` and checks it against the + // recorded creator inside the registry; unknown / wrong-owner / + // wrong-capsule / reaped collapse to `no-such-process` with no oracle. + // + // Still deferred (and honest about it): `attach` (resource-handle + // materialisation), `watch` / `unwatch` (host-published lifecycle events + // — an OPEN publish-authority question in RFC host_abi; `status` + bounded + // `wait` is the working alternative), and the `(NOT YET ...)` items the + // WIT itself flags (resource-limit enforcement, cpu/mem stats, pollables). + // ================================================================ + + fn spawn_persistent(&mut self, request: SpawnRequest) -> Result { + let cmd_for_audit = request.cmd.clone(); + let handle = self.runtime_handle.clone(); + let semaphore = self.blocking_semaphore.clone(); + + // Capability gate FIRST — a capsule lacking `host_process` gets + // `capability-denied` (consistent with `spawn` / `spawn-background` + // and the WIT "Security-gated" header), BEFORE any persistence- + // feasibility checks. Otherwise an ungranted capsule with no caller in + // scope would observe `persist-unsupported` instead of the capability + // error. + let Some(sec) = self.security.clone() else { + let result: Result = Err(ErrorCode::CapabilityDenied); + audit_process( + self, + "astrid:process/host.spawn-persistent", + &cmd_for_audit, + &result, + ); + return result; + }; + { + let cmd = request.cmd.to_string(); + let cid = self.capsule_id.as_str().to_owned(); + let check = util::bounded_block_on(&handle, &semaphore, async move { + sec.check_host_process(&cid, &cmd).await + }); + if check.is_err() { + let result: Result = Err(ErrorCode::CapabilityDenied); + audit_process( + self, + "astrid:process/host.spawn-persistent", + &cmd_for_audit, + &result, + ); + return result; + } + } + + // Persistence feasibility: refuse the owner-fallback principal — a + // persistent id must be scoped to an authenticated principal, else + // unauthenticated paths would share a `default` namespace that + // `list-processes` enumerates. + let Some(principal) = authenticated_principal(self) else { + return Err(ErrorCode::PersistUnsupported); + }; + // `some(0)` idle timeout is rejected per the WIT. + if request.idle_timeout_ms == Some(0) { + return Err(ErrorCode::InvalidInput); + } + if self.cancel_token.is_cancelled() { + return Err(ErrorCode::Cancelled); + } + + let capsule_id_arc: Arc = Arc::from(self.capsule_id.as_str()); + let workspace_root = self.workspace_root.clone(); + // Per-principal concurrent cap, SHARED with `spawn-background`: subtract + // this instance's live ephemeral handles so the registry's own check + // (`registry-live < effective`) bounds the COMBINED count to the cap. + let concurrent_cap = + usize::try_from(self.effective_profile().quotas.max_background_processes) + .unwrap_or(MAX_BACKGROUND_PROCESSES) + .min(MAX_BACKGROUND_PROCESSES); + let ephemeral_used = self + .process_count_by_principal + .get(&principal) + .copied() + .unwrap_or(0); + let effective_cap = concurrent_cap.saturating_sub(ephemeral_used); + + // Reject an oversized stdin prelude BEFORE spawning (avoids orphaning). + if request + .stdin + .as_ref() + .is_some_and(|s| s.len() > MAX_SPAWN_STDIN_BYTES) + { + let result: Result = Err(ErrorCode::TooLarge); + audit_process( + self, + "astrid:process/host.spawn-persistent", + &cmd_for_audit, + &result, + ); + return result; + } + + let want_stdin = request.keep_stdin_open.unwrap_or(false) || request.stdin.is_some(); + let mut child = match build_persistent_child(&request, &workspace_root, want_stdin) { + Ok(c) => c, + Err(e) => { + let result: Result = Err(e); + audit_process( + self, + "astrid:process/host.spawn-persistent", + &cmd_for_audit, + &result, + ); + return result; + }, + }; + // Reject a missing/zero pid: `killpg(0)` / `kill(0)` would target the + // daemon's OWN process group. A reaped child surfaces `None`; drop it + // (kill_on_drop reaps) and fail rather than store an unsignalable entry. + let Some(os_pid) = child.id().filter(|&p| p != 0) else { + let result: Result = Err(ErrorCode::Unknown( + "spawn-persistent: child has no usable pid".to_string(), + )); + audit_process( + self, + "astrid:process/host.spawn-persistent", + &cmd_for_audit, + &result, + ); + return result; + }; + let (Some(stdout), Some(stderr)) = (child.stdout.take(), child.stderr.take()) else { + return Err(ErrorCode::Unknown( + "spawn-persistent: missing stdio pipes".to_string(), + )); + }; + let mut stdin = child.stdin.take(); + + // Write the optional stdin prelude; on failure, fail the spawn (the + // child drops on return → kill_on_drop reaps the orphan). Retain the + // pipe ONLY when the guest asked to keep stdin open. + if let (Some(prelude), Some(pipe)) = (request.stdin.clone(), stdin.take()) { + let (pipe, write_res) = util::bounded_block_on(&handle, &semaphore, async move { + use tokio::io::AsyncWriteExt as _; + let mut pipe = pipe; + let r = pipe.write_all(&prelude).await; + (pipe, r) + }); + if write_res.is_err() { + let result: Result = Err(ErrorCode::Unknown( + "spawn-persistent: stdin prelude write failed".to_string(), + )); + audit_process( + self, + "astrid:process/host.spawn-persistent", + &cmd_for_audit, + &result, + ); + return result; + } + stdin = Some(pipe); + } + let stdin_for_registry = if request.keep_stdin_open.unwrap_or(false) { + stdin + } else { + None + }; + + let command = format!("{} {}", request.cmd, request.args.join(" ")); + let result = self.persistent_processes.spawn(persistent::SpawnParams { + creator: principal, + capsule_id: capsule_id_arc, + command, + os_pid, + child, + stdout, + stderr, + stdin: stdin_for_registry, + concurrent_cap: effective_cap, + label: request.label.clone(), + overflow: request.overflow, + log_ring_bytes: request.log_ring_bytes, + max_lifetime_ms: request.max_lifetime_ms, + idle_timeout_ms: request.idle_timeout_ms, + exit_retention_ms: request.exit_retention_ms, + }); + audit_process( + self, + "astrid:process/host.spawn-persistent", + &cmd_for_audit, + &result, + ); + result + } + + fn attach(&mut self, id: String) -> Result, ErrorCode> { + // Deferred: materialising a `process-handle` resource over a registry + // entry needs dual-typed dispatch in the resource table. The id-keyed + // free functions below ARE the documented `attach(id)?.method()` + // equivalents, so the persistent tier is fully usable without it. + let result: Result, ErrorCode> = Err(ErrorCode::Unknown( + "attach: resource-handle materialisation pending — use the id-keyed ops".to_string(), + )); + audit_process_id(self, "astrid:process/host.attach", &id, &result); + result + } + + fn list_processes( + &mut self, + label_filter: Option, + ) -> Result, ErrorCode> { + let principal = self.effective_principal(); + let capsule_id = self.capsule_id.as_str().to_owned(); + let result = + Ok(self + .persistent_processes + .list(&principal, &capsule_id, label_filter.as_deref())); + // Not id-keyed: audit the op + (non-secret) label filter, no id. + audit_process( + self, + "astrid:process/host.list-processes", + label_filter.as_deref().unwrap_or("*"), + &result, + ); + result + } + + fn status(&mut self, id: String) -> Result { + let principal = self.effective_principal(); + let capsule_id = self.capsule_id.as_str().to_owned(); + let result = self + .persistent_processes + .status(&id, &principal, &capsule_id); + audit_process_id(self, "astrid:process/host.status", &id, &result); + result + } + + fn status_many(&mut self, ids: Vec) -> Result, ErrorCode> { + let principal = self.effective_principal(); + let capsule_id = self.capsule_id.as_str().to_owned(); + let result = Ok(self + .persistent_processes + .status_many(&ids, &principal, &capsule_id)); + audit_process( + self, + "astrid:process/host.status-many", + &format!("{} ids", ids.len()), + &result, + ); + result + } + + fn read_logs(&mut self, id: String) -> Result { + let principal = self.effective_principal(); + let capsule_id = self.capsule_id.as_str().to_owned(); + let result = self + .persistent_processes + .read_logs(&id, &principal, &capsule_id); + audit_process_id(self, "astrid:process/host.read-logs", &id, &result); + result + } + + fn read_since( + &mut self, + id: String, + which_stream: LogStream, + cursor: LogCursor, + max_bytes: u32, + ) -> Result { + let principal = self.effective_principal(); + let capsule_id = self.capsule_id.as_str().to_owned(); + let result = self.persistent_processes.read_since( + &id, + &principal, + &capsule_id, + which_stream, + &cursor, + max_bytes, + ); + audit_process_id(self, "astrid:process/host.read-since", &id, &result); + result + } + + fn write_stdin(&mut self, id: String, data: Vec) -> Result { + let principal = self.effective_principal(); + let capsule_id = self.capsule_id.as_str().to_owned(); + let handle = self.runtime_handle.clone(); + let semaphore = self.io_semaphore.clone(); + let registry = self.persistent_processes.clone(); + let id_for_audit = id.clone(); + let result = util::bounded_block_on(&handle, &semaphore, async move { + registry + .write_stdin(&id, &principal, &capsule_id, &data) + .await + }); + audit_process_id( + self, + "astrid:process/host.write-stdin", + &id_for_audit, + &result, + ); + result + } + + fn close_stdin(&mut self, id: String) -> Result<(), ErrorCode> { + let principal = self.effective_principal(); + let capsule_id = self.capsule_id.as_str().to_owned(); + let result = self + .persistent_processes + .close_stdin(&id, &principal, &capsule_id); + audit_process_id(self, "astrid:process/host.close-stdin", &id, &result); + result + } + + fn signal(&mut self, id: String, sig: ProcessSignal) -> Result<(), ErrorCode> { + let principal = self.effective_principal(); + let capsule_id = self.capsule_id.as_str().to_owned(); + let result = self + .persistent_processes + .signal(&id, &principal, &capsule_id, sig); + audit_process_id(self, "astrid:process/host.signal", &id, &result); + result + } + + fn wait(&mut self, id: String, timeout_ms: u64) -> Result { + let principal = self.effective_principal(); + let capsule_id = self.capsule_id.as_str().to_owned(); + let handle = self.runtime_handle.clone(); + let semaphore = self.blocking_semaphore.clone(); + let cancel = self.cancel_token.clone(); + let registry = self.persistent_processes.clone(); + let timeout = std::time::Duration::from_millis(timeout_ms); + let id_for_audit = id.clone(); + let result = util::bounded_block_on_cancellable(&handle, &semaphore, &cancel, async move { + registry.wait(&id, &principal, &capsule_id, timeout).await + }) + .unwrap_or(Err(ErrorCode::Cancelled)); + audit_process_id(self, "astrid:process/host.wait", &id_for_audit, &result); + result + } + + fn stop(&mut self, id: String, grace_ms: Option) -> Result { + let principal = self.effective_principal(); + let capsule_id = self.capsule_id.as_str().to_owned(); + let handle = self.runtime_handle.clone(); + let semaphore = self.blocking_semaphore.clone(); + let cancel = self.cancel_token.clone(); + let registry = self.persistent_processes.clone(); + let grace = grace_ms.map(std::time::Duration::from_millis); + let id_for_audit = id.clone(); + let result = util::bounded_block_on_cancellable(&handle, &semaphore, &cancel, async move { + registry.stop(&id, &principal, &capsule_id, grace).await + }) + .unwrap_or(Err(ErrorCode::Cancelled)); + audit_process_id(self, "astrid:process/host.stop", &id_for_audit, &result); + result + } + + fn release_process(&mut self, id: String) -> Result<(), ErrorCode> { + let principal = self.effective_principal(); + let capsule_id = self.capsule_id.as_str().to_owned(); + let result = self + .persistent_processes + .release(&id, &principal, &capsule_id); + audit_process_id(self, "astrid:process/host.release-process", &id, &result); + result + } + + fn watch(&mut self, id: String, _suffix: Option) -> Result<(), ErrorCode> { + // Deferred by design: host-published lifecycle events raise an OPEN + // publish-authority question (manifest `[publish]` vs kernel-authored + // topic class) tracked in RFC host_abi. `status` + bounded `wait` is + // the working polling alternative until that resolves. + let result: Result<(), ErrorCode> = Err(ErrorCode::Unknown( + "watch: host lifecycle events deferred (publish-authority — RFC host_abi)".to_string(), + )); + audit_process_id(self, "astrid:process/host.watch", &id, &result); + result + } + + fn unwatch(&mut self, id: String) -> Result<(), ErrorCode> { + // Idempotent: nothing is armed while `watch` is deferred. + let result: Result<(), ErrorCode> = Ok(()); + audit_process_id(self, "astrid:process/host.unwatch", &id, &result); + result + } } diff --git a/crates/astrid-capsule/src/engine/wasm/host/process/persistent/config.rs b/crates/astrid-capsule/src/engine/wasm/host/process/persistent/config.rs new file mode 100644 index 000000000..328d86902 --- /dev/null +++ b/crates/astrid-capsule/src/engine/wasm/host/process/persistent/config.rs @@ -0,0 +1,116 @@ +//! Tunables and request-normalisation helpers for the persistent tier. +//! +//! Guest-supplied sizes / TTLs are clamped DOWN to host ceilings here so a +//! capsule can never request an unbounded ring, lifetime, or label. + +use std::time::Duration; + +use crate::engine::wasm::bindings::astrid::process::host::OverflowPolicy; + +use super::ring::Overflow; + +/// Default per-stream output ring capacity (stdout and stderr each). +const DEFAULT_LOG_RING_BYTES: usize = 1024 * 1024; +/// Hard ceiling on a guest-requested per-stream ring. +const MAX_LOG_RING_BYTES: usize = 8 * 1024 * 1024; +/// Floor on a guest-requested per-stream ring (a 0-byte ring is useless). +const MIN_LOG_RING_BYTES: usize = 4096; +/// Per-`write-stdin` call byte cap (matches the WIT contract). +pub(super) const MAX_STDIN_WRITE: usize = 1024 * 1024; +/// Per-principal RETAINED-id cap (live + exited-but-unreleased). Distinct +/// from the CONCURRENT cap (the profile's `max_background_processes`). +pub(super) const MAX_RETAINED_PER_PRINCIPAL: usize = 32; +/// Global registry-entry ceiling across all principals of one capsule. +pub(super) const MAX_REGISTRY_ENTRIES: usize = 256; +/// Default wall-clock lifetime ceiling, and the hard cap a guest request is +/// clamped DOWN to — a guest cannot request an unbounded lifetime. +const MAX_LIFETIME: Duration = Duration::from_secs(60 * 60 * 6); +/// Default idle reap interval (no read / wait / signal / write). +const DEFAULT_IDLE_TIMEOUT: Duration = Duration::from_secs(60 * 30); +/// Default post-exit retention of the id + log tail. +const DEFAULT_EXIT_RETENTION: Duration = Duration::from_secs(60 * 5); +/// Hard cap on post-exit retention. +const MAX_EXIT_RETENTION: Duration = Duration::from_secs(60 * 60); +/// SIGTERM→SIGKILL grace when `stop` is called with `grace-ms: none`. +pub(super) const DEFAULT_STOP_GRACE: Duration = Duration::from_secs(5); +/// Upper bound the `stop` grace is clamped to so a guest cannot pin a slot. +pub(super) const MAX_STOP_GRACE: Duration = Duration::from_secs(30); +/// Max bytes a single `read-since` chunk returns (host hard cap). +pub(super) const MAX_READ_SINCE_BYTES: usize = 4 * 1024 * 1024; +/// Operator label length clamp. +const MAX_LABEL_BYTES: usize = 128; + +/// Map the WIT `overflow-policy` (and its `none` default) to the internal +/// enum. +pub(super) fn overflow_from_wit(o: Option) -> Overflow { + match o { + Some(OverflowPolicy::Backpressure) => Overflow::Backpressure, + _ => Overflow::DropOldest, + } +} + +/// Clamp a guest-requested per-stream ring size to `[MIN, MAX]`, or the +/// default when unset. +pub(super) fn clamp_log_ring(bytes: Option) -> usize { + bytes + .map(|b| (b as usize).clamp(MIN_LOG_RING_BYTES, MAX_LOG_RING_BYTES)) + .unwrap_or(DEFAULT_LOG_RING_BYTES) +} + +/// Clamp a guest label (strip control chars, cap at `MAX_LABEL_BYTES` +/// **bytes**), or derive from `cmd`. The label is NOT an identity — only the +/// `process-id` is. Truncation is byte-aware (never splits a UTF-8 char) so a +/// non-ASCII label cannot exceed the documented byte ceiling. +pub(super) fn clamp_label(label: Option, cmd: &str) -> String { + let raw = label.unwrap_or_else(|| cmd.to_string()); + let mut out = String::with_capacity(MAX_LABEL_BYTES); + for c in raw.chars().filter(|c| !c.is_control()) { + if out.len() + c.len_utf8() > MAX_LABEL_BYTES { + break; + } + out.push(c); + } + out +} + +/// Resolve the effective `(lifetime, idle, retention)` durations from the +/// guest request, applying defaults and DOWN-clamping to host ceilings. +pub(super) fn resolve_ttls( + max_lifetime_ms: Option, + idle_timeout_ms: Option, + exit_retention_ms: Option, +) -> (Duration, Duration, Duration) { + let lifetime = max_lifetime_ms + .map(Duration::from_millis) + .unwrap_or(MAX_LIFETIME) + .min(MAX_LIFETIME); + let idle = idle_timeout_ms + .map(Duration::from_millis) + .unwrap_or(DEFAULT_IDLE_TIMEOUT) + .min(MAX_LIFETIME); + let retention = exit_retention_ms + .map(Duration::from_millis) + .unwrap_or(DEFAULT_EXIT_RETENTION) + .min(MAX_EXIT_RETENTION); + (lifetime, idle, retention) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn clamp_label_caps_bytes_not_chars() { + // Multi-byte chars must not push the label past the BYTE ceiling. + let long = "é".repeat(MAX_LABEL_BYTES); // 2 bytes each → 256 bytes + let clamped = clamp_label(Some(long), "cmd"); + assert!(clamped.len() <= MAX_LABEL_BYTES); + assert!(clamped.is_char_boundary(clamped.len())); // never split a char + } + + #[test] + fn clamp_label_strips_control_and_derives_from_cmd() { + assert_eq!(clamp_label(Some("a\nb\tc".into()), "cmd"), "abc"); + assert_eq!(clamp_label(None, "my-cmd"), "my-cmd"); + } +} diff --git a/crates/astrid-capsule/src/engine/wasm/host/process/persistent/entry.rs b/crates/astrid-capsule/src/engine/wasm/host/process/persistent/entry.rs new file mode 100644 index 000000000..942bc85c2 --- /dev/null +++ b/crates/astrid-capsule/src/engine/wasm/host/process/persistent/entry.rs @@ -0,0 +1,291 @@ +//! Per-process state, the monitor / reader tasks, and reaping primitives. + +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; + +use astrid_core::principal::PrincipalId; +use tokio::sync::watch; + +use crate::engine::wasm::bindings::astrid::process::host::{ + ErrorCode, ExitInfo, ProcessInfo, ProcessPhase, ProcessSignal, +}; + +use super::ring::{LogRing, Stream}; + +/// Terminal state recorded by the monitor task when the child exits. +#[derive(Clone, Copy)] +pub(super) struct ExitRecord { + pub(super) exit_code: Option, + pub(super) signal: Option, +} + +impl From for ExitInfo { + fn from(e: ExitRecord) -> Self { + ExitInfo { + exit_code: e.exit_code, + signal: e.signal, + } + } +} + +/// Lifecycle phase. Maps to the WIT `process-phase`, which deliberately has +/// no `reaped` — a reaped id resolves to `no-such-process` instead. The host +/// spawns synchronously, so it never reports the WIT's transient `starting`. +#[derive(Clone, Copy, PartialEq, Eq)] +pub(super) enum Phase { + Running, + Exited, +} + +impl From for ProcessPhase { + fn from(p: Phase) -> Self { + match p { + Phase::Running => ProcessPhase::Running, + Phase::Exited => ProcessPhase::Exited, + } + } +} + +/// Mutable inner state, guarded by one `Mutex` shared between the monitor +/// task, the reader tasks, and host calls. The lock is held only for short, +/// non-`await` critical sections. +pub(super) struct ProcessCore { + pub(super) phase: Phase, + pub(super) exit: Option, + pub(super) exited_at: Option, + pub(super) stdout: LogRing, + pub(super) stderr: LogRing, + pub(super) stdin: Option, + pub(super) stdin_open: bool, + pub(super) last_touch: Instant, +} + +/// One persistent process. Metadata is immutable after spawn; live state +/// lives behind `core`. +pub(super) struct PersistentEntry { + /// The raw `process-id`. Stored so `status` / `list-processes` can return + /// the reattach key (the WIT requires `process-info.id`); the map is still + /// *keyed* by the BLAKE3 hash of the id for lookup. Only ever returned to + /// the owning `(principal, capsule)` — never logged (audit uses a hash). + pub(super) id: String, + pub(super) creator: PrincipalId, + pub(super) capsule_id: Arc, + pub(super) label: String, + pub(super) command: String, + pub(super) os_pid: u32, + pub(super) spawned_at: Instant, + pub(super) max_lifetime: Duration, + pub(super) idle_timeout: Duration, + pub(super) exit_retention: Duration, + pub(super) core: Arc>, + /// Latches the exit so `wait` / `stop` await it without racing the + /// monitor task or holding the core lock across an `await`. + pub(super) exit_rx: watch::Receiver>, + /// The monitor task owning the `Child`. Aborting it drops the `Child`, + /// whose `kill_on_drop(true)` SIGKILLs the process — the reap backstop. + pub(super) monitor: tokio::task::JoinHandle<()>, +} + +impl PersistentEntry { + pub(super) fn is_live(&self) -> bool { + self.core + .lock() + .map(|c| c.phase != Phase::Exited) + .unwrap_or(false) + } + + /// Non-draining status snapshot. Returns the reattach `id` (the WIT + /// `process-info.id`); the caller is always the owning principal+capsule. + pub(super) fn info(&self) -> ProcessInfo { + let c = self.core.lock().unwrap_or_else(|e| e.into_inner()); + let now = Instant::now(); + let running = c.phase != Phase::Exited; + ProcessInfo { + id: self.id.clone(), + label: self.label.clone(), + command: self.command.clone(), + os_pid: running.then_some(self.os_pid), + phase: c.phase.into(), + exit: c.exit.map(Into::into), + age_ms: now.saturating_duration_since(self.spawned_at).as_millis() as u64, + idle_ms: now.saturating_duration_since(c.last_touch).as_millis() as u64, + buffered_bytes: (c.stdout.len() + c.stderr.len()) as u64, + bytes_dropped: c.stdout.overflow_dropped + c.stderr.overflow_dropped, + stdin_open: c.stdin_open, + cpu_ms: None, // (NOT YET POPULATED — see WIT) + mem_bytes_peak: None, // (NOT YET POPULATED — see WIT) + } + } +} + +/// Handles cloned out of an entry so a host call can act without holding the +/// registry map lock. +pub(super) struct Resolved { + pub(super) key: [u8; 32], + pub(super) core: Arc>, + pub(super) exit_rx: watch::Receiver>, + pub(super) os_pid: u32, +} + +/// Read the current exit (if any) without holding a lock across `await`. +pub(super) fn current_exit(core: &Arc>) -> Option { + core.lock().ok().and_then(|c| c.exit) +} + +/// Await the next non-`None` exit value on a watch receiver. +pub(super) async fn wait_for_exit( + rx: &mut watch::Receiver>, +) -> Option { + if let Some(e) = *rx.borrow() { + return Some(e); + } + loop { + if rx.changed().await.is_err() { + return None; + } + if let Some(e) = *rx.borrow() { + return Some(e); + } + } +} + +/// Spawn the monitor task that owns the `Child`, records its exit into +/// `core`, and notifies `exit_tx`. Returns the join handle (aborting it +/// drops the `Child`, whose `kill_on_drop` is the reap backstop). +pub(super) fn spawn_monitor( + runtime: &tokio::runtime::Handle, + mut child: tokio::process::Child, + core: Arc>, + exit_tx: watch::Sender>, +) -> tokio::task::JoinHandle<()> { + runtime.spawn(async move { + let status = child.wait().await; + let record = match status { + Ok(st) => ExitRecord { + exit_code: st.code(), + signal: exit_signal(&st), + }, + Err(_) => ExitRecord { + exit_code: Some(-1), + signal: None, + }, + }; + if let Ok(mut c) = core.lock() { + c.phase = Phase::Exited; + c.exit = Some(record); + c.exited_at = Some(Instant::now()); + c.stdin = None; + c.stdin_open = false; + } + let _ = exit_tx.send(Some(record)); + // `child` drops here: already exited, so `kill_on_drop` is a no-op. + // If the task is ABORTED before exit, that drop SIGKILLs instead. + }) +} + +#[cfg(unix)] +fn exit_signal(st: &std::process::ExitStatus) -> Option { + use std::os::unix::process::ExitStatusExt as _; + st.signal() +} + +#[cfg(not(unix))] +fn exit_signal(_st: &std::process::ExitStatus) -> Option { + None +} + +/// Reader read size. MUST be `<=` the minimum ring capacity +/// (`config::MIN_LOG_RING_BYTES` = 4096) so a full chunk always fits in an +/// *empty* `backpressure` ring — otherwise an over-cap chunk could never be +/// accepted (all-or-nothing) and the reader would spin forever. +pub(super) const READER_CHUNK_BYTES: usize = 4096; + +/// Spawn a reader task draining a child pipe into the in-core ring. Honors +/// `backpressure` by parking (not reading) when the ring is full — the OS +/// pipe fills and the child blocks on write; the WASM task is never parked. +pub(super) fn spawn_ring_reader( + runtime: &tokio::runtime::Handle, + mut pipe: R, + core: Arc>, + which: Stream, +) where + R: tokio::io::AsyncReadExt + Unpin + Send + 'static, +{ + runtime.spawn(async move { + let mut chunk = vec![0u8; READER_CHUNK_BYTES]; + loop { + match pipe.read(&mut chunk).await { + Ok(0) => break, + Ok(n) => { + let mut accepted = false; + while !accepted { + { + let mut c = core.lock().unwrap_or_else(|e| e.into_inner()); + let ring = match which { + Stream::Out => &mut c.stdout, + Stream::Err => &mut c.stderr, + }; + accepted = ring.push(&chunk[..n]); + } + if !accepted { + tokio::time::sleep(Duration::from_millis(25)).await; + } + } + }, + Err(_) => break, + } + } + }); +} + +/// Reap an entry removed from the map: SIGKILL the group (best effort) and +/// abort the monitor (dropping its `Child`, the `kill_on_drop` backstop). +pub(super) fn reap_entry(entry: PersistentEntry) { + if entry.is_live() { + let _ = send_signal(entry.os_pid, nix::sys::signal::Signal::SIGKILL); + } + entry.monitor.abort(); +} + +/// Map the WIT `process-signal` to a Unix signal. +pub(super) fn map_signal(sig: ProcessSignal) -> nix::sys::signal::Signal { + use nix::sys::signal::Signal; + match sig { + ProcessSignal::Term => Signal::SIGTERM, + ProcessSignal::Hup => Signal::SIGHUP, + ProcessSignal::Usr1 => Signal::SIGUSR1, + ProcessSignal::Usr2 => Signal::SIGUSR2, + ProcessSignal::Int => Signal::SIGINT, + ProcessSignal::Stop => Signal::SIGSTOP, + ProcessSignal::Cont => Signal::SIGCONT, + } +} + +/// Send a signal to the child's PROCESS GROUP (it is spawned with +/// `process_group(0)`, so descendants are signalled too), falling back to +/// the bare pid if the group send fails. +pub(super) fn send_signal(pid: u32, sig: nix::sys::signal::Signal) -> Result<(), ErrorCode> { + // Refuse pid 0: `killpg(0)` / `kill(0)` target the CALLER's (daemon's) own + // process group — never the child. A reaped child surfaces pid `None` + // (stored as 0); guard here as defense-in-depth (spawn also rejects it). + if pid == 0 { + return Err(ErrorCode::Closed); + } + #[cfg(unix)] + { + let raw = i32::try_from(pid).map_err(|_| ErrorCode::InvalidInput)?; + let target = nix::unistd::Pid::from_raw(raw); + if nix::sys::signal::killpg(target, sig).is_err() { + nix::sys::signal::kill(target, sig) + .map_err(|e| ErrorCode::Unknown(format!("signal {sig:?}: {e}")))?; + } + Ok(()) + } + #[cfg(not(unix))] + { + let _ = (pid, sig); + Err(ErrorCode::Unknown( + "process signals unsupported on this platform".to_string(), + )) + } +} diff --git a/crates/astrid-capsule/src/engine/wasm/host/process/persistent/ids.rs b/crates/astrid-capsule/src/engine/wasm/host/process/persistent/ids.rs new file mode 100644 index 000000000..dcd648db5 --- /dev/null +++ b/crates/astrid-capsule/src/engine/wasm/host/process/persistent/ids.rs @@ -0,0 +1,76 @@ +//! `process-id` minting and encoding. +//! +//! An id is 256 bits of OS entropy rendered as lowercase base32 (RFC 4648, +//! no padding). Lowercase base32 is a subset of the IPC topic-suffix grammar +//! (`[a-z0-9._-]+`), so the id doubles as a `watch` topic suffix without +//! sanitisation. Treat the wire form as an opaque secret. + +use rand::RngCore; +use rand::rngs::OsRng; + +const B32: &[u8; 32] = b"abcdefghijklmnopqrstuvwxyz234567"; + +/// Mint a fresh `process-id`: 256 bits of OS CSPRNG entropy, lowercase +/// base32. +pub(super) fn mint_id() -> String { + let mut bytes = [0u8; 32]; + OsRng.fill_bytes(&mut bytes); + base32_lower(&bytes) +} + +/// Lowercase base32 encode (RFC 4648 alphabet, no padding). +fn base32_lower(bytes: &[u8]) -> String { + let mut out = String::with_capacity(bytes.len() * 8 / 5 + 1); + let mut acc: u64 = 0; + let mut bits: u32 = 0; + for &b in bytes { + acc = (acc << 8) | u64::from(b); + bits += 8; + while bits >= 5 { + bits -= 5; + out.push(B32[((acc >> bits) & 0x1f) as usize] as char); + // Keep only the not-yet-emitted low bits so `acc` stays bounded. + acc &= (1u64 << bits) - 1; + } + } + if bits > 0 { + out.push(B32[((acc << (5 - bits)) & 0x1f) as usize] as char); + } + out +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn base32_is_lowercase_topic_safe() { + let s = base32_lower(&[0u8; 32]); + assert_eq!(s.len(), 52); // ceil(256 / 5) + assert!(s.chars().all(|c| B32.contains(&(c as u8)))); + // Subset of the topic-suffix grammar [a-z0-9._-]+. + assert!( + s.chars() + .all(|c| c.is_ascii_lowercase() || ('2'..='7').contains(&c)) + ); + } + + #[test] + fn base32_distinct_for_distinct_input() { + assert_ne!(base32_lower(&[1u8; 32]), base32_lower(&[2u8; 32])); + } + + #[test] + fn base32_known_vector() { + // All-zero input → all 'a'. + assert_eq!(base32_lower(&[0u8; 5]), "aaaaaaaa"); + } + + #[test] + fn mint_id_unique_and_sized() { + let a = mint_id(); + let b = mint_id(); + assert_ne!(a, b); + assert_eq!(a.len(), 52); + } +} diff --git a/crates/astrid-capsule/src/engine/wasm/host/process/persistent/mod.rs b/crates/astrid-capsule/src/engine/wasm/host/process/persistent/mod.rs new file mode 100644 index 000000000..7d3b9e34d --- /dev/null +++ b/crates/astrid-capsule/src/engine/wasm/host/process/persistent/mod.rs @@ -0,0 +1,584 @@ +//! `PersistentProcessRegistry` — host-owned storage for the PERSISTENT tier +//! of `astrid:process@1.0.0`. +//! +//! # Why this exists +//! +//! The EPHEMERAL tier (`spawn-background`) stores its `ManagedProcess` inside +//! the spawning instance's wasmtime resource table, so the child is reaped +//! when that instance is reset on return to the dynamic pool. A background +//! process started in one tool invocation therefore cannot survive to the +//! next — the split `spawn → read-logs → stop` pattern is impossible on a +//! pooled, stateless instance. +//! +//! The persistent tier relocates ownership OFF the instance: the child, its +//! log rings, and its stdin pipe live in this registry, which is an `Arc` +//! cloned into every pooled `HostState` of a capsule exactly like +//! [`ProcessTracker`](super::ProcessTracker). The registry outlives any +//! single instance, so a `process-id` minted on instance A is reattachable +//! from instance B. +//! +//! # Lifetime & reaping +//! +//! Reaped by explicit [`stop`](Self::stop) / [`release`](Self::release); the +//! per-entry idle / max-lifetime / exit-retention TTLs (the +//! [`reap_sweep`](Self::reap_sweep) the engine drives on a timer); or capsule +//! unload / daemon graceful shutdown ([`shutdown`](Self::shutdown)). NOT by +//! instance reset. +//! +//! On Linux the child is spawned under `bwrap --unshare-all --die-with-parent` +//! so the kernel reaps it even on a daemon SIGKILL. macOS Seatbelt has no +//! die-with-parent, so a daemon *hard crash* can orphan a still-sandboxed +//! child; graceful shutdown and capsule unload reap correctly. A weaker +//! cleanup guarantee, not a containment gap — the orphan stays inside its +//! Seatbelt profile. +//! +//! # Security model +//! +//! The `process-id` is a 256-bit host-minted CSPRNG token (lowercase base32, +//! see [`ids`]). The registry stores only a keyed BLAKE3 hash of the id, +//! never the raw token. Possession is necessary but NOT sufficient: every +//! id-keyed call re-resolves the live `(principal, capsule)` and checks them +//! against the recorded creator before touching the entry, so a leaked token +//! is inert across the principal/capsule boundary. Unknown / wrong-owner / +//! wrong-capsule / reaped / malformed all collapse to `no-such-process` with +//! no distinguishing oracle. + +mod config; +mod entry; +mod ids; +#[cfg(test)] +mod registry_tests; +mod ring; + +use std::collections::HashMap; +use std::sync::{Arc, Mutex, MutexGuard}; +use std::time::{Duration, Instant}; + +use astrid_core::principal::PrincipalId; +use rand::RngCore; +use rand::rngs::OsRng; +use tokio::io::AsyncWriteExt; +use tokio::sync::watch; + +use crate::engine::wasm::bindings::astrid::process::host::{ + ErrorCode, ExitInfo, LogChunk, LogCursor, LogStream, OverflowPolicy, ProcessInfo, + ProcessSignal, ReadLogsResult, +}; + +use config::{ + DEFAULT_STOP_GRACE, MAX_READ_SINCE_BYTES, MAX_REGISTRY_ENTRIES, MAX_RETAINED_PER_PRINCIPAL, + MAX_STDIN_WRITE, MAX_STOP_GRACE, clamp_label, clamp_log_ring, overflow_from_wit, resolve_ttls, +}; +use entry::{ + PersistentEntry, Phase, ProcessCore, current_exit, map_signal, reap_entry, send_signal, + spawn_monitor, spawn_ring_reader, wait_for_exit, +}; +use ids::mint_id; +use ring::{LogRing, Stream, decode_cursor, encode_cursor}; + +/// Keyed-hash map key derived from a `process-id` token. +type IdHash = [u8; 32]; + +/// Everything the registry needs to take ownership of a freshly-spawned +/// child. The caller (`spawn-persistent` host fn) has already run the +/// `host_process` capability gate and built the sandboxed `Child`; the +/// registry normalises the request knobs and enforces the caps. +pub(in crate::engine::wasm::host::process) struct SpawnParams { + pub(in crate::engine::wasm::host::process) creator: PrincipalId, + pub(in crate::engine::wasm::host::process) capsule_id: Arc, + /// cmd + args, as the capsule requested it (for display / label default). + pub(in crate::engine::wasm::host::process) command: String, + pub(in crate::engine::wasm::host::process) os_pid: u32, + pub(in crate::engine::wasm::host::process) child: tokio::process::Child, + pub(in crate::engine::wasm::host::process) stdout: tokio::process::ChildStdout, + pub(in crate::engine::wasm::host::process) stderr: tokio::process::ChildStderr, + pub(in crate::engine::wasm::host::process) stdin: Option, + /// Per-principal CONCURRENT cap (the profile's `max_background_processes`). + pub(in crate::engine::wasm::host::process) concurrent_cap: usize, + // ---- raw request knobs (normalised inside `spawn`) ---- + pub(in crate::engine::wasm::host::process) label: Option, + pub(in crate::engine::wasm::host::process) overflow: Option, + pub(in crate::engine::wasm::host::process) log_ring_bytes: Option, + pub(in crate::engine::wasm::host::process) max_lifetime_ms: Option, + pub(in crate::engine::wasm::host::process) idle_timeout_ms: Option, + pub(in crate::engine::wasm::host::process) exit_retention_ms: Option, +} + +/// Host-owned registry of a capsule's persistent processes. Cloned (`Arc`) +/// into every pooled `HostState` so an id survives instance churn. +pub struct PersistentProcessRegistry { + entries: Mutex>, + /// Per-registry random key for the keyed BLAKE3 id hash, so the stored + /// map keys are not a precomputable function of the token alone. + hash_key: [u8; 32], + /// The daemon runtime — children must NOT be owned by an instance's + /// executor or they would die with the instance. + runtime: tokio::runtime::Handle, +} + +impl std::fmt::Debug for PersistentProcessRegistry { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + let n = self.entries.lock().map(|e| e.len()).unwrap_or(0); + f.debug_struct("PersistentProcessRegistry") + .field("entries", &n) + .finish() + } +} + +impl PersistentProcessRegistry { + /// Construct a registry bound to a tokio runtime handle. + #[must_use] + pub fn new(runtime: tokio::runtime::Handle) -> Self { + let mut hash_key = [0u8; 32]; + OsRng.fill_bytes(&mut hash_key); + Self { + entries: Mutex::new(HashMap::new()), + hash_key, + runtime, + } + } + + fn lock(&self) -> MutexGuard<'_, HashMap> { + self.entries.lock().unwrap_or_else(|e| e.into_inner()) + } + + fn key_of(&self, id: &str) -> IdHash { + *blake3::keyed_hash(&self.hash_key, id.as_bytes()).as_bytes() + } + + /// Resolve a token to the shared handles needed to act WITHOUT holding + /// the map lock, IF the caller owns it. Touches the idle clock. + fn resolve( + &self, + id: &str, + principal: &PrincipalId, + capsule_id: &str, + ) -> Result { + let key = self.key_of(id); + let map = self.lock(); + let entry = map.get(&key).ok_or(ErrorCode::NoSuchProcess)?; + if &entry.creator != principal || &*entry.capsule_id != capsule_id { + return Err(ErrorCode::NoSuchProcess); + } + if let Ok(mut core) = entry.core.lock() { + core.last_touch = Instant::now(); + } + Ok(entry::Resolved { + key, + core: Arc::clone(&entry.core), + exit_rx: entry.exit_rx.clone(), + os_pid: entry.os_pid, + }) + } + + /// Spawn a persistent process, returning its `process-id`. Enforces the + /// concurrent + retained + global caps atomically under the map lock. + pub(in crate::engine::wasm::host::process) fn spawn( + &self, + p: SpawnParams, + ) -> Result { + let label = clamp_label(p.label, &p.command); + let log_ring = clamp_log_ring(p.log_ring_bytes); + let overflow = overflow_from_wit(p.overflow); + let (max_lifetime, idle_timeout, exit_retention) = + resolve_ttls(p.max_lifetime_ms, p.idle_timeout_ms, p.exit_retention_ms); + + let mut map = self.lock(); + if map.len() >= MAX_REGISTRY_ENTRIES { + return Err(ErrorCode::RegistryFull); + } + let (mut live, mut retained) = (0usize, 0usize); + for e in map.values() { + if e.creator == p.creator { + retained += 1; + if e.is_live() { + live += 1; + } + } + } + if retained >= MAX_RETAINED_PER_PRINCIPAL { + return Err(ErrorCode::RegistryFull); + } + if live >= p.concurrent_cap { + return Err(ErrorCode::Quota); + } + + let core = Arc::new(Mutex::new(ProcessCore { + phase: Phase::Running, + exit: None, + exited_at: None, + stdout: LogRing::new(log_ring, overflow), + stderr: LogRing::new(log_ring, overflow), + stdin: p.stdin, + stdin_open: false, + last_touch: Instant::now(), + })); + { + let mut c = core.lock().unwrap_or_else(|e| e.into_inner()); + c.stdin_open = c.stdin.is_some(); + } + spawn_ring_reader(&self.runtime, p.stdout, Arc::clone(&core), Stream::Out); + spawn_ring_reader(&self.runtime, p.stderr, Arc::clone(&core), Stream::Err); + let (exit_tx, exit_rx) = watch::channel::>(None); + let monitor = spawn_monitor(&self.runtime, p.child, Arc::clone(&core), exit_tx); + + let mut id = mint_id(); + let mut key = self.key_of(&id); + let mut tries = 0; + while map.contains_key(&key) { + id = mint_id(); + key = self.key_of(&id); + tries += 1; + if tries > 8 { + // 256-bit space: unreachable in practice. Fail closed. + monitor.abort(); + return Err(ErrorCode::Unknown( + "process-id collision space exhausted".to_string(), + )); + } + } + map.insert( + key, + PersistentEntry { + id: id.clone(), + creator: p.creator, + capsule_id: p.capsule_id, + label, + command: p.command, + os_pid: p.os_pid, + spawned_at: Instant::now(), + max_lifetime, + idle_timeout, + exit_retention, + core, + exit_rx, + monitor, + }, + ); + Ok(id) + } + + /// Number of LIVE (not-yet-exited) persistent processes a principal owns. + /// Lets the ephemeral `spawn-background` tier share the per-principal + /// concurrent cap with the persistent tier (and vice-versa). + pub(in crate::engine::wasm::host::process) fn live_count( + &self, + principal: &PrincipalId, + ) -> usize { + self.lock() + .values() + .filter(|e| &e.creator == principal && e.is_live()) + .count() + } + + /// Non-draining status snapshot of one process. + pub(in crate::engine::wasm::host::process) fn status( + &self, + id: &str, + principal: &PrincipalId, + capsule_id: &str, + ) -> Result { + let key = self.key_of(id); + let map = self.lock(); + let entry = map.get(&key).ok_or(ErrorCode::NoSuchProcess)?; + if &entry.creator != principal || &*entry.capsule_id != capsule_id { + return Err(ErrorCode::NoSuchProcess); + } + Ok(entry.info()) + } + + /// `status` for many ids in one pass; unknown / unowned ids are absent. + pub(in crate::engine::wasm::host::process) fn status_many( + &self, + ids: &[String], + principal: &PrincipalId, + capsule_id: &str, + ) -> Vec { + let map = self.lock(); + ids.iter() + .filter_map(|id| { + let entry = map.get(&self.key_of(id))?; + (entry.creator == *principal && &*entry.capsule_id == capsule_id) + .then(|| entry.info()) + }) + .collect() + } + + /// List the caller `(capsule, principal)`'s processes, optional label + /// substring filter. + pub(in crate::engine::wasm::host::process) fn list( + &self, + principal: &PrincipalId, + capsule_id: &str, + label_filter: Option<&str>, + ) -> Vec { + let map = self.lock(); + map.values() + .filter(|e| e.creator == *principal && &*e.capsule_id == capsule_id) + .filter(|e| label_filter.is_none_or(|f| e.label.contains(f))) + .map(PersistentEntry::info) + .collect() + } + + /// Drain both rings (the `read-logs` semantics). + pub(in crate::engine::wasm::host::process) fn read_logs( + &self, + id: &str, + principal: &PrincipalId, + capsule_id: &str, + ) -> Result { + let r = self.resolve(id, principal, capsule_id)?; + let mut core = r.core.lock().unwrap_or_else(|e| e.into_inner()); + let stdout = String::from_utf8_lossy(&core.stdout.drain()).into_owned(); + let stderr = String::from_utf8_lossy(&core.stderr.drain()).into_owned(); + let running = core.phase != Phase::Exited; + let exit = core.exit.map(Into::into); + Ok(ReadLogsResult { + stdout, + stderr, + running, + exit, + }) + } + + /// Non-draining, cursor-addressed read over one stream. + pub(in crate::engine::wasm::host::process) fn read_since( + &self, + id: &str, + principal: &PrincipalId, + capsule_id: &str, + which: LogStream, + cursor: &LogCursor, + max_bytes: u32, + ) -> Result { + let r = self.resolve(id, principal, capsule_id)?; + let max = (max_bytes as usize).min(MAX_READ_SINCE_BYTES); + let from = decode_cursor(cursor)?; + let core = r.core.lock().unwrap_or_else(|e| e.into_inner()); + let ring = match which { + LogStream::Stdout => &core.stdout, + LogStream::Stderr => &core.stderr, + }; + let exited = core.phase == Phase::Exited; + let (data, next, dropped) = ring.read_since(from, max); + let drained_eof = exited && next >= ring.end_offset(); + Ok(LogChunk { + data, + next: encode_cursor(next), + bytes_dropped: dropped, + drained_eof, + }) + } + + /// Fire-and-forget signal to the child's process group. A signal to an + /// already-exited (but unreaped) id is an idempotent success. + pub(in crate::engine::wasm::host::process) fn signal( + &self, + id: &str, + principal: &PrincipalId, + capsule_id: &str, + sig: ProcessSignal, + ) -> Result<(), ErrorCode> { + let r = self.resolve(id, principal, capsule_id)?; + if r.core + .lock() + .map(|c| c.phase == Phase::Exited) + .unwrap_or(true) + { + return Ok(()); + } + send_signal(r.os_pid, map_signal(sig)) + } + + /// Write to stdin (requires `keep-stdin-open`). + pub(in crate::engine::wasm::host::process) async fn write_stdin( + &self, + id: &str, + principal: &PrincipalId, + capsule_id: &str, + data: &[u8], + ) -> Result { + if data.len() > MAX_STDIN_WRITE { + return Err(ErrorCode::TooLarge); + } + let core = self.resolve(id, principal, capsule_id)?.core; + // Take the pipe out under the lock, write outside it, restore it. + let mut pipe = { + let mut c = core.lock().unwrap_or_else(|e| e.into_inner()); + if !c.stdin_open { + return Err(ErrorCode::Closed); + } + c.stdin.take().ok_or(ErrorCode::Closed)? + }; + let res = pipe.write_all(data).await; + let mut c = core.lock().unwrap_or_else(|e| e.into_inner()); + match res { + Ok(()) if c.stdin_open => { + c.stdin = Some(pipe); + Ok(data.len() as u32) + }, + Ok(()) => Ok(data.len() as u32), // closed concurrently; drop pipe + Err(_) => { + c.stdin_open = false; + Err(ErrorCode::Closed) + }, + } + } + + /// Close stdin (child sees EOF). Idempotent. + pub(in crate::engine::wasm::host::process) fn close_stdin( + &self, + id: &str, + principal: &PrincipalId, + capsule_id: &str, + ) -> Result<(), ErrorCode> { + let core = self.resolve(id, principal, capsule_id)?.core; + let mut c = core.lock().unwrap_or_else(|e| e.into_inner()); + c.stdin = None; + c.stdin_open = false; + Ok(()) + } + + /// Await exit up to a bounded timeout. Does NOT reap. + pub(in crate::engine::wasm::host::process) async fn wait( + &self, + id: &str, + principal: &PrincipalId, + capsule_id: &str, + timeout: Duration, + ) -> Result { + let r = self.resolve(id, principal, capsule_id)?; + if let Some(e) = current_exit(&r.core) { + return Ok(e.into()); + } + let mut rx = r.exit_rx; + match tokio::time::timeout(timeout, wait_for_exit(&mut rx)).await { + Ok(Some(e)) => Ok(e.into()), + Ok(None) => Err(ErrorCode::Unknown("exit channel closed".to_string())), + Err(_) => Err(ErrorCode::WaitTimeout), + } + } + + /// Graceful terminal stop: SIGTERM → grace → SIGKILL, then REMOVE the id + /// (frees the concurrent + retained slot). + pub(in crate::engine::wasm::host::process) async fn stop( + &self, + id: &str, + principal: &PrincipalId, + capsule_id: &str, + grace: Option, + ) -> Result { + let r = self.resolve(id, principal, capsule_id)?; + let grace = grace.unwrap_or(DEFAULT_STOP_GRACE).min(MAX_STOP_GRACE); + + let exit = if let Some(e) = current_exit(&r.core) { + e + } else { + let _ = send_signal(r.os_pid, nix::sys::signal::Signal::SIGTERM); + let mut rx = r.exit_rx.clone(); + match tokio::time::timeout(grace, wait_for_exit(&mut rx)).await { + Ok(Some(e)) => e, + _ => { + let _ = send_signal(r.os_pid, nix::sys::signal::Signal::SIGKILL); + let mut rx2 = r.exit_rx.clone(); + match tokio::time::timeout(MAX_STOP_GRACE, wait_for_exit(&mut rx2)).await { + Ok(Some(e)) => e, + _ => entry::ExitRecord { + exit_code: None, + signal: Some(9), + }, + } + }, + } + }; + // Remove under the lock, reap (killpg + abort) OUTSIDE it so the + // syscall never stalls other registry ops. + let removed = self.lock().remove(&r.key); + if let Some(entry) = removed { + reap_entry(entry); + } + Ok(exit.into()) + } + + /// Drop retention of an ALREADY-EXITED process. `invalid-input` if still + /// running. Idempotent (an unknown id is success — already gone). + pub(in crate::engine::wasm::host::process) fn release( + &self, + id: &str, + principal: &PrincipalId, + capsule_id: &str, + ) -> Result<(), ErrorCode> { + let key = self.key_of(id); + let mut map = self.lock(); + let Some(entry) = map.get(&key) else { + return Ok(()); + }; + if entry.creator != *principal || &*entry.capsule_id != capsule_id { + return Err(ErrorCode::NoSuchProcess); + } + if entry.is_live() { + return Err(ErrorCode::InvalidInput); + } + let removed = map.remove(&key); + drop(map); + if let Some(entry) = removed { + reap_entry(entry); + } + Ok(()) + } + + /// Sweep idle / over-lifetime / exit-retention-elapsed entries. The + /// engine drives this on a timer. Returns the count reaped. + pub fn reap_sweep(&self) -> usize { + let now = Instant::now(); + let mut to_remove: Vec = Vec::new(); + { + let map = self.lock(); + for (key, e) in map.iter() { + let (phase, exited_at, idle) = { + let c = e.core.lock().unwrap_or_else(|p| p.into_inner()); + ( + c.phase, + c.exited_at, + now.saturating_duration_since(c.last_touch), + ) + }; + let reap = match phase { + Phase::Exited => exited_at + .map(|t| now.saturating_duration_since(t) >= e.exit_retention) + .unwrap_or(true), + _ => { + now.saturating_duration_since(e.spawned_at) >= e.max_lifetime + || idle >= e.idle_timeout + }, + }; + if reap { + to_remove.push(*key); + } + } + } + let mut reaped = Vec::new(); + { + let mut map = self.lock(); + for key in to_remove { + if let Some(entry) = map.remove(&key) { + reaped.push(entry); + } + } + } + // Reap (killpg + abort) outside the map lock. + let n = reaped.len(); + for entry in reaped { + reap_entry(entry); + } + n + } + + /// Kill + clear every entry (capsule unload / daemon graceful shutdown). + pub fn shutdown(&self) { + let drained: Vec = self.lock().drain().map(|(_, e)| e).collect(); + for entry in drained { + reap_entry(entry); + } + } +} 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 new file mode 100644 index 000000000..99afe8e9f --- /dev/null +++ b/crates/astrid-capsule/src/engine/wasm/host/process/persistent/registry_tests.rs @@ -0,0 +1,206 @@ +//! Integration tests for `PersistentProcessRegistry` against REAL child +//! processes (no sandbox wrap — these exercise the registry's lifecycle: +//! reader tasks, the monitor task, ownership re-checks, caps, and reaping). + +use std::process::Stdio; +use std::sync::Arc; +use std::time::Duration; + +use astrid_core::principal::PrincipalId; + +use super::{PersistentProcessRegistry, SpawnParams}; + +/// Spawn a real child running `sh -c