xs/nu/
engine.rs

1use nu_cli::{add_cli_context, gather_parent_env_vars};
2use nu_cmd_lang::create_default_context;
3use nu_command::add_shell_command_context;
4use nu_engine::eval_block_with_early_return;
5use nu_parser::parse;
6use nu_protocol::debugger::WithoutDebug;
7use nu_protocol::engine::{Closure, Command, EngineState, Redirection, Stack, StateWorkingSet};
8use nu_protocol::engine::{Job, ThreadJob};
9use nu_protocol::shell_error::generic::GenericError;
10use nu_protocol::{OutDest, PipelineData, ShellError, Span, Value};
11
12use crate::error::Error;
13use crate::nu::commands;
14use crate::store::Store;
15
16#[derive(Clone)]
17pub struct Engine {
18    pub state: EngineState,
19}
20
21impl Engine {
22    pub fn new() -> Result<Self, Error> {
23        let mut engine_state = create_default_context();
24        engine_state = add_shell_command_context(engine_state);
25        engine_state = add_cli_context(engine_state);
26
27        let init_cwd = std::env::current_dir()?;
28        gather_parent_env_vars(&mut engine_state, init_cwd.as_ref());
29
30        Ok(Self {
31            state: engine_state,
32        })
33    }
34
35    pub fn add_commands(&mut self, commands: Vec<Box<dyn Command>>) -> Result<(), Error> {
36        let mut working_set = StateWorkingSet::new(&self.state);
37        for command in commands {
38            working_set.add_decl(command);
39        }
40        self.state.merge_delta(working_set.render())?;
41        Ok(())
42    }
43
44    pub fn add_alias(&mut self, name: &str, target: &str) -> Result<(), Error> {
45        let mut working_set = StateWorkingSet::new(&self.state);
46        let _ = parse(
47            &mut working_set,
48            None,
49            format!("alias {name} = {target}").as_bytes(),
50            false,
51        );
52        self.state.merge_delta(working_set.render())?;
53        Ok(())
54    }
55
56    pub fn eval(&self, input: PipelineData, expression: String) -> Result<PipelineData, String> {
57        let mut working_set = StateWorkingSet::new(&self.state);
58        let block = parse(&mut working_set, None, expression.as_bytes(), false);
59
60        if !working_set.parse_errors.is_empty() {
61            let first_error = &working_set.parse_errors[0];
62            let formatted = nu_protocol::format_cli_error(None, &working_set, first_error, None);
63            return Err(formatted);
64        }
65
66        let mut engine_state = self.state.clone();
67        engine_state
68            .merge_delta(working_set.render())
69            .map_err(|e| {
70                let working_set = StateWorkingSet::new(&self.state);
71                nu_protocol::format_cli_error(None, &working_set, &e, None)
72            })?;
73
74        let mut stack = Stack::new();
75        let mut stack =
76            stack.push_redirection(Some(Redirection::Pipe(OutDest::PipeSeparate)), None);
77
78        eval_block_with_early_return::<WithoutDebug>(&engine_state, &mut stack, &block, input)
79            .map(|exec_data| exec_data.body)
80            .map_err(|e| {
81                let working_set = StateWorkingSet::new(&engine_state);
82                nu_protocol::format_cli_error(None, &working_set, &e, None)
83            })
84    }
85
86    pub fn parse_closure(&mut self, script: &str) -> Result<Closure, Box<ShellError>> {
87        let mut working_set = StateWorkingSet::new(&self.state);
88        let block = parse(&mut working_set, None, script.as_bytes(), false);
89        self.state
90            .merge_delta(working_set.render())
91            .map_err(Box::new)?;
92
93        let mut stack = Stack::new();
94        let result = eval_block_with_early_return::<WithoutDebug>(
95            &self.state,
96            &mut stack,
97            &block,
98            PipelineData::empty(),
99        )
100        .map_err(Box::new)?;
101        let closure = result
102            .body
103            .into_value(Span::unknown())
104            .map_err(Box::new)?
105            .into_closure()
106            .map_err(Box::new)?;
107
108        self.state.merge_env(&mut stack).map_err(Box::new)?;
109
110        Ok(closure)
111    }
112
113    pub fn add_module(&mut self, name: &str, content: &str) -> Result<(), Box<ShellError>> {
114        let mut working_set = StateWorkingSet::new(&self.state);
115
116        // Create temporary file with .nu extension that will be cleaned up when temp_dir is dropped
117        let temp_dir = tempfile::TempDir::new().map_err(|e| {
118            Box::new(ShellError::Generic(GenericError::new_internal(
119                "I/O Error",
120                format!("Failed to create temporary directory for module '{name}': {e}"),
121            )))
122        })?;
123        let module_path = temp_dir.path().join(format!("{name}.nu"));
124        std::fs::write(&module_path, content).map_err(|e| {
125            Box::new(ShellError::Generic(GenericError::new_internal(
126                "I/O Error",
127                e.to_string(),
128            )))
129        })?;
130
131        // Parse the use statement
132        let use_stmt = format!("use {}", module_path.display());
133        let _block = parse(&mut working_set, None, use_stmt.as_bytes(), false);
134
135        // Check for parse errors
136        if !working_set.parse_errors.is_empty() {
137            let first_error = &working_set.parse_errors[0];
138            return Err(Box::new(ShellError::Generic(GenericError::new(
139                "Parse error",
140                first_error.to_string(),
141                first_error.span(),
142            ))));
143        }
144
145        // Merge changes into engine state
146        self.state
147            .merge_delta(working_set.render())
148            .map_err(Box::new)?;
149
150        // Create a temporary stack and evaluate the module
151        let mut stack = Stack::new();
152        let _ = eval_block_with_early_return::<WithoutDebug>(
153            &self.state,
154            &mut stack,
155            &_block,
156            PipelineData::empty(),
157        )
158        .map_err(Box::new)?;
159
160        // Merge environment variables into engine state
161        self.state.merge_env(&mut stack).map_err(Box::new)?;
162
163        Ok(())
164    }
165
166    pub fn with_env_vars(
167        mut self,
168        vars: impl IntoIterator<Item = (String, String)>,
169    ) -> Result<Self, Error> {
170        for (key, value) in vars {
171            self.state
172                .add_env_var(key, nu_protocol::Value::string(value, Span::unknown()));
173        }
174
175        Ok(self)
176    }
177
178    pub fn run_closure_in_job(
179        &mut self,
180        closure: &nu_protocol::engine::Closure,
181        args: Vec<Value>,
182        pipeline_input: Option<PipelineData>,
183        job_name: impl Into<String>,
184    ) -> Result<PipelineData, Box<ShellError>> {
185        let job_display_name = job_name.into(); // Convert job_name early for error messages
186
187        // -- create & register job (boilerplate) ---
188        let (sender, _rx) = std::sync::mpsc::channel();
189        let job = ThreadJob::new(
190            self.state.signals().clone(),
191            Some(job_display_name.clone()),
192            sender,
193        );
194        let _job_id = {
195            let mut j = self.state.jobs.lock().unwrap();
196            j.add_job(Job::Thread(job.clone()))
197        };
198
199        // -- temporarily attach the job to self.state (boilerplate) ---
200        let saved_bg_job = self.state.current_job.background_thread_job.clone();
201        self.state.current_job.background_thread_job = Some(job.clone());
202
203        // -- prepare stack & validate/inject positional arguments ---
204        let block = self.state.get_block(closure.block_id);
205        let mut stack = Stack::new();
206        let mut stack =
207            stack.push_redirection(Some(Redirection::Pipe(OutDest::PipeSeparate)), None);
208
209        let num_required = block.signature.required_positional.len();
210        let num_optional = block.signature.optional_positional.len();
211        let total_positional = num_required + num_optional;
212
213        if args.len() > total_positional {
214            return Err(Box::new(ShellError::Generic(GenericError::new(
215                format!(
216                    "Too many arguments for job '{job_display_name}': got {}, closure accepts at most {total_positional}.",
217                    args.len()
218                ),
219                format!("Closure signature: {name}", name = block.signature.name),
220                block.span.unwrap_or_else(Span::unknown),
221            ))));
222        }
223
224        if args.len() < num_required {
225            return Err(Box::new(ShellError::Generic(GenericError::new(
226                format!(
227                    "Job '{job_display_name}' run closure expects {num_required} required argument(s), but {} were provided.",
228                    args.len()
229                ),
230                format!("Closure signature: {name}", name = block.signature.name),
231                block.span.unwrap_or_else(Span::unknown),
232            ))));
233        }
234
235        // Inject provided positional args
236        for (i, val) in args.iter().enumerate() {
237            let param = if i < num_required {
238                &block.signature.required_positional[i]
239            } else {
240                &block.signature.optional_positional[i - num_required]
241            };
242            if let Some(var_id) = param.var_id {
243                stack.add_var(var_id, val.clone());
244            }
245        }
246
247        // Set default values for optional params not covered by provided args
248        let optional_covered = args.len().saturating_sub(num_required);
249        for i in optional_covered..num_optional {
250            let param = &block.signature.optional_positional[i];
251            if let Some(var_id) = param.var_id {
252                let default = param
253                    .default_value
254                    .clone()
255                    .unwrap_or_else(|| Value::nothing(Span::unknown()));
256                stack.add_var(var_id, default);
257            }
258        }
259
260        // Determine the actual pipeline input for eval_block_with_early_return
261        let eval_pipeline_input = pipeline_input.unwrap_or_else(PipelineData::empty);
262
263        // -- run using eval_block_with_early_return ---
264        let eval_res = nu_engine::eval_block_with_early_return::<WithoutDebug>(
265            &self.state,
266            &mut stack,
267            block,
268            eval_pipeline_input,
269        );
270
271        // -- merge env, restore job, cleanup job (boilerplate, same as before) ---
272        if eval_res.is_ok() {
273            if let Err(e) = self.state.merge_env(&mut stack) {
274                tracing::error!(
275                    "Failed to merge environment from job '{}': {}",
276                    job_display_name,
277                    e
278                );
279            }
280        }
281
282        self.state.current_job.background_thread_job = saved_bg_job;
283        eval_res.map(|exec_data| exec_data.body).map_err(Box::new)
284    }
285
286    /// Kill the background ThreadJob whose name equals `name`.
287    pub fn kill_job_by_name(&self, name: &str) {
288        if let Ok(mut jobs) = self.state.jobs.lock() {
289            let job_id = {
290                jobs.iter().find_map(|(jid, job)| {
291                    job.description()
292                        .and_then(|desc| if desc == name { Some(jid) } else { None })
293                })
294            };
295            if let Some(job_id) = job_id {
296                let _ = jobs.kill_and_remove(job_id);
297            }
298        }
299    }
300}
301
302/// Add core cross.stream commands that are common across all Nushell pipeline runners
303pub fn add_core_commands(engine: &mut Engine, store: &Store) -> Result<(), Error> {
304    engine.add_commands(vec![
305        Box::new(commands::cas_command::CasCommand::new(store.clone())),
306        Box::new(commands::get_command::GetCommand::new(store.clone())),
307        Box::new(commands::remove_command::RemoveCommand::new(store.clone())),
308        Box::new(commands::scru128_command::Scru128Command::new()),
309    ])
310}
xs/nu/engine.rs

xs/nu/
engine.rs
