monty-pool
An async pool of monty worker subprocesses: crash isolation, hard per-turn timeouts and elastic scaling for running untrusted Python. This is the recommended Rust embedding surface — see the Rust quickstart.
Documented with the telemetry-adapter feature(s) enabled.
pub struct Pool { /* private fields */ }
An elastic pool of monty subprocess workers.
min_processes workers spawn eagerly so the first checkout is fast;
further workers spawn on demand up to max_processes, and dead workers
are detected and replaced transparently. See the crate docs for the full
lifecycle.
Pool is safe to share across tasks and threads. Pool::close asks idle
workers to exit cleanly; merely dropping the pool kills them instead (via
their kill-on-drop handles). Workers held by live Checkouts die when
those are finished or dropped.
pub async fn new(config: PoolConfig) -> Result<Self, PoolError>
Creates the pool and eagerly spawns min_processes workers, failing
fast if the binary cannot be spawned. Must be called within a tokio
runtime (worker process and pipe I/O is driven by the runtime).
pub async fn checkout(&self, repl: &ReplConfig) -> Result<Checkout, PoolError>
Dedicates a worker to one REPL session created from repl.
Takes an idle worker when one exists, spawns a new one while below
max_processes, and otherwise waits up to checkout_timeout
(forever when None) before failing with PoolError::Exhausted.
pub async fn checkout_with_telemetry(
&self,
repl: &ReplConfig,
context: TelemetryContext,
) -> Result<Checkout, PoolError>
Checks out a session with distributed context captured by a host adapter.
pub async fn close(&self)
Asks idle workers to exit cleanly and reaps them, capping the wait per worker. Sessions still checked out keep their workers until they finish.
Optional: dropping the pool kills idle workers instead, which is just as safe — this only trades a SIGKILL for a clean protocol goodbye.
Telemetry exporter shutdown remains the configuring application’s responsibility and should happen after checked-out sessions finish.
pub fn idle_workers(&self) -> usize
Number of idle workers right now (diagnostics/tests only — the value is stale the moment it is returned).
pub fn idle_worker_pids(&self) -> Vec<u32>
PIDs of the idle workers (diagnostics/tests only).
pub struct PoolConfig {
/// Workers spawned eagerly at pool creation and kept warm. Forced to 0 for
/// the `MontyTransport::Websocket` transport (connections are made
/// per-checkout, not pre-warmed).
pub min_processes: usize,
/// Hard cap on live workers; checkouts beyond this wait.
pub max_processes: usize,
/// How workers are reached (spawned locally or connected to remotely).
pub transport: MontyTransport,
/// How long `Pool::checkout` waits for a free worker before
/// `PoolError::Exhausted`. `None` waits forever.
pub checkout_timeout: Option<std::time::Duration>,
/// Parent-side hard deadline per protocol turn: when it expires the
/// worker is killed and the call fails with `PoolError::Timeout`. This
/// backstops the child-side `ResourceLimits` — it also catches a child
/// that hangs in ways the sandbox limits cannot see. Synchronous host
/// telemetry callbacks prevent the timer from being polled while they run.
pub request_timeout: Option<std::time::Duration>,
/// Grace period for the automatic `max_duration` backstop.
///
/// When a session has a `ResourceLimits::max_duration` budget, the worker
/// reports its cumulative execution time on every turn-ending event (the
/// sandbox clock is the single source of truth: it runs only while the
/// interpreter executes, never during suspensions waiting on the host or
/// between feeds), and the parent bounds each execution turn by the
/// remaining budget plus this grace.
pub duration_limit_grace: Option<std::time::Duration>,
/// Recycle (kill and respawn) a worker after this many checkouts, to
/// bound the impact of any slow leak in a long-lived child.
pub max_checkouts_per_worker: Option<u32>,
}
Configuration for a Pool.
pub fn subprocess(binary_path: impl Into<PathBuf>) -> Self
Creates a subprocess-transport config with defaults: min_processes = 1,
max_processes = available parallelism, no timeouts, a 1s
duration_limit_grace, no recycling.
pub fn websocket(url: impl Into<String>) -> Self
Creates a WebSocket-transport config dialing url verbatim per checkout.
min_processes is 0 (no pre-warming — connections are made per-checkout).
Implements: Clone, Debug.
pub struct Checkout { /* private fields */ }
One worker dedicated to one REPL session.
Obtained from Pool::checkout. Checkout::finish returns the
worker to the pool; dropping without finishing kills the worker instead —
mid-execution state cannot be trusted back into the pool.
Turn futures (feed, resume*, dump, …) are not resumable after
being dropped mid-flight: the request may already be on the wire (or a
mount’s host I/O abandoned mid-service), so the protocol state is
unknowable. The checkout notices on its next call,
discards the worker, and fails with PoolError::Protocol; finish on
such a session likewise discards the worker rather than returning it.
pub async fn restore(
&mut self,
state: Vec<u8>,
mounts: Vec<MountSpec>,
on_print: OnPrint<'_>,
) -> Result<(Option<TurnEvent>, Option<String>), PoolError>
Restores a dumped session into this checkout’s freshly configured (but
not-yet-fed) worker, returning the re-announced suspension event when the
dump was taken mid-feed (None for an idle, between-feeds dump).
This is the low-level restore both session.load_session (idle dumps) and
session.load_snapshot (suspended dumps) drive: the caller inspects the
returned Option to tell which kind of dump it was and reject a
mismatch. Only valid before the worker has been fed (the child rejects a
Load once a repl exists).
mounts re-establish a suspended feed’s mounts, which are never part of
the dump (they are host configuration the parent services itself). Pass
the same mounts the original feed used, so the resumed feed’s covered
calls can still be answered by Checkout::resume_from_mounts. A dump
taken mid-OS-call re-announces the call in full, so the returned event
is that same TurnEvent::OsCall — restoring never answers it here.
The session’s resource budget is taken from the dump, so the prior
Configure limits are dropped here and re-adopted from the worker’s
reply.
Returns the re-announced suspension (Some — a suspended dump) or None
(an idle dump), paired with the worker’s adopted script name (the dump’s,
not the Configure one), which the parent surfaces in restored snapshots.
pub async fn feed(
&mut self,
code: &str,
inputs: Vec<(String, MontyObject)>,
mounts: Vec<MountSpec>,
skip_type_check: bool,
on_print: OnPrint<'_>,
) -> Result<TurnEvent, PoolError>
Executes one snippet against the session. Inputs become sandbox
globals; mounts apply to this feed only and are serviced by the parent
(an invalid host path fails here, before any frame is sent, as a
session-preserving PoolError::Runtime). Returns the first
suspension (or completion); print() output streams to on_print
throughout.
PoolError::Runtime / PoolError::Typing leave the session
usable; all other errors mean the worker was discarded.
pub async fn resume(
&mut self,
value: ResumeValue,
on_print: OnPrint<'_>,
) -> Result<TurnEvent, PoolError>
Answers a TurnEvent::FunctionCall or TurnEvent::OsCall.
pub async fn resume_from_mounts(
&mut self,
on_print: OnPrint<'_>,
) -> Result<Option<TurnEvent>, PoolError>
Answers a pending TurnEvent::OsCall from this feed’s mounts, when
they cover it.
Ok(None) means no mount covers the call (or the feed has none): the
suspension is left intact for the caller to answer itself, typically via
its own os handler and then Checkout::resume. Ok(Some(event))
means a mount serviced the call — including servicing it into an error
such as PermissionError — and the feed ran on to event.
This is how mounts are reached now that every OS call surfaces: an
auto-answering driver tries mounts first and falls back to its handler,
while a caller driving suspensions by hand can ignore mounts entirely.
Path containment inside covered calls is enforced by the MountTable.
Covered calls perform real host filesystem I/O, serviced on tokio’s blocking pool so a stalled volume cannot pin a runtime worker. Dropping this future mid-servicing abandons the feed’s mount state and is treated exactly like a cancellation mid-turn: the worker is discarded on the next call.
pub async fn resume_name_lookup(
&mut self,
result: impl Into<NameLookupResult>,
on_print: OnPrint<'_>,
) -> Result<TurnEvent, PoolError>
Answers a TurnEvent::NameLookup with a NameLookupResult (or a
MontyObject, an Option<MontyObject> where None is Undefined, or
a MontyException for Error): a value resolves the name; Undefined
makes the sandbox raise NameError for a plain lookup, or
AttributeError when the lookup carried an object_id (a lazy
attribute on a host-backed object — a class instance or class type);
Error raises the host’s exception in the sandbox, bypassing
hasattr() / getattr() defaults the way a raising property does.
pub async fn resume_futures(
&mut self,
results: Vec<(u32, ResumeValue)>,
on_print: OnPrint<'_>,
) -> Result<TurnEvent, PoolError>
Answers a TurnEvent::ResolveFutures with results for some or all
pending call ids. Each result must be Return or Error — a future
cannot resolve to another future or to “not found”.
pub async fn install_dependencies(
&mut self,
requirements: Vec<String>,
) -> Result<(), PoolError>
Installs third-party Python packages into the session, making them
importable by subsequent feeds. Session-scoped and repeatable; an empty
requirements list is a no-op.
Only an embedded-CPython worker honors this. The
monty sandbox worker has no host interpreter to install for and a uv
install failure both surface as PoolError::Runtime (the latter
carrying uv’s stderr); the session stays usable in either case. Bounded
by the pool’s request_timeout, so raise it for large dependency sets.
Each requirement is validated here, at the pool boundary, before any
frame is sent: a string that uv would parse as an option rather than a
package specifier is rejected with PoolError::Runtime (a
ValueError). See validate_requirement for the rationale.
pub async fn dump(&mut self) -> Result<Vec<u8>, PoolError>
Serializes the session (idle or suspended) into opaque bytes that
Checkout::restore can restore — including into a
different worker after this one crashes. The session stays live.
pub async fn finish(self) -> Result<(), PoolError>
Ends the session and returns the worker to the pool.
Consumes the checkout. On error the worker is discarded (and the error reported), but the pool remains healthy either way.
pub fn pid(&self) -> Option<u32>
OS process id of the worker, when it is a local subprocess (None for a
remote WebSocket worker, or a finished checkout). Diagnostics/tests.
pub async fn turn_raw(
&mut self,
request: &pb::ParentRequest,
on_event: OnRawEvent<'_>,
) -> Result<pb::ChildEvent, PoolError>
Sends request and returns the child’s turn-ending event, as protobuf —
no conversion to TurnEvent.
For callers that already speak the wire (a relay bridging a remote
client): rebuilding a ChildEvent frame from a TurnEvent means
hand-inverting the whole protocol, so this hands back what the child
actually sent. on_event sees each streamed Print before the
turn-ender.
Bypasses this checkout’s suspension bookkeeping (pending,
feed_mounts, restored_script_name) — never interleave with
feed/resume/restore. Worker lifecycle, poisoning and the
max_duration backstop work as on the typed path; a raw Load
re-adopts the dump’s budget like Checkout::restore. A FatalError
(or WebSocket ShutdownDump) turn-ender is returned so the driver can
forward it, but discards the worker first — later calls report
PoolError::Finished.
request is typically a remote client’s, so it is treated as hostile:
Configure/Reset/Shutdown are refused here (a client could
otherwise Reset away the operator-chosen resource limits and
re-Configure its own). A Load’s bytes DO reach the worker’s
deserialiser — the driver must verify a dump is one it issued
(monty-server signs and checks them) before passing it in.
As Checkout::feed: a dead worker, a protocol violation, or a turn
that outlived request_timeout or the remaining max_duration budget.
Implements: Drop.
pub struct ReplConfig {
/// Script name used in tracebacks and type-check diagnostics.
pub script_name: String,
/// Sandbox resource limits enforced inside the worker. `None` means
/// unlimited (except monty's standard recursion-depth default).
pub limits: Option<monty_types::ResourceLimits>,
/// Type-check every fed snippet before executing it.
pub type_check: bool,
/// Stub declarations made available to type checking.
pub type_check_stubs: Option<String>,
/// How the worker renders typing diagnostics. Chosen here rather than on
/// the raised error because the structured diagnostics never leave the
/// worker — only the rendered text crosses the wire.
pub type_check_config: monty_types::TypeCheckingConfig,
/// Give failed `assert` statements pytest-style introspected messages
/// (see `limitations/assert.md`). On by default with a 120-byte
/// operand-repr truncation; `MaxBytes` customizes the truncation.
pub assert_message_annotations: monty_types::AssertMessageAnnotations,
}
Arguments for the REPL session a checkout creates — mirrors
MontyRepl’s constructor surface.
Implements: Clone, Debug, Default.
pub enum TurnEvent {
/// The sandbox called an external function — answer with
/// `Checkout::resume`. When `object_id` is set this is a method call on
/// a host-backed object, routed by uuid — a class instance, or a class
/// type (a classmethod call, or construction of a host class, which is
/// spelled `__call__`); the receiver is NOT included in `args`.
FunctionCall { function_name: String, args: Vec<monty_types::MontyObject>, kwargs: Vec<(monty_types::MontyObject, monty_types::MontyObject)>, call_id: u32, object_id: Option<monty_types::MontyUuid> },
/// The sandbox performed an OS operation (e.g. `"Path.read_text"`).
/// Answer it from this feed's mounts with
/// `Checkout::resume_from_mounts`, or directly with
/// `Checkout::resume`. A caller with no handler should resume with
/// `ResumeValue::NotHandled`; the sandbox then raises the call's own
/// no-handler default.
OsCall { function_name: String, args: Vec<monty_types::MontyObject>, kwargs: Vec<(monty_types::MontyObject, monty_types::MontyObject)>, call_id: u32 },
/// The sandbox read an undefined name, or — when `object_id` is set — a
/// lazy attribute on the host-backed object with that uuid (a class
/// instance, or a class type) — answer with
/// `Checkout::resume_name_lookup`. An `Undefined` (or `None`) answer
/// raises `NameError` for plain lookups, `AttributeError` for attribute
/// lookups; an `Error` answer raises the host's exception in the sandbox.
NameLookup { name: String, object_id: Option<monty_types::MontyUuid> },
/// Every sandbox task is blocked on external futures — answer with
/// `Checkout::resume_futures`.
ResolveFutures { pending_call_ids: Vec<u32> },
/// The fed snippet completed with this value; the session is ready for
/// the next `Checkout::feed`.
Complete(monty_types::MontyObject),
}
How a protocol turn ended: a suspension that needs an answer from the caller, or completion of the fed snippet.
Implements: Debug.
pub enum ResumeValue {
/// The call returned this value.
Return(monty_types::MontyObject),
/// The call raised this exception.
Error(monty_types::MontyException),
/// The call is asynchronous: register an external future and continue
/// other tasks; resolve later via `Checkout::resume_futures`.
Future,
/// No handler exists for the called name — the sandbox raises
/// `NameError`.
NotFound,
/// No handler accepted this OS call — the sandbox raises the call's own
/// no-handler default (`PermissionError` naming the path for filesystem
/// calls, `RuntimeError` for the rest). Only valid answering a
/// `TurnEvent::OsCall`.
NotHandled,
}
The caller’s answer to a TurnEvent::FunctionCall or
TurnEvent::OsCall.
Implements: Debug.
pub type OnPrint<'a> = &'a mut dyn FnMut(monty_types::PrintStream, &str) -> PrintFuture + Send;
Callback receiving sandbox print() output streamed during a turn.
The callback returns a future so genuinely async sinks (a JS callback, a
socket) can be awaited per fragment; synchronous sinks wrap themselves
with on_print_sync. The future must be 'static, so it captures owned
copies of whatever it needs (including the text, if consumed async).
pub type OnRawEvent<'a> = &'a mut dyn FnMut(&pb::ChildEvent) -> PrintFuture + Send;
Callback for the events a Checkout::turn_raw streams before the
turn-ender — Prints today. Returns a future for the same reason
OnPrint does: a slow sink backpressures the worker.
pub type PrintFuture = std::pin::Pin<Box<dyn Future<Output = ()> + Send>>;
The (boxed) future an OnPrint callback returns for one fragment; the
turn awaits it before reading on, so a slow sink backpressures the worker.
pub fn on_print_sync<F>(
sink: F,
) -> impl FnMut(monty_types::PrintStream, &str) -> PrintFuture + Send
where
F: FnMut(monty_types::PrintStream, &str) + Send;
Adapts a synchronous print sink to the OnPrint callback shape.
let mut on_print = on_print_sync(|_stream, text| print!("{text}"));
// session.feed("print('hi')", vec![], vec![], false, &mut on_print).await?;
pub struct MountSpec {
/// Access mode.
pub mode: MountSpecMode,
/// Cap on total bytes written through this mount.
pub write_bytes_limit: Option<u64>,
/// Aggregate budget for retained overlay data and transient results.
pub memory_usage_limit: u64,
/* private fields */
}
A host directory mounted into the sandbox for one feed. Mounts are handled
entirely on the parent: the checkout services covered filesystem OS calls
from the host path itself (so mounts work even when the worker runs on a
remote machine). Every OS call still surfaces as a TurnEvent::OsCall;
mounts are consulted only when the caller asks, via
Checkout::resume_from_mounts.
pub fn new(
virtual_path: &str,
host_path: impl AsRef<Path>,
mode: MountSpecMode,
) -> Result<Self, PoolError>
Opens host_path and creates mount configuration with the default
100 MB memory budget and no cumulative write limit.
Build this once and reuse it; each call resolves the path afresh. The
open is blocking filesystem I/O, so an async caller opening a directory
that may stall (NFS, FUSE) should either build the spec before entering
the runtime or open the MountRoot under spawn_blocking and pass it
to Self::from_root. Feeds never reopen it, so this cost is paid once
per mount rather than once per feed.
Returns PoolError::Runtime if the virtual path is not absolute, or
the host path cannot be opened as a directory.
pub fn from_root(root: MountRoot, mode: MountSpecMode) -> Self
Creates mount configuration from an already-opened MountRoot, for
hosts that open it themselves to map failures their own way.
pub fn virtual_path(&self) -> &str
Returns the normalized virtual path this mount answers on.
pub fn host_path(&self) -> &Path
Returns the host directory path. Diagnostics only — operations run against the descriptor, not this path.
Implements: Clone, Debug.
pub enum MountSpecMode {
/// Reads only; writes raise `PermissionError` in the sandbox.
ReadOnly,
/// Files written by sandboxed code persist on the host and are untrusted;
/// the host must not execute them, including indirectly via a Python
/// `import` when the directory is on `sys.path`. `Self::Overlay` keeps
/// writes in memory instead.
ReadWrite,
/// Copy-on-write overlay in parent memory; writes are discarded when the
/// feed ends.
Overlay,
}
Access mode for a MountSpec.
Implements: Clone, Copy, Debug, Eq, PartialEq, StructuralPartialEq.
pub enum MontyTransport {
/// Spawn a local `monty subprocess` child and talk to it over framed
/// stdio pipes. Takes path to the `monty` (or compatible child) binary.
Subprocess(std::path::PathBuf),
/// Connect *out* to a remote child over a WebSocket — either a relay (which
/// pairs this connection with a child that dialed in with the same session
/// id) or a child running a server. One binary message per protocol frame.
///
/// The URL is dialed verbatim — if a relay needs the two ends to share a
/// session id in the path (`/<uuid>/parent`), the caller is responsible for
/// putting it there. Takes full `ws://`/`wss://` URL to dial.
Websocket(String),
}
How the pool reaches its workers.
Implements: Clone, Debug.
pub enum PoolError {
/// The worker is gone: it died (segfault, abort, external kill, or EOF on
/// its pipes), or it announced a `FatalError` and exited — which a serving
/// relay also uses to report that it could not start a worker at all. The
/// worker has been discarded; the pool stays usable.
Crashed { status: Option<std::process::ExitStatus>, cause: CrashCause },
/// The worker was killed after its turn outlived `request_timeout` (or
/// the `max_duration` backstop deadline).
Timeout { timeout: std::time::Duration },
/// The worker violated the wire protocol, or the caller violated the
/// checkout state machine. Worker-originated protocol failures discard the
/// worker; caller misuse leaves it intact — except a turn cancelled
/// mid-flight, where the abandoned worker is discarded too.
Protocol(std::borrow::Cow<'static, str>),
/// The sandboxed code raised a Python exception. The worker and its
/// session remain alive and usable — except for the one `MemoryError` a
/// worker exceeding its memory limit produces, where the worker is already
/// dead and the checkout finished (its distinct message distinguishes it
/// from the interpreter's own `MemoryError`).
Runtime(monty_types::MontyException),
/// Type checking rejected the fed snippet (sessions created with
/// `type_check`). The worker and session remain alive; the snippet did
/// not run.
Typing(String),
/// No worker became available within `checkout_timeout`.
Exhausted,
/// A worker process could not be spawned.
Spawn(String),
/// The checkout was already finished or its worker already discarded.
Finished,
/// The remote worker's connection dropped without a turn-ending event
/// (WebSocket transport only — the local analogue is `Self::Crashed`).
/// The sandbox may have died, or the server may have dropped the session
/// by policy (idle/session/turn timeout, capacity); the two are
/// indistinguishable to a client that only learns of the close, so this
/// deliberately says no more than "the connection went away".
Disconnected { context: String },
/// The remote server is shutting down and did **not** run the request —
/// re-running it on a fresh session is safe. `dump` carries the session
/// state captured just before shutdown, restorable via
/// `Checkout::restore` on a fresh checkout.
Shutdown { dump: Option<Vec<u8>> },
}
Why a pool operation failed.
Implements: Debug, Display, Error, From<Error>.
pub enum CrashCause {
/// It vanished — segfault, abort, external kill, EOF on its pipes — so
/// all the pool can say is what it was doing at the time.
Vanished { context: String },
/// It announced a `FatalError` and exited, so its own account replaces
/// the pool's. A serving relay also uses this to report that it could not
/// start a worker at all. A worker that exceeded its memory limit is not
/// here at all: that exit code classifies into
/// `PoolError::Runtime` carrying a `MemoryError`.
Announced { reason: String },
}
How the pool learned a worker had died — the two are mutually exclusive,
which is why they are one field rather than two Options: a worker that
states its own reason makes the pool’s note of what it was doing
redundant, and the message uses one or the other, never both.
Orthogonal to PoolError::Crashed’s status, which is present or not
in either case (a worker can announce a reason and be reaped for its
exit code — the usual shape of a version-skew death).
Implements: Debug.
pub mod telemetry_adapter;
Host-neutral bridge from Monty’s Rust spans to language SDK adapters.
pub const TELEMETRY_ADAPTER_VERSION: u8 = 1;
Current host-adapter protocol version.
pub struct TelemetryAdapterHandle { /* private fields */ }
Configured exporter-free pipeline shared by all checkouts using one adapter.
pub fn context(
&self,
trace_id: &str,
span_id: &str,
trace_flags: u8,
trace_state: &str,
) -> Result<TelemetryContext, String>
Validates host W3C fields and couples them to this adapter pipeline.
pub fn unparented_context(&self) -> TelemetryContext
Creates adapter context when no host span is active.
pub struct TelemetryContext { /* private fields */ }
Distributed parent context and isolated recorder for one checkout root.
pub fn for_logfire(logfire: Logfire) -> Self
Records the pool’s spans straight into logfire, with no adapter.
The rest of this module exists to hand spans to a foreign SDK, so its
pipeline is exporter-free and the host owns delivery. A Rust host has no
boundary to cross: it already has a configured Logfire, and the
recorder only needs one to write into. Without this it would have to
implement TelemetryAdapter and stand up a second OTLP exporter to
receive its own spans — or define a second span vocabulary of its own.
Unparented, like TelemetryAdapterHandle::unparented_context: a Rust
host’s ambient span is in the same process, so the pool’s roots attach
to it through tracing rather than through W3C fields.
pub trait TelemetryAdapter: Send + Sync + 'static {
fn start_span(&self, span: &SpanData) -> bool;
fn end_span(&self, span: &SpanData) -> bool;
fn emit_log(&self, parent_span_id: SpanId, record: &SdkLogRecord) -> bool;
fn disable_root(&self, trace_id: TraceId, root_span_id: SpanId);
}
Receives Monty records without owning exporters or credentials.
Returning false disables the affected Monty root and calls disable_root.
Starts a reconstructed host span keyed by its Rust OTel span ID.
Applies final data and ends a reconstructed host span.
Emits a log under the reconstructed span identified by parent_span_id.
Discards host-side state after delivery for one Monty root becomes unreliable.
pub fn configure_telemetry_adapter(
adapter: std::sync::Arc<dyn TelemetryAdapter>,
) -> Result<TelemetryAdapterHandle, logfire::ConfigureError>;
Configures the exporter-free Rust pipeline shared by language adapters.
MAX_VALUE_DEPTH— re-exported frommonty_proto.exceeds_max_value_depth— re-exported frommonty_proto.