tui/tabs/nms_eval.rs
1//! Async eval bridge for the Console tab.
2//!
3//! `ConsoleEval` is a concrete (no `dyn`) actor: it owns an mpsc to a
4//! per-console worker that drives `rpc::Session::handle_form`, plus the
5//! cancel handles (`EpochBumper` + `InterruptHandle`). The UI side is
6//! sync — `submit` is a non-blocking send, `drain` a `try_recv` loop —
7//! so the 200 ms `run_loop` never awaits eval.
8//!
9//! `Session::handle_form` parses an RPC request **envelope**
10//! (`(:id N :form <form>)`), not a bare form, so `submit` wraps each
11//! input with a monotonic id before handoff — mirroring `nms`'s
12//! `RpcEval::eval` and the sshd `eval_channel` worker. The worker
13//! receives already-wrapped envelopes and only forwards them.
14//!
15//! Two constructors form the test seam: [`ConsoleEval::spawn`] builds a
16//! real DB-backed `Session` actor, while [`ConsoleEval::echo`] spawns a
17//! runtime-only worker that echoes frames back (no `Session`, no DB)
18//! for plumbing tests.
19
20use rpc::session::SessionError;
21use rpc::{EpochBumper, InterruptHandle, RequestId, ScriptCtx, Session, parse_response};
22use sqlx::types::Uuid;
23
24/// Scrollback notice emitted once when the eval worker's result channel closes.
25pub(crate) const WORKER_STOPPED_NOTICE: &str = "eval worker stopped";
26use tokio::runtime::Handle;
27use tokio::sync::mpsc;
28use tokio::sync::mpsc::error::TryRecvError;
29use tokio::task::JoinHandle;
30
31/// A single drained item from the eval worker's result channel.
32#[derive(Debug)]
33pub enum DrainItem {
34 /// A response whose `:id` is a positive integer — route by that id.
35 Routed { id: i64, wire: String },
36 /// A response with `:id 0` (envelope/parse error from the server)
37 /// or a non-integer id; cannot be routed to a specific pending request.
38 Unroutable(String),
39 /// A plain-text notice generated by the drain loop itself (e.g.
40 /// "eval worker stopped") — not a wire frame.
41 Notice(String),
42}
43
44/// Async bridge between the sync TUI and an `rpc::Session` eval worker.
45pub struct ConsoleEval {
46 forms_tx: mpsc::UnboundedSender<String>,
47 results_rx: mpsc::UnboundedReceiver<String>,
48 /// Epoch cancel handle for the worker's `Session`. `None` for the
49 /// `echo` test actor, which has no engine to trip.
50 bumper: Option<EpochBumper>,
51 interrupt: InterruptHandle,
52 /// Monotonic envelope id; the wire `:id` of the next submission. An `i64`
53 /// (the wire/`RequestId` type) that starts at 1 and only ever increments
54 /// via `checked_add`, so it can never reach 0 (reserved for server
55 /// envelope/parse errors, routed to [`DrainItem::Unroutable`]) or a
56 /// negative value — the id space is simply refused once exhausted.
57 next_id: i64,
58 /// Set once `drain` observes the result channel disconnected, so the
59 /// "worker stopped" notice reaches scrollback exactly once rather than
60 /// on every subsequent tick.
61 worker_stopped_reported: bool,
62 /// Held so the worker stays alive for the eval's lifetime. Dropping
63 /// `ConsoleEval` closes `forms_tx`; the worker's `recv` returns
64 /// `None` and the task exits. Read only by the test seam.
65 _worker: JoinHandle<()>,
66}
67
68impl ConsoleEval {
69 /// Wraps `form` in a fresh `(:id {next_id} :form {form})` envelope
70 /// and hands it to the worker. Non-blocking. Returns `Some(id)` with
71 /// the allocated envelope id on success, or `None` when the channel
72 /// is closed (the worker has stopped).
73 pub fn submit(&mut self, form: String) -> Option<u64> {
74 let id = self.next_id;
75 // Refuse rather than wrap back toward the reserved 0 once the i64 id
76 // space is exhausted (practically unreachable).
77 let next = id.checked_add(1)?;
78 let envelope = format!("(:id {id} :form {form})");
79 self.forms_tx.send(envelope).ok()?;
80 self.next_id = next;
81 Some(id as u64)
82 }
83
84 /// Drains every ready response without blocking. Each received wire
85 /// frame is parsed to extract its `:id` and classified as
86 /// [`DrainItem::Routed`] (positive id), [`DrainItem::Unroutable`]
87 /// (id 0 / non-integer id / parse error), or [`DrainItem::Notice`]
88 /// (worker-stopped plain-text, emitted at most once).
89 pub fn drain(&mut self) -> Vec<DrainItem> {
90 let mut out = Vec::new();
91 loop {
92 match self.results_rx.try_recv() {
93 Ok(wire) => out.push(classify_response(wire)),
94 Err(TryRecvError::Empty) => break,
95 Err(TryRecvError::Disconnected) => {
96 if !self.worker_stopped_reported {
97 self.worker_stopped_reported = true;
98 out.push(DrainItem::Notice(WORKER_STOPPED_NOTICE.to_string()));
99 }
100 break;
101 }
102 }
103 }
104 out
105 }
106
107 /// Requests a cooperative cancel of the in-flight eval. Advances the
108 /// interrupt generation first (caught by `handle_form`'s pre-start
109 /// `check_interrupt` if the eval has not entered Wasm yet) then bumps
110 /// the engine epoch (traps an eval already running) — the
111 /// eval_channel ordering that closes the worker-pickup window.
112 pub fn interrupt(&self) {
113 self.interrupt.interrupt();
114 if let Some(bumper) = &self.bumper {
115 bumper.bump();
116 }
117 }
118
119 /// Builds a real `Session` for `user_id` and spawns its eval worker.
120 pub fn spawn(handle: &Handle, user_id: Uuid) -> Result<Self, SessionError> {
121 Self::from_ctx(handle, ScriptCtx::new(user_id))
122 }
123
124 /// Builds the worker from a caller-supplied context.
125 fn from_ctx(handle: &Handle, ctx: ScriptCtx) -> Result<Self, SessionError> {
126 let mut session = Session::new(ctx)?;
127 let bumper = session.epoch_bumper();
128 let interrupt = session.interrupt_handle();
129 let (forms_tx, mut forms_rx) = mpsc::unbounded_channel::<String>();
130 let (results_tx, results_rx) = mpsc::unbounded_channel::<String>();
131 let worker = handle.spawn(async move {
132 while let Some(frame) = forms_rx.recv().await {
133 let response = session.handle_form(&frame).await;
134 if results_tx.send(response).is_err() {
135 break;
136 }
137 }
138 });
139 Ok(Self {
140 forms_tx,
141 results_rx,
142 bumper: Some(bumper),
143 interrupt,
144 next_id: 1,
145 worker_stopped_reported: false,
146 _worker: worker,
147 })
148 }
149
150 /// Spawns a runtime-only worker that echoes each received frame back
151 /// verbatim. No `Session`, no DB; used to exercise the channel/drain
152 /// plumbing in isolation.
153 pub fn echo(handle: &Handle) -> Self {
154 let (forms_tx, mut forms_rx) = mpsc::unbounded_channel::<String>();
155 let (results_tx, results_rx) = mpsc::unbounded_channel::<String>();
156 let worker = handle.spawn(async move {
157 while let Some(frame) = forms_rx.recv().await {
158 if results_tx.send(frame).is_err() {
159 break;
160 }
161 }
162 });
163 Self {
164 forms_tx,
165 results_rx,
166 bumper: None,
167 interrupt: InterruptHandle::default(),
168 next_id: 1,
169 worker_stopped_reported: false,
170 _worker: worker,
171 }
172 }
173}
174
175/// Parse a wire frame and classify it as routed, unroutable, or notice.
176///
177/// Envelope parse errors result in `Unroutable`; id 0 (server-assigned
178/// for malformed requests) results in `Unroutable`; positive integer ids
179/// result in `Routed`; string ids result in `Unroutable`.
180fn classify_response(wire: String) -> DrainItem {
181 match parse_response(&wire) {
182 Ok(response) => match response.id {
183 RequestId::Int(n) if n > 0 => DrainItem::Routed { id: n, wire },
184 _ => DrainItem::Unroutable(wire),
185 },
186 Err(_) => DrainItem::Unroutable(wire),
187 }
188}
189
190#[cfg(test)]
191impl ConsoleEval {
192 /// Build the worker from a caller-supplied context — the test seam
193 /// that injects custom `ScriptLimits` (e.g. unbounded fuel for the
194 /// interrupt test). Not part of the production surface.
195 pub(crate) fn spawn_with_ctx(handle: &Handle, ctx: ScriptCtx) -> Result<Self, SessionError> {
196 Self::from_ctx(handle, ctx)
197 }
198
199 /// An owned abort handle for the worker task, so a test can observe
200 /// the worker finishing after the eval itself is dropped or abort it
201 /// to simulate a stopped worker.
202 pub(crate) fn worker_handle(&self) -> tokio::task::AbortHandle {
203 self._worker.abort_handle()
204 }
205}
206
207#[cfg(test)]
208use rpc::ScriptLimits;
209
210#[cfg(test)]
211mod tests;