Input output checking (#9680)

# Description

This PR tights input/output type-checking a bit more. There are a lot of
commands that don't have correct input/output types, so part of the
effort is updating them.

This PR now contains updates to commands that had wrong input/output
signatures. It doesn't add examples for these new signatures, but that
can be follow-up work.

# User-Facing Changes

BREAKING CHANGE BREAKING CHANGE

This work enforces many more checks on pipeline type correctness than
previous nushell versions. This strictness may uncover incompatibilities
in existing scripts or shortcomings in the type information for internal
commands.

# Tests + Formatting
<!--
Don't forget to add tests that cover your changes.

Make sure you've run and fixed any issues with these commands:

- `cargo fmt --all -- --check` to check standard code formatting (`cargo
fmt --all` applies these changes)
- `cargo clippy --workspace -- -D warnings -D clippy::unwrap_used -A
clippy::needless_collect -A clippy::result_large_err` to check that
you're using the standard code style
- `cargo test --workspace` to check that all tests pass
- `cargo run -- -c "use std testing; testing run-tests --path
crates/nu-std"` to run the tests for the standard library

> **Note**
> from `nushell` you can also use the `toolkit` as follows
> ```bash
> use toolkit.nu # or use an `env_change` hook to activate it
automatically
> toolkit check pr
> ```
-->

# After Submitting
<!-- If your PR had any user-facing changes, update [the
documentation](https://github.com/nushell/nushell.github.io) after the
PR is merged, if necessary. This will help us keep the docs up to date.
-->

crates/nu-parser/src/parser.rs

@@ -4,7 +4,7 @@ use crate::{
     lite_parser::{lite_parse, LiteCommand, LiteElement, LitePipeline},
     parse_mut,
     parse_patterns::{parse_match_pattern, parse_pattern},
-    type_check::{math_result_type, type_compatible},
+    type_check::{self, math_result_type, type_compatible},
     Token, TokenContents,
 };
 
@@ -5582,6 +5582,8 @@ pub fn parse_block(
 
     block.span = Some(span);
 
+    type_check::check_block_input_output(working_set, &block);
+
     block
 }
 

crates/nu-parser/src/type_check.rs

@@ -1,5 +1,8 @@
 use nu_protocol::{
-    ast::{Bits, Boolean, Comparison, Expr, Expression, Math, Operator},
+    ast::{
+        Bits, Block, Boolean, Comparison, Expr, Expression, Math, Operator, Pipeline,
+        PipelineElement,
+    },
     engine::StateWorkingSet,
     ParseError, Type,
 };
@@ -24,7 +27,30 @@ pub fn type_compatible(lhs: &Type, rhs: &Type) -> bool {
 
     match (lhs, rhs) {
         (Type::List(c), Type::List(d)) => type_compatible(c, d),
-        (Type::List(c), Type::Table(_)) => matches!(**c, Type::Any),
+        (Type::ListStream, Type::List(_)) => true,
+        (Type::List(_), Type::ListStream) => true,
+        (Type::List(c), Type::Table(table_fields)) => {
+            if matches!(**c, Type::Any) {
+                return true;
+            }
+
+            if let Type::Record(fields) = &**c {
+                is_compatible(fields, table_fields)
+            } else {
+                false
+            }
+        }
+        (Type::Table(table_fields), Type::List(c)) => {
+            if matches!(**c, Type::Any) {
+                return true;
+            }
+
+            if let Type::Record(fields) = &**c {
+                is_compatible(table_fields, fields)
+            } else {
+                false
+            }
+        }
         (Type::Number, Type::Int) => true,
         (Type::Int, Type::Number) => true,
         (Type::Number, Type::Float) => true,
@@ -921,3 +947,108 @@ pub fn math_result_type(
         }
     }
 }
+
+pub fn check_pipeline_type(
+    working_set: &mut StateWorkingSet,
+    pipeline: &Pipeline,
+    input_type: Type,
+) -> Type {
+    let mut current_type = input_type;
+
+    'elem: for elem in &pipeline.elements {
+        match elem {
+            PipelineElement::Expression(
+                _,
+                Expression {
+                    expr: Expr::Call(call),
+                    ..
+                },
+            ) => {
+                let decl = working_set.get_decl(call.decl_id);
+
+                if current_type == Type::Any {
+                    let mut new_current_type = None;
+                    for (_, call_output) in decl.signature().input_output_types {
+                        if let Some(inner_current_type) = &new_current_type {
+                            if inner_current_type == &Type::Any {
+                                break;
+                            } else if inner_current_type != &call_output {
+                                // Union unequal types to Any for now
+                                new_current_type = Some(Type::Any)
+                            }
+                        } else {
+                            new_current_type = Some(call_output.clone())
+                        }
+                    }
+
+                    if let Some(new_current_type) = new_current_type {
+                        current_type = new_current_type
+                    } else {
+                        current_type = Type::Any;
+                    }
+                    continue 'elem;
+                } else {
+                    for (call_input, call_output) in decl.signature().input_output_types {
+                        if type_compatible(&call_input, &current_type) {
+                            current_type = call_output.clone();
+                            continue 'elem;
+                        }
+                    }
+                }
+
+                if !decl.signature().input_output_types.is_empty() {
+                    working_set.error(ParseError::InputMismatch(current_type, call.head))
+                }
+                current_type = Type::Any;
+            }
+            PipelineElement::Expression(_, Expression { ty, .. }) => {
+                current_type = ty.clone();
+            }
+            _ => {
+                current_type = Type::Any;
+            }
+        }
+    }
+
+    current_type
+}
+
+pub fn check_block_input_output(working_set: &mut StateWorkingSet, block: &Block) {
+    // let inputs = block.input_types();
+
+    for (input_type, output_type) in &block.signature.input_output_types {
+        let mut current_type = input_type.clone();
+        let mut current_output_type = Type::Nothing;
+
+        for pipeline in &block.pipelines {
+            current_output_type = check_pipeline_type(working_set, pipeline, current_type);
+            current_type = Type::Nothing;
+        }
+
+        if !type_compatible(output_type, &current_output_type)
+            && output_type != &Type::Any
+            && current_output_type != Type::Any
+        {
+            working_set.error(ParseError::OutputMismatch(
+                output_type.clone(),
+                block
+                    .pipelines
+                    .last()
+                    .expect("internal error: we should have pipelines")
+                    .elements
+                    .last()
+                    .expect("internal error: we should have elements")
+                    .span(),
+            ))
+        }
+    }
+
+    if block.signature.input_output_types.is_empty() {
+        let mut current_type = Type::Any;
+
+        for pipeline in &block.pipelines {
+            let _ = check_pipeline_type(working_set, pipeline, current_type);
+            current_type = Type::Nothing;
+        }
+    }
+}