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

xs/nu/
engine.rs
