Lines
90.41 %
Functions
40.23 %
Branches
100 %
use std::cell::RefCell;
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
use nomiscript::{
Compiler, Error as NomiError, Expr, HostFnSpec, Program, Reader, SymbolTable, Value,
};
use scripting::runtime::{
EngineError, EngineOpts, ModuleCache, build_engine, classify_runtime_error, decode_eval_result,
use thiserror::Error;
use tracing::debug;
use wasmtime::{AnyRef, Engine, Linker, Rooted, Store, Val};
use crate::ctx::{EpochBumper, InterruptHandle, ScriptCtx};
use crate::envelope::{
EnvelopeError, ErrorCode, Request, RequestId, Response, ResponsePayload, format_response,
parse_request,
const EPOCH_DEADLINE_TICKS: u64 = 1;
/// Wasmtime Store data type for the rpc eval channel. Carries the
/// per-session user context (`ScriptCtx`) so native fns reach
/// `caller.data().ctx().user_id` directly. The legacy
/// `EvalContext` capture-protocol companion field retired alongside
/// the capture imports in P4 A6.c — `nomi-eval` now returns its
/// final value via the function's `(ref null any)` return slot.
///
/// Native fns are async (`Linker::func_wrap_async`) — they `.await`
/// `server::command::*` futures directly on whatever runtime drives the
/// surrounding `Session::handle_form` call. No owned runtime, no
/// `spawn_blocking`, no thread-local pool concerns.
pub struct SessionData {
ctx: ScriptCtx,
/// Per-request stdout sink. `env.log` (PRINT/DISPLAY/NEWLINE) appends here so
/// the mREPL (`nms --slynk-port`) can surface script output as
/// `:write-string` separately from the final value. Shared with the owning
/// `Session`, which drains it per request. The non-mrepl paths (sshd / the
/// rpc text REPL) ignore it; `env.log` still tees to `tracing` for them.
output: Arc<Mutex<String>>,
/// Render-only draft accumulator. `Some` only on the restricted template
/// render path ([`crate::template`]); the draft natives mutate it and the
/// render entry point reads it back via `store.into_data()`. `None` on the
/// normal eval channel, where the draft natives are not even linked.
draft: Option<RefCell<crate::draft::TransactionDraft>>,
}
impl SessionData {
pub(crate) fn new(ctx: ScriptCtx, output: Arc<Mutex<String>>) -> Self {
Self {
ctx,
output,
draft: None,
/// Builds session data with a draft accumulator armed — the render path's
/// constructor. Draft natives require `draft` to be `Some`.
pub(crate) fn for_render(ctx: ScriptCtx, output: Arc<Mutex<String>>) -> Self {
draft: Some(RefCell::new(crate::draft::TransactionDraft::new())),
#[must_use]
pub fn ctx(&self) -> &ScriptCtx {
&self.ctx
/// Mutates the draft accumulator if armed (render path). Returns an error
/// on the normal eval path where no draft is present — a draft native
/// reaching a non-render Store is a wiring bug, surfaced as a trap.
pub fn with_draft<F>(&self, f: F) -> wasmtime::Result<()>
where
F: FnOnce(&mut crate::draft::TransactionDraft),
{
let cell = self
.draft
.as_ref()
.ok_or_else(|| wasmtime::Error::msg("draft native invoked outside render mode"))?;
f(&mut cell.borrow_mut());
Ok(())
/// Consumes the accumulated draft, if any. Called after a render run via
/// `store.into_data()`.
pub fn into_draft(self) -> Option<crate::draft::TransactionDraft> {
self.draft.map(RefCell::into_inner)
/// Appends a script-output line to the per-request buffer. Called by the
/// `env.log` host fn. A poisoned lock is swallowed (output capture is
/// best-effort telemetry, never a reason to fail an eval).
pub fn push_output(&self, msg: &str) {
if let Ok(mut buf) = self.output.lock() {
buf.push_str(msg);
/// Per-channel evaluator. Owns mutable state across forms (defun-defined symbols
/// persist between requests in the same session via `SymbolTable`; the wasm
/// `ModuleCache` reuses compilations of structurally identical forms) and an
/// interrupt handle that cooperatively short-circuits the next form on demand.
/// Eval pipeline: `nomiscript::Compiler::compile_with_mode(Eval)` emits a
/// module exporting `nomi-eval`; the form's final value rides the function's
/// `(ref null any)` return slot and the host walks it into an `EvalValue`
/// via [`decode_eval_result`].
pub struct Session {
engine: Engine,
compiler: Compiler,
cache: ModuleCache,
symbols: SymbolTable,
interrupt: InterruptHandle,
/// Watermark of the highest interrupt generation already attributed to a
/// finished request (see [`Session::check_interrupt`] / [`Session::ack_interrupt`]).
/// A `C-g` counts only while `interrupt.generation() > interrupt_ack`, so one
/// signal aborts exactly one request and can never linger to poison a later
/// form.
interrupt_ack: u64,
/// Shared with each per-request `SessionData` so `env.log` output lands
/// where [`Session::handle_request`] can drain it.
/// `None` on an unguarded session; otherwise when the account status may
/// next go unchecked until. See [`Session::guarded`].
status_check_after: Option<Instant>,
/// The same 30s as the tenant pool's `QUARANTINE_RECHECK`, kept in step by hand.
const STATUS_RECHECK: Duration = Duration::from_secs(30);
/// A structured eval result: the captured script output plus the typed
/// value/error payload. The SLYNK mREPL maps `output` → `:write-string` and
/// `payload` → `:write-values` / `:evaluation-aborted`. `handle_form` (the text
/// wire path) does not use this — it formats the payload alone.
#[derive(Debug, Clone, PartialEq)]
pub struct EvalOutcome {
pub output: String,
pub payload: ResponsePayload,
#[derive(Debug, Error)]
pub enum SessionError {
#[error("engine init failed: {0}")]
Engine(#[from] EngineError),
impl Session {
pub fn new(ctx: ScriptCtx) -> Result<Self, SessionError> {
let engine = build_engine(EngineOpts::baseline().with_fuel())?;
let host_fns = crate::natives::all_compiler_specs();
let mut symbols = SymbolTable::with_builtins();
symbols.register_host_fns(&host_fns);
// Host-dependent prelude (ADR-0029): loaded only here, after the RPC
// host fns it calls are registered. The universal prelude already rode
// in via `with_builtins`.
crate::host_prelude::load(&mut symbols);
let mut session = Self {
engine,
compiler: Compiler::with_host_fns(host_fns.clone()),
cache: ModuleCache::new(),
symbols,
interrupt: InterruptHandle::new(),
interrupt_ack: 0,
output: Arc::new(Mutex::new(String::new())),
status_check_after: None,
// Phase 4: pre-warm the per-Session ModuleCache with the
// bare-call wasm for every zero-arg host fn that has a
// return type. The cache is keyed by full bytecode bytes so
// when a request like `(:id N :form (rpc-protocol-version))`
// lands, `cache.get_or_compile` finds the pre-compiled
// module and `handle_form`'s critical path is
// instantiate-only. Composed forms / arg-bearing calls /
// unseen forms still hit the cold compile path.
session.warm_bare_call_cache(&host_fns);
Ok(session)
/// Pre-compile and cache the bare-call wasm for every zero-arg
/// host fn whose result type is non-None. Skips no-arg fns with
/// `result: None` (their bare call would error at value position)
/// and skips arg-bearing fns (their wasm embeds the literal
/// args, so pre-warming would be wrong-keyed).
/// Failures are silently ignored — pre-warming is an optimization,
/// not a correctness step. If a spec fails to compile here, the
/// runtime path will surface the same error when the form lands.
fn warm_bare_call_cache(&mut self, host_fns: &[HostFnSpec]) {
for spec in host_fns {
if !spec.params.is_empty() || spec.result.is_none() {
continue;
let form = Expr::List(vec![Expr::Symbol(spec.nomi_name.clone())]);
let program = Program::new(vec![form]);
let Ok((bytes, _ty)) = self
.compiler
.compile_eval_with_type(&program, &mut self.symbols)
else {
let _ = self.cache.get_or_compile(&self.engine, &bytes);
pub fn interrupt_handle(&self) -> InterruptHandle {
self.interrupt.clone()
/// Symbol names completing `prefix`, sorted and deduplicated, for the SLYNK
/// completion rex (`M-x sly-complete-symbol` / mREPL TAB). The reader folds
/// symbols with `make_ascii_uppercase` at read time, so the match uppercases
/// the prefix the SAME way (ASCII-only — matching the reader, not full
/// Unicode) and returns the canonical name (it re-reads identically). Skips
/// `(SETF …)` setf-place names and compiler-internal `$…` / `__…` symbols —
/// none are head symbols a user types. An empty `prefix` lists every
/// completable symbol.
pub fn completions(&self, prefix: &str) -> Vec<String> {
let needle = prefix.to_ascii_uppercase();
let mut names: Vec<String> = self
.symbols
.iter()
.map(|(name, _)| name.as_str())
.filter(|name| {
!name.starts_with('$') && !name.starts_with("__") && !name.starts_with("(SETF")
})
.filter(|name| name.starts_with(&needle))
.map(str::to_owned)
.collect();
names.sort_unstable();
names.dedup();
names
/// Cooperative cancel handle for an in-flight `nomi-eval`. Clone
/// and hand to the transport layer: when the client sends a
/// cancel signal (e.g. emacs `C-g`), call `.bump()` and the
/// awaiting evaluation traps with `EngineError::EpochInterrupt`,
/// surfacing on the wire as `(:error (:code interrupted ...))`.
pub fn epoch_bumper(&self) -> EpochBumper {
EpochBumper::new(self.engine.clone())
/// Number of pre-compiled wasm modules currently cached.
/// Used by tests to confirm the phase-4 pre-warm step ran and
/// that subsequent `handle_form` calls of cached forms don't
/// trigger a cold compile.
pub fn cache_size(&self) -> Result<usize, EngineError> {
self.cache.len()
pub async fn handle_form(&mut self, frame: &str) -> String {
let response = match self.evaluate(frame).await {
Ok(resp) => resp,
Err(err) => err.into_response(),
format_response(&response)
/// Structured eval of a single bare `form` (not an envelope frame): clears
/// the output buffer, runs the form, and returns the captured script output
/// alongside the typed value/error payload. The SLYNK mREPL uses this so it
/// can render `:write-string` (output) and `:write-values` /
/// `:evaluation-aborted` (payload) separately. `handle_form` is unchanged.
pub async fn handle_request(&mut self, source: &str) -> EvalOutcome {
buf.clear();
let payload = match self.eval_source(source).await {
Ok(value) => ResponsePayload::Value(value),
Err(err) => err.into_response().payload,
let output = self
.output
.lock()
.map(|buf| buf.clone())
.unwrap_or_default();
EvalOutcome { output, payload }
/// Parses `source` to a single top-level form and evaluates it. Unlike
/// `handle_form`, the source is parsed to an AST directly (NOT interpolated
/// into an envelope string), so plist-shaped input can't hijack the
/// envelope: `1 :form (+ 2 3)` is rejected as "more than one form", not
/// silently re-read as a `(:id 1 :form …)` plist. The fixed `RequestId::Int(0)`
/// is unused downstream — the SLYNK layer tracks its own channel ids.
async fn eval_source(&mut self, source: &str) -> Result<Value, EvalFailure> {
let id = RequestId::Int(0);
let program = Reader::parse(source).map_err(|err| EvalFailure::Eval(id.clone(), err))?;
let mut exprs = program.exprs;
let form = match exprs.len() {
0 => return Ok(Value::Nil),
1 => exprs.remove(0),
_ => {
return Err(EvalFailure::Eval(
id,
NomiError::Compile("expected a single form".to_string()),
));
self.eval_one_form(form).await
/// Reads `path`, evaluates every top-level form in source order (state —
/// `defun`s etc. — accumulates across forms, like the sshd channel), and
/// returns a short summary. Powers SLY's `M-x sly-load-file`
/// (`slynk:load-file`). Captured output across all forms is returned for the
/// caller to surface; a failing form aborts the load at that point.
pub async fn handle_file(&mut self, path: &str) -> EvalOutcome {
let payload = match self.load_path(path).await {
Ok(summary) => ResponsePayload::Value(Value::String(summary)),
/// Attributes all interrupts up to `observed` to the current request, so
/// they aren't counted again. Crucially the caller passes the SAME
/// generation it used to decide an interrupt is pending — never a fresh
/// reload — so a `C-g` that lands strictly after that observation point is
/// NOT folded into this request; it stays pending for the next form. The
/// guard keeps the watermark monotonic.
/// Coalescing is intended: several `C-g`s observed together (generation
/// jumped by >1) cancel the one in-flight form, they do not pre-arm
/// cancellation of distinct future forms — that is the REPL `C-g` contract.
fn ack_interrupt(&mut self, observed: u64) {
if observed > self.interrupt_ack {
self.interrupt_ack = observed;
/// If an interrupt is pending, ack it (using the one generation snapshot
/// that decided it) and yield the interrupted error; otherwise `None`. The
/// single checkpoint used everywhere a form can abort on a `C-g` before the
/// Wasm call.
fn check_interrupt(&mut self, id: &RequestId) -> Option<EvalFailure> {
let observed = self.interrupt.generation();
(observed > self.interrupt_ack).then(|| {
self.ack_interrupt(observed);
EvalFailure::Interrupted(id.clone())
async fn load_path(&mut self, path: &str) -> Result<String, EvalFailure> {
// The synchronous file read + parse aren't covered by `run`'s checks, so
// guard them explicitly; a `C-g` during a later form is caught by that
// form's `run`, and the failing form aborts the whole load via `?`.
if let Some(err) = self.check_interrupt(&id) {
return Err(err);
let source = std::fs::read_to_string(path).map_err(|err| {
EvalFailure::Eval(
id.clone(),
NomiError::Compile(format!("cannot read {path}: {err}")),
)
})?;
let program = Reader::parse(&source).map_err(|err| EvalFailure::Eval(id.clone(), err))?;
let count = program.exprs.len();
for form in program.exprs {
self.run(&Request {
id: id.clone(),
form,
.await?;
Ok(format!("loaded {path} ({count} forms)"))
/// Evaluates a single already-parsed form (mREPL input). The interrupt
/// pre-start check lives in `run`.
async fn eval_one_form(&mut self, form: Expr) -> Result<Value, EvalFailure> {
id: RequestId::Int(0),
.await
async fn evaluate(&mut self, frame: &str) -> Result<Response, EvalFailure> {
let request = parse_request(frame).map_err(EvalFailure::Envelope)?;
let value = self.run(&request).await?;
Ok(Response {
id: request.id,
payload: ResponsePayload::Value(value),
/// A session for a remote user, re-checking per form that the account may
/// still be served.
/// The daemon's `refuse_if_withheld` runs only when a channel is *opened*
/// and most natives never touch the tenant pool, so without this an open SSH
/// session outlives a disable indefinitely. [`Session::new`] stays unguarded
/// for the local CLI and for tests, whose users have no directory row.
/// # Errors
/// As [`Session::new`].
pub fn guarded(ctx: ScriptCtx) -> Result<Self, SessionError> {
let mut session = Self::new(ctx)?;
session.status_check_after = Some(Instant::now());
/// Bounded by [`STATUS_RECHECK`] rather than read per form: this is the hot
/// eval path and the global pool's budget is 5.
async fn check_account(&mut self, id: &RequestId) -> Result<(), EvalFailure> {
let Some(due) = self.status_check_after else {
return Ok(());
if Instant::now() < due {
// Fails closed: an unreadable directory is not evidence of health.
let refusal = server::command::user::account_refusal(self.ctx.user_id)
.unwrap_or(Some(server::command::user::Refusal::Quarantined));
match refusal {
None => {
self.status_check_after = Some(Instant::now() + STATUS_RECHECK);
Some(reason) => Err(EvalFailure::Refused(id.clone(), reason)),
async fn run(&mut self, request: &Request) -> Result<Value, EvalFailure> {
debug!(user_id = %self.ctx.user_id, "evaluating form");
self.check_account(&request.id).await?;
// A `C-g` that landed before this form started (pre-armed, or during an
// earlier form of a load) aborts here.
if let Some(err) = self.check_interrupt(&request.id) {
let program = Program::new(vec![request.form.clone()]);
let (bytes, result_ty) = self
.map_err(|err| EvalFailure::Eval(request.id.clone(), err))?;
let module = self
.cache
.get_or_compile(&self.engine, &bytes)
.map_err(|err| EvalFailure::Engine(request.id.clone(), err))?;
let mut linker: Linker<SessionData> = Linker::new(&self.engine);
crate::natives::link(&mut linker).map_err(|err| {
EvalFailure::Engine(
request.id.clone(),
EngineError::Instantiate(err.to_string()),
let mut store: Store<SessionData> = Store::new(
&self.engine,
SessionData::new(self.ctx.clone(), Arc::clone(&self.output)),
);
store.set_fuel(self.ctx.limits.fuel).map_err(|err| {
EvalFailure::Engine(request.id.clone(), EngineError::Fuel(err.to_string()))
store.set_epoch_deadline(EPOCH_DEADLINE_TICKS);
let instance = linker
.instantiate_async(&mut store, &module)
.map_err(|err| EvalFailure::Engine(request.id.clone(), classify_runtime_error(&err)))?;
let func = instance.get_func(&mut store, "nomi-eval").ok_or_else(|| {
EngineError::MissingExport("nomi-eval".into()),
// An interrupt that arrived during compile/link (the epoch bump only
// cancels a running call) — abort before entering Wasm.
let mut results = [Val::AnyRef(None)];
let call_result = func.call_async(&mut store, &[], &mut results).await;
// The reader bumps the epoch AND signals the interrupt together on
// `(:emacs-interrupt)`. If a C-g arrived during THIS call it cancelled it
// (epoch trap → `call_result` is `Err`); attribute it to this request so
// it can't also abort the next one. We ack only on the error path, so a
// C-g just after a clean success stays pending for the next form. This
// covers every terminal exit uniformly (epoch cancel, the
// out-of-fuel/runtime-trap-wins-the-race case) with no per-exit cleanup.
//
// Attribution cutoff (deliberate): the ack snapshot is taken AFTER the
// call returns, so a C-g landing in the sub-µs window between the call
// returning `Err` and this load is attributed to THIS (failing) form,
// not the next one. That is correct REPL semantics — at that instant the
// failing form is still the in-flight request (its error hasn't reached
// the client), so the user can't yet have meant the C-g for a later
// form. A C-g pressed after the error is surfaced necessarily arrives
// after `run` returns, advancing the generation again, and the next
// form's pre-start `check_interrupt` catches it. The exact instant of
// call return is unobservable, so this boundary is irreducible, not a
// lost interrupt.
if call_result.is_err() {
call_result
let any: Option<Rooted<AnyRef>> = match &results[0] {
Val::AnyRef(a) => *a,
return Err(EvalFailure::Engine(
EngineError::Trap("nomi-eval did not return anyref".into()),
let captured = decode_eval_result(&mut store, any, result_ty).map_err(|err| {
EngineError::Trap(format!("decoding nomi-eval result: {err}")),
Ok(Value::from(captured))
enum EvalFailure {
Envelope(EnvelopeError),
Eval(RequestId, NomiError),
Engine(RequestId, EngineError),
Interrupted(RequestId),
Refused(RequestId, server::command::user::Refusal),
impl EvalFailure {
fn into_response(self) -> Response {
match self {
EvalFailure::Envelope(err) => Response {
payload: ResponsePayload::Error {
code: envelope_error_code(&err),
message: err.to_string(),
detail: Some(format!("{err:?}")),
},
EvalFailure::Eval(id, err) => Response {
code: nomi_error_code(&err),
EvalFailure::Engine(id, err) => Response {
code: engine_error_code(&err),
EvalFailure::Interrupted(id) => Response {
code: ErrorCode::new(ErrorCode::INTERRUPTED),
message: "evaluation interrupted before start".into(),
detail: None,
EvalFailure::Refused(id, reason) => Response {
code: ErrorCode::new(ErrorCode::AUTH),
message: match reason {
server::command::user::Refusal::Quarantined => {
"This account is undergoing maintenance and is temporarily unavailable"
server::command::user::Refusal::Disabled => "This account is disabled",
.into(),
fn envelope_error_code(err: &EnvelopeError) -> ErrorCode {
let symbol = match err {
EnvelopeError::Parse(_) => ErrorCode::PARSE,
EnvelopeError::NotSingleExpr
| EnvelopeError::NotPlist
| EnvelopeError::MissingKey(_)
| EnvelopeError::InvalidValue(_, _) => ErrorCode::ARGS,
ErrorCode::new(symbol)
fn nomi_error_code(err: &NomiError) -> ErrorCode {
NomiError::Parse(_) => ErrorCode::PARSE,
NomiError::Compile(_) | NomiError::UndefinedSymbol(_) => ErrorCode::COMPILE,
NomiError::Runtime(_) => ErrorCode::RUNTIME,
NomiError::Type { .. } | NomiError::Arity { .. } => ErrorCode::ARGS,
fn engine_error_code(err: &EngineError) -> ErrorCode {
match err {
EngineError::Compile(_) => ErrorCode::new(ErrorCode::COMPILE),
EngineError::OutOfFuel | EngineError::Trap(_) => ErrorCode::new(ErrorCode::RUNTIME),
EngineError::EpochInterrupt => ErrorCode::new(ErrorCode::INTERRUPTED),
EngineError::Instantiate(_) | EngineError::MissingExport(_) => {
ErrorCode::new(ErrorCode::SERVER)
EngineError::Fuel(_) | EngineError::Config(_) | EngineError::CachePoisoned => {
EngineError::NoConversion(_) => ErrorCode::new(ErrorCode::NO_CONVERSION),
// Commodity mismatch now arrives as a `ScriptRaised` carrying the
// reader-folded symbol `COMMODITY-MISMATCH` (ADR-0026): it `throw`s
// `$nomi_error` in-guest and the boundary wrapper bridges the uncaught
// throw to `__nomi_raise`. The code flows through verbatim like any
// script raise; the wire `:code` is the upper-cased symbol form.
EngineError::ScriptRaised { code, .. } => ErrorCode::new(code.clone()),
#[cfg(test)]
mod tests;