fix: report missing closure blocks instead of panicking (#19159)

## Description

When a job is spawned before a later input is parsed, the job retains an
older `EngineState`. The later closure's `BlockId` is therefore
unavailable in that job, which previously panicked in
`EngineState::get_block`.

- remove the infallible `ClosureEval` / `ClosureEvalOnce` constructors
and make every `try_new*` constructor take a span pointing at the
failing call or closure value, so a stale `BlockId` can no longer panic
through any entry point (`do`, `each`, `zip`, `where`, …)
- add `EngineState::get_closure_block` as the single fallible lookup
with a span and actionable help (shared `MISSING_CLOSURE_BLOCK_HELP`
const); direct `get_block` callers on runtime closure values (`tee`,
`with-env`, `export-env`, `explain`, `view source`, `record where`, `job
spawn`, `debug profile`, the hook runner and the tui `closure_arity`)
now report the same error instead of panicking
- `debug profile` checks the block before installing the profiler, and
`job spawn` checks before registering the job, so the error reaches the
caller's `try`/`catch` instead of the background thread
- best-effort consumers degrade instead of panicking: `describe
--detailed` omits the `signature` key, `to text` / `to toml --serialize`
fall back to the existing placeholder string, `to yaml --serialize`
returns the `closure_block_not_found` error, and
`Closure::coerce_into_string` (`to msgpack`, `to kdl`) returns
`CantConvert`
- `par-each` builds the evaluator once in `run` (skipping empty input
like `each`) and clones it into the workers, returning a single error
instead of one error value per item
- `default` propagates the construction error instead of storing it as
the default value
- `generate` reads the signature from the evaluator's `block()` instead
of looking the block up twice
- `ClosureEval` / `ClosureEvalOnce` expose `block()`, removing the
remaining second lookups in `record where`, the tui `call_closure` and
`generate`
- convert the `record` commands from #19190 to the fallible constructors
- add regression tests that reproduce the interactive, multi-input `job
spawn` / `job send` flow from #19147 through `do`, `each`, `zip`, `tee`,
`par-each`, `record each`, `view source`, `describe --detailed`, `to
text --serialize`, `debug profile`, `job spawn`, and a combined test for
`with-env`, `explain`, `generate`, `record where`, `record walk` and
`sort-by`

Removing the infallible constructors is a breaking change to nu-engine's
public API without a deprecation step: a deprecated shim cannot avoid
the panic, which is what this PR removes, so the break is intentional.

## User-facing changes (Release notes)

- Fixed an issue where a closure created after a job was spawned
panicked the job when used (`do`, `each`, `zip`, `tee`, `par-each`,
`sort-by`, `explain`, `with-env`, `job spawn`, …). Nushell now reports
`Closure cannot be evaluated` with a span and a hint instead, and `job
spawn` / `debug profile` now surface the error to the caller.
- Serializing such a closure no longer crashes: `describe --detailed`
and `to text` / `to toml --serialize` degrade to the existing
placeholder, while `to yaml --serialize` and `to msgpack` report a
conversion error.

## Additional notes

- fixes #19147
- rebase note: this branch now includes the conversion of the `record`
commands merged in #19190, which called the removed constructors

## Validation

- `cargo test -p nu-command --test tests job` (44 passed; do / each /
zip / tee / par-each / record each / record where / record walk / view
source / describe / to text / debug profile / job spawn / with-env /
explain / generate / sort-by scenarios)
- `cargo test -p nu-command --test tests` (3438 passed; the 21 failures
are pre-existing Windows symlink-privilege environment failures,
confirmed identical on upstream main)
- `cargo test` for nu-engine, nu-cmd-lang, nu-protocol, nu-tui,
nu-cmd-base, nu-color-config, nu-heavy-utils, nu-cli, nu-cmd-extra,
nu-lsp, nu-dap, nu-plugin-engine (only the same symlink-privilege
failures, confirmed on upstream main)
- `cargo test -p nu-engine --doc`, `cargo test -p nu-protocol --doc`
- `cargo fmt --all -- --check`
- `cargo clippy -p nu-engine -p nu-cmd-lang -p nu-command -p nu-cli -p
nu-cmd-extra -p nu-color-config -p nu-plugin-engine -p nu-tui -p
nu-protocol -p nu-cmd-base -p nu-heavy-utils --tests --no-deps -- -D
warnings`

crates/nu-cli/src/hints/external_hinter.rs

@@ -69,10 +69,15 @@ impl ExternalHinter {
         .map_err(|err| format!("failed to build context record: {err}"))?;
 
         let stack = Stack::with_parent(self.stack.clone());
-        let result = ClosureEvalOnce::new(self.engine_state.as_ref(), &stack, self.closure.clone())
-            .add_arg(Value::record(context, span))
-            .and_then(|closure| closure.run_with_input(PipelineData::empty()))
-            .and_then(|data| data.into_value(span));
+        let result = ClosureEvalOnce::try_new(
+            self.engine_state.as_ref(),
+            &stack,
+            self.closure.clone(),
+            span,
+        )
+        .and_then(|eval| eval.add_arg(Value::record(context, span)))
+        .and_then(|closure| closure.run_with_input(PipelineData::empty()))
+        .and_then(|data| data.into_value(span));
 
         match result {
             Ok(Value::String { val, .. }) => Ok(Some(HintResult {

crates/nu-cli/src/prompt_update.rs

@@ -57,9 +57,12 @@ fn get_prompt_string(
 ) -> Option<String> {
     let mut output = match stack.get_env_var(engine_state, prompt)? {
         Value::String { val, .. } => val.clone(),
-        Value::Closure { val, .. } => {
-            let result = ClosureEvalOnce::new(engine_state, stack, val.as_ref().clone())
-                .run_with_input(PipelineData::empty());
+        Value::Closure {
+            val, internal_span, ..
+        } => {
+            let result =
+                ClosureEvalOnce::try_new(engine_state, stack, val.as_ref().clone(), *internal_span)
+                    .and_then(|eval| eval.run_with_input(PipelineData::empty()));
 
             trace!(
                 "get_prompt_string (block) {}:{}:{}",

crates/nu-cmd-base/src/hook.rs

@@ -327,7 +327,7 @@ fn run_hook(
     arguments: Vec<(String, Value)>,
     span: Span,
 ) -> Result<PipelineData, ShellError> {
-    let block = engine_state.get_block(closure.block_id);
+    let block = engine_state.get_closure_block(closure, span)?;
 
     let input = optional_input.unwrap_or_else(PipelineData::empty);
 

crates/nu-cmd-extra/src/extra/filters/each_while.rs

@@ -76,7 +76,7 @@ impl Command for EachWhile {
             PipelineData::Value(Value::Range { .. }, ..)
             | PipelineData::Value(Value::List { .. }, ..)
             | PipelineData::ListStream(..) => {
-                let mut closure = ClosureEval::new(engine_state, stack, closure);
+                let mut closure = ClosureEval::try_new(engine_state, stack, closure, head)?;
                 Ok(input
                     .into_iter()
                     .map_while(move |value| {
@@ -94,7 +94,7 @@ impl Command for EachWhile {
             PipelineData::ByteStream(stream, ..) => {
                 let span = stream.span();
                 if let Some(chunks) = stream.chunks() {
-                    let mut closure = ClosureEval::new(engine_state, stack, closure);
+                    let mut closure = ClosureEval::try_new(engine_state, stack, closure, head)?;
                     Ok(chunks
                         .map_while(move |value| match value {
                             Ok(value) => match closure
@@ -115,7 +115,7 @@ impl Command for EachWhile {
             // This match allows non-iterables to be accepted,
             // which is currently considered undesirable (Nov 2022).
             PipelineData::Value(value, ..) => {
-                ClosureEvalOnce::new(engine_state, stack, closure).run_with_value(value)
+                ClosureEvalOnce::try_new(engine_state, stack, closure, head)?.run_with_value(value)
             }
         }
         .map(|data| data.set_metadata(metadata))

crates/nu-cmd-extra/src/extra/filters/update_cells.rs

@@ -142,7 +142,7 @@ With `--recursive`, nested records and lists are descended into instead and the
                 let val = val.to_mut();
                 update_record(
                     val,
-                    &mut ClosureEval::new(engine_state, stack, closure),
+                    &mut ClosureEval::try_new(engine_state, stack, closure, head)?,
                     span,
                     columns.as_ref(),
                     recursive,
@@ -153,7 +153,7 @@ With `--recursive`, nested records and lists are descended into instead and the
                 let metadata = input.take_metadata();
                 Ok(UpdateCellIterator {
                     iter: input.into_iter(),
-                    closure: ClosureEval::new(engine_state, stack, closure),
+                    closure: ClosureEval::try_new(engine_state, stack, closure, head)?,
                     columns,
                     recursive,
                     span: head,

crates/nu-cmd-lang/src/core_commands/describe.rs

@@ -417,7 +417,8 @@ fn describe_value_inner(
             })
         }
         Value::Closure { ref val, .. } => {
-            let block = engine_state.map(|engine_state| engine_state.get_block(val.block_id));
+            let block =
+                engine_state.and_then(|engine_state| engine_state.try_get_block(val.block_id));
 
             let mut record = record! {
                 "type" => Value::string("closure", head),

crates/nu-cmd-lang/src/core_commands/do_.rs

@@ -65,9 +65,14 @@ impl Command for Do {
         let has_env = call.has_flag(engine_state, caller_stack, "env")?;
 
         let result = if has_env {
-            ClosureEvalOnce::new_env_preserve_out_dest(engine_state, caller_stack, closure)
+            ClosureEvalOnce::try_new_env_preserve_out_dest(
+                engine_state,
+                caller_stack,
+                closure,
+                head,
+            )?
         } else {
-            ClosureEvalOnce::new_preserve_out_dest(engine_state, caller_stack, closure)
+            ClosureEvalOnce::try_new_preserve_out_dest(engine_state, caller_stack, closure, head)?
         }
         .add_args(rest)?
         .run_with_input(input);

crates/nu-color-config/src/style_computer.rs

@@ -75,10 +75,10 @@ impl<'a> StyleComputer<'a> {
             Some(ComputableStyle::Static(s)) => *s,
             // Closures are run here.
             Some(ComputableStyle::Closure(closure, span)) => {
-                let result = ClosureEvalOnce::new(self.engine_state, self.stack, closure.clone())
-                    .debug(false)
-                    .run_with_value(value.clone())
-                    .and_then(|data| data.into_value(*span));
+                let result =
+                    ClosureEvalOnce::try_new(self.engine_state, self.stack, closure.clone(), *span)
+                        .and_then(|eval| eval.debug(false).run_with_value(value.clone()))
+                        .and_then(|data| data.into_value(*span));
 
                 match result {
                     Ok(value) => {

crates/nu-command/src/debug/explain.rs

@@ -38,7 +38,7 @@ impl Command for Explain {
         let head = call.head;
         // This was all delightfully stolen from benchmark :)
         let capture_block: Closure = call.req(engine_state, stack, 0)?;
-        let block = engine_state.get_block(capture_block.block_id);
+        let block = engine_state.get_closure_block(&capture_block, head)?;
         let mut stack = stack.captures_to_stack(capture_block.captures);
         let elements = get_pipeline_elements(engine_state, &mut stack, block, head);
         Ok(Value::list(elements, head).into_pipeline_data())
@@ -282,8 +282,8 @@ pub fn debug_string_without_formatting(engine_state: &EngineState, value: &Value
                 .join(" ")
         ),
         Value::Closure { val, .. } => {
-            let block = engine_state.get_block(val.block_id);
-            if let Some(span) = block.span {
+            let block = engine_state.try_get_block(val.block_id);
+            if let Some(span) = block.and_then(|block| block.span) {
                 let contents_bytes = engine_state.get_span_contents(span);
                 let contents_string = String::from_utf8_lossy(contents_bytes);
                 contents_string.to_string()

crates/nu-command/src/debug/metadata_access.rs

@@ -39,9 +39,14 @@ impl Command for MetadataAccess {
         let closure: Closure = call.req(engine_state, caller_stack, 0)?;
         let metadata_record = Value::record(build_metadata_record(&input, call.head), call.head);
 
-        ClosureEvalOnce::new_env_preserve_out_dest(engine_state, caller_stack, closure)
-            .add_arg(metadata_record)?
-            .run_with_input(input)
+        ClosureEvalOnce::try_new_env_preserve_out_dest(
+            engine_state,
+            caller_stack,
+            closure,
+            call.head,
+        )?
+        .add_arg(metadata_record)?
+        .run_with_input(input)
     }
 
     fn examples(&self) -> Vec<Example<'_>> {

crates/nu-command/src/debug/metadata_set.rs

@@ -94,7 +94,7 @@ impl Command for MetadataSet {
             let record = extend_record_with_metadata(Record::new(), Some(&metadata), head);
             let metadata_value = record.into_value(head);
 
-            let result = ClosureEvalOnce::new(engine_state, stack, closure)
+            let result = ClosureEvalOnce::try_new(engine_state, stack, closure, head)?
                 .run_with_value(metadata_value)?
                 .into_value(head)?;
 

crates/nu-command/src/debug/profile.rs

@@ -146,11 +146,16 @@ confusing the id/parent_id hierarchy. The --expr flag is helpful for investigati
             ))
         };
 
+        // Check the closure before the profiler is installed, so a stale
+        // closure can't leave the job in debugging mode.
+        engine_state.get_closure_block(&closure, call.head)?;
+
         engine_state
             .activate_debugger(Box::new(profiler))
             .map_err(lock_err)?;
 
-        let result = ClosureEvalOnce::new(engine_state, stack, closure).run_with_input(input);
+        let result = ClosureEvalOnce::try_new(engine_state, stack, closure, call.head)
+            .and_then(|eval| eval.run_with_input(input));
 
         // Return potential errors
         let pipeline_data = result?;

crates/nu-command/src/debug/timeit.rs

@@ -68,7 +68,8 @@ This command will bubble up any errors encountered when running the closure. The
 
         let include_output = call.has_flag(engine_state, stack, "output")?;
         let closure: Closure = call.req(engine_state, stack, 0)?;
-        let closure = ClosureEvalOnce::new_preserve_out_dest(engine_state, stack, closure);
+        let closure =
+            ClosureEvalOnce::try_new_preserve_out_dest(engine_state, stack, closure, call.head)?;
 
         // Get the start time after all other computation has been done.
         let start_time = Instant::now();

crates/nu-command/src/debug/view_source.rs

@@ -264,7 +264,7 @@ and an explicit type annotation is lost. `$nu` and the record bound by
             }
             value => {
                 if let Ok(closure) = value.as_closure() {
-                    let block = engine_state.get_block(closure.block_id);
+                    let block = engine_state.get_closure_block(closure, call.head)?;
 
                     if let Some(span) = block.span {
                         let contents = engine_state.get_span_contents(span);

crates/nu-command/src/env/export_env.rs

@@ -42,7 +42,7 @@ impl Command for ExportEnv {
     ) -> Result<PipelineData, ShellError> {
         // Blocks compile to closure values under IR; use the block id from the evaluated arg.
         let closure: Closure = call.req(engine_state, caller_stack, 0)?;
-        let block = engine_state.get_block(closure.block_id);
+        let block = engine_state.get_closure_block(&closure, call.head)?;
         let mut callee_stack = caller_stack
             .gather_captures(engine_state, &block.captures)
             .reset_pipes();

crates/nu-command/src/env/with_env.rs

@@ -59,7 +59,7 @@ fn with_env(
 ) -> Result<PipelineData, ShellError> {
     let env: Record = call.req(engine_state, stack, 0)?;
     let capture_block: Closure = call.req(engine_state, stack, 1)?;
-    let block = engine_state.get_block(capture_block.block_id);
+    let block = engine_state.get_closure_block(&capture_block, call.head)?;
     let mut stack = stack.captures_to_stack_preserve_out_dest(capture_block.captures);
 
     // TODO: factor list of prohibited env vars into common place

crates/nu-command/src/experimental/job_spawn.rs

@@ -58,6 +58,11 @@ impl Command for JobSpawn {
 
         let closure: Closure = call.req(engine_state, stack, 0)?;
 
+        // The job thread's engine state is a clone of this one, so an unavailable
+        // block can be reported here, where the caller's `try`/`catch` sees it,
+        // instead of from the background thread after the job was registered.
+        engine_state.get_closure_block(&closure, head)?;
+
         let description: Option<String> = call.get_flag(engine_state, stack, "description")?;
         let job_stack = stack.clone();
 
@@ -107,8 +112,8 @@ impl Command for JobSpawn {
                     Some(Redirection::Pipe(OutDest::Null)),
                     Some(Redirection::Pipe(OutDest::Null)),
                 );
-                ClosureEvalOnce::new_preserve_out_dest(&job_state, &stack, closure)
-                    .run_with_input(PipelineData::Empty)
+                ClosureEvalOnce::try_new_preserve_out_dest(&job_state, &stack, closure, head)
+                    .and_then(|eval| eval.run_with_input(PipelineData::Empty))
                     .and_then(|data| data.drain())
                     .unwrap_or_else(|err| {
                         if !job_state.signals().interrupted() {

crates/nu-command/src/filesystem/watch.rs

@@ -331,7 +331,7 @@ fn run_closure(
         eprintln!("Now watching files at {path:?}. Press ctrl+c to abort.");
     }
 
-    let mut closure = ClosureEval::new(engine_state, stack, closure);
+    let mut closure = ClosureEval::try_new(engine_state, stack, closure, head)?;
     for events in iter {
         for event in events? {
             let matches_glob = glob_filter(glob_pattern.as_ref(), &event);

crates/nu-command/src/filters/chunk_by.rs

@@ -201,7 +201,7 @@ pub fn chunk_by(
         PipelineData::Value(Value::Range { .. }, ..)
         | PipelineData::Value(Value::List { .. }, ..)
         | PipelineData::ListStream(..) => {
-            let closure = ClosureEval::new(engine_state, stack, closure);
+            let closure = ClosureEval::try_new(engine_state, stack, closure, head)?;
 
             let result = chunk_value_stream(
                 input.into_iter(),

crates/nu-command/src/filters/default.rs

@@ -48,7 +48,7 @@ impl Command for Default {
         let columns: Vec<CellPath> = call.rest(engine_state, stack, 1)?;
         let empty = call.has_flag(engine_state, stack, "empty")?;
 
-        let default_value = DefaultValue::new(engine_state, stack, default_value);
+        let default_value = DefaultValue::new(engine_state, stack, default_value, call.head)?;
 
         default(
             engine_state,
@@ -243,7 +243,12 @@ enum DefaultValue {
 }
 
 impl DefaultValue {
-    fn new(engine_state: &EngineState, stack: &Stack, value: Value) -> Self {
+    fn new(
+        engine_state: &EngineState,
+        stack: &Stack,
+        value: Value,
+        head: Span,
+    ) -> Result<Self, ShellError> {
         let span = value.span();
 
         // FIXME temporary workaround to warn people of breaking change from #15654.
@@ -251,15 +256,14 @@ impl DefaultValue {
         // with `$` (no AST required under IR).
         let value = match closure_variable_warning(stack, engine_state, value) {
             Ok(val) => val,
-            Err(default_value) => return default_value,
+            Err(default_value) => return Ok(default_value),
         };
 
         match value {
-            Value::Closure { val, .. } => {
-                let closure_eval = ClosureEval::new(engine_state, stack, *val);
-                DefaultValue::Uncalculated(Box::new(closure_eval.into_spanned(span)))
-            }
-            _ => DefaultValue::Calculated(value),
+            Value::Closure { val, .. } => Ok(DefaultValue::Uncalculated(Box::new(
+                ClosureEval::try_new(engine_state, stack, *val, head)?.into_spanned(span),
+            ))),
+            _ => Ok(DefaultValue::Calculated(value)),
         }
     }
 

crates/nu-command/src/filters/each.rs

@@ -173,7 +173,7 @@ they represent."#
             | PipelineData::Value(Value::List { .. }, ..)
             | PipelineData::ListStream(..) => {
                 let metadata = input.take_metadata();
-                let mut closure = ClosureEval::new(engine_state, stack, closure);
+                let mut closure = ClosureEval::try_new(engine_state, stack, closure, head)?;
 
                 let out = if flatten {
                     input
@@ -201,7 +201,7 @@ they represent."#
                 if val.is_iterable() && val.type_name() != "matrix" =>
             {
                 let metadata = input.take_metadata();
-                let mut closure = ClosureEval::new(engine_state, stack, closure);
+                let mut closure = ClosureEval::try_new(engine_state, stack, closure, head)?;
 
                 let out = if flatten {
                     input
@@ -228,7 +228,7 @@ they represent."#
                     return Ok(PipelineData::empty());
                 };
 
-                let mut closure = ClosureEval::new(engine_state, stack, closure);
+                let mut closure = ClosureEval::try_new(engine_state, stack, closure, head)?;
                 let out = if flatten {
                     chunks
                         .flat_map(move |result| {
@@ -264,7 +264,7 @@ they represent."#
             // This match allows non-iterables to be accepted,
             // which is currently considered undesirable (Nov 2022).
             PipelineData::Value(value, metadata) => {
-                ClosureEvalOnce::new(engine_state, stack, closure)
+                ClosureEvalOnce::try_new(engine_state, stack, closure, head)?
                     .run_with_value_with_metadata(value, metadata)
             }
         };

crates/nu-command/src/filters/group_by.rs

@@ -525,7 +525,7 @@ fn group_closure(
     stack: &mut Stack,
 ) -> Result<IndexMap<GroupKey, Vec<Value>>, ShellError> {
     let mut groups = IndexMap::<_, Vec<_>>::new();
-    let mut closure = ClosureEval::new(engine_state, stack, closure);
+    let mut closure = ClosureEval::try_new(engine_state, stack, closure, span)?;
 
     for value in values {
         let key_val = closure.run_with_value(value.clone())?.into_value(span)?;

crates/nu-command/src/filters/insert.rs

@@ -150,7 +150,7 @@ fn insert_recursive(
             if let Value::Closure { val, .. } = replacement {
                 insert_single_value_by_closure(
                     &mut value,
-                    ClosureEvalOnce::new(engine_state, stack, *val),
+                    ClosureEvalOnce::try_new(engine_state, stack, *val, head_span)?,
                     head_span,
                     cell_paths,
                     false,
@@ -189,9 +189,10 @@ fn insert_recursive(
                         let value = stream.next();
                         let end_of_stream = value.is_none();
                         let value = value.unwrap_or(Value::nothing(head_span));
-                        let new_value = ClosureEvalOnce::new(engine_state, stack, *val)
-                            .run_with_value(value.clone())?
-                            .into_value(head_span)?;
+                        let new_value =
+                            ClosureEvalOnce::try_new(engine_state, stack, *val, head_span)?
+                                .run_with_value(value.clone())?
+                                .into_value(head_span)?;
 
                         pre_elems.push(new_value);
                         if !end_of_stream {
@@ -204,7 +205,7 @@ fn insert_recursive(
                     if let Value::Closure { val, .. } = replacement {
                         insert_single_value_by_closure(
                             &mut value,
-                            ClosureEvalOnce::new(engine_state, stack, *val),
+                            ClosureEvalOnce::try_new(engine_state, stack, *val, head_span)?,
                             head_span,
                             path,
                             true,
@@ -240,7 +241,7 @@ fn insert_recursive(
                     &new_cell_paths,
                 )
             } else if let Value::Closure { val, .. } = replacement {
-                let mut closure = ClosureEval::new(engine_state, stack, *val);
+                let mut closure = ClosureEval::try_new(engine_state, stack, *val, head_span)?;
                 let cell_paths = cell_paths.to_vec();
                 let stream = stream.map(move |mut value| {
                     let err =

crates/nu-command/src/filters/interleave.rs

@@ -119,8 +119,8 @@ interleave
             .then(|| Ok(input))
             .into_iter()
             .chain(closures.into_iter().map(|closure| {
-                ClosureEvalOnce::new(engine_state, stack, closure)
-                    .run_with_input(PipelineData::empty())
+                ClosureEvalOnce::try_new(engine_state, stack, closure, call.head)
+                    .and_then(|eval| eval.run_with_input(PipelineData::empty()))
             }))
             .try_for_each(|stream| {
                 stream.and_then(|stream| {

crates/nu-command/src/filters/items.rs

@@ -47,7 +47,7 @@ impl Command for Items {
                 let span = value.span();
                 match value {
                     Value::Record { val, .. } => {
-                        let mut closure = ClosureEval::new(engine_state, stack, closure);
+                        let mut closure = ClosureEval::try_new(engine_state, stack, closure, head)?;
                         Ok(val
                             .into_owned()
                             .into_iter()

crates/nu-command/src/filters/par_each.rs

@@ -1,5 +1,5 @@
 use super::utils::wrap_error_with_input;
-use nu_engine::{ClosureEval, ClosureEvalOnce, command_prelude::*};
+use nu_engine::{ClosureEval, command_prelude::*};
 use nu_protocol::{Signals, engine::Closure, shell_error::generic::GenericError};
 use rayon::prelude::*;
 use std::{
@@ -46,6 +46,9 @@ fn lock_pool_cache(
 fn build_pool(num_threads: usize, head: Span) -> Result<Arc<rayon::ThreadPool>, ShellError> {
     rayon::ThreadPoolBuilder::new()
         .num_threads(num_threads)
+        // Match a typical main-thread stack so the default `recursion_limit` (50)
+        // of closure calls fits on a worker instead of overflowing it.
+        .stack_size(8 * 1024 * 1024)
         .build()
         .map(Arc::new)
         .map_err(|e| {
@@ -214,6 +217,15 @@ impl Command for ParEach {
             ));
         }
 
+        // Construct the evaluator once here, so a stale closure reports a single
+        // error like `each` does instead of one error value per item. Empty input
+        // never constructs an evaluator, so it stays a no-op like in `each`.
+        // Every worker clones this evaluator instead of building its own.
+        if matches!(input, PipelineData::Empty) {
+            return Ok(PipelineData::empty());
+        }
+        let mut closure_eval = ClosureEval::try_new(engine_state, stack, closure, head)?;
+
         let mut input = input.into_stream_or_original(engine_state);
         let metadata = input.take_metadata();
 
@@ -238,17 +250,14 @@ impl Command for ParEach {
                         if keep_order {
                             Ok(pool.install(|| {
                                 let par_iter = vals.into_owned().into_par_iter().enumerate();
-                                let mapped =
-                                    parallel_closure_map(engine_state, stack, &closure, par_iter);
+                                let mapped = parallel_closure_map(closure_eval, par_iter);
                                 apply_order(mapped.collect())
                                     .into_pipeline_data(span, signals.clone())
                             }))
                         } else {
                             let par_iter = vals.into_owned().into_par_iter();
                             Ok(stream_parallel_values(
-                                engine_state,
-                                stack,
-                                closure.clone(),
+                                closure_eval,
                                 pool,
                                 span,
                                 signals.clone(),
@@ -264,17 +273,14 @@ impl Command for ParEach {
                                     .into_range_iter(span, signals.clone())
                                     .enumerate()
                                     .par_bridge();
-                                let mapped =
-                                    parallel_closure_map(engine_state, stack, &closure, par_iter);
+                                let mapped = parallel_closure_map(closure_eval, par_iter);
                                 apply_order(mapped.collect())
                                     .into_pipeline_data(span, signals.clone())
                             }))
                         } else {
                             let par_iter = val.into_range_iter(span, signals.clone()).par_bridge();
                             Ok(stream_parallel_values(
-                                engine_state,
-                                stack,
-                                closure.clone(),
+                                closure_eval,
                                 pool,
                                 span,
                                 signals.clone(),
@@ -284,9 +290,7 @@ impl Command for ParEach {
                     }
                     // This match allows non-iterables to be accepted,
                     // which is currently considered undesirable (Nov 2022).
-                    value => {
-                        ClosureEvalOnce::new(engine_state, stack, closure).run_with_value(value)
-                    }
+                    value => closure_eval.run_with_value(value),
                 }
             }
             PipelineData::ListStream(stream, ..) => {
@@ -294,15 +298,13 @@ impl Command for ParEach {
                 if keep_order {
                     Ok(pool.install(|| {
                         let par_iter = stream.into_iter().enumerate().par_bridge();
-                        let mapped = parallel_closure_map(engine_state, stack, &closure, par_iter);
+                        let mapped = parallel_closure_map(closure_eval, par_iter);
                         apply_order(mapped.collect()).into_pipeline_data(head, signals.clone())
                     }))
                 } else {
                     let par_iter = stream.into_iter().par_bridge();
                     Ok(stream_parallel_values(
-                        engine_state,
-                        stack,
-                        closure.clone(),
+                        closure_eval,
                         pool,
                         head,
                         signals.clone(),
@@ -321,8 +323,7 @@ impl Command for ParEach {
                                     (idx, val.unwrap_or_else(|err| Value::error(err, head)))
                                 })
                                 .par_bridge();
-                            let mapped =
-                                parallel_closure_map(engine_state, stack, &closure, par_iter);
+                            let mapped = parallel_closure_map(closure_eval, par_iter);
                             apply_order(mapped.collect()).into_pipeline_data(head, signals.clone())
                         }))
                     } else {
@@ -330,9 +331,7 @@ impl Command for ParEach {
                             .map(move |val| val.unwrap_or_else(|err| Value::error(err, head)))
                             .par_bridge();
                         Ok(stream_parallel_values(
-                            engine_state,
-                            stack,
-                            closure.clone(),
+                            closure_eval,
                             pool,
                             head,
                             signals.clone(),
@@ -350,19 +349,17 @@ impl Command for ParEach {
 }
 
 fn stream_parallel_values(
-    engine_state: &EngineState,
-    stack: &Stack,
-    closure: Closure,
+    closure_eval: ClosureEval,
     pool: Arc<rayon::ThreadPool>,
     span: Span,
     signals: Signals,
     input: impl ParallelIterator<Item = Value> + 'static,
 ) -> PipelineData {
     let (tx, rx) = mpsc::sync_channel(STREAM_BUFFER_SIZE);
-    let worker_engine_state = engine_state.clone();
-    // The worker outlives the borrow of `stack`, so it gets its own copy. `ClosureEval::new`
-    // builds the closure's stack from it, the same as the `--keep-order` path does from `stack`.
-    let worker_stack = stack.clone();
+    // Each worker clones the evaluator, whose stack already carries the
+    // `recursion_count + 1` from `captures_to_stack`, so closure calls in
+    // workers count toward `recursion_limit` exactly once, like the
+    // `--keep-order` path.
     let worker_signals = signals.clone();
 
     // Keep an Arc for the lifetime of the spawned producer. For cached pools this keeps
@@ -374,7 +371,7 @@ fn stream_parallel_values(
 
         let _ = input
             .map_init(
-                move || ClosureEval::new(&worker_engine_state, &worker_stack, closure.clone()),
+                move || closure_eval.clone(),
                 move |closure_eval, value| {
                     if map_signals.interrupted() {
                         return Err(());
@@ -447,18 +444,12 @@ fn run_closure_on_value(closure_eval: &mut ClosureEval, value: Value) -> Value {
 }
 
 fn parallel_closure_map(
-    engine_state: &EngineState,
-    stack: &mut Stack,
-    closure: &Closure,
+    closure_eval: ClosureEval,
     input: impl ParallelIterator<Item = (usize, Value)>,
 ) -> impl ParallelIterator<Item = (usize, Value)> {
     input.map_init(
-        move || ClosureEval::new(engine_state, stack, closure.clone()),
-        |closure_eval, (index, value)| {
-            let value = run_closure_on_value(closure_eval, value);
-
-            (index, value)
-        },
+        move || closure_eval.clone(),
+        |closure_eval, (index, value)| (index, run_closure_on_value(closure_eval, value)),
     )
 }
 

crates/nu-command/src/filters/reduce.rs

@@ -122,7 +122,7 @@ impl Command for Reduce {
             ShellError::Generic(GenericError::new("Expected input", "needs input", head))
         })?;
 
-        let mut closure = ClosureEval::new(engine_state, stack, closure);
+        let mut closure = ClosureEval::try_new(engine_state, stack, closure, head)?;
 
         for value in iter {
             engine_state.signals().check(&head)?;

crates/nu-command/src/filters/rename.rs

@@ -138,7 +138,9 @@ fn rename(
     };
     let closure: Option<Closure> = call.get_flag(engine_state, stack, "block")?;
 
-    let mut closure = closure.map(|closure| ClosureEval::new(engine_state, stack, closure));
+    let mut closure = closure
+        .map(|closure| ClosureEval::try_new(engine_state, stack, closure, call.head))
+        .transpose()?;
 
     let metadata = input.take_metadata();
     input

crates/nu-command/src/filters/skip/skip_until.rs

@@ -77,7 +77,7 @@ impl Command for SkipUntil {
         let head = call.head;
         let closure: Closure = call.req(engine_state, stack, 0)?;
 
-        let mut closure = ClosureEval::new(engine_state, stack, closure);
+        let mut closure = ClosureEval::try_new(engine_state, stack, closure, head)?;
 
         let metadata = input.take_metadata();
         Ok(input

crates/nu-command/src/filters/skip/skip_while.rs

@@ -82,7 +82,7 @@ impl Command for SkipWhile {
         let head = call.head;
         let closure: Closure = call.req(engine_state, stack, 0)?;
 
-        let mut closure = ClosureEval::new(engine_state, stack, closure);
+        let mut closure = ClosureEval::try_new(engine_state, stack, closure, head)?;
 
         let metadata = input.take_metadata();
         Ok(input

crates/nu-command/src/filters/sort_by.rs

@@ -149,7 +149,7 @@ impl Command for SortBy {
             .map(|val| match val {
                 Value::CellPath { val, .. } => Ok(Comparator::CellPath(val)),
                 Value::Closure { val, .. } => {
-                    let closure_eval = ClosureEval::new(engine_state, stack, *val);
+                    let closure_eval = ClosureEval::try_new(engine_state, stack, *val, head)?;
                     if custom {
                         Ok(Comparator::CustomClosure(closure_eval))
                     } else {

crates/nu-command/src/filters/take/take_until.rs

@@ -79,7 +79,7 @@ impl Command for TakeUntil {
 
         let metadata = input.take_metadata();
 
-        let mut closure = ClosureEval::new(engine_state, stack, closure);
+        let mut closure = ClosureEval::try_new(engine_state, stack, closure, head)?;
         let predicate = move |value: &Value| {
             closure
                 .run_with_value(value.clone())

crates/nu-command/src/filters/take/take_while.rs

@@ -79,7 +79,7 @@ impl Command for TakeWhile {
 
         let metadata = input.take_metadata();
 
-        let mut closure = ClosureEval::new(engine_state, stack, closure);
+        let mut closure = ClosureEval::try_new(engine_state, stack, closure, head)?;
         let predicate = move |value: &Value| {
             closure
                 .run_with_value(value.clone())

crates/nu-command/src/filters/tee.rs

@@ -86,13 +86,16 @@ use it in your pipeline."
         let use_stderr = call.has_flag(engine_state, stack, "stderr")?;
 
         let closure: Spanned<Closure> = call.req(engine_state, stack, 0)?;
-        let closure_span = closure.span;
+        let closure_span = call.head;
         let closure = closure.item;
 
         let engine_state_arc = Arc::new(engine_state.clone());
 
         let mut eval_block = {
             let closure_engine_state = engine_state_arc.clone();
+            let block = closure_engine_state
+                .get_closure_block(&closure, closure_span)?
+                .clone();
             let mut closure_stack = stack
                 .captures_to_stack_preserve_out_dest(closure.captures)
                 .reset_pipes();
@@ -102,7 +105,7 @@ use it in your pipeline."
                 let result = eval_block_with_early_return(
                     &closure_engine_state,
                     &mut closure_stack,
-                    closure_engine_state.get_block(closure.block_id),
+                    &block,
                     input,
                 )
                 .map(|p| p.body);

crates/nu-command/src/filters/update.rs

@@ -122,13 +122,13 @@ fn update_recursive(
                 if matches!(cell_paths.first(), Some(PathMember::Int { .. })) {
                     update_single_value_by_closure(
                         &mut value,
-                        ClosureEvalOnce::new(engine_state, stack, *val),
+                        ClosureEvalOnce::try_new(engine_state, stack, *val, head_span)?,
                         head_span,
                         cell_paths,
                         false,
                     )?;
                 } else {
-                    let mut closure = ClosureEval::new(engine_state, stack, *val);
+                    let mut closure = ClosureEval::try_new(engine_state, stack, *val, head_span)?;
                     update_value_by_closure(&mut value, &mut closure, head_span, cell_paths)?;
                 }
             } else {
@@ -177,7 +177,7 @@ fn update_recursive(
                 if let Value::Closure { val, .. } = replacement {
                     update_single_value_by_closure(
                         value,
-                        ClosureEvalOnce::new(engine_state, stack, *val),
+                        ClosureEvalOnce::try_new(engine_state, stack, *val, head_span)?,
                         head_span,
                         path,
                         true,
@@ -204,7 +204,7 @@ fn update_recursive(
                     &new_cell_paths,
                 )
             } else if let Value::Closure { val, .. } = replacement {
-                let mut closure = ClosureEval::new(engine_state, stack, *val);
+                let mut closure = ClosureEval::try_new(engine_state, stack, *val, head_span)?;
                 let cell_paths = cell_paths.to_vec();
                 let stream = stream.map(move |mut value| {
                     let err =

crates/nu-command/src/filters/upsert.rs

@@ -178,7 +178,7 @@ fn upsert_recursive(
             if let Value::Closure { val, .. } = replacement {
                 upsert_single_value_by_closure(
                     &mut value,
-                    ClosureEvalOnce::new(engine_state, stack, *val),
+                    ClosureEvalOnce::try_new(engine_state, stack, *val, head_span)?,
                     head_span,
                     cell_paths,
                     false,
@@ -215,7 +215,7 @@ fn upsert_recursive(
                 let value = if path.is_empty() {
                     let value = stream.next().unwrap_or(Value::nothing(head_span));
                     if let Value::Closure { val, .. } = replacement {
-                        ClosureEvalOnce::new(engine_state, stack, *val)
+                        ClosureEvalOnce::try_new(engine_state, stack, *val, head_span)?
                             .run_with_value(value)?
                             .into_value(head_span)?
                     } else {
@@ -225,7 +225,7 @@ fn upsert_recursive(
                     if let Value::Closure { val, .. } = replacement {
                         upsert_single_value_by_closure(
                             &mut value,
-                            ClosureEvalOnce::new(engine_state, stack, *val),
+                            ClosureEvalOnce::try_new(engine_state, stack, *val, head_span)?,
                             head_span,
                             path,
                             true,
@@ -263,7 +263,7 @@ fn upsert_recursive(
                     &new_cell_paths,
                 )
             } else if let Value::Closure { val, .. } = replacement {
-                let mut closure = ClosureEval::new(engine_state, stack, *val);
+                let mut closure = ClosureEval::try_new(engine_state, stack, *val, head_span)?;
                 let cell_paths = cell_paths.to_vec();
                 let stream = stream.map(move |mut value| {
                     let err =

crates/nu-command/src/filters/utils.rs

@@ -97,7 +97,7 @@ pub fn boolean_fold(
     let head = call.head;
     let closure: Closure = call.req(engine_state, stack, 0)?;
 
-    let mut closure = ClosureEval::new(engine_state, stack, closure);
+    let mut closure = ClosureEval::try_new(engine_state, stack, closure, head)?;
 
     for value in input {
         engine_state.signals().check(&head)?;

crates/nu-command/src/filters/where_.rs

@@ -64,7 +64,7 @@ Row conditions cannot be stored in a variable. To pass a condition with a variab
         let head = call.head;
         let closure: Closure = call.req(engine_state, stack, 0)?;
 
-        let mut closure = ClosureEval::new(engine_state, stack, closure);
+        let mut closure = ClosureEval::try_new(engine_state, stack, closure, head)?;
 
         let metadata = input.take_metadata();
         Ok(input

crates/nu-command/src/filters/zip.rs

@@ -103,7 +103,8 @@ impl Command for Zip {
         let metadata = input.take_metadata();
         let other = if let Value::Closure { val, .. } = other {
             // If a closure was provided, evaluate it and consume its stream output
-            ClosureEvalOnce::new(engine_state, stack, *val).run_with_input(PipelineData::empty())?
+            ClosureEvalOnce::try_new(engine_state, stack, *val, head)?
+                .run_with_input(PipelineData::empty())?
         } else {
             other.into_pipeline_data()
         };

crates/nu-command/src/formats/to/text.rs

@@ -217,8 +217,8 @@ fn local_into_string(
             .join(separator),
         Value::Closure { val, .. } => {
             if serialize_types {
-                let block = engine_state.get_block(val.block_id);
-                if let Some(span) = block.span {
+                let block = engine_state.try_get_block(val.block_id);
+                if let Some(span) = block.and_then(|block| block.span) {
                     let contents_bytes = engine_state.get_span_contents(span);
                     let contents_string = String::from_utf8_lossy(contents_bytes);
                     contents_string.to_string()

crates/nu-command/src/formats/to/toml.rs

@@ -79,8 +79,8 @@ pub(crate) fn nu_value_to_toml_value(
         }
         Value::Closure { val, .. } => {
             if serialize_types {
-                let block = engine_state.get_block(val.block_id);
-                if let Some(span) = block.span {
+                let block = engine_state.try_get_block(val.block_id);
+                if let Some(span) = block.and_then(|block| block.span) {
                     let contents_bytes = engine_state.get_span_contents(span);
                     let contents_string = String::from_utf8_lossy(contents_bytes);
                     toml::Value::String(contents_string.to_string())

crates/nu-command/src/generators/generate.rs

@@ -99,8 +99,8 @@ In this case, generation also stops when the input stream stops."
         let head = call.head;
         let closure: Closure = call.req(engine_state, stack, 0)?;
         let initial: Option<Value> = call.opt(engine_state, stack, 1)?;
-        let block = engine_state.get_block(closure.block_id);
-        let mut closure = ClosureEval::new(engine_state, stack, closure);
+        let mut closure = ClosureEval::try_new(engine_state, stack, closure, head)?;
+        let block = closure.block();
 
         match input {
             PipelineData::Empty | PipelineData::Value(Value::Nothing { .. }, ..) => {

crates/nu-command/src/matrix/map.rs

@@ -50,7 +50,7 @@ impl Command for MatrixMap {
         let mut vals = Vec::with_capacity(matrix.array.len());
         for &val in matrix.array.iter() {
             let element = Value::float(val, head);
-            let result = ClosureEvalOnce::new(engine_state, stack, closure.clone())
+            let result = ClosureEvalOnce::try_new(engine_state, stack, closure.clone(), head)?
                 .run_with_value(element)?
                 .into_value(head)?;
             vals.push(value_to_f64(&result, head).map_err(|_| {

crates/nu-command/src/matrix/reduce.rs

@@ -66,7 +66,7 @@ impl Command for MatrixReduce {
             }
         };
 
-        let mut closure_eval = ClosureEval::new(engine_state, stack, closure);
+        let mut closure_eval = ClosureEval::try_new(engine_state, stack, closure, head)?;
 
         // Match `reduce`: first arg is the element, second is the accumulator.
         for &val in iter {

crates/nu-command/src/platform/input/list.rs

@@ -903,10 +903,8 @@ impl InputList {
                     .map(|v| v.to_expanded_string(", ", config))
                     .unwrap_or_else(|_| value.to_expanded_string(", ", config)),
                 DisplayMode::Closure(closure) => {
-                    let mut closure_eval =
-                        ClosureEval::new(engine_state, stack, Closure::clone(closure));
-                    closure_eval
-                        .run_with_value(value.clone())
+                    ClosureEval::try_new(engine_state, stack, Closure::clone(closure), span)
+                        .and_then(|mut closure_eval| closure_eval.run_with_value(value.clone()))
                         .and_then(|data| data.into_value(span))
                         .map(|v| v.to_expanded_string(", ", config))
                         .unwrap_or_else(|_| value.to_expanded_string(", ", config))

crates/nu-command/src/record/apply.rs

@@ -99,7 +99,7 @@ Given a list of records, the transforms are applied to each record in turn and t
         let transforms: Record = call.req(engine_state, stack, 0)?;
         let strict = call.has_flag(engine_state, stack, "strict")?;
 
-        let mut transforms = prepare(engine_state, stack, transforms)?;
+        let mut transforms = prepare(engine_state, stack, transforms, head)?;
         map_records(
             input,
             head,
@@ -129,28 +129,30 @@ fn prepare(
     engine_state: &EngineState,
     stack: &Stack,
     transforms: Record,
+    head: Span,
 ) -> Result<Transforms, ShellError> {
     transforms
         .into_iter()
         .map(|(key, transform)| {
             let span = transform.span();
-            let transform = match transform {
-                Value::Closure { val, .. } => {
-                    Transform::Closure(Box::new(ClosureEval::new(engine_state, stack, *val)))
-                }
-                Value::Record { val, .. } => {
-                    Transform::Record(prepare(engine_state, stack, val.into_owned())?)
-                }
-                other => {
-                    return Err(ShellError::TypeMismatch {
-                        err_message: format!(
-                            "expected a closure or a record of closures, found {}",
-                            other.get_type()
-                        ),
-                        span,
-                    });
-                }
-            };
+            let transform =
+                match transform {
+                    Value::Closure { val, .. } => Transform::Closure(Box::new(
+                        ClosureEval::try_new(engine_state, stack, *val, head)?,
+                    )),
+                    Value::Record { val, .. } => {
+                        Transform::Record(prepare(engine_state, stack, val.into_owned(), head)?)
+                    }
+                    other => {
+                        return Err(ShellError::TypeMismatch {
+                            err_message: format!(
+                                "expected a closure or a record of closures, found {}",
+                                other.get_type()
+                            ),
+                            span,
+                        });
+                    }
+                };
             Ok((key, span, transform))
         })
         .collect()

crates/nu-command/src/record/each.rs

@@ -93,7 +93,7 @@ Given a list of records, each record is rebuilt on its own and the results are s
         let head = call.head;
         let closure: Closure = call.req(engine_state, stack, 0)?;
 
-        let mut closure = ClosureEval::new(engine_state, stack, closure);
+        let mut closure = ClosureEval::try_new(engine_state, stack, closure, head)?;
         map_records(input, head, engine_state.signals(), move |record, span| {
             // Like `Record::insert` and `into record`, `IndexMap::insert` keeps the
             // position of an existing key, but finds it without a linear search.

crates/nu-command/src/record/walk.rs

@@ -131,7 +131,7 @@ Given a list of records, each record is walked on its own, with cell paths start
         let closure: Closure = call.req(engine_state, stack, 0)?;
 
         let mut walker = Walker {
-            closure: ClosureEval::new(engine_state, stack, closure),
+            closure: ClosureEval::try_new(engine_state, stack, closure, head)?,
             containers: call.has_flag(engine_state, stack, "containers")?,
             lists: !call.has_flag(engine_state, stack, "no-lists")?,
             path: Vec::new(),

crates/nu-command/src/record/where_.rs

@@ -105,13 +105,6 @@ Given a list of records, each record is filtered on its own and the results are
     ) -> Result<PipelineData, ShellError> {
         let head = call.head;
         let closure: Closure = call.req(engine_state, stack, 0)?;
-        // The parser turns a row condition such as `value > 1` into a block whose
-        // only parameter is `$it` (see `parse_row_condition`), while a closure
-        // literal or a closure in a variable keeps the parameters it declares.
-        let row_condition = matches!(
-            engine_state.get_block(closure.block_id).signature.required_positional.as_slice(),
-            [param] if param.name == "$it"
-        );
         let keys_only = call.has_flag(engine_state, stack, "keys")?;
         let values_only = call.has_flag(engine_state, stack, "values")?;
         if keys_only && values_only {
@@ -122,8 +115,14 @@ Given a list of records, each record is filtered on its own and the results are
                 right_span: call.get_flag_span(stack, "values").unwrap_or(head),
             });
         }
-
-        let mut closure = ClosureEval::new(engine_state, stack, closure);
+        let mut closure = ClosureEval::try_new(engine_state, stack, closure, head)?;
+        // The parser turns a row condition such as `value > 1` into a block whose
+        // only parameter is `$it` (see `parse_row_condition`), while a closure
+        // literal or a closure in a variable keeps the parameters it declares.
+        let row_condition = matches!(
+            closure.block().signature.required_positional.as_slice(),
+            [param] if param.name == "$it"
+        );
         map_records(input, head, engine_state.signals(), move |record, span| {
             let mut kept = Record::new();
             for (key, value) in record {

crates/nu-command/src/strings/split/list.rs

@@ -191,7 +191,7 @@ impl Command for SubCommand {
         let split = split.unwrap_or(Split::On);
         let matcher = match separator {
             Value::Closure { val, .. } => {
-                Matcher::from_closure(ClosureEval::new(engine_state, stack, *val))
+                Matcher::from_closure(ClosureEval::try_new(engine_state, stack, *val, call.head)?)
             }
             _ => Matcher::new(engine_state, has_regex, separator)?,
         };

crates/nu-command/src/strings/str_/replace.rs

@@ -129,10 +129,8 @@ groups as its argument. It must return a string that will be used as a replaceme
     ) -> Result<PipelineData, ShellError> {
         let find: Spanned<String> = call.req(engine_state, stack, 0)?;
         let replace = match call.req(engine_state, stack, 1)? {
-            Value::Closure {
-                val, internal_span, ..
-            } => Ok(ReplacementValue::Closure(Box::new(
-                ClosureEval::new(engine_state, stack, *val).into_spanned(internal_span),
+            Value::Closure { val, .. } => Ok(ReplacementValue::Closure(Box::new(
+                ClosureEval::try_new(engine_state, stack, *val, call.head)?.into_spanned(call.head),
             ))),
             Value::String {
                 val, internal_span, ..

crates/nu-command/tests/commands/job.rs

@@ -38,6 +38,335 @@ fn job_send_background_job_works() -> Result {
     test().run(code).expect_value_eq("boop")
 }
 
+#[test]
+#[cfg_attr(ci, serial)]
+fn job_send_closure_parsed_after_spawn_returns_an_error() -> Result {
+    let mut tester = test();
+
+    tester.run::<()>(
+        "
+        let queue = job spawn {
+            job recv | try { do $in } catch { |err| $err | job send 0 }
+        }
+    ",
+    )?;
+
+    tester.run::<()>("{ 1 + 1 } | job send $queue")?;
+
+    tester
+        .run("job recv --timeout 10sec | get msg")
+        .expect_value_eq("Closure cannot be evaluated")
+}
+
+#[test]
+#[cfg_attr(ci, serial)]
+fn job_send_closure_parsed_after_spawn_zip_returns_an_error() -> Result {
+    let mut tester = test();
+
+    tester.run::<()>(
+        "
+        let queue = job spawn {
+            let closure = job recv
+            try { [1 2 3] | zip $closure } catch { |err| $err | job send 0 }
+        }
+    ",
+    )?;
+
+    tester.run::<()>("{ |it| $it + 1 } | job send $queue")?;
+
+    tester
+        .run("job recv --timeout 10sec | get msg")
+        .expect_value_eq("Closure cannot be evaluated")
+}
+
+#[test]
+#[cfg_attr(ci, serial)]
+fn job_send_closure_parsed_after_spawn_each_returns_an_error() -> Result {
+    let mut tester = test();
+
+    tester.run::<()>(
+        "
+        let queue = job spawn {
+            job recv | try { $in | each $in } catch { |err| $err | job send 0 }
+        }
+    ",
+    )?;
+
+    tester.run::<()>("{ |it| $it + 1 } | job send $queue")?;
+
+    tester
+        .run("job recv --timeout 10sec | get msg")
+        .expect_value_eq("Closure cannot be evaluated")
+}
+
+#[test]
+#[cfg_attr(ci, serial)]
+fn job_send_closure_parsed_after_spawn_tee_returns_an_error() -> Result {
+    let mut tester = test();
+
+    tester.run::<()>(
+        "
+        let queue = job spawn {
+            let closure = job recv
+            try { [1 2 3] | tee $closure | ignore } catch { |err| $err | job send 0 }
+        }
+    ",
+    )?;
+
+    tester.run::<()>("{ |it| $it + 1 } | job send $queue")?;
+
+    tester
+        .run("job recv --timeout 10sec | get msg")
+        .expect_value_eq("Closure cannot be evaluated")
+}
+
+#[test]
+#[cfg_attr(ci, serial)]
+fn job_send_closure_parsed_after_spawn_par_each_returns_one_error() -> Result {
+    let mut tester = test();
+
+    tester.run::<()>(
+        "
+        let queue = job spawn {
+            let closure = job recv
+            try { [1 2 3] | par-each $closure | ignore } catch { |err| $err | job send 0 }
+        }
+    ",
+    )?;
+
+    tester.run::<()>("{ |it| $it + 1 } | job send $queue")?;
+
+    tester
+        .run("job recv --timeout 10sec | get msg")
+        .expect_value_eq("Closure cannot be evaluated")
+}
+
+#[test]
+#[cfg_attr(ci, serial)]
+fn job_send_closure_parsed_after_spawn_record_each_returns_an_error() -> Result {
+    let mut tester = test();
+
+    tester.run::<()>(
+        "
+        let queue = job spawn {
+            let closure = job recv
+            try { {a: 1} | record each $closure | ignore } catch { |err| $err | job send 0 }
+        }
+    ",
+    )?;
+
+    tester.run::<()>("{ |key, value| {($key): $value} } | job send $queue")?;
+
+    tester
+        .run("job recv --timeout 10sec | get msg")
+        .expect_value_eq("Closure cannot be evaluated")
+}
+
+#[test]
+#[cfg_attr(ci, serial)]
+fn job_send_closure_parsed_after_spawn_view_source_returns_an_error() -> Result {
+    let mut tester = test();
+
+    tester.run::<()>(
+        "
+        let queue = job spawn {
+            let closure = job recv
+            try { view source $closure | ignore } catch { |err| $err | job send 0 }
+        }
+    ",
+    )?;
+
+    tester.run::<()>("{ |it| $it + 1 } | job send $queue")?;
+
+    tester
+        .run("job recv --timeout 10sec | get msg")
+        .expect_value_eq("Closure cannot be evaluated")
+}
+
+#[test]
+#[cfg_attr(ci, serial)]
+fn job_send_closure_parsed_after_spawn_describe_detailed_does_not_panic() -> Result {
+    let mut tester = test();
+
+    tester.run::<()>(
+        "
+        let queue = job spawn {
+            let closure = job recv
+            try { $closure | describe --detailed | get type | job send 0 } catch { |err| $err.msg | job send 0 }
+        }
+    ",
+    )?;
+
+    tester.run::<()>("{ |it| $it + 1 } | job send $queue")?;
+
+    tester
+        .run("job recv --timeout 10sec")
+        .expect_value_eq("closure")
+}
+
+#[test]
+#[cfg_attr(ci, serial)]
+fn job_send_closure_parsed_after_spawn_debug_profile_does_not_leave_debugger_active() -> Result {
+    let mut tester = test();
+
+    tester.run::<()>(
+        "
+        let queue = job spawn {
+            let closure = job recv
+            try { debug profile $closure | ignore } catch { |err| $err.msg | job send 0 }
+        }
+    ",
+    )?;
+
+    tester.run::<()>("{ |it| $it + 1 } | job send $queue")?;
+
+    // Once the error arrives, the `debug profile` call has finished. Its failure
+    // must not leave the profiler installed: the debugger slot is shared with
+    // the job, so a leaked profiler shows up here as a non-Noop debugger.
+    let msg = tester.run::<String>("job recv --timeout 10sec")?;
+    assert_eq!(msg, "Closure cannot be evaluated");
+
+    let debugger = tester
+        .engine_state
+        .debugger
+        .lock()
+        .expect("debugger lock is poisoned");
+    assert!(
+        format!("{debugger:?}").contains("NoopDebugger"),
+        "debug profile left an active debugger installed: {debugger:?}"
+    );
+    Ok(())
+}
+
+#[test]
+#[cfg_attr(ci, serial)]
+fn job_send_closure_parsed_after_spawn_job_spawn_is_catchable() -> Result {
+    let mut tester = test();
+
+    tester.run::<()>(
+        "
+        let queue = job spawn {
+            let closure = job recv
+            try { job spawn $closure | ignore; '' } catch { |err| $err.msg } | job send 0
+        }
+    ",
+    )?;
+
+    tester.run::<()>("{ |it| $it + 1 } | job send $queue")?;
+
+    tester
+        .run("job recv --timeout 10sec")
+        .expect_value_eq("Closure cannot be evaluated")
+}
+
+#[test]
+#[cfg_attr(ci, serial)]
+fn job_send_closure_parsed_after_spawn_error_labels_render() -> Result {
+    let mut tester = test();
+
+    // The stale closure's own span points into source this job's engine state
+    // does not have. Errors must label themselves with a call-site span
+    // instead, or the label fails to render (`OutOfBounds`).
+    tester.run::<()>(
+        "
+        let queue = job spawn {
+            let closure = job recv
+            [
+                (try { [{a: 1}] | update a $closure | ignore; '' } catch { |err| $err | to text })
+                (try { [1 2 3] | tee $closure | ignore; '' } catch { |err| $err | to text })
+                (try { ['b'] | str replace 'b' $closure | ignore; '' } catch { |err| $err | to text })
+                (try { 5 | default (job recv) | ignore; '' } catch { |err| $err | to text })
+            ] | job send 0
+        }
+    ",
+    )?;
+
+    // The second closure feeds the inline `(job recv)` in the `default` site.
+    tester.run::<()>("{ |it| $it + 1 } | job send $queue")?;
+    tester.run::<()>("{ |it| $it + 1 } | job send $queue")?;
+
+    let rendered: Vec<String> = tester.run("job recv --timeout 10sec")?;
+    for text in &rendered {
+        assert!(
+            text.contains("Closure cannot be evaluated"),
+            "error not rendered: {text}"
+        );
+        // A label span pointing into the sender's source fails to render.
+        assert!(
+            !text.contains("OutOfBounds"),
+            "error label failed to render: {text}"
+        );
+    }
+    Ok(())
+}
+
+#[test]
+#[cfg_attr(ci, serial)]
+fn job_send_closure_parsed_after_spawn_default_returns_an_error() -> Result {
+    let mut tester = test();
+
+    tester.run::<()>(
+        "
+        let queue = job spawn {
+            try { 5 | default (job recv) | ignore; '' } catch { |err| $err.msg } | job send 0
+        }
+    ",
+    )?;
+
+    tester.run::<()>("{ 1 + 1 } | job send $queue")?;
+
+    tester
+        .run("job recv --timeout 10sec")
+        .expect_value_eq("Closure cannot be evaluated")
+}
+
+#[test]
+#[cfg_attr(ci, serial)]
+fn job_send_closure_parsed_after_spawn_other_consumers_return_errors() -> Result {
+    let mut tester = test();
+
+    tester.run::<()>(
+        "
+        let queue = job spawn {
+            let closure = job recv
+            [
+                (try { with-env {X: 1} $closure | ignore; '' } catch { |err| $err.msg })
+                (try { explain $closure | ignore; '' } catch { |err| $err.msg })
+                (try { generate $closure 1 | ignore; '' } catch { |err| $err.msg })
+                (try { {a: 1} | record where $closure | ignore; '' } catch { |err| $err.msg })
+                (try { {a: {b: 1}} | record walk $closure | ignore; '' } catch { |err| $err.msg })
+                (try { [2 1] | sort-by $closure | ignore; '' } catch { |err| $err.msg })
+            ] | job send 0
+        }
+    ",
+    )?;
+
+    tester.run::<()>("{ |it| $it + 1 } | job send $queue")?;
+
+    tester
+        .run("job recv --timeout 10sec")
+        .expect_value_eq(vec!["Closure cannot be evaluated"; 6])
+}
+
+#[test]
+#[cfg_attr(ci, serial)]
+fn job_send_closure_parsed_after_spawn_to_text_serialize_does_not_panic() -> Result {
+    let mut tester = test();
+
+    tester.run::<()>(
+        "
+        let queue = job spawn {
+            let closure = job recv
+            try { $closure | to text --serialize | str starts-with 'unable to retrieve' | job send 0 } catch { |err| $err.msg | job send 0 }
+        }
+    ",
+    )?;
+
+    tester.run::<()>("{ |it| $it + 1 } | job send $queue")?;
+
+    tester.run("job recv --timeout 10sec").expect_value_eq(true)
+}
+
 #[test]
 #[cfg_attr(ci, serial)]
 fn job_send_to_self_works() -> Result {

crates/nu-engine/src/closure_eval.rs

@@ -13,7 +13,7 @@ use std::{
 /// [`ClosureEval`] is used to repeatedly evaluate a closure with different values/inputs.
 ///
 /// [`ClosureEval`] has a builder API.
-/// It is first created via [`ClosureEval::new`],
+/// It is first created via [`ClosureEval::try_new`],
 /// then has arguments added via [`ClosureEval::add_arg`],
 /// and then can be run using [`ClosureEval::run_with_input`].
 ///
@@ -23,7 +23,8 @@ use std::{
 /// # let engine_state = unimplemented!();
 /// # let stack = unimplemented!();
 /// # let closure = unimplemented!();
-/// let mut closure = ClosureEval::new(engine_state, stack, closure);
+/// # let span = unimplemented!();
+/// let mut closure = ClosureEval::try_new(engine_state, stack, closure, span).unwrap();
 /// let iter = Vec::<Value>::new()
 ///     .into_iter()
 ///     .map(move |value| closure.add_arg(value).unwrap().run_with_input(PipelineData::empty()));
@@ -38,7 +39,8 @@ use std::{
 /// # let engine_state = unimplemented!();
 /// # let stack = unimplemented!();
 /// # let closure = unimplemented!();
-/// let mut closure = ClosureEval::new(engine_state, stack, closure);
+/// # let span = unimplemented!();
+/// let mut closure = ClosureEval::try_new(engine_state, stack, closure, span).unwrap();
 /// let iter = Vec::<Value>::new()
 ///     .into_iter()
 ///     .map(move |value| closure.run_with_value(value));
@@ -61,37 +63,48 @@ pub struct ClosureEval {
 }
 
 impl ClosureEval {
-    /// Create a new [`ClosureEval`].
-    pub fn new(engine_state: &EngineState, stack: &Stack, closure: Closure) -> Self {
-        let engine_state = engine_state.clone();
-        let callee_stack = stack.captures_to_stack(closure.captures);
-        let block = engine_state.get_block(closure.block_id).clone();
-        let env_vars = stack.env_vars.clone();
-        let env_hidden = stack.env_hidden.clone();
-        let call_eval = CallEval::new(
-            callee_stack,
-            Span::unknown(),
-            block.span.unwrap_or(Span::unknown()),
-            get_eval_block_with_early_return(&engine_state),
-        );
-
-        Self {
-            engine_state,
-            block,
-            env_vars,
-            env_hidden,
-            call_eval,
-        }
-    }
-
-    pub fn new_preserve_out_dest(
+    /// Create a new [`ClosureEval`], or return an error when the closure's
+    /// block is not available in this engine state.
+    ///
+    /// Closures can be passed between jobs. A job may have been spawned before
+    /// the closure was parsed, so its cloned engine state may not contain the
+    /// referenced block. Keep that case as a regular shell error instead of
+    /// panicking while looking up the block. The `span` points at the call or
+    /// closure value that triggered the error.
+    pub fn try_new(
         engine_state: &EngineState,
         stack: &Stack,
         closure: Closure,
-    ) -> Self {
+        span: Span,
+    ) -> Result<Self, ShellError> {
+        Self::try_new_impl(engine_state, stack, closure, span, false)
+    }
+
+    /// Create a new [`ClosureEval`] while preserving the caller's output
+    /// destination, or return an error when the closure's block is unavailable.
+    pub fn try_new_preserve_out_dest(
+        engine_state: &EngineState,
+        stack: &Stack,
+        closure: Closure,
+        span: Span,
+    ) -> Result<Self, ShellError> {
+        Self::try_new_impl(engine_state, stack, closure, span, true)
+    }
+
+    fn try_new_impl(
+        engine_state: &EngineState,
+        stack: &Stack,
+        closure: Closure,
+        span: Span,
+        preserve_out_dest: bool,
+    ) -> Result<Self, ShellError> {
+        let block = engine_state.get_closure_block(&closure, span)?.clone();
         let engine_state = engine_state.clone();
-        let callee_stack = stack.captures_to_stack_preserve_out_dest(closure.captures);
-        let block = engine_state.get_block(closure.block_id).clone();
+        let callee_stack = if preserve_out_dest {
+            stack.captures_to_stack_preserve_out_dest(closure.captures)
+        } else {
+            stack.captures_to_stack(closure.captures)
+        };
         let env_vars = stack.env_vars.clone();
         let env_hidden = stack.env_hidden.clone();
         let call_eval = CallEval::new(
@@ -101,13 +114,18 @@ impl ClosureEval {
             get_eval_block_with_early_return(&engine_state),
         );
 
-        Self {
+        Ok(Self {
             engine_state,
             block,
             env_vars,
             env_hidden,
             call_eval,
-        }
+        })
+    }
+
+    /// The block this evaluator runs.
+    pub fn block(&self) -> &Block {
+        &self.block
     }
 
     /// Sets whether to enable debugging when evaluating the closure.
@@ -165,7 +183,7 @@ impl ClosureEval {
 /// [`ClosureEvalOnce`] is used to evaluate a closure a single time.
 ///
 /// [`ClosureEvalOnce`] has a builder API.
-/// It is first created via [`ClosureEvalOnce::new`],
+/// It is first created via [`ClosureEvalOnce::try_new`],
 /// then has arguments added via [`ClosureEvalOnce::add_arg`],
 /// and then can be run using [`ClosureEvalOnce::run_with_input`].
 ///
@@ -176,7 +194,9 @@ impl ClosureEval {
 /// # let stack = unimplemented!();
 /// # let closure = unimplemented!();
 /// # let value = unimplemented!();
-/// let result = ClosureEvalOnce::new(engine_state, stack, closure)
+/// # let span = unimplemented!();
+/// let result = ClosureEvalOnce::try_new(engine_state, stack, closure, span)
+///     .unwrap()
 ///     .add_arg(value)
 ///     .unwrap()
 ///     .run_with_input(PipelineData::empty());
@@ -192,7 +212,10 @@ impl ClosureEval {
 /// # let stack = unimplemented!();
 /// # let closure = unimplemented!();
 /// # let value = unimplemented!();
-/// let result = ClosureEvalOnce::new(engine_state, stack, closure).run_with_value(value);
+/// # let span = unimplemented!();
+/// let result = ClosureEvalOnce::try_new(engine_state, stack, closure, span)
+///     .unwrap()
+///     .run_with_value(value);
 /// ```
 ///
 /// In contrast to [`ClosureEval`], the lifetime of [`ClosureEvalOnce`] is bound
@@ -205,10 +228,12 @@ pub struct ClosureEvalOnce<'a> {
 }
 
 impl<'a> ClosureEvalOnce<'a> {
-    /// Create a new [`ClosureEvalOnce`].
-    pub fn new(engine_state: &'a EngineState, stack: &Stack, closure: Closure) -> Self {
-        let block = engine_state.get_block(closure.block_id);
-        let callee_stack = stack.captures_to_stack(closure.captures);
+    fn with_block(
+        engine_state: &'a EngineState,
+        block: &'a Block,
+        callee_stack: Stack,
+        caller_stack: Option<&'a mut Stack>,
+    ) -> Self {
         let call_eval = CallEval::new(
             callee_stack,
             Span::unknown(),
@@ -219,50 +244,64 @@ impl<'a> ClosureEvalOnce<'a> {
             engine_state,
             block,
             call_eval,
-            caller_stack: None,
+            caller_stack,
         }
     }
 
-    pub fn new_preserve_out_dest(
+    /// Create a new [`ClosureEvalOnce`] or return an error when the closure's
+    /// block is not available in this engine state.
+    ///
+    /// Closures can be passed between jobs. A job may have been spawned before
+    /// the closure was parsed, so its cloned engine state may not contain the
+    /// referenced block. Keep that case as a regular shell error instead of
+    /// panicking while looking up the block. The `span` points at the call or
+    /// closure value that triggered the error.
+    pub fn try_new(
         engine_state: &'a EngineState,
         stack: &Stack,
         closure: Closure,
-    ) -> Self {
-        let block = engine_state.get_block(closure.block_id);
-        let callee_stack = stack.captures_to_stack_preserve_out_dest(closure.captures);
-        let call_eval = CallEval::new(
-            callee_stack,
-            Span::unknown(),
-            block.span.unwrap_or(Span::unknown()),
-            get_eval_block_with_early_return(engine_state),
-        );
-        Self {
-            engine_state,
-            block,
-            call_eval,
-            caller_stack: None,
-        }
+        span: Span,
+    ) -> Result<Self, ShellError> {
+        let block = engine_state.get_closure_block(&closure, span)?;
+        let callee_stack = stack.captures_to_stack(closure.captures);
+        Ok(Self::with_block(engine_state, block, callee_stack, None))
     }
 
-    pub fn new_env_preserve_out_dest(
+    /// Create a new [`ClosureEvalOnce`] while preserving the caller's output
+    /// destination, or return an error when the closure's block is unavailable.
+    pub fn try_new_preserve_out_dest(
+        engine_state: &'a EngineState,
+        stack: &Stack,
+        closure: Closure,
+        span: Span,
+    ) -> Result<Self, ShellError> {
+        let block = engine_state.get_closure_block(&closure, span)?;
+        let callee_stack = stack.captures_to_stack_preserve_out_dest(closure.captures);
+        Ok(Self::with_block(engine_state, block, callee_stack, None))
+    }
+
+    /// Create a new [`ClosureEvalOnce`] while preserving the caller's
+    /// environment and output destination, or return an error when the
+    /// closure's block is unavailable.
+    pub fn try_new_env_preserve_out_dest(
         engine_state: &'a EngineState,
         stack: &'a mut Stack,
         closure: Closure,
-    ) -> Self {
-        let block = engine_state.get_block(closure.block_id);
+        span: Span,
+    ) -> Result<Self, ShellError> {
+        let block = engine_state.get_closure_block(&closure, span)?;
         let callee_stack = stack.captures_to_stack_preserve_out_dest(closure.captures);
-        let call_eval = CallEval::new(
-            callee_stack,
-            Span::unknown(),
-            block.span.unwrap_or(Span::unknown()),
-            get_eval_block_with_early_return(engine_state),
-        );
-        Self {
+        Ok(Self::with_block(
             engine_state,
             block,
-            call_eval,
-            caller_stack: Some(stack),
-        }
+            callee_stack,
+            Some(stack),
+        ))
+    }
+
+    /// The block this evaluator runs.
+    pub fn block(&self) -> &Block {
+        self.block
     }
 
     /// Sets whether to enable debugging when evaluating the closure.

crates/nu-engine/src/env.rs

@@ -52,10 +52,11 @@ pub fn convert_env_vars(
                 })?
                 .as_closure()?;
 
-            let new_val = ClosureEvalOnce::new(engine_state, stack, conversion.clone())
-                .debug(false)
-                .run_with_value(val.clone())?
-                .into_value(val.span())?;
+            let new_val =
+                ClosureEvalOnce::try_new(engine_state, stack, conversion.clone(), val.span())?
+                    .debug(false)
+                    .run_with_value(val.clone())?
+                    .into_value(val.span())?;
 
             stack.add_env_var(key.to_string(), new_val);
         }
@@ -314,7 +315,7 @@ fn get_converted_value(
         .as_closure()?;
 
     Ok(
-        ClosureEvalOnce::new(engine_state, stack, conversion.clone())
+        ClosureEvalOnce::try_new(engine_state, stack, conversion.clone(), orig_val.span())?
             .debug(false)
             .run_with_value(orig_val.clone())?
             .into_value(orig_val.span())?,

crates/nu-heavy-utils/src/yaml/error.rs

@@ -572,6 +572,15 @@ impl From<SerializeError> for ShellError {
                 .with_code("shell::yaml::serialize::non_roundtrippable::closure")
                 .with_help("Try `--serialize` or `--non-roundtrip`")
             }
+            SerError::Message { msg } if msg == SerializeError::CLOSURE_BLOCK_NOT_FOUND => {
+                GenericError::new(
+                    "Closure block not found",
+                    "The closure refers to a block that is not available in this engine state",
+                    span,
+                )
+                .with_code("shell::yaml::serialize::closure_block_not_found")
+                .with_help(nu_protocol::engine::MISSING_CLOSURE_BLOCK_HELP)
+            }
             SerError::Message { msg } if msg == SerializeError::NON_ROUNDTRIPPABLE_ERROR => {
                 GenericError::new(
                     "Found non-roundtrippable error",
@@ -612,4 +621,5 @@ impl SerializeError {
         concat!(module_path!(), "::non_roundtrippable::closure");
     pub const NON_ROUNDTRIPPABLE_ERROR: &str =
         concat!(module_path!(), "::non_roundtrippable::error");
+    pub const CLOSURE_BLOCK_NOT_FOUND: &str = concat!(module_path!(), "::closure_block_not_found");
 }

crates/nu-heavy-utils/src/yaml/serialize.rs

@@ -353,13 +353,15 @@ impl Serialize for YamlValue<'_> {
             OPTIONS.with_borrow(|options| match &options.non_roundtrip {
                 NonRoundtrip::Null => serializer.serialize_unit(),
                 NonRoundtrip::Lossy { engine_state } => {
-                    let block = engine_state.get_block(closure.block_id);
-                    if let Some(span) = block.span {
+                    let block = engine_state.try_get_block(closure.block_id);
+                    if let Some(span) = block.and_then(|block| block.span) {
                         let contents = engine_state.get_span_contents(span);
                         let contents = String::from_utf8_lossy(contents);
                         serialize_with_tag(serializer, tag, contents)
-                    } else {
+                    } else if block.is_some() {
                         Err(S::Error::custom(SerializeError::CLOSURE_SPAN_NOT_FOUND))
+                    } else {
+                        Err(S::Error::custom(SerializeError::CLOSURE_BLOCK_NOT_FOUND))
                     }
                 }
                 NonRoundtrip::Error => {

crates/nu-plugin-engine/src/context.rs

@@ -299,8 +299,8 @@ fn plugin_config(
     config.plugins.get(plugin_name).cloned().map(|value| {
         let span = value.span();
         match value {
-            Value::Closure { val, .. } => ClosureEvalOnce::new(engine_state, stack, *val)
-                .run_with_input(PipelineData::empty())
+            Value::Closure { val, .. } => ClosureEvalOnce::try_new(engine_state, stack, *val, span)
+                .and_then(|eval| eval.run_with_input(PipelineData::empty()))
                 .and_then(|data| data.into_value(span))
                 .unwrap_or_else(|err| Value::error(err, head)),
             _ => value.clone(),

crates/nu-protocol/src/engine/closure.rs

@@ -41,14 +41,19 @@ impl Hash for Closure {
     }
 }
 
+/// Help text shared by every error reporting a closure whose block is not
+/// available in the engine state, most commonly because the closure was
+/// parsed after the running job was spawned.
+pub const MISSING_CLOSURE_BLOCK_HELP: &str = "the closure was likely parsed after this job was spawned; define it before `job spawn`, or pass plain data instead of a closure";
+
 impl Closure {
     pub fn coerce_into_string<'a>(
         &self,
         engine_state: &'a EngineState,
         span: Span,
     ) -> Result<Cow<'a, str>, ShellError> {
-        let block = engine_state.get_block(self.block_id);
-        if let Some(span) = block.span {
+        let block = engine_state.try_get_block(self.block_id);
+        if let Some(span) = block.and_then(|block| block.span) {
             let contents_bytes = engine_state.get_span_contents(span);
             Ok(String::from_utf8_lossy(contents_bytes))
         } else {

crates/nu-protocol/src/engine/engine_state.rs

@@ -37,7 +37,7 @@ type PoisonDebuggerError<'a> = PoisonError<MutexGuard<'a, Box<dyn Debugger>>>;
 #[cfg(feature = "plugin")]
 use crate::{PluginRegistryFile, PluginRegistryItem, RegisteredPlugin};
 
-use super::{CurrentJob, Jobs, Mail, Mailbox, ThreadJob};
+use super::{Closure, CurrentJob, Jobs, Mail, Mailbox, ThreadJob};
 
 /// Configure whether the current working directory may be updated when [`EngineState::merge_env`]
 /// is called.
@@ -1098,13 +1098,39 @@ impl EngineState {
 
     /// Optionally get a block by id, if it exists
     ///
-    /// Prefer to use [`.get_block()`](Self::get_block) in most cases - `BlockId`s that don't exist
-    /// are normally a compiler error. This only exists to stop plugins from crashing the engine if
-    /// they send us something invalid.
+    /// Use this instead of [`.get_block()`](Self::get_block) whenever the
+    /// `BlockId` comes from a runtime value instead of the current parse,
+    /// because such a block may legitimately be missing (see
+    /// [`get_closure_block`](Self::get_closure_block)).
     pub fn try_get_block(&self, block_id: BlockId) -> Option<&Arc<Block>> {
         self.blocks.get(block_id.get())
     }
 
+    /// Get the block a closure refers to, or an error when that block is not
+    /// available in this engine state.
+    ///
+    /// Closures can be passed between jobs. A job may have been spawned before
+    /// the closure was parsed, so its cloned engine state may not contain the
+    /// referenced block. Unlike [`get_block`](Self::get_block), this reports
+    /// that case as a regular shell error instead of panicking.
+    pub fn get_closure_block(
+        &self,
+        closure: &Closure,
+        span: Span,
+    ) -> Result<&Arc<Block>, ShellError> {
+        self.try_get_block(closure.block_id).ok_or_else(|| {
+            ShellError::Generic(GenericError::new(
+                "Closure cannot be evaluated",
+                format!(
+                    "the closure refers to a block that is not available in this engine state (block {})",
+                    closure.block_id.get()
+                ),
+                span,
+            )
+            .with_help(super::MISSING_CLOSURE_BLOCK_HELP))
+        })
+    }
+
     pub fn get_module(&self, module_id: ModuleId) -> &Module {
         self.modules
             .get(module_id.get())

crates/nu-tui/src/hooks.rs

@@ -13,7 +13,7 @@
 
 use nu_engine::ClosureEvalOnce;
 use nu_protocol::engine::{Closure, EngineState, Stack};
-use nu_protocol::{IntoPipelineData, PipelineData, Span, Value};
+use nu_protocol::{IntoPipelineData, PipelineData, ShellError, Span, Value};
 
 #[derive(Debug, Clone)]
 pub enum HookOutcome {
@@ -26,8 +26,19 @@ pub enum HookOutcome {
 
 /// Whether the closure declares a positional parameter. Hooks and source
 /// closures pass their argument both as `$in` and, when declared, as `$row`.
-pub fn closure_arity(engine_state: &EngineState, closure: &Closure) -> usize {
-    let block = engine_state.get_block(closure.block_id);
+///
+/// Fails when the closure's block is not available in this engine state,
+/// which happens for a closure parsed after the running job was spawned.
+/// The error is only ever formatted into preview text, so the label span
+/// is left unknown.
+pub fn closure_arity(engine_state: &EngineState, closure: &Closure) -> Result<usize, ShellError> {
+    Ok(block_arity(
+        engine_state.get_closure_block(closure, Span::unknown())?,
+    ))
+}
+
+/// How many positional parameters the closure's block declares.
+fn block_arity(block: &nu_protocol::ast::Block) -> usize {
     block.signature.required_positional.len() + block.signature.optional_positional.len()
 }
 
@@ -38,9 +49,10 @@ pub fn call_closure(
     stack: &Stack,
     closure: Closure,
     input: Value,
-) -> Result<PipelineData, nu_protocol::ShellError> {
-    let arity = closure_arity(engine_state, &closure);
-    let mut eval = ClosureEvalOnce::new(engine_state, stack, closure);
+) -> Result<PipelineData, ShellError> {
+    let span = input.span();
+    let mut eval = ClosureEvalOnce::try_new(engine_state, stack, closure, span)?;
+    let arity = block_arity(eval.block());
     if arity > 0 {
         eval = eval.add_arg(input.clone())?;
     }

crates/nu-tui/src/widgets/preview.rs

@@ -37,11 +37,19 @@ impl PreviewWidget {
             return;
         };
         let engine = session.engine.as_ref().zip(stack);
-        let wants_row = self
-            .transform
-            .as_ref()
-            .zip(engine)
-            .is_some_and(|(c, ((engine_state, _), _))| closure_arity(engine_state, c) > 0);
+        // A transform closure whose block is unavailable (for example one parsed
+        // after this job was spawned) fails for every row, so report it once
+        // instead of falling through to the file preview.
+        let wants_row = match self.transform.as_ref().zip(engine) {
+            Some((c, ((engine_state, _), _))) => match closure_arity(engine_state, c) {
+                Ok(arity) => arity > 0,
+                Err(err) => {
+                    preview.text = format!("preview error: {err}");
+                    return;
+                }
+            },
+            None => false,
+        };
         let file = match (row, wants_row) {
             // One-parameter closure: it is the source, no file is read.
             (Some(row), true) => FilePreview {
@@ -267,7 +275,10 @@ fn apply_transform(
         }
     };
 
-    let mut eval = ClosureEvalOnce::new(engine_state, stack, closure);
+    let mut eval = match ClosureEvalOnce::try_new(engine_state, stack, closure, Span::unknown()) {
+        Ok(eval) => eval,
+        Err(err) => return format!("preview error: {err}"),
+    };
     if let Some(row) = row {
         eval = match eval.add_arg(row.clone()) {
             Ok(eval) => eval,

tests/repl/test_custom_commands.rs

@@ -326,6 +326,36 @@ fn closure_calls_count_toward_recursion_limit(#[case] depth: i64, #[case] ok: bo
     Ok(())
 }
 
+#[test]
+fn infinite_closure_recursion_in_par_each_does_not_panic() -> Result {
+    // A par-each worker that recurses without end must hit the recursion
+    // limit like the main thread does. This only holds when the worker
+    // threads have a stack big enough for the limit; an overflow would
+    // abort the whole process instead of raising this error.
+    // The default limit (50) is the case that needs the deeper worker stack:
+    // with Rayon's smaller default stack the worker overflows and aborts the
+    // process before the limit check can fire.
+    let err = test_with_recursion_limit(50)
+        .run("[1] | par-each {|_| let M = {|x| do $x $x}; do $M $M}")
+        .expect_shell_error()?;
+    let hit_limit = matches!(
+        &err,
+        ShellError::RecursionLimitReached {
+            recursion_limit: 50,
+            ..
+        }
+    ) || matches!(
+        &err,
+        ShellError::EvalBlockWithInput(wrapped)
+            if matches!(
+                wrapped.error.as_ref(),
+                ShellError::RecursionLimitReached { recursion_limit: 50, .. }
+            )
+    );
+    assert!(hit_limit, "expected the recursion limit, got: {err}");
+    Ok(())
+}
+
 #[test]
 fn recursion_limit_set_in_script_applies() -> Result {
     // The limit is read from the stack's config, so setting it earlier in the same script