tui/tabs/

nms_eval.rs

//! Async eval bridge for the Console tab.
//!
//! `ConsoleEval` is a concrete (no `dyn`) actor: it owns an mpsc to a
//! per-console worker that drives `rpc::Session::handle_form`, plus the
//! cancel handles (`EpochBumper` + `InterruptHandle`). The UI side is
//! sync — `submit` is a non-blocking send, `drain` a `try_recv` loop —
//! so the 200 ms `run_loop` never awaits eval.
//!
//! `Session::handle_form` parses an RPC request **envelope**
//! (`(:id N :form <form>)`), not a bare form, so `submit` wraps each
//! input with a monotonic id before handoff — mirroring `nms`'s
//! `RpcEval::eval` and the sshd `eval_channel` worker. The worker
//! receives already-wrapped envelopes and only forwards them.
//!
//! Two constructors form the test seam: [`ConsoleEval::spawn`] builds a
//! real DB-backed `Session` actor, while [`ConsoleEval::echo`] spawns a
//! runtime-only worker that echoes frames back (no `Session`, no DB)
//! for plumbing tests.

use rpc::session::SessionError;
use rpc::{EpochBumper, InterruptHandle, RequestId, ScriptCtx, Session, parse_response};
use sqlx::types::Uuid;

/// Scrollback notice emitted once when the eval worker's result channel closes.
pub(crate) const WORKER_STOPPED_NOTICE: &str = "eval worker stopped";
use tokio::runtime::Handle;
use tokio::sync::mpsc;
use tokio::sync::mpsc::error::TryRecvError;
use tokio::task::JoinHandle;

/// A single drained item from the eval worker's result channel.
#[derive(Debug)]
pub enum DrainItem {
    /// A response whose `:id` is a positive integer — route by that id.
    Routed { id: i64, wire: String },
    /// A response with `:id 0` (envelope/parse error from the server)
    /// or a non-integer id; cannot be routed to a specific pending request.
    Unroutable(String),
    /// A plain-text notice generated by the drain loop itself (e.g.
    /// "eval worker stopped") — not a wire frame.
    Notice(String),
}

/// Async bridge between the sync TUI and an `rpc::Session` eval worker.
pub struct ConsoleEval {
    forms_tx: mpsc::UnboundedSender<String>,
    results_rx: mpsc::UnboundedReceiver<String>,
    /// Epoch cancel handle for the worker's `Session`. `None` for the
    /// `echo` test actor, which has no engine to trip.
    bumper: Option<EpochBumper>,
    interrupt: InterruptHandle,
    /// Monotonic envelope id; the wire `:id` of the next submission. An `i64`
    /// (the wire/`RequestId` type) that starts at 1 and only ever increments
    /// via `checked_add`, so it can never reach 0 (reserved for server
    /// envelope/parse errors, routed to [`DrainItem::Unroutable`]) or a
    /// negative value — the id space is simply refused once exhausted.
    next_id: i64,
    /// Set once `drain` observes the result channel disconnected, so the
    /// "worker stopped" notice reaches scrollback exactly once rather than
    /// on every subsequent tick.
    worker_stopped_reported: bool,
    /// Held so the worker stays alive for the eval's lifetime. Dropping
    /// `ConsoleEval` closes `forms_tx`; the worker's `recv` returns
    /// `None` and the task exits. Read only by the test seam.
    _worker: JoinHandle<()>,
}

impl ConsoleEval {
    /// Wraps `form` in a fresh `(:id {next_id} :form {form})` envelope
    /// and hands it to the worker. Non-blocking. Returns `Some(id)` with
    /// the allocated envelope id on success, or `None` when the channel
    /// is closed (the worker has stopped).
    pub fn submit(&mut self, form: String) -> Option<u64> {
        let id = self.next_id;
        // Refuse rather than wrap back toward the reserved 0 once the i64 id
        // space is exhausted (practically unreachable).
        let next = id.checked_add(1)?;
        let envelope = format!("(:id {id} :form {form})");
        self.forms_tx.send(envelope).ok()?;
        self.next_id = next;
        Some(id as u64)
    }

