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, with
default CheckoutOptions — see Self::checkout_with.
pub async fn checkout_with(
&self,
repl: &ReplConfig,
options: CheckoutOptions,
) -> Result<Checkout, PoolError>
Dedicates a worker to one REPL session created from repl, with the
host context options carries.
Takes an idle worker when one exists, spawns or dials a new one (with
options.connect_headers, preceded by the W3C trace headers of
options.telemetry) while below max_processes, and otherwise waits
up to checkout_timeout (forever when None) before failing with
PoolError::Exhausted.
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>,
/// Where pool and turn metrics are recorded, from
/// `TelemetryAdapterHandle::metrics`.
/// `None` records nothing at all. Independent of tracing: metrics cover
/// every checkout, traced or not. Worker up/down counters total over all
/// pools that record into the same host meter.
pub metrics: Option<Metrics>,
}
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 — except that this checkout’s configured max_suspensions stays
as a ceiling on the re-adopted one; the count restarts at zero.
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: impl Into<String>,
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). The session’s first feed
sets the sandbox working directory to its first mount’s virtual path,
or / without mounts; later feeds keep the directory (os.chdir
included) unless Checkout::feed_with_cwd switches it. 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 feed_with_cwd(
&mut self,
code: impl Into<String>,
inputs: Vec<(String, MontyObject)>,
mounts: Vec<MountSpec>,
cwd: Option<&str>,
skip_type_check: bool,
on_print: OnPrint<'_>,
) -> Result<TurnEvent, PoolError>
Checkout::feed with an explicit switch of the sandbox working
directory before the feed: an absolute POSIX virtual path that
os.getcwd() reports and relative paths resolve against. None keeps
the session’s current directory, which the first feed defaults to its
first mount (else /).
A relative or NUL-containing cwd is a session-preserving
PoolError::Runtime (ValueError), raised before any frame is sent.
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”.
Also accepts exactly one matching result for a call with allow_eager_await set.
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 parent-side
limits work as on the typed path; a raw Load re-adopts its 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.
pub fn callback_context(&self) -> opentelemetry::Context
The innermost telemetry context for host callbacks answering this checkout.
Implements: Drop.
pub struct ReplConfig {
/// Script name used in tracebacks and type-check diagnostics, and the
/// basis of the sandbox's `__file__`: its final path component placed
/// under the working directory a feed starts in (`/main.py` at the root).
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,
/// How long the worker may hold buffered `print()` output before sending
/// it, batching a burst of prints into one `Print` event instead of one
/// each. `None` takes the worker's default
/// (`DEFAULT_PRINT_FLUSH_INTERVAL`); `Duration::ZERO` restores line
/// buffering, delivering each completed line on its own.
///
/// Output is always flushed before a suspension or a turn ends, so this
/// only sets how long live output may lag — never what arrives, or in
/// what order. The wire carries whole milliseconds, so a positive interval
/// below 1 ms is sent as 1 ms rather than rounding down into the
/// line-buffering sentinel.
pub print_flush_interval: Option<std::time::Duration>,
}
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>, allow_eager_await: bool },
/// 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;
Telemetry for the pool: spans describing what each session did, and aggregate metrics over the fleet running them.
This root is the host-facing surface both are delivered through — the
TelemetryAdapter bridge for a host whose SDK is in another language,
and TelemetryContext::for_logfire / Metrics::for_logfire for a Rust
host that already owns one. A foreign-language binding can either export
metrics aggregated by the statically linked SDK or stream raw measurements
into its host SDK. The recorders behind this surface are internal:
tracing mirrors the protocol into spans, metrics records instruments,
and tracing_json encodes span attributes the way the Logfire SDKs do.
pub struct Measurement<'a> {
/// Which kind of instrument records this measurement.
pub kind: MetricKind,
/// Dotted instrument name, e.g. `monty.pool.checkout.wait`.
pub name: &'static str,
/// UCUM unit: `s`, `By`, `1`, or a `{thing}` annotation for counts.
pub unit: &'static str,
/// One-line description of what the instrument measures.
pub description: &'static str,
/// The measured value.
pub value: MetricValue,
/// Low-cardinality dimensions to record it under.
pub attributes: &'a [opentelemetry::KeyValue],
}
One raw measurement and the metadata needed to create its instrument.
pub enum MetricKind {
/// Monotonic sum: the value is an increment.
Counter,
/// Non-monotonic sum: the value adjusts a current count.
UpDownCounter,
/// Distribution: the value is one sample.
Histogram,
}
The kind of host instrument a definition builds.
Implements: Clone, Copy, Debug, Eq, PartialEq, StructuralPartialEq.
pub enum MetricValue {
/// An integral count.
I64(i64),
/// A duration in seconds, or a ratio.
F64(f64),
}
A measured value, integral for counts and byte sizes, floating for durations and ratios.
Implements: Clone, Copy, Debug, PartialEq, StructuralPartialEq.
pub struct Metrics(/* private */);
Records pool measurements into a Rust or foreign-host metrics SDK.
Put on PoolConfig::metrics; cheap to clone
(one Arc). Foreign-language bindings obtain it from
TelemetryAdapterHandle::metrics,
while Rust hosts construct one with Metrics::for_logfire.
pub fn for_logfire(logfire: Logfire) -> Self
Builds Monty’s instruments on a configured Logfire meter.
Create one and clone it per pool: the worker counters then sum over all pools and sessions that record through the handle.
Implements: Clone, Debug.
pub const TELEMETRY_ADAPTER_VERSION: u8 = 1;
Current host-adapter protocol version used outside the Python binding.
pub struct TelemetryAdapterHandle { /* private fields */ }
Configured pipeline shared by all checkouts using one language 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 context_with_remote(
&self,
trace_id: &str,
span_id: &str,
trace_flags: u8,
trace_state: &str,
is_remote: bool,
) -> Result<TelemetryContext, String>
Validates host W3C fields while preserving whether the parent is remote.
pub fn context_from_ids(
&self,
trace_id: TraceId,
span_id: SpanId,
trace_flags: u8,
trace_state: &str,
is_remote: bool,
) -> Result<TelemetryContext, String>
Validates numeric W3C fields without a text round-trip.
pub fn unparented_context(&self) -> TelemetryContext
Creates adapter context when no host span is active.
pub fn metrics(&self) -> Metrics
Metrics handle for PoolConfig::metrics.
Unlike spans, metrics do not travel per checkout: they are aggregates covering untraced sessions and worker lifecycle events that belong to no session at all. Every call returns the same shared recorder, so the worker counters sum over all pools configured from this handle.
pub fn force_flush(&self) -> OTelSdkResult
Flushes the statically linked telemetry pipeline.
Adapters receiving aggregated metrics should call this from their flush path. Raw measurements are delivered synchronously and instead follow the foreign SDK’s flush lifecycle.
pub struct TelemetryContext { /* private fields */ }
Distributed parent context and recorder for one checkout root.
pub fn for_logfire(logfire: Logfire) -> Self
Records the pool’s spans straight into logfire: a Rust host already
owns a configured SDK, so it needs no TelemetryAdapter bridge.
Unparented, like TelemetryAdapterHandle::unparented_context — an
in-process ambient span attaches through tracing, not W3C fields.
pub trait TelemetryAdapter: Send + Sync + 'static {
fn start_span(&self, span: &SpanData) -> bool;
fn start_span_with_parent(&self, span: &SpanData, parent: &Context) -> 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);
fn record_metric(&self, measurement: &Measurement<'_>) { ... }
fn export_metrics(&self, payload: &[u8]) { ... }
}
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.
Starts a span with its complete parent context when the adapter needs it.
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. Called after every in-flight callback for that root has returned.
Records one raw metric measurement in the foreign host’s metrics SDK.
Defaults to dropping the measurement because version-1 adapters receive aggregated OTLP batches instead.
Exports one aggregated OTLP ExportMetricsServiceRequest protobuf.
Version-1 adapters receive these batches. New adapters can instead use
configure_telemetry_adapter_with_host_metrics so the foreign SDK
owns aggregation and receives Self::record_metric calls.
pub fn configure_telemetry_adapter(
adapter: std::sync::Arc<dyn TelemetryAdapter>,
) -> Result<TelemetryAdapterHandle, logfire::ConfigureError>;
Configures the statically linked Logfire pipeline used by a language binding.
pub fn configure_telemetry_adapter_with_host_metrics(
adapter: std::sync::Arc<dyn TelemetryAdapter>,
) -> Result<TelemetryAdapterHandle, logfire::ConfigureError>;
Configures tracing in Rust while the foreign host owns metrics aggregation.
Every measurement is delivered through TelemetryAdapter::record_metric
and therefore follows the host SDK’s views, readers, and exporters.
pub mod telemetry_adapter;
Telemetry for the pool: spans describing what each session did, and aggregate metrics over the fleet running them.
This root is the host-facing surface both are delivered through — the
TelemetryAdapter bridge for a host whose SDK is in another language,
and TelemetryContext::for_logfire / Metrics::for_logfire for a Rust
host that already owns one. A foreign-language binding can either export
metrics aggregated by the statically linked SDK or stream raw measurements
into its host SDK. The recorders behind this surface are internal:
tracing mirrors the protocol into spans, metrics records instruments,
and tracing_json encodes span attributes the way the Logfire SDKs do.
pub struct Measurement<'a> {
/// Which kind of instrument records this measurement.
pub kind: MetricKind,
/// Dotted instrument name, e.g. `monty.pool.checkout.wait`.
pub name: &'static str,
/// UCUM unit: `s`, `By`, `1`, or a `{thing}` annotation for counts.
pub unit: &'static str,
/// One-line description of what the instrument measures.
pub description: &'static str,
/// The measured value.
pub value: MetricValue,
/// Low-cardinality dimensions to record it under.
pub attributes: &'a [opentelemetry::KeyValue],
}
One raw measurement and the metadata needed to create its instrument.
pub enum MetricKind {
/// Monotonic sum: the value is an increment.
Counter,
/// Non-monotonic sum: the value adjusts a current count.
UpDownCounter,
/// Distribution: the value is one sample.
Histogram,
}
The kind of host instrument a definition builds.
Implements: Clone, Copy, Debug, Eq, PartialEq, StructuralPartialEq.
pub enum MetricValue {
/// An integral count.
I64(i64),
/// A duration in seconds, or a ratio.
F64(f64),
}
A measured value, integral for counts and byte sizes, floating for durations and ratios.
Implements: Clone, Copy, Debug, PartialEq, StructuralPartialEq.
pub struct Metrics(/* private */);
Records pool measurements into a Rust or foreign-host metrics SDK.
Put on PoolConfig::metrics; cheap to clone
(one Arc). Foreign-language bindings obtain it from
TelemetryAdapterHandle::metrics,
while Rust hosts construct one with Metrics::for_logfire.
pub fn for_logfire(logfire: Logfire) -> Self
Builds Monty’s instruments on a configured Logfire meter.
Create one and clone it per pool: the worker counters then sum over all pools and sessions that record through the handle.
Implements: Clone, Debug.
pub const TELEMETRY_ADAPTER_VERSION: u8 = 1;
Current host-adapter protocol version used outside the Python binding.
pub struct TelemetryAdapterHandle { /* private fields */ }
Configured pipeline shared by all checkouts using one language 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 context_with_remote(
&self,
trace_id: &str,
span_id: &str,
trace_flags: u8,
trace_state: &str,
is_remote: bool,
) -> Result<TelemetryContext, String>
Validates host W3C fields while preserving whether the parent is remote.
pub fn context_from_ids(
&self,
trace_id: TraceId,
span_id: SpanId,
trace_flags: u8,
trace_state: &str,
is_remote: bool,
) -> Result<TelemetryContext, String>
Validates numeric W3C fields without a text round-trip.
pub fn unparented_context(&self) -> TelemetryContext
Creates adapter context when no host span is active.
pub fn metrics(&self) -> Metrics
Metrics handle for PoolConfig::metrics.
Unlike spans, metrics do not travel per checkout: they are aggregates covering untraced sessions and worker lifecycle events that belong to no session at all. Every call returns the same shared recorder, so the worker counters sum over all pools configured from this handle.
pub fn force_flush(&self) -> OTelSdkResult
Flushes the statically linked telemetry pipeline.
Adapters receiving aggregated metrics should call this from their flush path. Raw measurements are delivered synchronously and instead follow the foreign SDK’s flush lifecycle.
pub struct TelemetryContext { /* private fields */ }
Distributed parent context and recorder for one checkout root.
pub fn for_logfire(logfire: Logfire) -> Self
Records the pool’s spans straight into logfire: a Rust host already
owns a configured SDK, so it needs no TelemetryAdapter bridge.
Unparented, like TelemetryAdapterHandle::unparented_context — an
in-process ambient span attaches through tracing, not W3C fields.
pub trait TelemetryAdapter: Send + Sync + 'static {
fn start_span(&self, span: &SpanData) -> bool;
fn start_span_with_parent(&self, span: &SpanData, parent: &Context) -> 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);
fn record_metric(&self, measurement: &Measurement<'_>) { ... }
fn export_metrics(&self, payload: &[u8]) { ... }
}
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.
Starts a span with its complete parent context when the adapter needs it.
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. Called after every in-flight callback for that root has returned.
Records one raw metric measurement in the foreign host’s metrics SDK.
Defaults to dropping the measurement because version-1 adapters receive aggregated OTLP batches instead.
Exports one aggregated OTLP ExportMetricsServiceRequest protobuf.
Version-1 adapters receive these batches. New adapters can instead use
configure_telemetry_adapter_with_host_metrics so the foreign SDK
owns aggregation and receives Self::record_metric calls.
pub fn configure_telemetry_adapter(
adapter: std::sync::Arc<dyn TelemetryAdapter>,
) -> Result<TelemetryAdapterHandle, logfire::ConfigureError>;
Configures the statically linked Logfire pipeline used by a language binding.
pub fn configure_telemetry_adapter_with_host_metrics(
adapter: std::sync::Arc<dyn TelemetryAdapter>,
) -> Result<TelemetryAdapterHandle, logfire::ConfigureError>;
Configures tracing in Rust while the foreign host owns metrics aggregation.
Every measurement is delivered through TelemetryAdapter::record_metric
and therefore follows the host SDK’s views, readers, and exporters.
pub struct CheckoutOptions {
/// Distributed trace context captured by a host adapter. Its span also
/// goes out as `traceparent`/`tracestate` on a WebSocket dial, so a remote
/// worker's own spans join the host's trace.
pub telemetry: Option<TelemetryContext>,
/// Extra headers for this checkout's WebSocket upgrade request; the
/// subprocess transport makes no request and ignores them. Duplicate
/// names are last-wins, even against the default `user-agent`
/// (`monty-pool/<version>`), the `traceparent` from `telemetry`, `host`
/// and the other handshake headers, and a malformed name or value fails
/// the dial.
pub connect_headers: Vec<(String, String)>,
}
Host-side context for one checkout, as opposed to the ReplConfig the
worker is sent.
pub fn with_telemetry(self, telemetry: Option<TelemetryContext>) -> Self
Sets the distributed trace context, if the host captured one.
pub fn with_connect_headers(self, connect_headers: Vec<(String, String)>) -> Self
Sets the headers for this checkout’s WebSocket upgrade request.
Implements: Debug, Default.
DEFAULT_PRINT_FLUSH_INTERVAL— re-exported frommonty_proto.MAX_VALUE_DEPTH— re-exported frommonty_proto.exceeds_max_value_depth— re-exported frommonty_proto.