Move external closer to internal (#1611)

* Refactor InputStream and affected commands.

First, making `values` private and leaning on the `Stream` implementation makes
consumes of `InputStream` less likely to have to change in the future, if we
change what an `InputStream` is internally.

Second, we're dropping `Option<InputStream>` as the input to pipelines,
internals, and externals. Instead, `InputStream.is_empty` can be used to check
for "emptiness". Empty streams are typically only ever used as the first input
to a pipeline.

* Add run_external internal command.

We want to push external commands closer to internal commands, eventually
eliminating the concept of "external" completely. This means we can consolidate
a couple of things:

- Variable evaluation (for example, `$it`, `$nu`, alias vars)
- Behaviour of whole stream vs per-item external execution

It should also make it easier for us to start introducing argument signatures
for external commands,

* Update run_external.rs

* Update run_external.rs

* Update run_external.rs

* Update run_external.rs

Co-authored-by: Jonathan Turner <jonathandturner@users.noreply.github.com>

crates/nu-cli/src/cli.rs

@@ -10,7 +10,7 @@ use crate::prelude::*;
 use futures_codec::FramedRead;
 
 use nu_errors::ShellError;
-use nu_protocol::hir::{ClassifiedCommand, ExternalCommand};
+use nu_protocol::hir::{ClassifiedCommand, Expression, InternalCommand, Literal, NamedArguments};
 use nu_protocol::{Primitive, ReturnSuccess, Scope, Signature, UntaggedValue, Value};
 
 use log::{debug, trace};
@@ -344,6 +344,8 @@ pub fn create_default_context(
             whole_stream_command(FromYML),
             whole_stream_command(FromIcs),
             whole_stream_command(FromVcf),
+            // "Private" commands (not intended to be accessed directly)
+            whole_stream_command(RunExternalCommand),
         ]);
 
         cfg_if::cfg_if! {
@@ -724,14 +726,39 @@ async fn process_line(
             // ...and we're in the CLI
             // ...then change to this directory
             if cli_mode && pipeline.commands.list.len() == 1 {
-                if let ClassifiedCommand::External(ExternalCommand {
+                if let ClassifiedCommand::Internal(InternalCommand {
                     ref name, ref args, ..
                 }) = pipeline.commands.list[0]
                 {
-                    if dunce::canonicalize(&name).is_ok()
+                    let internal_name = name;
+                    let name = args
+                        .positional
+                        .as_ref()
+                        .and_then(|potionals| {
+                            potionals.get(0).map(|e| {
+                                if let Expression::Literal(Literal::String(ref s)) = e.expr {
+                                    &s
+                                } else {
+                                    ""
+                                }
+                            })
+                        })
+                        .unwrap_or("");
+
+                    if internal_name == "run_external"
+                        && args
+                            .positional
+                            .as_ref()
+                            .map(|ref v| v.len() == 1)
+                            .unwrap_or(true)
+                        && args
+                            .named
+                            .as_ref()
+                            .map(NamedArguments::is_empty)
+                            .unwrap_or(true)
+                        && dunce::canonicalize(&name).is_ok()
                         && PathBuf::from(&name).is_dir()
                         && ichwh::which(&name).await.unwrap_or(None).is_none()
-                        && args.list.is_empty()
                     {
                         // Here we work differently if we're in Windows because of the expected Windows behavior
                         #[cfg(windows)]
@@ -773,6 +800,7 @@ async fn process_line(
                     }
                 }
             }
+
             let input_stream = if redirect_stdin {
                 let file = futures::io::AllowStdIo::new(std::io::stdin());
                 let stream = FramedRead::new(file, MaybeTextCodec).map(|line| {
@@ -793,13 +821,13 @@ async fn process_line(
                         panic!("Internal error: could not read lines of text from stdin")
                     }
                 });
-                Some(stream.to_input_stream())
+                stream.to_input_stream()
             } else {
-                None
+                InputStream::empty()
             };
 
             match run_pipeline(pipeline, ctx, input_stream, &Scope::empty()).await {
-                Ok(Some(input)) => {
+                Ok(input) => {
                     // Running a pipeline gives us back a stream that we can then
                     // work through. At the top level, we just want to pull on the
                     // values to compute them.
@@ -810,7 +838,7 @@ async fn process_line(
                         shell_manager: ctx.shell_manager.clone(),
                         host: ctx.host.clone(),
                         ctrl_c: ctx.ctrl_c.clone(),
-                        commands: ctx.registry.clone(),
+                        registry: ctx.registry.clone(),
                         name: Tag::unknown(),
                     };
 
@@ -834,7 +862,6 @@ async fn process_line(
 
                     LineResult::Success(line.to_string())
                 }
-                Ok(None) => LineResult::Success(line.to_string()),
                 Err(err) => LineResult::Error(line.to_string(), err),
             }
         }

crates/nu-cli/src/commands.rs

@@ -77,6 +77,7 @@ pub(crate) mod rename;
 pub(crate) mod reverse;
 pub(crate) mod rm;
 pub(crate) mod run_alias;
+pub(crate) mod run_external;
 pub(crate) mod save;
 pub(crate) mod shells;
 pub(crate) mod shuffle;
@@ -189,6 +190,7 @@ pub(crate) use reject::Reject;
 pub(crate) use rename::Rename;
 pub(crate) use reverse::Reverse;
 pub(crate) use rm::Remove;
+pub(crate) use run_external::RunExternalCommand;
 pub(crate) use save::Save;
 pub(crate) use shells::Shells;
 pub(crate) use shuffle::Shuffle;

crates/nu-cli/src/commands/append.rs

@@ -45,5 +45,5 @@ fn append(
     after.push_back(row);
     let after = futures::stream::iter(after);
 
-    Ok(OutputStream::from_input(input.values.chain(after)))
+    Ok(OutputStream::from_input(input.chain(after)))
 }

crates/nu-cli/src/commands/autoview.rs

@@ -29,7 +29,7 @@ impl WholeStreamCommand for Autoview {
     ) -> Result<OutputStream, ShellError> {
         autoview(RunnableContext {
             input: args.input,
-            commands: registry.clone(),
+            registry: registry.clone(),
             shell_manager: args.shell_manager,
             host: args.host,
             ctrl_c: args.ctrl_c,
@@ -42,7 +42,7 @@ pub struct RunnableContextWithoutInput {
     pub shell_manager: ShellManager,
     pub host: Arc<parking_lot::Mutex<Box<dyn Host>>>,
     pub ctrl_c: Arc<AtomicBool>,
-    pub commands: CommandRegistry,
+    pub registry: CommandRegistry,
     pub name: Tag,
 }
 
@@ -52,7 +52,7 @@ impl RunnableContextWithoutInput {
             shell_manager: context.shell_manager,
             host: context.host,
             ctrl_c: context.ctrl_c,
-            commands: context.commands,
+            registry: context.registry,
             name: context.name,
         };
         (context.input, new_context)
@@ -92,7 +92,7 @@ pub fn autoview(context: RunnableContext) -> Result<OutputStream, ShellError> {
 
                         if let Some(table) = table {
                             let command_args = create_default_command_args(&context).with_input(stream);
-                            let result = table.run(command_args, &context.commands);
+                            let result = table.run(command_args, &context.registry);
                             result.collect::<Vec<_>>().await;
                         }
                     }
@@ -106,7 +106,7 @@ pub fn autoview(context: RunnableContext) -> Result<OutputStream, ShellError> {
                                     let mut stream = VecDeque::new();
                                     stream.push_back(UntaggedValue::string(s).into_value(Tag { anchor, span }));
                                     let command_args = create_default_command_args(&context).with_input(stream);
-                                    let result = text.run(command_args, &context.commands);
+                                    let result = text.run(command_args, &context.registry);
                                     result.collect::<Vec<_>>().await;
                                 } else {
                                     out!("{}", s);
@@ -126,7 +126,7 @@ pub fn autoview(context: RunnableContext) -> Result<OutputStream, ShellError> {
                                     let mut stream = VecDeque::new();
                                     stream.push_back(UntaggedValue::string(s).into_value(Tag { anchor, span }));
                                     let command_args = create_default_command_args(&context).with_input(stream);
-                                    let result = text.run(command_args, &context.commands);
+                                    let result = text.run(command_args, &context.registry);
                                     result.collect::<Vec<_>>().await;
                                 } else {
                                     out!("{}\n", s);
@@ -168,7 +168,7 @@ pub fn autoview(context: RunnableContext) -> Result<OutputStream, ShellError> {
                                     let mut stream = VecDeque::new();
                                     stream.push_back(x);
                                     let command_args = create_default_command_args(&context).with_input(stream);
-                                    let result = binary.run(command_args, &context.commands);
+                                    let result = binary.run(command_args, &context.registry);
                                     result.collect::<Vec<_>>().await;
                                 } else {
                                     use pretty_hex::*;
@@ -254,7 +254,7 @@ pub fn autoview(context: RunnableContext) -> Result<OutputStream, ShellError> {
                                     let mut stream = VecDeque::new();
                                     stream.push_back(x);
                                     let command_args = create_default_command_args(&context).with_input(stream);
-                                    let result = table.run(command_args, &context.commands);
+                                    let result = table.run(command_args, &context.registry);
                                     result.collect::<Vec<_>>().await;
                                 } else {
                                     out!("{:?}", item);
@@ -291,6 +291,7 @@ fn create_default_command_args(context: &RunnableContextWithoutInput) -> RawComm
                 positional: None,
                 named: None,
                 span,
+                is_last: true,
             },
             name_tag: context.name.clone(),
             scope: Scope::empty(),

crates/nu-cli/src/commands/classified/expr.rs

@@ -1,39 +1,29 @@
 use crate::evaluate::evaluate_baseline_expr;
 use crate::prelude::*;
+
 use log::{log_enabled, trace};
+
 use nu_errors::ShellError;
 use nu_protocol::hir::SpannedExpression;
-
-use futures_util::pin_mut;
 use nu_protocol::Scope;
 
 pub(crate) fn run_expression_block(
     expr: SpannedExpression,
     context: &mut Context,
-    input: Option<InputStream>,
+    input: InputStream,
     scope: &Scope,
-) -> Result<Option<InputStream>, ShellError> {
+) -> Result<InputStream, ShellError> {
-    let scope = scope.clone();
     if log_enabled!(log::Level::Trace) {
         trace!(target: "nu::run::expr", "->");
         trace!(target: "nu::run::expr", "{:?}", expr);
     }
 
+    let scope = scope.clone();
     let registry = context.registry().clone();
+    let stream = input.map(move |row| {
+        let scope = scope.clone().set_it(row);
+        evaluate_baseline_expr(&expr, &registry, &scope)
+    });
 
-    let stream = async_stream! {
+    Ok(stream.to_input_stream())
-        if let Some(input) = input {
-            let values = input.values;
-            pin_mut!(values);
-
-            while let Some(row) = values.next().await {
-                let scope = scope.clone().set_it(row);
-                yield evaluate_baseline_expr(&expr, &registry, &scope);
-            }
-        } else {
-            yield evaluate_baseline_expr(&expr, &registry, &scope);
-        }
-    };
-
-    Ok(Some(stream.to_input_stream()))
 }

crates/nu-cli/src/commands/classified/external.rs

@@ -1,19 +1,22 @@
 use crate::futures::ThreadedReceiver;
 use crate::prelude::*;
+
+use std::io::Write;
+use std::ops::Deref;
+use std::process::{Command, Stdio};
+use std::sync::mpsc;
+
 use bytes::{BufMut, Bytes, BytesMut};
 use futures::executor::block_on_stream;
 use futures::stream::StreamExt;
 use futures_codec::FramedRead;
 use log::trace;
+
 use nu_errors::ShellError;
 use nu_protocol::hir::{ExternalArg, ExternalCommand};
 use nu_protocol::{ColumnPath, Primitive, Scope, ShellTypeName, UntaggedValue, Value};
 use nu_source::{Tag, Tagged};
 use nu_value_ext::as_column_path;
-use std::io::Write;
-use std::ops::Deref;
-use std::process::{Command, Stdio};
-use std::sync::mpsc;
 
 pub enum StringOrBinary {
     String(String),
@@ -96,10 +99,10 @@ pub fn nu_value_to_string(command: &ExternalCommand, from: &Value) -> Result<Str
 pub(crate) async fn run_external_command(
     command: ExternalCommand,
     context: &mut Context,
-    input: Option<InputStream>,
+    input: InputStream,
     _scope: &Scope,
     is_last: bool,
-) -> Result<Option<InputStream>, ShellError> {
+) -> Result<InputStream, ShellError> {
     trace!(target: "nu::run::external", "-> {}", command.name);
 
     if !did_find_command(&command.name).await {
@@ -110,7 +113,7 @@ pub(crate) async fn run_external_command(
         ));
     }
 
-    if command.has_it_argument() || command.has_nu_argument() {
+    if command.has_it_argument() {
         run_with_iterator_arg(command, context, input, is_last)
     } else {
         run_with_stdin(command, context, input, is_last)
@@ -166,16 +169,13 @@ fn to_column_path(
 fn run_with_iterator_arg(
     command: ExternalCommand,
     context: &mut Context,
-    input: Option<InputStream>,
+    input: InputStream,
     is_last: bool,
-) -> Result<Option<InputStream>, ShellError> {
+) -> Result<InputStream, ShellError> {
     let path = context.shell_manager.path();
 
-    let mut inputs: InputStream = if let Some(input) = input {
+    let mut inputs: InputStream =
-        trace_stream!(target: "nu::trace_stream::external::it", "input" = input)
+        trace_stream!(target: "nu::trace_stream::external::it", "input" = input);
-    } else {
-        InputStream::empty()
-    };
 
     let stream = async_stream! {
         while let Some(value) = inputs.next().await {
@@ -363,12 +363,10 @@ fn run_with_iterator_arg(
                 }
             }).collect::<Vec<String>>();
 
-            match spawn(&command, &path, &process_args[..], None, is_last) {
+            match spawn(&command, &path, &process_args[..], InputStream::empty(), is_last) {
-                Ok(res) => {
+                Ok(mut res) => {
-                    if let Some(mut res) = res {
+                    while let Some(item) = res.next().await {
-                        while let Some(item) = res.next().await {
+                        yield Ok(item)
-                            yield Ok(item)
-                        }
                     }
                 }
                 Err(reason) => {
@@ -382,19 +380,18 @@ fn run_with_iterator_arg(
         }
     };
 
-    Ok(Some(stream.to_input_stream()))
+    Ok(stream.to_input_stream())
 }
 
 fn run_with_stdin(
     command: ExternalCommand,
     context: &mut Context,
-    input: Option<InputStream>,
+    input: InputStream,
     is_last: bool,
-) -> Result<Option<InputStream>, ShellError> {
+) -> Result<InputStream, ShellError> {
     let path = context.shell_manager.path();
 
-    let input = input
+    let input = trace_stream!(target: "nu::trace_stream::external::stdin", "input" = input);
-        .map(|input| trace_stream!(target: "nu::trace_stream::external::stdin", "input" = input));
 
     let process_args = command
         .args
@@ -432,9 +429,9 @@ fn spawn(
     command: &ExternalCommand,
     path: &str,
     args: &[String],
-    input: Option<InputStream>,
+    input: InputStream,
     is_last: bool,
-) -> Result<Option<InputStream>, ShellError> {
+) -> Result<InputStream, ShellError> {
     let command = command.clone();
 
     let mut process = {
@@ -471,7 +468,7 @@ fn spawn(
     }
 
     // open since we have some contents for stdin
-    if input.is_some() {
+    if !input.is_empty() {
         process.stdin(Stdio::piped());
         trace!(target: "nu::run::external", "set up stdin pipe");
     }
@@ -490,7 +487,7 @@ fn spawn(
         let stdout_name_tag = command.name_tag;
 
         std::thread::spawn(move || {
-            if let Some(input) = input {
+            if !input.is_empty() {
                 let mut stdin_write = stdin
                     .take()
                     .expect("Internal error: could not get stdin pipe for external command");
@@ -632,7 +629,7 @@ fn spawn(
         });
 
         let stream = ThreadedReceiver::new(rx);
-        Ok(Some(stream.to_input_stream()))
+        Ok(stream.to_input_stream())
     } else {
         Err(ShellError::labeled_error(
             "Failed to spawn process",
@@ -717,7 +714,7 @@ fn shell_os_paths() -> Vec<std::path::PathBuf> {
 mod tests {
     use super::{
         add_quotes, argument_contains_whitespace, argument_is_quoted, expand_tilde, remove_quotes,
-        run_external_command, Context,
+        run_external_command, Context, InputStream,
     };
     use futures::executor::block_on;
     use nu_errors::ShellError;
@@ -740,10 +737,11 @@ mod tests {
     async fn non_existent_run() -> Result<(), ShellError> {
         let cmd = ExternalBuilder::for_name("i_dont_exist.exe").build();
 
+        let input = InputStream::empty();
         let mut ctx = Context::basic().expect("There was a problem creating a basic context.");
 
         assert!(
-            run_external_command(cmd, &mut ctx, None, &Scope::empty(), false)
+            run_external_command(cmd, &mut ctx, input, &Scope::empty(), false)
                 .await
                 .is_err()
         );

crates/nu-cli/src/commands/classified/internal.rs

@@ -10,20 +10,15 @@ use nu_protocol::{CommandAction, Primitive, ReturnSuccess, Scope, UntaggedValue,
 pub(crate) fn run_internal_command(
     command: InternalCommand,
     context: &mut Context,
-    input: Option<InputStream>,
+    input: InputStream,
     scope: &Scope,
-) -> Result<Option<InputStream>, ShellError> {
+) -> Result<InputStream, ShellError> {
     if log_enabled!(log::Level::Trace) {
         trace!(target: "nu::run::internal", "->");
         trace!(target: "nu::run::internal", "{}", command.name);
     }
 
-    let objects: InputStream = if let Some(input) = input {
+    let objects: InputStream = trace_stream!(target: "nu::trace_stream::internal", "input" = input);
-        trace_stream!(target: "nu::trace_stream::internal", "input" = input)
-    } else {
-        InputStream::empty()
-    };
-
     let internal_command = context.expect_command(&command.name);
 
     let result = {
@@ -36,8 +31,7 @@ pub(crate) fn run_internal_command(
         )
     };
 
-    let result = trace_out_stream!(target: "nu::trace_stream::internal", "output" = result);
+    let mut result = trace_out_stream!(target: "nu::trace_stream::internal", "output" = result);
-    let mut result = result.values;
     let mut context = context.clone();
 
     let stream = async_stream! {
@@ -69,7 +63,8 @@ pub(crate) fn run_internal_command(
                                         head: command.args.head,
                                         positional: None,
                                         named: None,
-                                        span: Span::unknown()
+                                        span: Span::unknown(),
+                                        is_last: false,
                                     },
                                     name_tag: Tag::unknown_anchor(command.name_span),
                                     scope: Scope::empty(),
@@ -186,5 +181,5 @@ pub(crate) fn run_internal_command(
         }
     };
 
-    Ok(Some(stream.to_input_stream()))
+    Ok(stream.to_input_stream())
 }

crates/nu-cli/src/commands/classified/pipeline.rs

@@ -10,9 +10,9 @@ use nu_protocol::Scope;
 pub(crate) async fn run_pipeline(
     pipeline: ClassifiedPipeline,
     ctx: &mut Context,
-    mut input: Option<InputStream>,
+    mut input: InputStream,
     scope: &Scope,
-) -> Result<Option<InputStream>, ShellError> {
+) -> Result<InputStream, ShellError> {
     let mut iter = pipeline.commands.list.into_iter().peekable();
 
     loop {

crates/nu-cli/src/commands/clip.rs

@@ -41,7 +41,7 @@ pub mod clipboard {
         RunnableContext { input, name, .. }: RunnableContext,
     ) -> Result<OutputStream, ShellError> {
         let stream = async_stream! {
-            let values: Vec<Value> = input.values.collect().await;
+            let values: Vec<Value> = input.collect().await;
 
             let mut clip_stream = inner_clip(values, name).await;
             while let Some(value) = clip_stream.next().await {

crates/nu-cli/src/commands/command.rs

@@ -179,7 +179,7 @@ impl CommandArgs {
             args: T::deserialize(&mut deserializer)?,
             context: RunnableContext {
                 input,
-                commands: registry.clone(),
+                registry: registry.clone(),
                 shell_manager,
                 name: name_tag,
                 host,
@@ -215,7 +215,7 @@ impl CommandArgs {
             args: T::deserialize(&mut deserializer)?,
             context: RunnableContext {
                 input,
-                commands: registry.clone(),
+                registry: registry.clone(),
                 shell_manager,
                 name: name_tag,
                 host,
@@ -238,13 +238,13 @@ pub struct RunnableContext {
     pub shell_manager: ShellManager,
     pub host: Arc<parking_lot::Mutex<Box<dyn Host>>>,
     pub ctrl_c: Arc<AtomicBool>,
-    pub commands: CommandRegistry,
+    pub registry: CommandRegistry,
     pub name: Tag,
 }
 
 impl RunnableContext {
     pub fn get_command(&self, name: &str) -> Option<Arc<Command>> {
-        self.commands.get_command(name)
+        self.registry.get_command(name)
     }
 }
 
@@ -530,7 +530,6 @@ impl Command {
 
         let out = args
             .input
-            .values
             .map(move |x| {
                 let call_info = UnevaluatedCallInfo {
                     args: raw_args.call_info.args.clone(),
@@ -597,7 +596,7 @@ impl WholeStreamCommand for FnFilterCommand {
         let registry: CommandRegistry = registry.clone();
         let func = self.func;
 
-        let result = input.values.map(move |it| {
+        let result = input.map(move |it| {
             let registry = registry.clone();
             let call_info = match call_info.clone().evaluate_with_new_it(&registry, &it) {
                 Err(err) => return OutputStream::from(vec![Err(err)]).values,

crates/nu-cli/src/commands/compact.rs

@@ -39,7 +39,7 @@ pub fn compact(
     CompactArgs { rest: columns }: CompactArgs,
     RunnableContext { input, .. }: RunnableContext,
 ) -> Result<OutputStream, ShellError> {
-    let objects = input.values.filter(move |item| {
+    let objects = input.filter(move |item| {
         let keep = if columns.is_empty() {
             item.is_some()
         } else {

crates/nu-cli/src/commands/config.rs

@@ -124,7 +124,7 @@ pub fn config(
             yield ReturnSuccess::value(UntaggedValue::Row(result.into()).into_value(&value.tag));
         }
         else if let Some(v) = set_into {
-            let rows: Vec<Value> = input.values.collect().await;
+            let rows: Vec<Value> = input.collect().await;
             let key = v.to_string();
 
             if rows.len() == 0 {

crates/nu-cli/src/commands/count.rs

@@ -37,7 +37,7 @@ pub fn count(
     RunnableContext { input, name, .. }: RunnableContext,
 ) -> Result<OutputStream, ShellError> {
     let stream = async_stream! {
-        let rows: Vec<Value> = input.values.collect().await;
+        let rows: Vec<Value> = input.collect().await;
 
         yield ReturnSuccess::value(UntaggedValue::int(rows.len()).into_value(name))
     };

crates/nu-cli/src/commands/debug.rs

@@ -37,7 +37,6 @@ fn debug_value(
     RunnableContext { input, .. }: RunnableContext,
 ) -> Result<impl ToOutputStream, ShellError> {
     Ok(input
-        .values
         .map(move |v| {
             if raw {
                 ReturnSuccess::value(

crates/nu-cli/src/commands/default.rs

@@ -47,7 +47,6 @@ fn default(
     RunnableContext { input, .. }: RunnableContext,
 ) -> Result<OutputStream, ShellError> {
     let stream = input
-        .values
         .map(move |item| {
             let mut result = VecDeque::new();
 

crates/nu-cli/src/commands/each.rs

@@ -1,8 +1,10 @@
 use crate::commands::classified::pipeline::run_pipeline;
-
 use crate::commands::PerItemCommand;
 use crate::context::CommandRegistry;
 use crate::prelude::*;
+
+use futures::stream::once;
+
 use nu_errors::ShellError;
 use nu_protocol::{
     hir::ClassifiedPipeline, CallInfo, ReturnSuccess, Scope, Signature, SyntaxShape, UntaggedValue,
@@ -46,37 +48,34 @@ impl PerItemCommand for Each {
                 } => {
                     let mut context = Context::from_raw(&raw_args, &registry);
                     let input_clone = input.clone();
-                    let input_stream = async_stream! {
+                    let input_stream = once(async { Ok(input) }).to_input_stream();
-                        yield Ok(input.clone())
-                    }.to_input_stream();
 
                     let result = run_pipeline(
                         ClassifiedPipeline::new(block.clone(), None),
                         &mut context,
-                        Some(input_stream),
+                        input_stream,
                         &Scope::new(input_clone),
                     ).await;
 
                     match result {
-                        Ok(Some(v)) => {
+                        Ok(stream) if stream.is_empty() => {
-                            let results: Vec<Value> = v.collect().await;
+                            yield Err(ShellError::labeled_error(
+                                "Expected a block",
+                                "each needs a block",
+                                tag,
+                            ));
+                        }
+                        Ok(mut stream) => {
                             let errors = context.get_errors();
                             if let Some(error) = errors.first() {
                                 yield Err(error.clone());
                                 return;
                             }
 
-                            for result in results {
+                            while let Some(result) = stream.next().await {
                                 yield Ok(ReturnSuccess::Value(result));
                             }
                         }
-                        Ok(None) => {
-                            yield Err(ShellError::labeled_error(
-                                "Expected a block",
-                                "each needs a block",
-                                tag,
-                            ));
-                        }
                         Err(e) => {
                             yield Err(e);
                         }

crates/nu-cli/src/commands/enter.rs

@@ -101,7 +101,8 @@ impl PerItemCommand for Enter {
                                                     head: raw_args.call_info.args.head,
                                                     positional: None,
                                                     named: None,
-                                                    span: Span::unknown()
+                                                    span: Span::unknown(),
+                                                    is_last: false,
                                                 },
                                                 name_tag: raw_args.call_info.name_tag,
                                                 scope: raw_args.call_info.scope.clone()

crates/nu-cli/src/commands/evaluate_by.rs

@@ -45,7 +45,7 @@ pub fn evaluate_by(
     RunnableContext { input, name, .. }: RunnableContext,
 ) -> Result<OutputStream, ShellError> {
     let stream = async_stream! {
-        let values: Vec<Value> = input.values.collect().await;
+        let values: Vec<Value> = input.collect().await;
 
         if values.is_empty() {
             yield Err(ShellError::labeled_error(

crates/nu-cli/src/commands/first.rs

@@ -48,7 +48,5 @@ fn first(
         1
     };
 
-    Ok(OutputStream::from_input(
+    Ok(OutputStream::from_input(context.input.take(rows_desired)))
-        context.input.values.take(rows_desired),
-    ))
 }

crates/nu-cli/src/commands/get.rs

@@ -197,7 +197,6 @@ pub fn get(
         let member = fields.remove(0);
         trace!("get {:?} {:?}", member, fields);
         let stream = input
-            .values
             .map(move |item| {
                 let mut result = VecDeque::new();
 

crates/nu-cli/src/commands/group_by.rs

@@ -43,7 +43,7 @@ pub fn group_by(
     RunnableContext { input, name, .. }: RunnableContext,
 ) -> Result<OutputStream, ShellError> {
     let stream = async_stream! {
-        let values: Vec<Value> = input.values.collect().await;
+        let values: Vec<Value> = input.collect().await;
 
         if values.is_empty() {
             yield Err(ShellError::labeled_error(

crates/nu-cli/src/commands/headers.rs

@@ -35,7 +35,7 @@ pub fn headers(
     RunnableContext { input, .. }: RunnableContext,
 ) -> Result<OutputStream, ShellError> {
     let stream = async_stream! {
-        let rows: Vec<Value> = input.values.collect().await;
+        let rows: Vec<Value> = input.collect().await;
 
         if rows.len() < 1 {
             yield Err(ShellError::untagged_runtime_error("Couldn't find headers, was the input a properly formatted, non-empty table?"));

crates/nu-cli/src/commands/histogram.rs

@@ -53,7 +53,7 @@ pub fn histogram(
     RunnableContext { input, name, .. }: RunnableContext,
 ) -> Result<OutputStream, ShellError> {
     let stream = async_stream! {
-        let values: Vec<Value> = input.values.collect().await;
+        let values: Vec<Value> = input.collect().await;
 
         let Tagged { item: group_by, .. } = column_name.clone();
 

crates/nu-cli/src/commands/lines.rs

@@ -47,7 +47,7 @@ fn lines(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream,
     let mut leftover_string = String::new();
     let stream = async_stream! {
         loop {
-            match input.values.next().await {
+            match input.next().await {
                 Some(Value { value: UntaggedValue::Primitive(Primitive::String(st)), ..}) => {
                     let mut st = leftover_string.clone() + &st;
                     leftover.clear();

crates/nu-cli/src/commands/map_max_by.rs

@@ -46,7 +46,7 @@ pub fn map_max_by(
     RunnableContext { input, name, .. }: RunnableContext,
 ) -> Result<OutputStream, ShellError> {
     let stream = async_stream! {
-        let values: Vec<Value> = input.values.collect().await;
+        let values: Vec<Value> = input.collect().await;
 
 
         if values.is_empty() {

crates/nu-cli/src/commands/nth.rs

@@ -49,7 +49,6 @@ fn nth(
     RunnableContext { input, .. }: RunnableContext,
 ) -> Result<OutputStream, ShellError> {
     let stream = input
-        .values
         .enumerate()
         .map(move |(idx, item)| {
             let row_number = vec![row_number.clone()];

crates/nu-cli/src/commands/pick.rs

@@ -1,7 +1,6 @@
 use crate::commands::WholeStreamCommand;
 use crate::context::CommandRegistry;
 use crate::prelude::*;
-use futures_util::pin_mut;
 use nu_errors::ShellError;
 use nu_protocol::{
     ColumnPath, PathMember, Primitive, ReturnSuccess, ReturnValue, Signature, SyntaxShape,
@@ -44,7 +43,9 @@ impl WholeStreamCommand for Pick {
 
 fn pick(
     PickArgs { rest: mut fields }: PickArgs,
-    RunnableContext { input, name, .. }: RunnableContext,
+    RunnableContext {
+        mut input, name, ..
+    }: RunnableContext,
 ) -> Result<OutputStream, ShellError> {
     if fields.is_empty() {
         return Err(ShellError::labeled_error(
@@ -64,13 +65,10 @@ fn pick(
         .collect::<Vec<ColumnPath>>();
 
     let stream = async_stream! {
-        let values = input.values;
-        pin_mut!(values);
-
         let mut empty = true;
         let mut bring_back: indexmap::IndexMap<String, Vec<Value>> = indexmap::IndexMap::new();
 
-        while let Some(value) = values.next().await {
+        while let Some(value) = input.next().await {
             for path in &column_paths {
                 let path_members_span = span_for_spanned_list(path.members().iter().map(|p| p.span));
 

crates/nu-cli/src/commands/plugin.rs

@@ -98,7 +98,7 @@ pub fn filter_plugin(
     trace!("filtering :: {:?}", call_info);
 
     let stream = bos
-        .chain(args.input.values)
+        .chain(args.input)
         .chain(eos)
         .map(move |v| match v {
             Value {
@@ -343,7 +343,7 @@ pub fn sink_plugin(
     let call_info = args.call_info.clone();
 
     let stream = async_stream! {
-        let input: Vec<Value> = args.input.values.collect().await;
+        let input: Vec<Value> = args.input.collect().await;
 
         let request = JsonRpc::new("sink", (call_info.clone(), input));
         let request_raw = serde_json::to_string(&request);

crates/nu-cli/src/commands/prepend.rs

@@ -43,5 +43,5 @@ fn prepend(
 ) -> Result<OutputStream, ShellError> {
     let prepend = futures::stream::iter(vec![row]);
 
-    Ok(OutputStream::from_input(prepend.chain(input.values)))
+    Ok(prepend.chain(input).to_output_stream())
 }

crates/nu-cli/src/commands/range.rs

@@ -50,7 +50,5 @@ fn range(
     let from = *from as usize;
     let to = *to as usize;
 
-    Ok(OutputStream::from_input(
+    Ok(input.skip(from).take(to - from + 1).to_output_stream())
-        input.values.skip(from).take(to - from + 1),
-    ))
 }

crates/nu-cli/src/commands/reduce_by.rs

@@ -45,7 +45,7 @@ pub fn reduce_by(
     RunnableContext { input, name, .. }: RunnableContext,
 ) -> Result<OutputStream, ShellError> {
     let stream = async_stream! {
-        let values: Vec<Value> = input.values.collect().await;
+        let values: Vec<Value> = input.collect().await;
 
         if values.is_empty() {
             yield Err(ShellError::labeled_error(

crates/nu-cli/src/commands/reject.rs

@@ -48,9 +48,7 @@ fn reject(
 
     let fields: Vec<_> = fields.iter().map(|f| f.item.clone()).collect();
 
-    let stream = input
+    let stream = input.map(move |item| reject_fields(&item, &fields, &item.tag));
-        .values
-        .map(move |item| reject_fields(&item, &fields, &item.tag));
 
     Ok(stream.from_input_stream())
 }

crates/nu-cli/src/commands/rename.rs

@@ -54,7 +54,6 @@ pub fn rename(
     let new_column_names = new_column_names.into_iter().flatten().collect::<Vec<_>>();
 
     let stream = input
-        .values
         .map(move |item| {
             let mut result = VecDeque::new();
 

crates/nu-cli/src/commands/reverse.rs

@@ -32,7 +32,7 @@ fn reverse(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream
     let args = args.evaluate_once(registry)?;
     let (input, _args) = args.parts();
 
-    let input = input.values.collect::<Vec<_>>();
+    let input = input.collect::<Vec<_>>();
 
     let output = input.map(move |mut vec| {
         vec.reverse();

crates/nu-cli/src/commands/run_alias.rs

@@ -56,37 +56,36 @@ impl PerItemCommand for AliasCommand {
 
         let stream = async_stream! {
             let mut context = Context::from_raw(&raw_args, &registry);
-            let input_stream = async_stream! {
+
-                yield Ok(input.clone())
+            let input_clone = Ok(input.clone());
-            }.to_input_stream();
+            let input_stream = futures::stream::once(async { input_clone }).boxed().to_input_stream();
 
             let result = run_pipeline(
                 ClassifiedPipeline::new(block.clone(), None),
                 &mut context,
-                Some(input_stream),
+                input_stream,
                 &scope
             ).await;
 
             match result {
-                Ok(Some(v)) => {
+                Ok(stream) if stream.is_empty() => {
-                    let results: Vec<Value> = v.collect().await;
-                    let errors = context.get_errors();
-                    if let Some(error) = errors.first() {
-                        yield Err(error.clone());
-                        return;
-                    }
-
-                    for result in results {
-                        yield Ok(ReturnSuccess::Value(result));
-                    }
-                }
-                Ok(None) => {
                     yield Err(ShellError::labeled_error(
                         "Expected a block",
                         "each needs a block",
                         tag,
                     ));
                 }
+                Ok(mut stream) => {
+                    // We collect first to ensure errors are put into the context
+                    while let Some(result) = stream.next().await {
+                        yield Ok(ReturnSuccess::Value(result));
+                    }
+
+                    let errors = context.get_errors();
+                    if let Some(error) = errors.first() {
+                        yield Err(error.clone());
+                    }
+                }
                 Err(e) => {
                     yield Err(e);
                 }

crates/nu-cli/src/commands/run_external.rs

@@ -0,0 +1,128 @@
+use crate::commands::classified::external;
+use crate::commands::WholeStreamCommand;
+use crate::prelude::*;
+
+use derive_new::new;
+use parking_lot::Mutex;
+
+use nu_errors::ShellError;
+use nu_protocol::hir::{
+    Expression, ExternalArg, ExternalArgs, ExternalCommand, Literal, SpannedExpression,
+};
+use nu_protocol::{ReturnSuccess, Scope, Signature, SyntaxShape};
+
+#[derive(Deserialize)]
+pub struct RunExternalArgs {}
+
+#[derive(new)]
+pub struct RunExternalCommand;
+
+fn spanned_expression_to_string(expr: &SpannedExpression) -> String {
+    if let SpannedExpression {
+        expr: Expression::Literal(Literal::String(s)),
+        ..
+    } = expr
+    {
+        s.clone()
+    } else {
+        "notacommand!!!".to_string()
+    }
+}
+
+impl WholeStreamCommand for RunExternalCommand {
+    fn name(&self) -> &str {
+        "run_external"
+    }
+
+    fn signature(&self) -> Signature {
+        Signature::build(self.name()).rest(SyntaxShape::Any, "external command arguments")
+    }
+
+    fn usage(&self) -> &str {
+        ""
+    }
+
+    fn run(
+        &self,
+        args: CommandArgs,
+        registry: &CommandRegistry,
+    ) -> Result<OutputStream, ShellError> {
+        let positionals = args.call_info.args.positional.ok_or_else(|| {
+            ShellError::untagged_runtime_error("positional arguments unexpectedly empty")
+        })?;
+
+        let mut command_args = positionals.iter();
+        let name = command_args
+            .next()
+            .map(spanned_expression_to_string)
+            .ok_or_else(|| {
+                ShellError::untagged_runtime_error(
+                    "run_external unexpectedly missing external name positional arg",
+                )
+            })?;
+
+        let command = ExternalCommand {
+            name,
+            name_tag: Tag::unknown(),
+            args: ExternalArgs {
+                list: command_args
+                    .map(|arg| ExternalArg {
+                        arg: spanned_expression_to_string(arg),
+                        tag: Tag::unknown(),
+                    })
+                    .collect(),
+                span: Default::default(),
+            },
+        };
+
+        let mut external_context;
+        #[cfg(windows)]
+        {
+            external_context = Context {
+                registry: registry.clone(),
+                host: args.host.clone(),
+                shell_manager: args.shell_manager.clone(),
+                ctrl_c: args.ctrl_c.clone(),
+                current_errors: Arc::new(Mutex::new(vec![])),
+                windows_drives_previous_cwd: Arc::new(Mutex::new(std::collections::HashMap::new())),
+            };
+        }
+        #[cfg(not(windows))]
+        {
+            external_context = Context {
+                registry: registry.clone(),
+                host: args.host.clone(),
+                shell_manager: args.shell_manager.clone(),
+                ctrl_c: args.ctrl_c.clone(),
+                current_errors: Arc::new(Mutex::new(vec![])),
+            };
+        }
+
+        let is_last = args.call_info.args.is_last;
+        let input = args.input;
+        let stream = async_stream! {
+            let scope = Scope::empty();
+            let result = external::run_external_command(
+                command,
+                &mut external_context,
+                input,
+                &scope,
+                is_last,
+            ).await;
+
+            match result {
+                Ok(mut stream) => {
+                    while let Some(value) = stream.next().await {
+                        yield Ok(ReturnSuccess::Value(value));
+                    }
+                },
+                Err(e) => {
+                    yield Err(e);
+                },
+                _ => {}
+            }
+        };
+
+        Ok(stream.to_output_stream())
+    }
+}

crates/nu-cli/src/commands/save.rs

@@ -168,7 +168,7 @@ fn save(
         shell_manager,
         host,
         ctrl_c,
-        commands: registry,
+        registry,
         ..
     }: RunnableContext,
     raw_args: RawCommandArgs,
@@ -177,7 +177,7 @@ fn save(
     let name_tag = name.clone();
 
     let stream = async_stream! {
-        let input: Vec<Value> = input.values.collect().await;
+        let input: Vec<Value> = input.collect().await;
         if path.is_none() {
             // If there is no filename, check the metadata for the anchor filename
             if input.len() > 0 {
@@ -232,7 +232,8 @@ fn save(
                                     head: raw_args.call_info.args.head,
                                     positional: None,
                                     named: None,
-                                    span: Span::unknown()
+                                    span: Span::unknown(),
+                                    is_last: false,
                                 },
                                 name_tag: raw_args.call_info.name_tag,
                                 scope: Scope::empty(), // FIXME?

crates/nu-cli/src/commands/shuffle.rs

@@ -48,7 +48,7 @@ fn shuffle(
     RunnableContext { input, .. }: RunnableContext,
 ) -> Result<OutputStream, ShellError> {
     let stream = async_stream! {
-        let mut values: Vec<Value> = input.values.collect().await;
+        let mut values: Vec<Value> = input.collect().await;
 
         let out = if let Some(n) = limit {
             let (shuffled, _) = values.partial_shuffle(&mut thread_rng(), *n as usize);

crates/nu-cli/src/commands/size.rs

@@ -33,7 +33,6 @@ fn size(args: CommandArgs, _registry: &CommandRegistry) -> Result<OutputStream,
     let name_span = tag.span;
 
     Ok(input
-        .values
         .map(move |v| {
             if let Ok(s) = v.as_string() {
                 ReturnSuccess::value(count(&s, &v.tag))

crates/nu-cli/src/commands/skip.rs

@@ -41,7 +41,5 @@ fn skip(SkipArgs { rows }: SkipArgs, context: RunnableContext) -> Result<OutputS
         1
     };
 
-    Ok(OutputStream::from_input(
+    Ok(OutputStream::from_input(context.input.skip(rows_desired)))
-        context.input.values.skip(rows_desired),
-    ))
 }

crates/nu-cli/src/commands/skip_while.rs

@@ -68,7 +68,7 @@ impl WholeStreamCommand for SkipWhile {
             }
         };
 
-        let objects = call_info.input.values.skip_while(move |item| {
+        let objects = call_info.input.skip_while(move |item| {
             let condition = condition.clone();
             trace!("ITEM = {:?}", item);
             let result = evaluate_baseline_expr(&*condition, &registry, &Scope::new(item.clone()));

crates/nu-cli/src/commands/split_by.rs

@@ -44,7 +44,7 @@ pub fn split_by(
     RunnableContext { input, name, .. }: RunnableContext,
 ) -> Result<OutputStream, ShellError> {
     let stream = async_stream! {
-        let values: Vec<Value> = input.values.collect().await;
+        let values: Vec<Value> = input.collect().await;
 
         if values.len() > 1 || values.is_empty() {
             yield Err(ShellError::labeled_error(

crates/nu-cli/src/commands/split_column.rs

@@ -57,7 +57,6 @@ fn split_column(
     let name_span = name.span;
 
     Ok(input
-        .values
         .map(move |v| {
             if let Ok(s) = v.as_string() {
                 let splitter = separator.replace("\\n", "\n");

crates/nu-cli/src/commands/split_row.rs

@@ -43,7 +43,6 @@ fn split_row(
     RunnableContext { input, name, .. }: RunnableContext,
 ) -> Result<OutputStream, ShellError> {
     let stream = input
-        .values
         .map(move |v| {
             if let Ok(s) = v.as_string() {
                 let splitter = separator.item.replace("\\n", "\n");

crates/nu-cli/src/commands/sum.rs

@@ -27,7 +27,7 @@ impl WholeStreamCommand for Sum {
     ) -> Result<OutputStream, ShellError> {
         sum(RunnableContext {
             input: args.input,
-            commands: registry.clone(),
+            registry: registry.clone(),
             shell_manager: args.shell_manager,
             host: args.host,
             ctrl_c: args.ctrl_c,

crates/nu-cli/src/commands/t_sort_by.rs

@@ -69,7 +69,7 @@ fn t_sort_by(
     RunnableContext { input, name, .. }: RunnableContext,
 ) -> Result<OutputStream, ShellError> {
     Ok(OutputStream::new(async_stream! {
-        let values: Vec<Value> = input.values.collect().await;
+        let values: Vec<Value> = input.collect().await;
 
         let column_grouped_by_name = if let Some(grouped_by) = group_by {
             Some(grouped_by.item().clone())

crates/nu-cli/src/commands/tags.rs

@@ -30,7 +30,6 @@ impl WholeStreamCommand for Tags {
 fn tags(args: CommandArgs, _registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
     Ok(args
         .input
-        .values
         .map(move |v| {
             let mut tags = TaggedDictBuilder::new(v.tag());
             {

crates/nu-cli/src/commands/to_bson.rs

@@ -266,7 +266,7 @@ fn to_bson(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream
     let name_span = name_tag.span;
 
     let stream = async_stream! {
-        let input: Vec<Value> = args.input.values.collect().await;
+        let input: Vec<Value> = args.input.collect().await;
 
         let to_process_input = if input.len() > 1 {
             let tag = input[0].tag.clone();

crates/nu-cli/src/commands/to_delimited_data.rs

@@ -175,7 +175,7 @@ pub fn to_delimited_data(
     let name_span = name_tag.span;
 
     let stream = async_stream! {
-         let input: Vec<Value> = input.values.collect().await;
+         let input: Vec<Value> = input.collect().await;
 
          let to_process_input = if input.len() > 1 {
              let tag = input[0].tag.clone();

crates/nu-cli/src/commands/to_html.rs

@@ -35,7 +35,7 @@ fn to_html(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream
     let name_tag = args.name_tag();
     //let name_span = name_tag.span;
     let stream = async_stream! {
-        let input: Vec<Value> = args.input.values.collect().await;
+        let input: Vec<Value> = args.input.collect().await;
         let headers = nu_protocol::merge_descriptors(&input);
         let mut output_string = "<html><body>".to_string();
 

crates/nu-cli/src/commands/to_json.rs

@@ -136,7 +136,7 @@ fn to_json(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream
     let name_tag = args.name_tag();
     let name_span = name_tag.span;
     let stream = async_stream! {
-        let input: Vec<Value> = args.input.values.collect().await;
+        let input: Vec<Value> = args.input.collect().await;
 
         let to_process_input = if input.len() > 1 {
             let tag = input[0].tag.clone();

crates/nu-cli/src/commands/to_md.rs

@@ -34,7 +34,7 @@ fn to_html(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream
     let name_tag = args.name_tag();
     //let name_span = name_tag.span;
     let stream = async_stream! {
-        let input: Vec<Value> = args.input.values.collect().await;
+        let input: Vec<Value> = args.input.collect().await;
         let headers = nu_protocol::merge_descriptors(&input);
         let mut output_string = String::new();
 

crates/nu-cli/src/commands/to_sqlite.rs

@@ -205,7 +205,7 @@ fn to_sqlite(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStre
     let args = args.evaluate_once(registry)?;
     let name_tag = args.name_tag();
     let stream = async_stream! {
-        let input: Vec<Value> = args.input.values.collect().await;
+        let input: Vec<Value> = args.input.collect().await;
 
         match sqlite_input_stream_to_bytes(input) {
             Ok(out) => yield ReturnSuccess::value(out),

crates/nu-cli/src/commands/to_toml.rs

@@ -98,7 +98,7 @@ fn to_toml(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream
     let name_tag = args.name_tag();
     let name_span = name_tag.span;
     let stream = async_stream! {
-        let input: Vec<Value> = args.input.values.collect().await;
+        let input: Vec<Value> = args.input.collect().await;
 
         let to_process_input = if input.len() > 1 {
             let tag = input[0].tag.clone();

crates/nu-cli/src/commands/to_url.rs

@@ -33,7 +33,7 @@ fn to_url(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream,
     let input = args.input;
 
     let stream = async_stream! {
-        let input: Vec<Value> = input.values.collect().await;
+        let input: Vec<Value> = input.collect().await;
 
         for value in input {
             match value {

crates/nu-cli/src/commands/to_yaml.rs

@@ -130,7 +130,7 @@ fn to_yaml(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream
     let name_span = name_tag.span;
 
     let stream = async_stream! {
-        let input: Vec<Value> = args.input.values.collect().await;
+        let input: Vec<Value> = args.input.collect().await;
 
         let to_process_input = if input.len() > 1 {
             let tag = input[0].tag.clone();

crates/nu-cli/src/commands/trim.rs

@@ -29,10 +29,8 @@ impl WholeStreamCommand for Trim {
 }
 
 fn trim(args: CommandArgs, _registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
-    let input = args.input;
+    Ok(args
-
+        .input
-    Ok(input
-        .values
         .map(move |v| {
             let string = String::extract(&v)?;
             ReturnSuccess::value(UntaggedValue::string(string.trim()).into_value(v.tag()))

crates/nu-cli/src/commands/uniq.rs

@@ -37,7 +37,7 @@ fn uniq(
     RunnableContext { input, .. }: RunnableContext,
 ) -> Result<OutputStream, ShellError> {
     let stream = async_stream! {
-        let uniq_values: IndexSet<_> = input.values.collect().await;
+        let uniq_values: IndexSet<_> = input.collect().await;
 
         for item in uniq_values.iter().map(|row| ReturnSuccess::value(row.clone())) {
             yield item;

crates/nu-cli/src/commands/what.rs

@@ -1,8 +1,6 @@
 use crate::commands::WholeStreamCommand;
-
 use crate::prelude::*;
-use futures::StreamExt;
+
-use futures_util::pin_mut;
 use nu_errors::ShellError;
 use nu_protocol::{ReturnSuccess, ReturnValue, Signature, UntaggedValue};
 
@@ -34,14 +32,11 @@ impl WholeStreamCommand for What {
 }
 
 pub fn what(
-    WhatArgs {}: WhatArgs,
+    _: WhatArgs,
-    RunnableContext { input, .. }: RunnableContext,
+    RunnableContext { mut input, .. }: RunnableContext,
 ) -> Result<OutputStream, ShellError> {
     let stream = async_stream! {
-        let values = input.values;
+        while let Some(row) = input.next().await {
-        pin_mut!(values);
-
-        while let Some(row) = values.next().await {
             let name = value::format_type(&row, 100);
             yield ReturnSuccess::value(UntaggedValue::string(name).into_value(Tag::unknown_anchor(row.tag.span)));
         }

crates/nu-cli/src/commands/which_.rs

@@ -78,7 +78,7 @@ struct WhichArgs {
 
 fn which(
     WhichArgs { application, all }: WhichArgs,
-    RunnableContext { commands, .. }: RunnableContext,
+    RunnableContext { registry, .. }: RunnableContext,
 ) -> Result<OutputStream, ShellError> {
     let external = application.starts_with('^');
     let item = if external {
@@ -89,7 +89,7 @@ fn which(
 
     let stream = async_stream! {
         if !external {
-            let builtin = commands.has(&item);
+            let builtin = registry.has(&item);
             if builtin {
                 yield ReturnSuccess::value(entry_builtin!(item, application.tag.clone()));
             }

crates/nu-cli/src/prelude.rs

@@ -27,7 +27,7 @@ macro_rules! trace_stream {
         if log::log_enabled!(target: $target, log::Level::Trace) {
             use futures::stream::StreamExt;
 
-            let objects = $expr.values.inspect(move |o| {
+            let objects = $expr.inspect(move |o| {
                 trace!(
                     target: $target,
                     "{} = {}",
@@ -49,7 +49,7 @@ macro_rules! trace_out_stream {
         if log::log_enabled!(target: $target, log::Level::Trace) {
             use futures::stream::StreamExt;
 
-            let objects = $expr.values.inspect(move |o| {
+            let objects = $expr.inspect(move |o| {
                 trace!(
                     target: $target,
                     "{} = {}",
@@ -132,17 +132,13 @@ where
     U: Into<Result<nu_protocol::Value, nu_errors::ShellError>>,
 {
     fn to_input_stream(self) -> InputStream {
-        InputStream {
+        InputStream::from_stream(self.map(|item| match item.into() {
-            values: self
+            Ok(result) => result,
-                .map(|item| match item.into() {
+            Err(err) => match HasFallibleSpan::maybe_span(&err) {
-                    Ok(result) => result,
+                Some(span) => nu_protocol::UntaggedValue::Error(err).into_value(span),
-                    Err(err) => match HasFallibleSpan::maybe_span(&err) {
+                None => nu_protocol::UntaggedValue::Error(err).into_untagged_value(),
-                        Some(span) => nu_protocol::UntaggedValue::Error(err).into_value(span),
+            },
-                        None => nu_protocol::UntaggedValue::Error(err).into_untagged_value(),
+        }))
-                    },
-                })
-                .boxed(),
-        }
     }
 }
 

crates/nu-cli/src/stream/input.rs

@@ -1,22 +1,32 @@
 use crate::prelude::*;
-use futures::stream::iter;
+use futures::stream::{iter, once};
 use nu_errors::ShellError;
 use nu_protocol::{Primitive, UntaggedValue, Value};
 use nu_source::{Tagged, TaggedItem};
 
 pub struct InputStream {
-    pub(crate) values: BoxStream<'static, Value>,
+    values: BoxStream<'static, Value>,
+
+    // Whether or not an empty stream was explicitly requeted via InputStream::empty
+    empty: bool,
 }
 
 impl InputStream {
     pub fn empty() -> InputStream {
-        vec![UntaggedValue::nothing().into_value(Tag::unknown())].into()
+        InputStream {
+            values: once(async { UntaggedValue::nothing().into_untagged_value() }).boxed(),
+            empty: true,
+        }
     }
 
     pub fn into_vec(self) -> impl Future<Output = Vec<Value>> {
         self.values.collect()
     }
 
+    pub fn is_empty(&self) -> bool {
+        self.empty
+    }
+
     pub fn drain_vec(&mut self) -> impl Future<Output = Vec<Value>> {
         let mut values: BoxStream<'static, Value> = iter(VecDeque::new()).boxed();
         std::mem::swap(&mut values, &mut self.values);
@@ -27,6 +37,7 @@ impl InputStream {
     pub fn from_stream(input: impl Stream<Item = Value> + Send + 'static) -> InputStream {
         InputStream {
             values: input.boxed(),
+            empty: false,
         }
     }
 
@@ -129,7 +140,10 @@ impl Stream for InputStream {
 
 impl From<BoxStream<'static, Value>> for InputStream {
     fn from(input: BoxStream<'static, Value>) -> InputStream {
-        InputStream { values: input }
+        InputStream {
+            values: input,
+            empty: false,
+        }
     }
 }
 
@@ -137,6 +151,7 @@ impl From<VecDeque<Value>> for InputStream {
     fn from(input: VecDeque<Value>) -> InputStream {
         InputStream {
             values: futures::stream::iter(input).boxed(),
+            empty: false,
         }
     }
 }
@@ -145,6 +160,7 @@ impl From<Vec<Value>> for InputStream {
     fn from(input: Vec<Value>) -> InputStream {
         InputStream {
             values: futures::stream::iter(input).boxed(),
+            empty: false,
         }
     }
 }

crates/nu-cli/src/stream/output.rs

@@ -52,7 +52,7 @@ impl Stream for OutputStream {
 impl From<InputStream> for OutputStream {
     fn from(input: InputStream) -> OutputStream {
         OutputStream {
-            values: input.values.map(ReturnSuccess::value).boxed(),
+            values: input.map(ReturnSuccess::value).boxed(),
         }
     }
 }

crates/nu-cli/tests/commands/cp.rs

@@ -188,7 +188,7 @@ fn copies_same_file_twice() {
 #[test]
 fn copy_files_using_glob_two_parents_up_using_multiple_dots() {
     Playground::setup("cp_test_9", |dirs, sandbox| {
-        sandbox.within("foo").mkdir("bar").with_files(vec![
+        sandbox.within("foo").within("bar").with_files(vec![
             EmptyFile("jonathan.json"),
             EmptyFile("andres.xml"),
             EmptyFile("yehuda.yaml"),

crates/nu-cli/tests/commands/sum.rs

@@ -32,3 +32,31 @@ fn all() {
         assert_eq!(actual, "448");
     })
 }
+
+#[test]
+fn outputs_zero_with_no_input() {
+    Playground::setup("sum_test_2", |dirs, sandbox| {
+        sandbox.with_files(vec![FileWithContentToBeTrimmed(
+            "meals.json",
+            r#"
+                {
+                    meals: [
+                        {description: "1 large egg", calories: 90},
+                        {description: "1 cup white rice", calories: 250},
+                        {description: "1 tablespoon fish oil", calories: 108}
+                    ]
+                }
+            "#,
+        )]);
+
+        let actual = nu!(
+            cwd: dirs.test(), pipeline(
+            r#"
+                sum
+                | echo $it
+            "#
+        ));
+
+        assert_eq!(actual, "0");
+    })
+}

crates/nu-parser/src/parse.rs

@@ -6,12 +6,11 @@ use crate::signature::SignatureRegistry;
 use log::trace;
 use nu_errors::{ArgumentError, ParseError};
 use nu_protocol::hir::{
-    self, Binary, ClassifiedCommand, ClassifiedPipeline, Commands, Expression, ExternalArg,
+    self, Binary, ClassifiedCommand, ClassifiedPipeline, Commands, Expression, Flag, FlagKind,
-    ExternalArgs, ExternalCommand, Flag, FlagKind, InternalCommand, Member, NamedArguments,
+    InternalCommand, Member, NamedArguments, Operator, SpannedExpression, Unit,
-    Operator, SpannedExpression, Unit,
 };
 use nu_protocol::{NamedType, PositionalType, Signature, SyntaxShape, UnspannedPathMember};
-use nu_source::{Span, Spanned, SpannedItem, Tag};
+use nu_source::{Span, Spanned, SpannedItem};
 use num_bigint::BigInt;
 
 /// Parses a simple column path, one without a variable (implied or explicit) at the head
@@ -980,66 +979,76 @@ pub fn classify_pipeline(
     let mut commands = Commands::new(Span::new(0, 0));
     let mut error = None;
 
-    for lite_cmd in lite_pipeline.commands.iter() {
+    let mut iter = lite_pipeline.commands.iter().peekable();
+    while let Some(lite_cmd) = iter.next() {
         if lite_cmd.name.item.starts_with('^') {
-            let cmd_name: String = lite_cmd.name.item.chars().skip(1).collect();
+            let name = lite_cmd
-            // This is an external command we should allow arguments to pass through with minimal parsing
+                .name
-            commands.push(ClassifiedCommand::External(ExternalCommand {
+                .clone()
-                name: cmd_name,
+                .map(|v| v.chars().skip(1).collect::<String>());
-                name_tag: Tag::unknown_anchor(lite_cmd.name.span),
+
-                args: ExternalArgs {
+            // TODO this is the same as the `else` branch below, only the name differs. Find a way
-                    list: lite_cmd
+            //      to share this functionality.
-                        .args
+            let name_iter = std::iter::once(name);
-                        .iter()
+            let args = name_iter.chain(lite_cmd.args.clone().into_iter());
-                        .map(|x| ExternalArg {
+            let args = arguments_from_string_iter(args);
-                            arg: x.item.clone(),
+
-                            tag: Tag::unknown_anchor(x.span),
+            commands.push(ClassifiedCommand::Internal(InternalCommand {
-                        })
+                name: "run_external".to_string(),
-                        .collect(),
+                name_span: Span::unknown(),
-                    span: Span::new(0, 0),
+                args: hir::Call {
+                    head: Box::new(SpannedExpression {
+                        expr: Expression::string("run_external".to_string()),
+                        span: Span::unknown(),
+                    }),
+                    positional: Some(args),
+                    named: None,
+                    span: Span::unknown(),
+                    is_last: iter.peek().is_none(),
                 },
             }))
         } else if lite_cmd.name.item == "=" {
             let expr = if !lite_cmd.args.is_empty() {
                 let (_, expr, err) = parse_math_expression(0, &lite_cmd.args[0..], registry, false);
-                if error.is_none() {
+                error = error.or(err);
-                    error = err;
-                }
                 expr
             } else {
-                if error.is_none() {
+                error = error.or_else(|| {
-                    error = Some(ParseError::argument_error(
+                    Some(ParseError::argument_error(
                         lite_cmd.name.clone(),
                         ArgumentError::MissingMandatoryPositional("an expression".into()),
                     ))
-                }
+                });
                 garbage(lite_cmd.span())
             };
             commands.push(ClassifiedCommand::Expr(Box::new(expr)))
         } else if let Some(signature) = registry.get(&lite_cmd.name.item) {
             let (internal_command, err) = parse_internal_command(&lite_cmd, registry, &signature);
 
-            if error.is_none() {
+            error = error.or(err);
-                error = err;
-            }
             commands.push(ClassifiedCommand::Internal(internal_command))
         } else {
-            let trimmed = trim_quotes(&lite_cmd.name.item);
+            let name = lite_cmd.name.clone().map(|v| {
-            let name = expand_path(&trimmed);
+                let trimmed = trim_quotes(&v);
-            // This is an external command we should allow arguments to pass through with minimal parsing
+                expand_path(&trimmed)
-            commands.push(ClassifiedCommand::External(ExternalCommand {
+            });
-                name,
+
-                name_tag: Tag::unknown_anchor(lite_cmd.name.span),
+            let name_iter = std::iter::once(name);
-                args: ExternalArgs {
+            let args = name_iter.chain(lite_cmd.args.clone().into_iter());
-                    list: lite_cmd
+            let args = arguments_from_string_iter(args);
-                        .args
+
-                        .iter()
+            commands.push(ClassifiedCommand::Internal(InternalCommand {
-                        .map(|x| ExternalArg {
+                name: "run_external".to_string(),
-                            arg: x.item.clone(),
+                name_span: Span::unknown(),
-                            tag: Tag::unknown_anchor(x.span),
+                args: hir::Call {
-                        })
+                    head: Box::new(SpannedExpression {
-                        .collect(),
+                        expr: Expression::string("run_external".to_string()),
-                    span: Span::new(0, 0),
+                        span: Span::unknown(),
+                    }),
+                    positional: Some(args),
+                    named: None,
+                    span: Span::unknown(),
+                    is_last: iter.peek().is_none(),
                 },
             }))
         }
@@ -1048,6 +1057,20 @@ pub fn classify_pipeline(
     ClassifiedPipeline::new(commands, error)
 }
 
+/// Parse out arguments from spanned expressions
+pub fn arguments_from_string_iter(
+    iter: impl Iterator<Item = Spanned<String>>,
+) -> Vec<SpannedExpression> {
+    iter.map(|v| {
+        // TODO parse_full_column_path
+        SpannedExpression {
+            expr: Expression::string(v.to_string()),
+            span: v.span,
+        }
+    })
+    .collect::<Vec<_>>()
+}
+
 /// Easy shorthand function to create a garbage expression at the given span
 pub fn garbage(span: Span) -> SpannedExpression {
     SpannedExpression::new(Expression::Garbage, span)

crates/nu-protocol/src/hir.rs

@@ -909,6 +909,7 @@ pub struct Call {
     pub positional: Option<Vec<SpannedExpression>>,
     pub named: Option<NamedArguments>,
     pub span: Span,
+    pub is_last: bool,
 }
 
 impl Call {
@@ -981,6 +982,7 @@ impl Call {
             positional: None,
             named: None,
             span,
+            is_last: false,
         }
     }
 }