    /// Drains every ready response without blocking. Each received wire
    /// frame is parsed to extract its `:id` and classified as
    /// [`DrainItem::Routed`] (positive id), [`DrainItem::Unroutable`]
    /// (id 0 / non-integer id / parse error), or [`DrainItem::Notice`]
    /// (worker-stopped plain-text, emitted at most once).
    pub fn drain(&mut self) -> Vec<DrainItem> {
        let mut out = Vec::new();
        loop {
            match self.results_rx.try_recv() {
                Ok(wire) => out.push(classify_response(wire)),
                Err(TryRecvError::Empty) => break,
                Err(TryRecvError::Disconnected) => {
                    if !self.worker_stopped_reported {
                        self.worker_stopped_reported = true;
                        out.push(DrainItem::Notice(WORKER_STOPPED_NOTICE.to_string()));
                    }
                    break;
                }
            }
        }
        out
    }

    /// Requests a cooperative cancel of the in-flight eval. Advances the
    /// interrupt generation first (caught by `handle_form`'s pre-start
    /// `check_interrupt` if the eval has not entered Wasm yet) then bumps
    /// the engine epoch (traps an eval already running) — the
    /// eval_channel ordering that closes the worker-pickup window.
    pub fn interrupt(&self) {
        self.interrupt.interrupt();
        if let Some(bumper) = &self.bumper {
            bumper.bump();
        }
    }

    /// Builds a real `Session` for `user_id` and spawns its eval worker.
    pub fn spawn(handle: &Handle, user_id: Uuid) -> Result<Self, SessionError> {
        Self::from_ctx(handle, ScriptCtx::new(user_id))
    }

    /// Builds the worker from a caller-supplied context.
    fn from_ctx(handle: &Handle, ctx: ScriptCtx) -> Result<Self, SessionError> {
        Ok(Self::spawn_session(handle, Session::new(ctx)?))
    }

    /// Spawns the worker around an already-built session, so the caller decides
    /// whether it is guarded.
    pub fn spawn_session(handle: &Handle, mut session: Session) -> Self {
        let bumper = session.epoch_bumper();
        let interrupt = session.interrupt_handle();
        let (forms_tx, mut forms_rx) = mpsc::unbounded_channel::<String>();
        let (results_tx, results_rx) = mpsc::unbounded_channel::<String>();
        let worker = handle.spawn(async move {
            while let Some(frame) = forms_rx.recv().await {
                let response = session.handle_form(&frame).await;
                if results_tx.send(response).is_err() {
                    break;
                }
            }
        });
        Self {
            forms_tx,
            results_rx,
            bumper: Some(bumper),
            interrupt,
            next_id: 1,
            worker_stopped_reported: false,
            _worker: worker,
        }
    }

    /// Spawns a runtime-only worker that echoes each received frame back
    /// verbatim. No `Session`, no DB; used to exercise the channel/drain
    /// plumbing in isolation.
    pub fn echo(handle: &Handle) -> Self {
        let (forms_tx, mut forms_rx) = mpsc::unbounded_channel::<String>();
        let (results_tx, results_rx) = mpsc::unbounded_channel::<String>();
        let worker = handle.spawn(async move {
            while let Some(frame) = forms_rx.recv().await {
                if results_tx.send(frame).is_err() {
                    break;
                }
            }
        });
        Self {
            forms_tx,
            results_rx,
            bumper: None,
            interrupt: InterruptHandle::default(),
            next_id: 1,
            worker_stopped_reported: false,
            _worker: worker,
        }
    }
}

/// Parse a wire frame and classify it as routed, unroutable, or notice.
///
/// Envelope parse errors result in `Unroutable`; id 0 (server-assigned
/// for malformed requests) results in `Unroutable`; positive integer ids
/// result in `Routed`; string ids result in `Unroutable`.
fn classify_response(wire: String) -> DrainItem {
    match parse_response(&wire) {
        Ok(response) => match response.id {
            RequestId::Int(n) if n > 0 => DrainItem::Routed { id: n, wire },
            _ => DrainItem::Unroutable(wire),
        },
        Err(_) => DrainItem::Unroutable(wire),
    }
}

#[cfg(test)]
impl ConsoleEval {
    /// Build the worker from a caller-supplied context — the test seam
    /// that injects custom `ScriptLimits` (e.g. unbounded fuel for the
    /// interrupt test). Not part of the production surface.
    pub(crate) fn spawn_with_ctx(handle: &Handle, ctx: ScriptCtx) -> Result<Self, SessionError> {
        Self::from_ctx(handle, ctx)
    }

    /// An owned abort handle for the worker task, so a test can observe
    /// the worker finishing after the eval itself is dropped or abort it
    /// to simulate a stopped worker.
    pub(crate) fn worker_handle(&self) -> tokio::task::AbortHandle {
        self._worker.abort_handle()
    }
}

#[cfg(test)]
use rpc::ScriptLimits;

#[cfg(test)]
mod tests;