diff --git a/changelog.d/1399.fix.md b/changelog.d/1399.fix.md
new file mode 100644
index 000000000..c6fae2a1f
--- /dev/null
+++ b/changelog.d/1399.fix.md
@@ -0,0 +1,3 @@
+Function closures now correctly update type state of the program.
+
+authors: zettroke
diff --git a/src/compiler/compiler.rs b/src/compiler/compiler.rs
index b1b618af6..f6dc01e61 100644
--- a/src/compiler/compiler.rs
+++ b/src/compiler/compiler.rs
@@ -84,6 +84,19 @@ impl CompilerError {
 }
 
 impl<'a> Compiler<'a> {
+    pub(crate) fn new(fns: &'a [Box<dyn Function>], config: CompileConfig) -> Self {
+        Self {
+            fns,
+            diagnostics: vec![],
+            fallible: false,
+            abortable: false,
+            external_queries: vec![],
+            external_assignments: vec![],
+            skip_missing_query_target: vec![],
+            fallible_expression_error: None,
+            config,
+        }
+    }
     /// Compiles a given source into the final [`Program`].
     ///
     /// # Arguments
@@ -106,17 +119,7 @@ impl<'a> Compiler<'a> {
         let initial_state = state.clone();
         let mut state = state.clone();
 
-        let mut compiler = Self {
-            fns,
-            diagnostics: vec![],
-            fallible: false,
-            abortable: false,
-            external_queries: vec![],
-            external_assignments: vec![],
-            skip_missing_query_target: vec![],
-            fallible_expression_error: None,
-            config,
-        };
+        let mut compiler = Compiler::new(fns, config);
         let expressions = compiler.compile_root_exprs(ast, &mut state);
 
         let (errors, warnings): (Vec<_>, Vec<_>) =
@@ -272,7 +275,7 @@ impl<'a> Compiler<'a> {
         Some(Group::new(expr))
     }
 
-    fn compile_root_exprs(
+    pub(crate) fn compile_root_exprs(
         &mut self,
         nodes: impl IntoIterator<Item = Node<ast::RootExpr>>,
         state: &mut TypeState,
diff --git a/src/compiler/expression/function_call.rs b/src/compiler/expression/function_call.rs
index 35579a6e4..bfcf4c4eb 100644
--- a/src/compiler/expression/function_call.rs
+++ b/src/compiler/expression/function_call.rs
@@ -501,30 +501,38 @@ impl<'a> Builder<'a> {
         state: &mut TypeState,
     ) -> Result<(Option<Closure>, bool), FunctionCallError> {
         // Check if we have a closure we need to compile.
-        if let Some((variables, input)) = self.closure.clone() {
+        if let Some((variables, input)) = &self.closure {
             // TODO: This assumes the closure will run exactly once, which is incorrect.
             // see: https://github.com/vectordotdev/vector/issues/13782
 
-            let block = closure_block.expect("closure must contain block");
-
+            let mut variables_types = vec![];
             // At this point, we've compiled the block, so we can remove the
             // closure variables from the compiler's local environment.
-            variables
-                .iter()
-                .for_each(|ident| match locals.remove_variable(ident) {
-                    Some(details) => state.local.insert_variable(ident.clone(), details),
-                    None => {
-                        state.local.remove_variable(ident);
-                    }
-                });
+            for ident in variables {
+                let variable_details = state
+                    .local
+                    .remove_variable(ident)
+                    .expect("Closure variable must be present");
+                variables_types.push(variable_details);
+
+                // If outer scope has this variable, restore its state
+                if let Some(details) = locals.remove_variable(ident) {
+                    state.local.insert_variable(ident.clone(), details);
+                }
+            }
 
-            let (block_span, (block, block_type_def)) = block.take();
+            let (block_span, (block, block_type_def)) = closure_block
+                .ok_or(FunctionCallError::MissingClosure {
+                    call_span: Span::default(), // TODO can we provide a better span?
+                    example: None,
+                })?
+                .take();
 
             let closure_fallible = block_type_def.is_fallible();
 
             // Check the type definition of the resulting block.This needs to match
             // whatever is configured by the closure input type.
-            let expected_kind = input.output.into_kind();
+            let expected_kind = input.clone().output.into_kind();
             let found_kind = block_type_def
                 .kind()
                 .union(block_type_def.returns().clone());
@@ -537,7 +545,7 @@ impl<'a> Builder<'a> {
                 });
             }
 
-            let fnclosure = Closure::new(variables, block, block_type_def);
+            let fnclosure = Closure::new(variables.clone(), variables_types, block, block_type_def);
             self.list.set_closure(fnclosure.clone());
 
             // closure = Some(fnclosure);
@@ -700,6 +708,27 @@ impl Expression for FunctionCall {
 
         let mut expr_result = self.expr.apply_type_info(&mut state);
 
+        // Closure can change state of locals in our `state`, so we need to update it.
+        if let Some(closure) = &self.closure {
+            // To get correct `type_info()` from closure we need to add closure arguments into current state
+            let mut closure_state = state.clone();
+            for (ident, details) in closure
+                .variables
+                .iter()
+                .cloned()
+                .zip(closure.variables_types.iter().cloned())
+            {
+                closure_state.local.insert_variable(ident, details);
+            }
+            let mut closure_info = closure.block.type_info(&closure_state);
+            // No interaction with closure arguments can't affect parent state, so remove them before merge
+            for ident in &closure.variables {
+                closure_info.state.local.remove_variable(ident);
+            }
+
+            state = state.merge(closure_info.state);
+        }
+
         // If one of the arguments only partially matches the function type
         // definition, then we mark the entire function as fallible.
         //
@@ -1224,9 +1253,10 @@ impl DiagnosticMessage for FunctionCallError {
 
 #[cfg(test)]
 mod tests {
-    use crate::compiler::{value::kind, FunctionExpression};
-
     use super::*;
+    use crate::compiler::{value::kind, Compiler, FunctionExpression};
+    use crate::parser::parse;
+    use crate::stdlib::ForEach;
 
     #[derive(Clone, Debug)]
     struct Fn;
@@ -1411,4 +1441,27 @@ mod tests {
 
         assert_eq!(Ok(expected), params);
     }
+
+    #[test]
+    fn closure_type_state() {
+        let program = parse(
+            r#"
+            v = ""
+
+            for_each({}) -> |key, value| {
+                v = 0
+            }
+        "#,
+        )
+        .unwrap();
+
+        let fns = vec![Box::new(ForEach) as Box<dyn Function>];
+        let mut compiler = Compiler::new(&fns, CompileConfig::default());
+
+        let mut state = TypeState::default();
+        compiler.compile_root_exprs(program, &mut state);
+        let var = state.local.variable(&Ident::new("v")).unwrap();
+
+        assert_eq!(var.type_def.kind(), &Kind::bytes().or_integer());
+    }
 }
diff --git a/src/compiler/function.rs b/src/compiler/function.rs
index 0e4262aec..931c3ceb6 100644
--- a/src/compiler/function.rs
+++ b/src/compiler/function.rs
@@ -1,6 +1,13 @@
 #![allow(clippy::missing_errors_doc)]
 pub mod closure;
 
+use super::{
+    expression::{container::Variant, Block, Container, Expr, Expression},
+    state::TypeState,
+    value::{kind, Kind},
+    CompileConfig, Span, TypeDef,
+};
+use crate::compiler::type_def::Details;
 use crate::diagnostic::{DiagnosticMessage, Label, Note};
 use crate::parser::ast::Ident;
 use crate::path::OwnedTargetPath;
@@ -10,13 +17,6 @@ use std::{
     fmt,
 };
 
-use super::{
-    expression::{container::Variant, Block, Container, Expr, Expression},
-    state::TypeState,
-    value::{kind, Kind},
-    CompileConfig, Span, TypeDef,
-};
-
 pub type Compiled = Result<Box<dyn Expression>, Box<dyn DiagnosticMessage>>;
 pub type CompiledArgument =
     Result<Option<Box<dyn std::any::Any + Send + Sync>>, Box<dyn DiagnosticMessage>>;
@@ -448,15 +448,22 @@ mod test_impls {
 #[derive(Debug, Clone, PartialEq)]
 pub struct Closure {
     pub variables: Vec<Ident>,
+    pub variables_types: Vec<Details>,
     pub block: Block,
     pub block_type_def: TypeDef,
 }
 
 impl Closure {
     #[must_use]
-    pub fn new<T: Into<Ident>>(variables: Vec<T>, block: Block, block_type_def: TypeDef) -> Self {
+    pub fn new(
+        variables: Vec<Ident>,
+        variables_types: Vec<Details>,
+        block: Block,
+        block_type_def: TypeDef,
+    ) -> Self {
         Self {
-            variables: variables.into_iter().map(Into::into).collect(),
+            variables,
+            variables_types,
             block,
             block_type_def,
         }
diff --git a/src/compiler/program.rs b/src/compiler/program.rs
index 18a8b2b96..a4f4bb806 100644
--- a/src/compiler/program.rs
+++ b/src/compiler/program.rs
@@ -48,7 +48,7 @@ pub struct ProgramInfo {
     /// Returns whether the compiled program can fail at runtime.
     ///
     /// A program can only fail at runtime if the fallible-function-call
-    /// (`foo!()`) is used within the source.
+    /// (`foo!()`) is used within the source.vrl
     pub fallible: bool,
 
     /// Returns whether the compiled program can be aborted at runtime.
diff --git a/src/compiler/type_def.rs b/src/compiler/type_def.rs
index 5352ad8b4..71a1d9af5 100644
--- a/src/compiler/type_def.rs
+++ b/src/compiler/type_def.rs
@@ -538,7 +538,7 @@ impl From<TypeDef> for Kind {
 }
 
 #[derive(Debug, Clone, PartialEq)]
-pub(crate) struct Details {
+pub struct Details {
     pub(crate) type_def: TypeDef,
     pub(crate) value: Option<Value>,
 }
diff --git a/src/stdlib/filter.rs b/src/stdlib/filter.rs
index 402b4b331..706e6a964 100644
--- a/src/stdlib/filter.rs
+++ b/src/stdlib/filter.rs
@@ -122,6 +122,7 @@ impl FunctionExpression for FilterFn {
             variables,
             block,
             block_type_def: _,
+            ..
         } = &self.closure;
         let runner = closure::Runner::new(variables, |ctx| block.resolve(ctx));
 
diff --git a/src/stdlib/for_each.rs b/src/stdlib/for_each.rs
index 4e4686086..ce021d817 100644
--- a/src/stdlib/for_each.rs
+++ b/src/stdlib/for_each.rs
@@ -98,6 +98,7 @@ impl FunctionExpression for ForEachFn {
             variables,
             block,
             block_type_def: _,
+            ..
         } = &self.closure;
         let runner = closure::Runner::new(variables, |ctx| block.resolve(ctx));
 
diff --git a/src/stdlib/map_keys.rs b/src/stdlib/map_keys.rs
index bb5bb9862..0ef24fe8b 100644
--- a/src/stdlib/map_keys.rs
+++ b/src/stdlib/map_keys.rs
@@ -122,6 +122,7 @@ impl FunctionExpression for MapKeysFn {
             variables,
             block,
             block_type_def: _,
+            ..
         } = &self.closure;
         let runner = closure::Runner::new(variables, |ctx| block.resolve(ctx));
 
diff --git a/src/stdlib/map_values.rs b/src/stdlib/map_values.rs
index a0e1f5040..a399aaa47 100644
--- a/src/stdlib/map_values.rs
+++ b/src/stdlib/map_values.rs
@@ -121,6 +121,7 @@ impl FunctionExpression for MapValuesFn {
             variables,
             block,
             block_type_def: _,
+            ..
         } = &self.closure;
         let runner = closure::Runner::new(variables, |ctx| block.resolve(ctx));