From 78f97177ed2623e580dc6de20df18291560b6ce7 Mon Sep 17 00:00:00 2001 From: "Joshua J. Bouw" Date: Sat, 6 Jun 2026 19:53:45 +0400 Subject: [PATCH 1/7] feat(process): wire the persistent-tier contract into the host (stubbed) Bumps the core/wit submodule to the astrid:process@1.0.0 persistent-tier WIT (unicity-astrid/wit#12) and regenerates the host bindings, so the kernel now speaks the full contract. Implements: - SIGSTOP / SIGCONT mapping in ProcessHandle.signal. - The 15 new persistent Host fns as fail-secure stubs matching the WIT's (NOT YET IMPLEMENTED) notes: spawn-persistent -> persist-unsupported; every id-keyed op -> no-such-process (no registry => no id resolves, which also denies a cross-principal existence oracle); list-processes / status-many -> empty. Nothing here can leak or escape. This is the contract-wiring increment: the surface is fixed and the SDK can generate against it. The host-owned PersistentProcessRegistry that makes spawn-persistent / read-logs / stop / status / list actually work (Tier 1, which unblocks the shell capsule's background-process tools) lands next. Workspace builds; clippy clean. --- .../src/engine/wasm/host/process/handle.rs | 2 + .../src/engine/wasm/host/process/mod.rs | 89 ++++- .../deps/astrid-process/process@1.0.0.wit | 339 +++++++++++++++++- wit | 2 +- 4 files changed, 418 insertions(+), 14 deletions(-) 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..e5adc66d3 100644 --- a/crates/astrid-capsule/src/engine/wasm/host/process/mod.rs +++ b/crates/astrid-capsule/src/engine/wasm/host/process/mod.rs @@ -30,7 +30,8 @@ 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, KillResult, LogChunk, LogCursor, LogStream, + ProcessHandle, ProcessInfo, ProcessResult, ProcessSignal, ReadLogsResult, SpawnRequest, }; use crate::engine::wasm::host::util; use crate::engine::wasm::host_state::HostState; @@ -288,4 +289,90 @@ impl process::Host for HostState { ); result } + + // ================================================================ + // PERSISTENT TIER — `astrid:process@1.0.0`. + // + // The WIT declares the full surface; the host fills it in + // incrementally (see the `(NOT YET IMPLEMENTED)` notes in + // `host/process@1.0.0.wit`). Until the host-owned + // `PersistentProcessRegistry` lands, these are fail-secure stubs: + // - `spawn-persistent` => `persist-unsupported` (persistence off). + // - every id-keyed op => `no-such-process` (no registry => no id + // resolves; honest + denies any cross-principal oracle). + // - `list-processes` / `status-many` => empty (the caller has no + // persistent processes). + // No behaviour here is reachable in a way that could leak or escape; + // the SHAPES are wired so the SDK can generate and the registry can + // drop in behind this contract without a WIT change. + // ================================================================ + + fn spawn_persistent(&mut self, _request: SpawnRequest) -> Result { + Err(ErrorCode::PersistUnsupported) + } + + fn attach(&mut self, _id: String) -> Result, ErrorCode> { + Err(ErrorCode::NoSuchProcess) + } + + fn list_processes( + &mut self, + _label_filter: Option, + ) -> Result, ErrorCode> { + Ok(Vec::new()) + } + + fn status(&mut self, _id: String) -> Result { + Err(ErrorCode::NoSuchProcess) + } + + fn status_many(&mut self, _ids: Vec) -> Result, ErrorCode> { + Ok(Vec::new()) + } + + fn read_logs(&mut self, _id: String) -> Result { + Err(ErrorCode::NoSuchProcess) + } + + fn read_since( + &mut self, + _id: String, + _which_stream: LogStream, + _cursor: LogCursor, + _max_bytes: u32, + ) -> Result { + Err(ErrorCode::NoSuchProcess) + } + + fn write_stdin(&mut self, _id: String, _data: Vec) -> Result { + Err(ErrorCode::NoSuchProcess) + } + + fn close_stdin(&mut self, _id: String) -> Result<(), ErrorCode> { + Err(ErrorCode::NoSuchProcess) + } + + fn signal(&mut self, _id: String, _sig: ProcessSignal) -> Result<(), ErrorCode> { + Err(ErrorCode::NoSuchProcess) + } + + fn wait(&mut self, _id: String, _timeout_ms: u64) -> Result { + Err(ErrorCode::NoSuchProcess) + } + + fn stop(&mut self, _id: String, _grace_ms: Option) -> Result { + Err(ErrorCode::NoSuchProcess) + } + + fn release_process(&mut self, _id: String) -> Result<(), ErrorCode> { + Err(ErrorCode::NoSuchProcess) + } + + fn watch(&mut self, _id: String, _suffix: Option) -> Result<(), ErrorCode> { + Err(ErrorCode::NoSuchProcess) + } + + fn unwatch(&mut self, _id: String) -> Result<(), ErrorCode> { + Err(ErrorCode::NoSuchProcess) + } } 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 a819553cb..04a23b478 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 @@ -12,15 +12,30 @@ /// child-process workflows on the unikernel target; remaining /// workloads stay desktop-only by design. /// -/// Child stdio is byte-oriented (`write-stdin` / `read-logs`) rather -/// than stream-based. The bus already handles high-throughput -/// capsule-to-capsule traffic; stream halves on `process-handle` would -/// pull weight only for niche "splice TCP into child into TCP" media- -/// gateway scenarios that haven't materialised yet. Add as -/// `process-handle@1.1.0` if/when concrete need arises. +/// Two tiers. EPHEMERAL — `spawn` / `spawn-background`, returning a +/// `process-handle` resource owned by the spawning instance's resource +/// table and reaped on resource (instance) drop. PERSISTENT — host-owned, +/// via `spawn-persistent`, returning an opaque, principal-scoped +/// `process-id` whose child lives in a host registry (lifetime = the +/// capsule, not the instance) and is reattachable from any later +/// invocation, including a different pooled instance. The persistent tier +/// exists because a pooled, stateless instance is reset between tool +/// invocations, so an instance-owned handle cannot survive the +/// start-then-read-then-stop pattern. See RFC: host_abi ("Process: +/// ephemeral and persistent tiers") for the full design, security model, +/// and lifecycle. /// -/// Frozen per the ABI evolution discipline (RFC: host_abi). Shape changes -/// ship as a new file at a new version path; never edit this file. +/// Child stdio is byte-oriented (`write-stdin` / `read-logs` / cursored +/// `read-since`) rather than stream-based; stream halves stay deferred to +/// a concrete need per the original rationale. +/// +/// IMPLEMENTATION STATUS. This file is the full 1.0 contract; the host +/// fills it in incrementally. Symbols tagged `(NOT YET IMPLEMENTED)`, +/// `(NOT YET ENFORCED)`, or `(NOT YET POPULATED)` parse and link but the +/// host currently stubs them (returns `unknown` / `persist-unsupported`, +/// leaves an `option` `none`, or skips enforcement) — the SHAPE is fixed +/// now so behaviour can land without a breaking change. Draft until the +/// 1.0 release freezes it per the ABI evolution discipline (RFC: host_abi). package astrid:process@1.0.0; @@ -36,7 +51,8 @@ interface host { invalid-input, /// `cwd` resolved outside the workspace. boundary-escape, - /// Per-capsule background-process cap exhausted (max 8). + /// Per-principal CONCURRENT background-process cap exhausted (shared + /// between ephemeral `spawn-background` and `spawn-persistent`). quota, /// Stdin payload exceeded the per-call 1 MB cap or the /// cumulative-per-process write quota. @@ -47,6 +63,24 @@ interface host { cancelled, /// Wait timed out before the process exited. wait-timeout, + /// No PERSISTENT process with this id is visible to the calling + /// principal: unknown, already released/reaped, owned by a different + /// principal, OR owned by a different capsule. The host MUST NOT + /// distinguish these — "exists but not yours" is a cross-principal + /// oracle. The post-restart / post-reap recovery signal is an empty + /// `list-processes`, not this error. + no-such-process, + /// The per-principal/per-capsule RETAINED-id cap is full (live + + /// exited-but-unreleased). Distinct from `quota` (the CONCURRENT cap). + registry-full, + /// The id is structurally malformed; returned WITHOUT a registry + /// lookup (timing-oracle defense). Treat as `no-such-process`. + invalid-id, + /// `spawn-persistent` refused: the host cannot keep a detached child + /// contained for this configuration (sandbox disabled; macOS without + /// the operator persistence opt-in; or an owner-fallback principal). + /// Fail-closed: never persist a process the host cannot reap. + persist-unsupported, /// Unspecific host error; detail is best-effort. unknown(string), } @@ -67,14 +101,57 @@ interface host { usr2, /// SIGINT — interrupt (Ctrl-C equivalent). int, + /// SIGSTOP — pause the child (cannot be caught/ignored). Lets a + /// supervisor throttle a runaway without killing it. + stop, + /// SIGCONT — resume a paused child. + cont, } /// Environment variable to pass to a spawned process. + /// + /// Environment is the ONLY path into the child: the sandbox strips the + /// host's ambient environment except a small kernel passthrough + /// allowlist. Carries operator-reviewable, non-secret configuration; a + /// SECRET reaches a child over `write-stdin` / `spawn-request.stdin`, + /// NOT `args` (which are recorded verbatim in the audit log). record env-var { key: string, value: string, } + /// Per-stream ring-overflow policy for a persistent process's buffered + /// stdout/stderr (the ephemeral tier is always `drop-oldest`). + enum overflow-policy { + /// Drop oldest bytes to make room; cumulative loss is surfaced as + /// `bytes-dropped` so the reader knows the stream is non-contiguous. + drop-oldest, + /// Stop draining the child's pipe when the ring is full, so the OS + /// pipe buffer fills and the child blocks on write. Parks the host + /// reader task only — never the WASM task. Correct for REPL / + /// MCP-stdio children where dropping output corrupts protocol framing. + backpressure, + } + + /// Per-child OS resource ceilings (RLIMIT / cgroup), applicable to every + /// tier. `none` per field => the principal's profile default. + /// + /// (NOT YET ENFORCED) — declared so the record shape is stable; the host + /// currently bounds only the process slot, log buffers, and lifetime, so + /// a single child can still consume unbounded CPU/RAM. Including the + /// fields now avoids a breaking record change when enforcement lands. + record resource-limits { + /// Address-space / RSS ceiling (RLIMIT_AS or cgroup memory.max). + max-memory-bytes: option, + /// Cumulative CPU-time ceiling (RLIMIT_CPU -> SIGXCPU then SIGKILL). + max-cpu-ms: option, + /// Max concurrent child PIDs (RLIMIT_NPROC / cgroup pids.max) — a + /// fork-bomb fence. + max-pids: option, + /// Max open file descriptors (RLIMIT_NOFILE). + max-open-files: option, + } + /// Request to spawn a host process. record spawn-request { /// Command to execute. @@ -92,6 +169,41 @@ interface host { /// inside the sandbox; absolute paths and `..` escapes are /// rejected with `boundary-escape`. cwd: option, + /// Per-child OS resource ceilings. Applies to EVERY tier. + /// (NOT YET ENFORCED — see `resource-limits`.) + limits: option, + + // ---- the fields below are honored ONLY by `spawn-persistent` + // and ignored by `spawn` / `spawn-background`. ---- + + /// Operator-readable label surfaced in `list-processes` / audit / + /// `astrid ps`-style tooling. Untrusted: host length-clamps (<= 128 + /// bytes) + strips control chars. NOT an identity — the `process-id` + /// is the only identity. `none` => derived from `cmd`. + label: option, + /// Keep a writable stdin pipe open after the optional `stdin` prelude + /// so `write-stdin` works across invocations (REPL / `psql` / + /// MCP-stdio). `none` => false (stdin closed after the prelude). + keep-stdin-open: option, + /// Ring-overflow policy. `none` => `drop-oldest`. + overflow: option, + /// Per-stream output ring capacity (bytes), stdout and stderr each. + /// `none` => host default (1 MiB). Host-clamped to the profile ceiling. + log-ring-bytes: option, + /// Wall-clock lifetime ceiling from spawn. On expiry: SIGTERM -> grace + /// -> SIGKILL -> reap. `none` => the profile ceiling. Any value is + /// clamped DOWN to that ceiling — a guest CANNOT request an unbounded + /// lifetime. + max-lifetime-ms: option, + /// Reap an UNTOUCHED persistent process (no read / wait / reattach / + /// signal / write) after this idle interval — the primary anti-leak + /// backstop for spawn-and-forget. `none` => profile default. + /// `some(0)` is rejected with `invalid-input`. + idle-timeout-ms: option, + /// After a persistent process EXITS, retain its id + drained log tail + /// this long before the host auto-reaps the registry entry (the "read + /// its output later" window). `none` => profile default. Host-clamped. + exit-retention-ms: option, } /// Exit information for a process that has terminated. @@ -139,11 +251,116 @@ interface host { stderr: string, } + // ================================================================ + // PERSISTENT TIER value types. (NOT YET IMPLEMENTED — the host stubs + // the persistent functions below; the SHAPES are fixed now.) + // ================================================================ + + /// Opaque, unforgeable, principal-scoped identity for a PERSISTENT + /// process that survives instance churn (pool reset / per-principal + /// multiplexing). + /// + /// Wire form: a 256-bit host-minted CSPRNG token (host entropy, NEVER + /// the guest), base32; the host stores only a keyed hash. Possession is + /// necessary but NOT sufficient — the host re-checks the calling + /// principal == creator on EVERY id-keyed call, so a token that leaks + /// across the principal boundary is inert. Treat as an opaque secret: + /// never parse, pattern-match, or synthesize it. + type process-id = string; + + /// Lifecycle phase of a persistent process, reported WITHOUT draining + /// log buffers (see `status` / `process-info`). + enum process-phase { + /// Spawn accepted; child not yet confirmed running. + starting, + /// Child is running. + running, + /// Terminated; `exit-info` available; logs readable until released + /// or the exit-retention TTL elapses. + exited, + /// Exited AND its retained id/buffers were reaped. Terminal; the id + /// is permanently dead. + reaped, + } + + /// Opaque, host-encoded, resumable cursor into a persistent process's + /// log streams, for NON-DRAINING reads (`read-since`). Encodes per-stream + /// offset + a generation, so a stale cursor after ring eviction / reap + /// resolves to reported drops or `no-such-process`, never to another + /// process's bytes. `token: none` on the first read = from the oldest + /// retained byte. Treat as opaque; do not construct or parse. + record log-cursor { + token: option, + } + + /// Which stream a cursor read addresses. + enum log-stream { + stdout, + stderr, + } + + /// A NON-DRAINING slice of a persistent process's stream, addressed by + /// cursor — the polling-safe complement to `read-logs` (which DRAINS). + /// Multiple independent readers each keep their own cursor and observe + /// the full retained stream. + record log-chunk { + /// Bytes in `[requested-cursor, next)`. `list` not `string`: + /// children emit non-UTF-8; the caller decodes. + data: list, + /// Cursor to pass to the NEXT `read-since` to resume exactly here. + next: log-cursor, + /// Cumulative bytes evicted from this stream's ring before they could + /// be delivered through this cursor (0 unless the reader fell behind). + bytes-dropped: u64, + /// `true` once the child has exited AND all retained output on this + /// stream has been delivered through this cursor — the clean EOF. + drained-eof: bool, + } + + /// A NON-DRAINING status snapshot of one persistent process — the unit of + /// `list-processes` and the return of `status`. Repeated calls never + /// consume log bytes and never mutate process state. + record process-info { + /// Reattach key. Stable across invocations / instances. + id: process-id, + /// Operator-supplied label (or one derived from `cmd`). + label: string, + /// cmd + args, pre-sandbox-wrap, as the capsule requested it. + command: string, + /// OS PID while running; `none` once reaped. Advisory ONLY (PIDs are + /// recycled); never use as cross-call identity — use `id`. + os-pid: option, + /// Lifecycle phase (no buffer drain). + phase: process-phase, + /// Present once `exited` / `reaped`. + exit: option, + /// Host-monotonic milliseconds since spawn (uptime). + age-ms: u64, + /// Milliseconds since the last operation touched this process (drives + /// `idle-timeout-ms`). + idle-ms: u64, + /// Bytes currently buffered and drainable (stdout + stderr). + buffered-bytes: u64, + /// Cumulative bytes evicted from the rings since spawn. + bytes-dropped: u64, + /// Whether stdin is still open for `write-stdin`. + stdin-open: bool, + /// Cumulative CPU time the child has consumed. (NOT YET POPULATED.) + cpu-ms: option, + /// Peak resident memory observed for the child. (NOT YET POPULATED.) + mem-bytes-peak: option, + } + /// A running or recently-terminated background process. Drop is /// automatic — capsules don't need to explicitly close; the /// host reaps the process on resource drop. /// - /// Per-capsule cap: 8 concurrent background processes. + /// Obtained from `spawn-background` (reap-on-drop) or `attach` + /// (DETACH-on-drop — dropping it does NOT reap the underlying persistent + /// process; it severs this instance's view). + /// + /// Per-principal cap: see `quota` (shared between ephemeral and + /// persistent spawns). resource process-handle { /// Read buffered logs since the last call. Drains the buffers /// (subsequent reads return only new data). Buffers are 1 MiB @@ -153,9 +370,11 @@ interface host { /// Write to the process's stdin. Useful for REPL-style /// children (`python -i`, `psql`, MCP stdio subprocesses). /// Returns bytes actually written; capped at 1 MB per call. + /// (NOT YET IMPLEMENTED — stdin pipe capture pending.) write-stdin: func(data: list) -> result; /// Close the stdin pipe (child observes EOF on read). + /// (NOT YET IMPLEMENTED — stdin pipe capture pending.) close-stdin: func() -> result<_, error-code>; /// Send a signal. Fire-and-forget; for graceful shutdown use @@ -189,11 +408,16 @@ interface host { /// Pollable that fires when the process has exited. Compose /// with other pollables to multiplex "wait on child OR - /// receive IPC event." + /// receive IPC event." Instance-local: dies on instance reset, so + /// for a PERSISTENT process the durable exit signal is the event + /// model (`watch`), not this pollable. + /// (NOT YET IMPLEMENTED — returns an always-ready stub pollable.) subscribe-exit: func() -> pollable; /// Pollable that fires when stdout / stderr has buffered - /// data ready to be drained via `read-logs`. + /// data ready to be drained via `read-logs`. Same instance-local + /// caveat as `subscribe-exit`. + /// (NOT YET IMPLEMENTED — returns an always-ready stub pollable.) subscribe-logs: func() -> pollable; } @@ -208,4 +432,95 @@ interface host { /// stdout/stderr are buffered (1 MiB per stream, ring buffer). /// Audit: recorded. 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.) + // + // Every id-keyed function re-resolves the live calling principal + + // capsule and checks them against the recorded creator BEFORE touching + // the entry; wrong-owner / unknown / reaped all collapse to + // `no-such-process` (no oracle). Audited (op + principal + capsule + + // id-hash; never the raw id, env, stdin, or log contents). + // ================================================================ + + /// Spawn a PERSISTENT background process whose lifetime is decoupled from + /// the spawning instance. Returns a `process-id` any LATER invocation of + /// the SAME capsule under the SAME principal can reattach to. Gated on + /// `host_process` AND the operator `allow_persistent` sub-grant; counts + /// against BOTH the per-principal concurrent cap (shared with + /// `spawn-background`) AND the retained-id cap. Fail-closed refusals + /// (`persist-unsupported`): owner-fallback principal; sandbox disabled; + /// macOS without the operator persistence opt-in. + spawn-persistent: func(request: spawn-request) -> result; + + /// Re-materialise a `process-handle` over a persistent process for THIS + /// invocation — the composition primitive (every ephemeral method works). + /// DETACH-on-drop: dropping it (incl. on instance reset) does NOT reap. + /// `no-such-process` if the id is unknown to the caller. + attach: func(id: process-id) -> result; + + /// List the caller `(capsule, principal)`'s persistent processes, + /// optionally filtered by a label substring. NEVER another principal's or + /// capsule's. Empty is normal (post-restart / post-reap recovery signal). + list-processes: func(label-filter: option) -> result, error-code>; + + /// Resolve an id to its `process-info` WITHOUT draining buffers — the safe + /// poll primitive. `no-such-process` if unknown to the caller. + status: func(id: process-id) -> result; + + /// Batch `status`: one registry pass + one audit record for a supervisor + /// polling N children on a timer. Unknown ids are absent from the result. + status-many: func(ids: list) -> result, error-code>; + + /// `attach(id)?.read-logs()` — DRAINS buffered logs since the last read. + read-logs: func(id: process-id) -> result; + + /// NON-DRAINING, cursor-addressed, resumable read from a persistent + /// process's stream. Works on an exited process's retained tail until + /// released / TTL. `max-bytes` bounds the chunk (host also hard-caps). + 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.) + write-stdin: func(id: process-id, data: list) -> result; + + /// `attach(id)?.close-stdin()`. (NOT YET IMPLEMENTED.) + close-stdin: func(id: process-id) -> result<_, error-code>; + + /// `attach(id)?.signal(sig)`. Fire-and-forget signal delivery. + signal: func(id: process-id, sig: process-signal) -> result<_, error-code>; + + /// `attach(id)?.wait(timeout-ms)` with a REQUIRED bounded timeout (an + /// unbounded wait would pin a pooled instance). Does NOT release on exit. + /// `wait-timeout` on elapse. For long-lived waits use `watch`. + wait: func(id: process-id, timeout-ms: u64) -> result; + + /// Graceful terminal stop: SIGTERM, wait up to `grace-ms`, then SIGKILL; + /// drain final buffers; REMOVE the id (frees the concurrent + retained + /// slot). `grace-ms: none` => host default. (For an immediate SIGKILL use + /// `attach(id)?.kill()`.) + stop: func(id: process-id, grace-ms: option) -> result; + + /// Drop the host's retention of an ALREADY-EXITED process: frees the + /// retained-id slot + discards the buffered tail WITHOUT signalling. + /// `invalid-input` if still running (use `stop`). Idempotent. + release-process: func(id: process-id) -> result<_, error-code>; + + /// Arm DURABLE lifecycle events for `id`, published by the host UNDER THE + /// CREATOR PRINCIPAL on + /// `astrid.process.v1.{exited,log-ready,stdin-drained}.` (a + /// pollable cannot survive instance reset, so the durable channel is the + /// IPC bus the capsule already subscribes to). The `exited` payload carries + /// a reason (self-exit / killed / reaped-ttl / reaped-shutdown). `suffix: + /// none` => the id. Idempotent. `no-such-process` if unknown to the caller. + /// OPEN (RFC: host_abi): whether the host-published topics need a manifest + /// `[publish]` declaration or are a kernel-authored topic class — `watch` + /// may be cut in favour of `status` + bounded `wait` polling. + watch: func(id: process-id, suffix: option) -> result<_, error-code>; + + /// Stop publishing lifecycle events for `id` (the child keeps running). + /// Idempotent. + unwatch: func(id: process-id) -> result<_, error-code>; } diff --git a/wit b/wit index 467240dee..c5630ab7e 160000 --- a/wit +++ b/wit @@ -1 +1 @@ -Subproject commit 467240deeba2cf7ff28ba32d12b5a88485be5b3d +Subproject commit c5630ab7e23d9d28042cca15edb01abcddac2271 From 960f864bd10c23119855429df29b2afcd5737ffa Mon Sep 17 00:00:00 2001 From: "Joshua J. Bouw" Date: Sat, 6 Jun 2026 20:11:07 +0400 Subject: [PATCH 2/7] chore(process): sync host stubs to reviewed WIT (stop -> exit-info) Bumps core/wit to wit@46704fd (review fixes) and updates the stop stub's return type to exit-info; drops the now-unused KillResult import. --- .../src/engine/wasm/host/process/mod.rs | 6 +- .../deps/astrid-process/process@1.0.0.wit | 78 ++++++++++++------- wit | 2 +- 3 files changed, 56 insertions(+), 30 deletions(-) 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 e5adc66d3..7b2f7a4d3 100644 --- a/crates/astrid-capsule/src/engine/wasm/host/process/mod.rs +++ b/crates/astrid-capsule/src/engine/wasm/host/process/mod.rs @@ -30,8 +30,8 @@ use tracing::warn; use wasmtime::component::Resource; use crate::engine::wasm::bindings::astrid::process::host::{ - self as process, EnvVar, ErrorCode, ExitInfo, KillResult, LogChunk, LogCursor, LogStream, - ProcessHandle, ProcessInfo, ProcessResult, ProcessSignal, ReadLogsResult, 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; @@ -360,7 +360,7 @@ impl process::Host for HostState { Err(ErrorCode::NoSuchProcess) } - fn stop(&mut self, _id: String, _grace_ms: Option) -> Result { + fn stop(&mut self, _id: String, _grace_ms: Option) -> Result { Err(ErrorCode::NoSuchProcess) } 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 04a23b478..cbc27ee96 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 @@ -25,9 +25,12 @@ /// ephemeral and persistent tiers") for the full design, security model, /// and lifecycle. /// -/// Child stdio is byte-oriented (`write-stdin` / `read-logs` / cursored -/// `read-since`) rather than stream-based; stream halves stay deferred to -/// a concrete need per the original rationale. +/// Child stdio is byte-oriented (pipe-based) rather than stream-based; stream +/// halves stay deferred to a concrete need per the original rationale. The +/// byte-faithful read path is `read-since` (`list`); the convenience +/// `read-logs` / `read-logs-result` (and the synchronous `process-result`) +/// surface stdout/stderr as UTF-8, which is LOSSY on non-UTF-8 output — fine +/// for text, use `read-since` for binary. /// /// IMPLEMENTATION STATUS. This file is the full 1.0 contract; the host /// fills it in incrementally. Symbols tagged `(NOT YET IMPLEMENTED)`, @@ -65,17 +68,17 @@ interface host { wait-timeout, /// No PERSISTENT process with this id is visible to the calling /// principal: unknown, already released/reaped, owned by a different - /// principal, OR owned by a different capsule. The host MUST NOT - /// distinguish these — "exists but not yours" is a cross-principal - /// oracle. The post-restart / post-reap recovery signal is an empty - /// `list-processes`, not this error. + /// principal, owned by a different capsule, OR structurally malformed. + /// The host MUST NOT distinguish these — "exists but not yours" and + /// "well-formed vs garbage" are both cross-principal/structural oracles. + /// (The host MAY fast-reject obvious garbage internally to avoid a + /// lookup, but the guest-visible result is identical.) The post-restart + /// / post-reap recovery signal is an empty `list-processes`, not this + /// error. no-such-process, /// The per-principal/per-capsule RETAINED-id cap is full (live + /// exited-but-unreleased). Distinct from `quota` (the CONCURRENT cap). registry-full, - /// The id is structurally malformed; returned WITHOUT a registry - /// lookup (timing-oracle defense). Treat as `no-such-process`. - invalid-id, /// `spawn-persistent` refused: the host cannot keep a detached child /// contained for this configuration (sandbox disabled; macOS without /// the operator persistence opt-in; or an owner-fallback principal). @@ -143,8 +146,10 @@ interface host { record resource-limits { /// Address-space / RSS ceiling (RLIMIT_AS or cgroup memory.max). max-memory-bytes: option, - /// Cumulative CPU-time ceiling (RLIMIT_CPU -> SIGXCPU then SIGKILL). - max-cpu-ms: option, + /// Cumulative CPU-time ceiling in WHOLE SECONDS (RLIMIT_CPU -> SIGXCPU + /// then SIGKILL; second-granularity is all RLIMIT_CPU enforces, so the + /// unit is seconds, not ms). + max-cpu-secs: option, /// Max concurrent child PIDs (RLIMIT_NPROC / cgroup pids.max) — a /// fork-bomb fence. max-pids: option, @@ -260,27 +265,32 @@ interface host { /// process that survives instance churn (pool reset / per-principal /// multiplexing). /// - /// Wire form: a 256-bit host-minted CSPRNG token (host entropy, NEVER - /// the guest), base32; the host stores only a keyed hash. Possession is + /// Wire form: a 256-bit host-minted CSPRNG token (host entropy, NEVER the + /// guest), **lowercase base32** (RFC 4648 `a-z2-7`, NO padding) — chosen so + /// the id is itself a valid IPC topic suffix (`[a-z0-9._-]+`), usable + /// directly as the `watch` suffix without sanitisation. The host stores + /// only a keyed hash. Possession is /// necessary but NOT sufficient — the host re-checks the calling /// principal == creator on EVERY id-keyed call, so a token that leaks /// across the principal boundary is inert. Treat as an opaque secret: /// never parse, pattern-match, or synthesize it. type process-id = string; - /// Lifecycle phase of a persistent process, reported WITHOUT draining - /// log buffers (see `status` / `process-info`). + /// Lifecycle phase of a persistent process, reported WITHOUT draining log + /// buffers (see `status` / `process-info`). There is deliberately NO + /// `reaped` phase: once a process is reaped, its id resolves to + /// `no-such-process` and drops out of `list-processes`, so `exited` is the + /// terminal observable phase. The reap (and its reason) is delivered via + /// the `watch` exit event, not `status`. enum process-phase { /// Spawn accepted; child not yet confirmed running. starting, /// Child is running. running, - /// Terminated; `exit-info` available; logs readable until released - /// or the exit-retention TTL elapses. + /// Terminated; `exit-info` available; logs readable until released or + /// the exit-retention TTL elapses, after which the id becomes + /// `no-such-process`. exited, - /// Exited AND its retained id/buffers were reaped. Terminal; the id - /// is permanently dead. - reaped, } /// Opaque, host-encoded, resumable cursor into a persistent process's @@ -391,6 +401,11 @@ interface host { /// patterns that mustn't hang on a runaway child. /// Returns exit-info on exit, `wait-timeout` error if the /// timeout elapsed first. + /// + /// On a handle obtained via `attach` (a persistent process), `none` is + /// REJECTED with `invalid-input`: an unbounded wait would pin the + /// pooled instance. Use a bounded timeout, the id-keyed `wait` (also + /// bounded), or the durable `watch` event for long-lived waits. wait: func(timeout-ms: option) -> result; /// Wait for the process to exit AND drain remaining stdout / @@ -474,7 +489,13 @@ interface host { /// polling N children on a timer. Unknown ids are absent from the result. status-many: func(ids: list) -> result, error-code>; - /// `attach(id)?.read-logs()` — DRAINS buffered logs since the last read. + /// `attach(id)?.read-logs()` — DRAINS the SINGLE shared ring per + /// `process-id`: whoever drains first consumes those bytes, so concurrent + /// drainers across invocations/attachments race over one cursor. For + /// independent multi-reader use (a UI tailer AND the capsule's own parser) + /// OR byte-faithful output, use `read-since` (per-cursor, non-draining, + /// `list`). `read-logs-result` stdout/stderr are UTF-8 (lossy on + /// non-UTF-8 child output). read-logs: func(id: process-id) -> result; /// NON-DRAINING, cursor-addressed, resumable read from a persistent @@ -498,10 +519,15 @@ interface host { wait: func(id: process-id, timeout-ms: u64) -> result; /// Graceful terminal stop: SIGTERM, wait up to `grace-ms`, then SIGKILL; - /// drain final buffers; REMOVE the id (frees the concurrent + retained - /// slot). `grace-ms: none` => host default. (For an immediate SIGKILL use - /// `attach(id)?.kill()`.) - stop: func(id: process-id, grace-ms: option) -> result; + /// REMOVE the id (frees the concurrent + retained slot). `grace-ms: none` + /// => host default. (For an immediate SIGKILL use `attach(id)?.kill()`.) + /// + /// Returns only `exit-info`, NOT the final buffers: `stop` reaps the id, so + /// to keep a child's last output drain it with `read-since` (byte-faithful) + /// BEFORE `stop`. Returning the buffers here would hand back stdout/stderr + /// the guest must UTF-8-decode (lossy on non-UTF-8 output); `read-since` is + /// the byte-exact path. + stop: func(id: process-id, grace-ms: option) -> result; /// Drop the host's retention of an ALREADY-EXITED process: frees the /// retained-id slot + discards the buffered tail WITHOUT signalling. diff --git a/wit b/wit index c5630ab7e..46704fd90 160000 --- a/wit +++ b/wit @@ -1 +1 @@ -Subproject commit c5630ab7e23d9d28042cca15edb01abcddac2271 +Subproject commit 46704fd90d114b1a939053e261d7ae5c0a0de525 From ed6e86b052736a468d99edd50920e6512a76e4c7 Mon Sep 17 00:00:00 2001 From: "Joshua J. Bouw" Date: Sat, 6 Jun 2026 20:42:05 +0400 Subject: [PATCH 3/7] chore(wit): bump submodule to merged main (process tier #12, hook #11) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Re-point the wit submodule from the pre-merge feature-branch commit (46704fd) to merged origin/main (8946e98). host/ content is byte-identical between the two, so the committed wit-staging/ stays in lockstep — only the gitlink moves. This puts any release on a real, pushed main ref rather than a feature branch that gets GC'd post-merge. --- wit | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/wit b/wit index 46704fd90..8946e98d1 160000 --- a/wit +++ b/wit @@ -1 +1 @@ -Subproject commit 46704fd90d114b1a939053e261d7ae5c0a0de525 +Subproject commit 8946e98d149cbd320636138c26a1333f3b16fb66 From a1a97a43d2a2bece299e28ebf9ff3da3d7907e40 Mon Sep 17 00:00:00 2001 From: "Joshua J. Bouw" Date: Sat, 6 Jun 2026 20:50:28 +0400 Subject: [PATCH 4/7] docs(changelog): record astrid:process persistent-tier host wiring (#866) --- CHANGELOG.md | 1 + 1 file changed, 1 insertion(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index e714817a1..276a2920a 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 — host ABI wired (stubbed).** The frozen `astrid:process` host package gains a persistent tier above the existing ephemeral `spawn` / `spawn-background` calls, so a background child can outlive the pooled, *stateless* WASM instance that started it — today a `process-handle` resource is reaped when its instance is reset on return to the pool, so a process started in one tool invocation cannot survive to the next. New surface: `spawn-persistent` (returns a host-owned, topic-safe `process-id`), `attach`, the id-keyed lifecycle ops (`read-logs` / `read-since` / `write-stdin` / `close-stdin` / `signal` / `wait` / `stop` / `release-process`), the registry queries (`list-processes` / `status` / `status-many`), and the `watch` / `unwatch` change feed — plus the supporting types (`process-id`, `process-phase`, `log-cursor` / `log-stream` / `log-chunk`, `process-info`, `overflow-policy`, `resource-limits`), the `stop` → `exit-info` return, the `stop` / `cont` signals, and the `no-such-process` / `registry-full` / `persist-unsupported` error codes. **This change is contract + host wiring only:** the host methods are present and audited but **stubbed** — `spawn-persistent` returns `persist-unsupported`, the id-keyed ops return `no-such-process`, the queries return empty — so guests and SDKs can link against the full surface while the host-owned `PersistentProcessRegistry` that actually survives instance-pool resets lands as a follow-up. The contract is frozen in the `wit` submodule (unicity-astrid/wit#12) and specified in the host-ABI RFC (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. From e12a649de49fa68635b28e6d2c83653a72c4f901 Mon Sep 17 00:00:00 2001 From: "Joshua J. Bouw" Date: Sat, 6 Jun 2026 21:34:14 +0400 Subject: [PATCH 5/7] feat(process): implement the persistent-process tier registry (#866) Replace the spawn-persistent / id-keyed stubs with a real host-owned PersistentProcessRegistry, shared across a capsule's pooled instances (cloned into every HostState like ProcessTracker) so a process-id survives instance reset. Implements: spawn-persistent (256-bit base32 id, keyed-BLAKE3-hashed, owner-fallback refused), status/status-many/list-processes, read-logs (drain) + read-since (non-draining cursor), signal/wait/stop/release-process, write-stdin/close-stdin. Per-call (principal,capsule) re-check vs creator; concurrent + retained caps; idle/lifetime/exit-retention reaper; kill-all on unload/shutdown. Linux + macOS (daemon-hard-crash orphan caveat documented). attach + watch remain deferred. --- CHANGELOG.md | 2 +- .../src/engine/wasm/host/process/mod.rs | 313 ++++++++-- .../wasm/host/process/persistent/config.rs | 90 +++ .../wasm/host/process/persistent/entry.rs | 274 +++++++++ .../wasm/host/process/persistent/ids.rs | 76 +++ .../wasm/host/process/persistent/mod.rs | 570 ++++++++++++++++++ .../host/process/persistent/registry_tests.rs | 198 ++++++ .../wasm/host/process/persistent/ring.rs | 189 ++++++ .../src/engine/wasm/host_state.rs | 9 + crates/astrid-capsule/src/engine/wasm/mod.rs | 39 ++ .../src/engine/wasm/test_fixtures.rs | 5 +- crates/astrid-hooks/src/handler/wasm.rs | 7 + 12 files changed, 1728 insertions(+), 44 deletions(-) create mode 100644 crates/astrid-capsule/src/engine/wasm/host/process/persistent/config.rs create mode 100644 crates/astrid-capsule/src/engine/wasm/host/process/persistent/entry.rs create mode 100644 crates/astrid-capsule/src/engine/wasm/host/process/persistent/ids.rs create mode 100644 crates/astrid-capsule/src/engine/wasm/host/process/persistent/mod.rs create mode 100644 crates/astrid-capsule/src/engine/wasm/host/process/persistent/registry_tests.rs create mode 100644 crates/astrid-capsule/src/engine/wasm/host/process/persistent/ring.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 276a2920a..7b3c3034d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -28,7 +28,7 @@ Changelog tracking starts with 0.2.0. Prior versions were not tracked. ### Added -- **`astrid:process@1.0.0` persistent-process tier — host ABI wired (stubbed).** The frozen `astrid:process` host package gains a persistent tier above the existing ephemeral `spawn` / `spawn-background` calls, so a background child can outlive the pooled, *stateless* WASM instance that started it — today a `process-handle` resource is reaped when its instance is reset on return to the pool, so a process started in one tool invocation cannot survive to the next. New surface: `spawn-persistent` (returns a host-owned, topic-safe `process-id`), `attach`, the id-keyed lifecycle ops (`read-logs` / `read-since` / `write-stdin` / `close-stdin` / `signal` / `wait` / `stop` / `release-process`), the registry queries (`list-processes` / `status` / `status-many`), and the `watch` / `unwatch` change feed — plus the supporting types (`process-id`, `process-phase`, `log-cursor` / `log-stream` / `log-chunk`, `process-info`, `overflow-policy`, `resource-limits`), the `stop` → `exit-info` return, the `stop` / `cont` signals, and the `no-such-process` / `registry-full` / `persist-unsupported` error codes. **This change is contract + host wiring only:** the host methods are present and audited but **stubbed** — `spawn-persistent` returns `persist-unsupported`, the id-keyed ops return `no-such-process`, the queries return empty — so guests and SDKs can link against the full surface while the host-owned `PersistentProcessRegistry` that actually survives instance-pool resets lands as a follow-up. The contract is frozen in the `wit` submodule (unicity-astrid/wit#12) and specified in the host-ABI RFC (unicity-astrid/rfcs#22). Closes #866. +- **`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/mod.rs b/crates/astrid-capsule/src/engine/wasm/host/process/mod.rs index 7b2f7a4d3..c6a51a700 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; @@ -37,6 +38,7 @@ 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. @@ -97,6 +99,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(); @@ -293,86 +332,276 @@ impl process::Host for HostState { // ================================================================ // PERSISTENT TIER — `astrid:process@1.0.0`. // - // The WIT declares the full surface; the host fills it in - // incrementally (see the `(NOT YET IMPLEMENTED)` notes in - // `host/process@1.0.0.wit`). Until the host-owned - // `PersistentProcessRegistry` lands, these are fail-secure stubs: - // - `spawn-persistent` => `persist-unsupported` (persistence off). - // - every id-keyed op => `no-such-process` (no registry => no id - // resolves; honest + denies any cross-principal oracle). - // - `list-processes` / `status-many` => empty (the caller has no - // persistent processes). - // No behaviour here is reachable in a way that could leak or escape; - // the SHAPES are wired so the SDK can generate and the registry can - // drop in behind this contract without a WIT change. + // 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 { - Err(ErrorCode::PersistUnsupported) + fn spawn_persistent(&mut self, request: SpawnRequest) -> Result { + // Refuse the owner-fallback principal (see `authenticated_principal`). + 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); + } + + let cmd_for_audit = request.cmd.clone(); + let capsule_id_arc: Arc = Arc::from(self.capsule_id.as_str()); + let workspace_root = self.workspace_root.clone(); + let handle = self.runtime_handle.clone(); + let semaphore = self.blocking_semaphore.clone(); + + // Capability gate — identical to `spawn-background`. + 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; + } + } + if self.cancel_token.is_cancelled() { + return Err(ErrorCode::Cancelled); + } + + // Per-principal concurrent cap (shared with `spawn-background`). + let concurrent_cap = + usize::try_from(self.effective_profile().quotas.max_background_processes) + .unwrap_or(MAX_BACKGROUND_PROCESSES) + .min(MAX_BACKGROUND_PROCESSES); + + 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; + }, + }; + let os_pid = child.id().unwrap_or(0); + 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, retaining the pipe ONLY when the + // guest asked to keep stdin open (else dropping it closes stdin = EOF). + if let (Some(prelude), Some(pipe)) = (request.stdin.clone(), stdin.take()) { + let written = util::bounded_block_on(&handle, &semaphore, async move { + use tokio::io::AsyncWriteExt as _; + let mut pipe = pipe; + let _ = pipe.write_all(&prelude).await; + pipe + }); + stdin = Some(written); + } + 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, + 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> { - Err(ErrorCode::NoSuchProcess) + // 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. + Err(ErrorCode::Unknown( + "attach: resource-handle materialisation pending — use the id-keyed ops".to_string(), + )) } fn list_processes( &mut self, - _label_filter: Option, + label_filter: Option, ) -> Result, ErrorCode> { - Ok(Vec::new()) + let principal = self.effective_principal(); + let capsule_id = self.capsule_id.as_str(); + Ok(self + .persistent_processes + .list(&principal, capsule_id, label_filter.as_deref())) } - fn status(&mut self, _id: String) -> Result { - Err(ErrorCode::NoSuchProcess) + fn status(&mut self, id: String) -> Result { + let principal = self.effective_principal(); + let capsule_id = self.capsule_id.as_str(); + self.persistent_processes + .status(&id, &principal, capsule_id) } - fn status_many(&mut self, _ids: Vec) -> Result, ErrorCode> { - Ok(Vec::new()) + fn status_many(&mut self, ids: Vec) -> Result, ErrorCode> { + let principal = self.effective_principal(); + let capsule_id = self.capsule_id.as_str(); + Ok(self + .persistent_processes + .status_many(&ids, &principal, capsule_id)) } - fn read_logs(&mut self, _id: String) -> Result { - Err(ErrorCode::NoSuchProcess) + fn read_logs(&mut self, id: String) -> Result { + let principal = self.effective_principal(); + let capsule_id = self.capsule_id.as_str(); + self.persistent_processes + .read_logs(&id, &principal, capsule_id) } fn read_since( &mut self, - _id: String, - _which_stream: LogStream, - _cursor: LogCursor, - _max_bytes: u32, + id: String, + which_stream: LogStream, + cursor: LogCursor, + max_bytes: u32, ) -> Result { - Err(ErrorCode::NoSuchProcess) + let principal = self.effective_principal(); + let capsule_id = self.capsule_id.as_str(); + self.persistent_processes.read_since( + &id, + &principal, + capsule_id, + which_stream, + &cursor, + max_bytes, + ) } - fn write_stdin(&mut self, _id: String, _data: Vec) -> Result { - Err(ErrorCode::NoSuchProcess) + 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(); + util::bounded_block_on(&handle, &semaphore, async move { + registry + .write_stdin(&id, &principal, &capsule_id, &data) + .await + }) } - fn close_stdin(&mut self, _id: String) -> Result<(), ErrorCode> { - Err(ErrorCode::NoSuchProcess) + fn close_stdin(&mut self, id: String) -> Result<(), ErrorCode> { + let principal = self.effective_principal(); + let capsule_id = self.capsule_id.as_str(); + self.persistent_processes + .close_stdin(&id, &principal, capsule_id) } - fn signal(&mut self, _id: String, _sig: ProcessSignal) -> Result<(), ErrorCode> { - Err(ErrorCode::NoSuchProcess) + fn signal(&mut self, id: String, sig: ProcessSignal) -> Result<(), ErrorCode> { + let principal = self.effective_principal(); + let capsule_id = self.capsule_id.as_str(); + self.persistent_processes + .signal(&id, &principal, capsule_id, sig) } - fn wait(&mut self, _id: String, _timeout_ms: u64) -> Result { - Err(ErrorCode::NoSuchProcess) + 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); + util::bounded_block_on_cancellable(&handle, &semaphore, &cancel, async move { + registry.wait(&id, &principal, &capsule_id, timeout).await + }) + .unwrap_or(Err(ErrorCode::Cancelled)) } - fn stop(&mut self, _id: String, _grace_ms: Option) -> Result { - Err(ErrorCode::NoSuchProcess) + 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); + util::bounded_block_on_cancellable(&handle, &semaphore, &cancel, async move { + registry.stop(&id, &principal, &capsule_id, grace).await + }) + .unwrap_or(Err(ErrorCode::Cancelled)) } - fn release_process(&mut self, _id: String) -> Result<(), ErrorCode> { - Err(ErrorCode::NoSuchProcess) + fn release_process(&mut self, id: String) -> Result<(), ErrorCode> { + let principal = self.effective_principal(); + let capsule_id = self.capsule_id.as_str(); + self.persistent_processes + .release(&id, &principal, capsule_id) } fn watch(&mut self, _id: String, _suffix: Option) -> Result<(), ErrorCode> { - Err(ErrorCode::NoSuchProcess) + // 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. + Err(ErrorCode::Unknown( + "watch: host lifecycle events deferred (publish-authority — RFC host_abi)".to_string(), + )) } fn unwatch(&mut self, _id: String) -> Result<(), ErrorCode> { - Err(ErrorCode::NoSuchProcess) + // Idempotent: nothing is armed while `watch` is deferred. + Ok(()) } } 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..4ab83f9f1 --- /dev/null +++ b/crates/astrid-capsule/src/engine/wasm/host/process/persistent/config.rs @@ -0,0 +1,90 @@ +//! 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, length-cap), or derive from +/// `cmd`. The label is NOT an identity — only the `process-id` is. +pub(super) fn clamp_label(label: Option, cmd: &str) -> String { + let raw = label.unwrap_or_else(|| cmd.to_string()); + raw.chars() + .filter(|c| !c.is_control()) + .take(MAX_LABEL_BYTES) + .collect() +} + +/// 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) +} 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..1650f154e --- /dev/null +++ b/crates/astrid-capsule/src/engine/wasm/host/process/persistent/entry.rs @@ -0,0 +1,274 @@ +//! 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 { + 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. The `id` is left empty — a token is + /// never echoed back out of the registry. + 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: String::new(), + 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 +} + +/// 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; 8192]; + 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> { + #[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..a38238fe1 --- /dev/null +++ b/crates/astrid-capsule/src/engine/wasm/host/process/persistent/mod.rs @@ -0,0 +1,570 @@ +//! `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 { + 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) + } + + /// 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..d3b590fd1 --- /dev/null +++ b/crates/astrid-capsule/src/engine/wasm/host/process/persistent/registry_tests.rs @@ -0,0 +1,198 @@ +//! 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